AidopT8KpiManualRefreshService.cs 13 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277
  1. using Admin.NET.Core.Service;
  2. using Admin.NET.Plugin.AiDOP.FinishedWarehouse;
  3. using Admin.NET.Plugin.AiDOP.Infrastructure;
  4. using Admin.NET.Plugin.AiDOP.Manufacturing;
  5. using Admin.NET.Plugin.AiDOP.MaterialWarehouse;
  6. using Microsoft.Extensions.Logging;
  7. namespace Admin.NET.Plugin.AiDOP.DataPlatform;
  8. /// <summary>
  9. /// S5/S6/S7 T8 KPI 手动刷新统一编排(看板"刷新"按钮真实跑批入口)。
  10. /// 单次刷新语义:抢全局刷新锁 → T8 基表入站(硬门,共享 6 表)→ 本模块 RunFull(MANUAL) → 汇总逐指标状态 → 释放锁。
  11. /// 与 Cron job_s5_t8_kpi_refresh 共享同一把锁(<see cref="RefreshLockKey"/>),保证手动刷新与定时跑批互斥、并发手动刷新排他。
  12. /// 入站为 S5/S6/S7 共用(T8BaseInboundMdpSyncService 6 张基表无法按模块拆),故刷任一模块均触发一次全量入站。
  13. /// S5_L1_002 / S5_L1_004 仍 legacy 直连 T8(本轮不改),随 S5 RunFull 一并执行,逐指标标 LEGACY_SUCCESS。
  14. ///
  15. /// 锁设计(加固):用持有者无关的硬 TTL 键(SysCacheService.Set 带过期),而非 BeginCacheLock。
  16. /// 原因:BeginCacheLock 在持有者线程存活期间会续租,一旦某次运行意外挂死(如源库慢查询无超时)会永久占锁、
  17. /// 阻塞后续所有刷新。改为硬 TTL 后:正常路径 finally 主动 Remove;即使挂死,键到 <see cref="RefreshLockTtlSeconds"/> 也自动过期。
  18. /// 配合 QueryT8Async 的 T8 命令超时(60s)与主库命令超时(30s),单次运行时长恒 &lt; TTL,正常不会误过期。
  19. /// </summary>
  20. public sealed class AidopT8KpiManualRefreshService : ITransient
  21. {
  22. /// <summary>手动刷新 + Cron 共享的全局互斥锁 key(覆盖 inbound + transform 整段)。</summary>
  23. public const string RefreshLockKey = "aidop:t8kpi:refresh:lock";
  24. /// <summary>刷新锁硬 TTL(秒,持有者无关):到点自动过期,即使某次运行意外挂死也不会永久阻塞后续刷新。
  25. /// 取值须大于加固后最坏运行时长(T8 查询各 ≤60s、主库命令 ≤30s,全程数分钟内),10 分钟为安全上限。</summary>
  26. public const int RefreshLockTtlSeconds = 600;
  27. // legacy 直连 T8 的 KPI(仅剩 S5_L1_004 TVF;S5_L1_002 已中台化),逐指标状态标 LEGACY_SUCCESS 以示区分。
  28. private static readonly HashSet<string> LegacyKpiCodes = new(StringComparer.OrdinalIgnoreCase) { "S5_L1_004" };
  29. private readonly T8BaseInboundMdpSyncService _inbound;
  30. private readonly S5MdpSyncTransformService _s5;
  31. private readonly S6MdpSyncTransformService _s6;
  32. private readonly S7MdpSyncTransformService _s7;
  33. private readonly SysCacheService _cache;
  34. private readonly ILogger<AidopT8KpiManualRefreshService> _logger;
  35. public AidopT8KpiManualRefreshService(
  36. T8BaseInboundMdpSyncService inbound,
  37. S5MdpSyncTransformService s5,
  38. S6MdpSyncTransformService s6,
  39. S7MdpSyncTransformService s7,
  40. SysCacheService cache,
  41. ILogger<AidopT8KpiManualRefreshService> logger)
  42. {
  43. _inbound = inbound;
  44. _s5 = s5;
  45. _s6 = s6;
  46. _s7 = s7;
  47. _cache = cache;
  48. _logger = logger;
  49. }
  50. /// <summary>执行一次模块级手动刷新(S5/S6/S7)。返回统一契约,绝不抛业务异常(取消除外)。</summary>
  51. public async Task<AidopKpiRefreshResult> RunModuleRefreshAsync(string moduleCode, CancellationToken cancellationToken)
  52. {
  53. var mc = string.IsNullOrWhiteSpace(moduleCode) ? string.Empty : moduleCode.Trim().ToUpperInvariant();
  54. var startedAt = DateTime.Now;
  55. var result = new AidopKpiRefreshResult
  56. {
  57. ModuleCode = mc,
  58. StartedAt = startedAt.ToString("yyyy-MM-dd HH:mm:ss")
  59. };
  60. if (mc is not ("S5" or "S6" or "S7"))
  61. {
  62. result.Ok = false;
  63. result.OverallStatus = "SKIPPED";
  64. result.Message = $"模块 {mc} 不支持 T8 KPI 手动刷新";
  65. Finish(result, startedAt);
  66. return result;
  67. }
  68. // 硬 TTL 锁:ExistKey 命中即判定已有刷新在跑(手动或定时);否则 Set(带 TTL) 占锁。
  69. // 单实例 + 人工点击/每日 5 次定时下,ExistKey→Set 的微小竞态概率可忽略;即便未主动释放,TTL 到点自动过期。
  70. if (_cache.ExistKey(RefreshLockKey))
  71. {
  72. result.Ok = false;
  73. result.OverallStatus = "REFRESHING";
  74. result.Message = "T8 KPI 刷新正在进行中(手动或定时任务占用),请稍后重试";
  75. Finish(result, startedAt);
  76. return result;
  77. }
  78. _cache.Set(RefreshLockKey, $"MANUAL:{mc}:{startedAt:yyyyMMddHHmmss}", TimeSpan.FromSeconds(RefreshLockTtlSeconds));
  79. try
  80. {
  81. // ① 入站硬门:共享 6 张 T8 基表全量入站;失败即中断,不用旧 std 伪装成功。
  82. try
  83. {
  84. // T8 源表无 tenant_id;手动刷新显式经账套映射解析真实租户。
  85. var t8TenantId = AidopSourceTenantMap.ResolveTenantId("pbxfxp", 0);
  86. var inbound = await _inbound.RunInboundAsync(t8TenantId, true, null, cancellationToken);
  87. result.Inbound.Ok = true;
  88. result.Inbound.StageRows = inbound.Tables.Sum(t => t.StgRows);
  89. result.Inbound.StandardRows = inbound.Tables.Sum(t => t.StdRows);
  90. }
  91. catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested)
  92. {
  93. throw;
  94. }
  95. catch (Exception ex)
  96. {
  97. _logger.LogError(ex, "[手动刷新] {Module} T8 基表入站失败", mc);
  98. result.Ok = false;
  99. result.OverallStatus = "FAILED";
  100. result.Inbound.Ok = false;
  101. result.Inbound.Message = ex.Message;
  102. result.Message = "源数据入站失败,未更新 KPI(保留原有看板数据)";
  103. Finish(result, startedAt);
  104. return result;
  105. }
  106. // ② 本模块 KPI 转换(RunFull 为全有全无:任一 KPI 异常即整体抛错)。
  107. string batchId;
  108. int dwdRows;
  109. int kpiRows;
  110. Dictionary<string, int> perKpiRows;
  111. List<string> denomStatus;
  112. try
  113. {
  114. switch (mc)
  115. {
  116. case "S5":
  117. var r5 = await _s5.RunFullAsync(cancellationToken, "MANUAL");
  118. batchId = r5.BatchId;
  119. dwdRows = r5.DwdRows;
  120. kpiRows = r5.KpiRows;
  121. perKpiRows = r5.PerKpiKpiRows;
  122. denomStatus = r5.KpiDenominatorStatus;
  123. break;
  124. case "S6":
  125. var r6 = await _s6.RunFullAsync(cancellationToken, "MANUAL");
  126. batchId = r6.BatchId;
  127. dwdRows = r6.DwdRows;
  128. kpiRows = r6.KpiRows;
  129. perKpiRows = r6.PerKpiKpiRows;
  130. denomStatus = r6.KpiDenominatorStatus;
  131. break;
  132. default:
  133. var r7 = await _s7.RunFullAsync(cancellationToken, "MANUAL");
  134. batchId = r7.BatchId;
  135. dwdRows = r7.DwdRows;
  136. kpiRows = r7.KpiRows;
  137. perKpiRows = r7.PerKpiKpiRows;
  138. denomStatus = r7.KpiDenominatorStatus;
  139. break;
  140. }
  141. }
  142. catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested)
  143. {
  144. throw;
  145. }
  146. catch (Exception ex)
  147. {
  148. _logger.LogError(ex, "[手动刷新] {Module} KPI 转换失败", mc);
  149. result.Ok = false;
  150. result.OverallStatus = "FAILED";
  151. result.Transform.Ok = false;
  152. result.Transform.Message = ex.Message;
  153. result.Message = "KPI 计算失败(源数据已入站,看板保留原有数据)";
  154. Finish(result, startedAt);
  155. return result;
  156. }
  157. result.BatchId = batchId;
  158. result.Transform.Ok = true;
  159. result.Transform.DwdRows = dwdRows;
  160. result.Transform.KpiRows = kpiRows;
  161. // ③ 逐指标状态:denom=OK → SUCCESS(legacy→LEGACY_SUCCESS);denom=NO_* → NO_DATA(真实无数据,非伪造 0)。
  162. var denomMap = ParseDenominator(denomStatus);
  163. foreach (var kv in perKpiRows.OrderBy(k => k.Key, StringComparer.Ordinal))
  164. {
  165. var code = kv.Key;
  166. var denom = denomMap.TryGetValue(code, out var d) ? d : "OK";
  167. var ok = string.Equals(denom, "OK", StringComparison.OrdinalIgnoreCase);
  168. var status = ok
  169. ? (LegacyKpiCodes.Contains(code) ? "LEGACY_SUCCESS" : "SUCCESS")
  170. : "NO_DATA";
  171. result.PerKpi.Add(new AidopKpiRefreshPerKpi
  172. {
  173. MetricCode = code,
  174. Status = status,
  175. Rows = kv.Value,
  176. Message = ok ? null : denom
  177. });
  178. }
  179. result.SuccessCount = result.PerKpi.Count(p => p.Status is "SUCCESS" or "LEGACY_SUCCESS");
  180. result.NoDataCount = result.PerKpi.Count(p => p.Status == "NO_DATA");
  181. result.FailedCount = result.PerKpi.Count(p => p.Status is "FAILED" or "LEGACY_FAILED");
  182. result.Ok = true;
  183. // RunFull 为全有全无,成功路径下 FailedCount 恒为 0;PARTIAL_SUCCESS 仅为契约保留态。
  184. result.OverallStatus = result.FailedCount > 0 ? "PARTIAL_SUCCESS" : "SUCCESS";
  185. Finish(result, startedAt);
  186. result.Message = $"刷新完成:成功 {result.SuccessCount},无数据 {result.NoDataCount},失败 {result.FailedCount},耗时 {result.DurationMs / 1000.0:0.0} 秒";
  187. return result;
  188. }
  189. finally
  190. {
  191. // 无论成功/失败/取消,主动释放锁(硬 TTL 是兜底,正常路径这里即时释放)。
  192. _cache.Remove(RefreshLockKey);
  193. }
  194. }
  195. /// <summary>把 "S5_L1_001:OK" 形式的分母状态列表解析为 metricCode → status 字典。</summary>
  196. private static Dictionary<string, string> ParseDenominator(List<string> denom)
  197. {
  198. var map = new Dictionary<string, string>(StringComparer.OrdinalIgnoreCase);
  199. foreach (var item in denom)
  200. {
  201. if (string.IsNullOrWhiteSpace(item)) continue;
  202. var idx = item.IndexOf(':');
  203. if (idx <= 0) continue;
  204. map[item.Substring(0, idx).Trim()] = item.Substring(idx + 1).Trim();
  205. }
  206. return map;
  207. }
  208. private static void Finish(AidopKpiRefreshResult result, DateTime startedAt)
  209. {
  210. var finishedAt = DateTime.Now;
  211. result.FinishedAt = finishedAt.ToString("yyyy-MM-dd HH:mm:ss");
  212. result.DurationMs = (long)(finishedAt - startedAt).TotalMilliseconds;
  213. }
  214. }
  215. /// <summary>手动刷新统一返回契约(PascalCase 属性经框架序列化为 camelCase:Ok→ok、BatchId→batchId)。</summary>
  216. public sealed class AidopKpiRefreshResult
  217. {
  218. public bool Ok { get; set; }
  219. public string ModuleCode { get; set; } = string.Empty;
  220. /// <summary>整体状态:SUCCESS / PARTIAL_SUCCESS / FAILED / REFRESHING / SKIPPED。</summary>
  221. public string OverallStatus { get; set; } = string.Empty;
  222. public string? BatchId { get; set; }
  223. public string StartedAt { get; set; } = string.Empty;
  224. public string FinishedAt { get; set; } = string.Empty;
  225. public long DurationMs { get; set; }
  226. public AidopKpiRefreshInbound Inbound { get; set; } = new();
  227. public AidopKpiRefreshTransform Transform { get; set; } = new();
  228. public int SuccessCount { get; set; }
  229. public int NoDataCount { get; set; }
  230. public int FailedCount { get; set; }
  231. public List<AidopKpiRefreshPerKpi> PerKpi { get; set; } = new();
  232. public string Message { get; set; } = string.Empty;
  233. }
  234. public sealed class AidopKpiRefreshInbound
  235. {
  236. public bool Ok { get; set; }
  237. public int StageRows { get; set; }
  238. public int StandardRows { get; set; }
  239. public string? Message { get; set; }
  240. }
  241. public sealed class AidopKpiRefreshTransform
  242. {
  243. public bool Ok { get; set; }
  244. public int DwdRows { get; set; }
  245. public int KpiRows { get; set; }
  246. public string? Message { get; set; }
  247. }
  248. /// <summary>单指标刷新状态:SUCCESS / LEGACY_SUCCESS / NO_DATA / FAILED / LEGACY_FAILED。</summary>
  249. public sealed class AidopKpiRefreshPerKpi
  250. {
  251. public string MetricCode { get; set; } = string.Empty;
  252. public string Status { get; set; } = string.Empty;
  253. public int Rows { get; set; }
  254. public string? Message { get; set; }
  255. }