AidopT8KpiManualRefreshService.cs 13 KB

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