S7MdpSyncTransformService.cs 46 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942
  1. using Admin.NET.Core.Service;
  2. using Admin.NET.Plugin.AiDOP.Infrastructure;
  3. using Admin.NET.Plugin.AiDOP.MaterialWarehouse;
  4. using Microsoft.Extensions.Logging;
  5. using System.Text.Json;
  6. namespace Admin.NET.Plugin.AiDOP.FinishedWarehouse;
  7. /// <summary>
  8. /// S7 成品仓储 — KPI 计算与刷新转换服务。
  9. /// 双模式:读本地中立标准层(由适配器从贴源→标准),不再直连源库。
  10. /// 计算逻辑沿用方老师 v5.4 KPI J 列口径(等价改写为 MySQL 读 std)。
  11. /// 包含 KPI:S7_L1_001 订单发货周期 / S7_L1_002 订单发货满足率 / S7_L1_003 成品仓储人效。
  12. /// </summary>
  13. public class S7MdpSyncTransformService : ITransient
  14. {
  15. private readonly ISqlSugarClient _db;
  16. private readonly TransformRunLogFinalizer _runLogFinalizer;
  17. private readonly SysNoticeService _sysNoticeService;
  18. private readonly ILogger<S7MdpSyncTransformService> _logger;
  19. private readonly SmartOps.KpiCalcDispatcher _kpiCalcDispatcher;
  20. private readonly SmartOps.KpiDimensionRunService _dimensionRun;
  21. private readonly SmartOps.IKpiTargetResolver _kpiTargetResolver;
  22. private readonly InventoryMdpSyncService _inventoryMdpSync;
  23. private readonly SmartOps.S9CompositeKpiWriter _s9Composite;
  24. private const string JobCode = "S7_MDP_SYNC_TRANSFORM";
  25. private const string JobName = "S7 成品仓储 MDP 同步与转换";
  26. private const string ModuleCode = "S7";
  27. private const string L2ValueTable = "ado_s9_kpi_value_l2_day";
  28. // FAILURE-NOTIFICATION-1:超级管理员 superAdmin.NET(AccountType=999)
  29. private const long NoticeReceiverUserId = 1300000000101L;
  30. private const string NoticeReceiverUserName = "超级管理员";
  31. public S7MdpSyncTransformService(
  32. ISqlSugarClient db,
  33. SysNoticeService sysNoticeService,
  34. ILogger<S7MdpSyncTransformService> logger,
  35. SmartOps.KpiCalcDispatcher kpiCalcDispatcher,
  36. SmartOps.KpiDimensionRunService dimensionRun,
  37. SmartOps.IKpiTargetResolver kpiTargetResolver,
  38. InventoryMdpSyncService inventoryMdpSync,
  39. TransformRunLogFinalizer runLogFinalizer,
  40. SmartOps.S9CompositeKpiWriter s9Composite)
  41. {
  42. _db = db;
  43. _runLogFinalizer = runLogFinalizer;
  44. _sysNoticeService = sysNoticeService;
  45. _logger = logger;
  46. _kpiCalcDispatcher = kpiCalcDispatcher;
  47. _dimensionRun = dimensionRun;
  48. _kpiTargetResolver = kpiTargetResolver;
  49. _inventoryMdpSync = inventoryMdpSync;
  50. _s9Composite = s9Composite;
  51. }
  52. public async Task<S7MdpSyncTransformResult> RunFullAsync(
  53. CancellationToken cancellationToken = default,
  54. string triggerType = "AUTO",
  55. S7MdpRefreshOption? option = null)
  56. {
  57. cancellationToken.ThrowIfCancellationRequested();
  58. option ??= S7MdpRefreshOption.Default();
  59. NormalizeOption(option);
  60. var now = DateTime.Now;
  61. var batchId = $"S7_MDP_FULL_{now:yyyyMMddHHmmss}";
  62. var normalizedTrigger = NormalizeTriggerType(triggerType);
  63. var runLogId = await InsertTransformRunLogAsync(batchId, now, normalizedTrigger, option);
  64. var result = new S7MdpSyncTransformResult
  65. {
  66. BatchId = batchId,
  67. RunLogId = runLogId,
  68. TriggerType = normalizedTrigger,
  69. SourceZtid = option.SourceZtid,
  70. TargetTenantId = option.TargetTenantId,
  71. TargetFactoryId = option.TargetFactoryId,
  72. BizDate = option.BizDate,
  73. BizMonth = option.BizMonth,
  74. MonthlyPeriodStart = option.MonthlyPeriodStart,
  75. MonthlyPeriodEnd = option.MonthlyPeriodEnd
  76. };
  77. try
  78. {
  79. result.StageRows = 0;
  80. result.StandardRows = await _inventoryMdpSync.TransformTransStdFromStgAsync(
  81. option.TargetTenantId, cancellationToken);
  82. var sub25 = await BuildS7L1001OrderShipmentCycleAsync(batchId, now, option, normalizedTrigger, cancellationToken);
  83. result.MergeSub("S7_L1_001", sub25);
  84. var sub26 = await BuildS7L1002OrderShipmentFulfillmentAsync(batchId, now, option, normalizedTrigger, cancellationToken);
  85. result.MergeSub("S7_L1_002", sub26);
  86. var sub27 = await BuildS7L1003FinishedWarehouseEfficiencyAsync(batchId, now, option, normalizedTrigger, cancellationToken);
  87. result.MergeSub("S7_L1_003", sub27);
  88. var currentBizDate = option.BizDate;
  89. const int backfillDays = 14;
  90. for (var dayOffset = backfillDays - 1; dayOffset >= 0; dayOffset--)
  91. {
  92. option.BizDate = currentBizDate.AddDays(-dayOffset);
  93. foreach (var metricCode in Enumerable.Range(1, 16).Select(x => $"S7_L2_{x:000}"))
  94. {
  95. var sub = await BuildS7StageKpiAsync(
  96. metricCode, batchId, now, option, normalizedTrigger, cancellationToken);
  97. result.MergeSub(metricCode, sub);
  98. }
  99. result.MergeSub("S9_L1_001", await BuildS9L1001QualityReturnRateAsync(
  100. batchId, now, option, normalizedTrigger, cancellationToken));
  101. }
  102. option.BizDate = currentBizDate;
  103. result.KpiRows += await _s9Composite.WriteRecentAsync(
  104. option.TargetTenantId, option.TargetFactoryId, option.BizDate, cancellationToken);
  105. await MarkTransformRunSuccessAsync(runLogId, now, result);
  106. return result;
  107. }
  108. catch (Exception ex)
  109. {
  110. // 宿主关停不是转换失败:交给 finally 收口为 ABORTED,不污染 FAILED 语义。
  111. if (!_runLogFinalizer.IsHostStopping)
  112. await MarkTransformRunFailedAsync(runLogId, now, ex.Message, batchId);
  113. throw;
  114. }
  115. finally
  116. {
  117. await _runLogFinalizer.FinalizeIfHostStoppingAsync(runLogId, now);
  118. }
  119. }
  120. // ─────────────────────────────────────────────────────────────────────────
  121. /// <summary>S7_L1_001 订单发货周期 = 最晚发货日期 - 最早 FQC 报检日期(5 表 JOIN)。</summary>
  122. private async Task<KpiBuildSubResult> BuildS7L1001OrderShipmentCycleAsync(
  123. string batchId, DateTime now, S7MdpRefreshOption option, string triggerType, CancellationToken ct)
  124. {
  125. var sub = new KpiBuildSubResult();
  126. const string sql = @"
  127. select order_no as order_no,
  128. datediff(max(ship_time), min(start_time)) as cycle_days
  129. from (
  130. select b.order_no as order_no, b.item_code as item_code,
  131. ifnull(c.apply_time, b.release_time) as start_time,
  132. (case when b.closed_flag=1 then b.closed_time else d.approved_time end) as ship_time,
  133. (case when b.closed_flag=1 or b.qty_planned<=b.qty_completed then 1 else 0 end) as completed
  134. from mdp_std_work_order_line b
  135. left join (
  136. select tenant_id as tenant_id, ref_task_no, item_num,
  137. max(approved_time) as approved_time, sum(qty_change) as qty_change
  138. from mdp_std_inv_trans
  139. where tenant_id=@tenantId and biz_doc_type='SALES_SHIP'
  140. and summary_flag=0 and void_flag=0 and approved_flag=1 AND (@sourceSystem='' OR source_system=@sourceSystem) AND (source_system<>'T8' OR domain=@sourceDomain)
  141. group by tenant_id, ref_task_no, item_num
  142. ) d on b.task_no=d.ref_task_no and b.item_code=d.item_num and d.tenant_id=b.tenant_id
  143. left join mdp_std_fqc_task c on c.sales_order_line=b.source_row_id and c.apply_flag=1 and c.tenant_id=b.tenant_id
  144. where b.tenant_id=@tenantId and b.doc_type='SALES_ORDER' and b.void_flag=0 and b.approved_flag=1
  145. ) n
  146. group by order_no
  147. having min(completed)=1";
  148. var rows = await _db.Ado.SqlQueryAsync<S7CycleRow>(sql, new[]
  149. {
  150. new SugarParameter("@tenantId", option.TargetTenantId),
  151. new SugarParameter("@sourceDomain", option.SourceZtid),
  152. new SugarParameter("@sourceSystem", "")
  153. });
  154. sub.T8Rows = rows.Count;
  155. var dwdAffected = 0;
  156. var cycleList = new List<int>();
  157. foreach (var r in rows)
  158. {
  159. ct.ThrowIfCancellationRequested();
  160. if (string.IsNullOrEmpty(r.order_no)) continue;
  161. if (r.cycle_days.HasValue) cycleList.Add(r.cycle_days.Value);
  162. dwdAffected += await _db.Ado.ExecuteCommandAsync(@"
  163. INSERT INTO dwd_order_shipment_cycle
  164. (tenant_id, factory_id, biz_date, source_ztid, order_no, cycle_days, batch_id, create_time)
  165. VALUES
  166. (@tenantId, @factoryId, @bizDate, @sourceDomain, @orderNo, @cycleDays, @batchId, @now)
  167. ON DUPLICATE KEY UPDATE
  168. cycle_days=VALUES(cycle_days),
  169. batch_id=VALUES(batch_id), update_time=@now",
  170. new SugarParameter("@tenantId", option.TargetTenantId),
  171. new SugarParameter("@factoryId", option.TargetFactoryId),
  172. new SugarParameter("@bizDate", option.BizDate),
  173. new SugarParameter("@sourceDomain", option.SourceZtid),
  174. new SugarParameter("@orderNo", r.order_no),
  175. new SugarParameter("@cycleDays", r.cycle_days),
  176. new SugarParameter("@batchId", batchId),
  177. new SugarParameter("@now", now));
  178. }
  179. sub.DwdRows = dwdAffected;
  180. // 数据准备(dwd 逐单 cycle_days)已完成。最终聚合交分发器(日 KPI:period 复用 BizDate,SQL 按 biz_date 圈选)。
  181. decimal? legacyValue = cycleList.Count > 0
  182. ? Math.Round((decimal)cycleList.Average(), 4)
  183. : null;
  184. var legacyDenom = cycleList.Count > 0 ? "OK" : "NO_COMPLETED_ORDER";
  185. var dispatch = await _kpiCalcDispatcher.DispatchAsync(
  186. "S7_L1_001", ModuleCode, option.TargetTenantId, option.TargetFactoryId,
  187. option.BizDate, option.BizDate, option.BizDate, option.SourceZtid,
  188. batchId, triggerType, legacyValue, legacyDenom, ct);
  189. sub.KpiRows = dispatch.ShouldUpsert
  190. ? await UpsertKpiValueAsync("S7_L1_001", option.BizDate, dispatch.MetricValue, now, option)
  191. : 0;
  192. sub.DenominatorStatus = dispatch.DenominatorStatus;
  193. // 调度:SUMMARY 成功/NO_DATA 后触发对应 DIMENSION 跑批(共享 BatchId;SUMMARY FAILED 不触发)。
  194. // 维度失败不影响汇总链路(内部已落 dimension_run_log)。
  195. if (dispatch.ShouldUpsert)
  196. {
  197. try
  198. {
  199. await _dimensionRun.RunDimensionAsync(
  200. "S7_L1_001", ModuleCode, option.TargetTenantId, option.BizDate, batchId, triggerType, ct);
  201. }
  202. catch (Exception ex)
  203. {
  204. _logger.LogWarning(ex, "S7_L1_001 维度跑批异常(不影响汇总链路)");
  205. }
  206. }
  207. return sub;
  208. }
  209. /// <summary>S7_L1_002 订单发货满足率 = (交期前发货行数 / 该订单总行数) × 100%。</summary>
  210. private async Task<KpiBuildSubResult> BuildS7L1002OrderShipmentFulfillmentAsync(
  211. string batchId, DateTime now, S7MdpRefreshOption option, string triggerType, CancellationToken ct)
  212. {
  213. var sub = new KpiBuildSubResult();
  214. const string sql = @"
  215. select order_no as order_no,
  216. count(order_no) as total_rows,
  217. sum(completed) as in_window_rows
  218. from (
  219. select b.order_no as order_no, b.task_no as task_no, b.item_code as item_code,
  220. (case when sum(d.qty_change)>=b.qty_planned then 1 else 0 end) as completed
  221. from mdp_std_work_order_line b
  222. left join (
  223. select tenant_id as tenant_id, ref_task_no, item_num,
  224. date(approved_time) as approved_date, qty_change
  225. from mdp_std_inv_trans
  226. where tenant_id=@tenantId and biz_doc_type='SALES_SHIP'
  227. and summary_flag=0 and void_flag=0 and approved_flag=1 AND (@sourceSystem='' OR source_system=@sourceSystem) AND (source_system<>'T8' OR domain=@sourceDomain)
  228. ) d on b.task_no=d.ref_task_no and b.item_code=d.item_num
  229. and d.approved_date<=b.plan_finish_date and d.tenant_id=b.tenant_id
  230. where b.tenant_id=@tenantId and b.doc_type='SALES_ORDER' and b.void_flag=0 and b.approved_flag=1
  231. group by b.order_no, b.task_no, b.item_code, b.qty_planned
  232. ) n
  233. group by order_no";
  234. var rows = await _db.Ado.SqlQueryAsync<S7FulfillmentRow>(sql, new[]
  235. {
  236. new SugarParameter("@tenantId", option.TargetTenantId),
  237. new SugarParameter("@sourceDomain", option.SourceZtid),
  238. new SugarParameter("@sourceSystem", "")
  239. });
  240. sub.T8Rows = rows.Count;
  241. var dwdAffected = 0;
  242. var rateList = new List<decimal>();
  243. foreach (var r in rows)
  244. {
  245. ct.ThrowIfCancellationRequested();
  246. if (string.IsNullOrEmpty(r.order_no)) continue;
  247. decimal? rate = (r.total_rows > 0)
  248. ? Math.Round((decimal)r.in_window_rows / r.total_rows * 100m, 4)
  249. : null;
  250. if (rate.HasValue) rateList.Add(rate.Value);
  251. dwdAffected += await _db.Ado.ExecuteCommandAsync(@"
  252. INSERT INTO dwd_order_shipment_fulfillment
  253. (tenant_id, factory_id, biz_date, source_ztid, order_no,
  254. total_rows, in_window_rows, fulfillment_rate, batch_id, create_time)
  255. VALUES
  256. (@tenantId, @factoryId, @bizDate, @sourceDomain, @orderNo,
  257. @total, @inWindow, @rate, @batchId, @now)
  258. ON DUPLICATE KEY UPDATE
  259. total_rows=VALUES(total_rows), in_window_rows=VALUES(in_window_rows),
  260. fulfillment_rate=VALUES(fulfillment_rate),
  261. batch_id=VALUES(batch_id), update_time=@now",
  262. new SugarParameter("@tenantId", option.TargetTenantId),
  263. new SugarParameter("@factoryId", option.TargetFactoryId),
  264. new SugarParameter("@bizDate", option.BizDate),
  265. new SugarParameter("@sourceDomain", option.SourceZtid),
  266. new SugarParameter("@orderNo", r.order_no),
  267. new SugarParameter("@total", r.total_rows),
  268. new SugarParameter("@inWindow", r.in_window_rows),
  269. new SugarParameter("@rate", rate),
  270. new SugarParameter("@batchId", batchId),
  271. new SugarParameter("@now", now));
  272. }
  273. sub.DwdRows = dwdAffected;
  274. // 数据准备(dwd 逐单 fulfillment_rate,已 ×100)已完成。最终聚合交分发器(CONFIG_SQL 直接 AVG 该列不再 ×100)。
  275. decimal? legacyValue = rateList.Count > 0
  276. ? Math.Round(rateList.Average(), 4)
  277. : null;
  278. var legacyDenom = rateList.Count > 0 ? "OK" : "NO_VALID_ORDER";
  279. var dispatch = await _kpiCalcDispatcher.DispatchAsync(
  280. "S7_L1_002", ModuleCode, option.TargetTenantId, option.TargetFactoryId,
  281. option.BizDate, option.BizDate, option.BizDate, option.SourceZtid,
  282. batchId, triggerType, legacyValue, legacyDenom, ct);
  283. sub.KpiRows = dispatch.ShouldUpsert
  284. ? await UpsertKpiValueAsync("S7_L1_002", option.BizDate, dispatch.MetricValue, now, option)
  285. : 0;
  286. sub.DenominatorStatus = dispatch.DenominatorStatus;
  287. // 调度:SUMMARY 成功/NO_DATA 后触发对应 DIMENSION 跑批(共享 BatchId;SUMMARY FAILED 不触发)。
  288. if (dispatch.ShouldUpsert)
  289. {
  290. try
  291. {
  292. await _dimensionRun.RunDimensionAsync(
  293. "S7_L1_002", ModuleCode, option.TargetTenantId, option.BizDate, batchId, triggerType, ct);
  294. }
  295. catch (Exception ex)
  296. {
  297. _logger.LogWarning(ex, "S7_L1_002 维度跑批异常(不影响汇总链路)");
  298. }
  299. }
  300. return sub;
  301. }
  302. /// <summary>S7_L1_003 成品仓储人效 = SALES_SHIP 数量 / WAREHOUSE ACTIVE 人数。</summary>
  303. private async Task<KpiBuildSubResult> BuildS7L1003FinishedWarehouseEfficiencyAsync(
  304. string batchId, DateTime now, S7MdpRefreshOption option, string triggerType, CancellationToken ct)
  305. {
  306. var sub = new KpiBuildSubResult();
  307. const string sqlNumer = @"
  308. select ref_task_no as task_no, item_num as item_code,
  309. date(approved_time) as approved_date, qty_change as qty_change
  310. from mdp_std_inv_trans
  311. where tenant_id=@tenantId and biz_doc_type='SALES_SHIP'
  312. and summary_flag=0 and void_flag=0 and approved_flag=1 AND (@sourceSystem='' OR source_system=@sourceSystem) AND (source_system<>'T8' OR domain=@sourceDomain)
  313. and date(approved_time) between @startDateText and @endDateText";
  314. const string sqlDenom = @"
  315. select count(*) as penum
  316. from mdp_std_employee
  317. where tenant_id=@tenantId and employment_status='ACTIVE' and position_code='WAREHOUSE' AND (@sourceSystem='' OR source_system=@sourceSystem) AND (source_system<>'T8' OR domain=@sourceDomain)";
  318. var pNumer = new[]
  319. {
  320. new SugarParameter("@tenantId", option.TargetTenantId),
  321. new SugarParameter("@startDateText", option.MonthlyPeriodStart.ToString("yyyy-MM-dd")),
  322. new SugarParameter("@endDateText", option.MonthlyPeriodEnd.ToString("yyyy-MM-dd")),
  323. new SugarParameter("@sourceDomain", option.SourceZtid),
  324. new SugarParameter("@sourceSystem", "")
  325. };
  326. var pDenom = new[]
  327. {
  328. new SugarParameter("@tenantId", option.TargetTenantId),
  329. new SugarParameter("@sourceDomain", option.SourceZtid),
  330. new SugarParameter("@sourceSystem", "")
  331. };
  332. var numerRows = await _db.Ado.SqlQueryAsync<S7ShipmentDetailRow>(sqlNumer, pNumer);
  333. var denomRows = await _db.Ado.SqlQueryAsync<S7PeNumRow>(sqlDenom, pDenom);
  334. sub.T8Rows = numerRows.Count + denomRows.Count;
  335. decimal? shipmentQty = numerRows.Sum(r => r.qty_change ?? 0m);
  336. if (numerRows.Count == 0) shipmentQty = null;
  337. int? headcount = denomRows.FirstOrDefault()?.penum;
  338. decimal? efficiency = null;
  339. string denomStatus;
  340. if (!headcount.HasValue || headcount.Value <= 0)
  341. denomStatus = "NO_HEADCOUNT";
  342. else if (!shipmentQty.HasValue)
  343. denomStatus = "NO_NUMERATOR";
  344. else
  345. {
  346. efficiency = Math.Round(shipmentQty.Value / headcount.Value, 4);
  347. denomStatus = "OK";
  348. }
  349. sub.DenominatorStatus = denomStatus;
  350. var dwdAffected = await _db.Ado.ExecuteCommandAsync(@"
  351. INSERT INTO dwd_finished_warehouse_efficiency
  352. (tenant_id, factory_id, biz_month, source_ztid, period_start, period_end,
  353. shipment_qty, warehouse_headcount, efficiency, denominator_status, batch_id, create_time)
  354. VALUES
  355. (@tenantId, @factoryId, @bizMonth, @sourceDomain, @periodStart, @periodEnd,
  356. @shipmentQty, @headcount, @efficiency, @denomStatus, @batchId, @now)
  357. ON DUPLICATE KEY UPDATE
  358. period_start=VALUES(period_start), period_end=VALUES(period_end),
  359. shipment_qty=VALUES(shipment_qty), warehouse_headcount=VALUES(warehouse_headcount),
  360. efficiency=VALUES(efficiency), denominator_status=VALUES(denominator_status),
  361. batch_id=VALUES(batch_id), update_time=@now",
  362. new SugarParameter("@tenantId", option.TargetTenantId),
  363. new SugarParameter("@factoryId", option.TargetFactoryId),
  364. new SugarParameter("@bizMonth", option.BizMonth),
  365. new SugarParameter("@sourceDomain", option.SourceZtid),
  366. new SugarParameter("@periodStart", option.MonthlyPeriodStart),
  367. new SugarParameter("@periodEnd", option.MonthlyPeriodEnd),
  368. new SugarParameter("@shipmentQty", shipmentQty),
  369. new SugarParameter("@headcount", headcount),
  370. new SugarParameter("@efficiency", efficiency),
  371. new SugarParameter("@denomStatus", denomStatus),
  372. new SugarParameter("@batchId", batchId),
  373. new SugarParameter("@now", now));
  374. sub.DwdRows = dwdAffected;
  375. // 月度 KPI 最终聚合交分发器;bizDate=月末、period=当月窗口;
  376. // legacyValue=efficiency、legacyDenom 保留 NO_HEADCOUNT/NO_NUMERATOR(CONFIG_SQL 下塌缩为 NO_DATA)。
  377. var dispatch = await _kpiCalcDispatcher.DispatchAsync(
  378. "S7_L1_003", ModuleCode, option.TargetTenantId, option.TargetFactoryId,
  379. option.MonthlyPeriodEnd, option.MonthlyPeriodStart, option.MonthlyPeriodEnd, option.SourceZtid,
  380. batchId, triggerType, efficiency, denomStatus, ct);
  381. sub.KpiRows = dispatch.ShouldUpsert
  382. ? await UpsertKpiValueAsync("S7_L1_003", option.MonthlyPeriodEnd, dispatch.MetricValue, now, option)
  383. : 0;
  384. sub.DenominatorStatus = dispatch.DenominatorStatus;
  385. // 调度:SUMMARY 成功/NO_DATA 后触发人效月度 DIMENSION 跑批(月度:@biz_date=月末派生 biz_month)。
  386. if (dispatch.ShouldUpsert)
  387. {
  388. try
  389. {
  390. await _dimensionRun.RunDimensionAsync(
  391. "S7_L1_003", ModuleCode, option.TargetTenantId, option.MonthlyPeriodEnd, batchId, triggerType, ct);
  392. }
  393. catch (Exception ex)
  394. {
  395. _logger.LogWarning(ex, "S7_L1_003 维度跑批异常(不影响汇总链路)");
  396. }
  397. }
  398. return sub;
  399. }
  400. private Task<KpiBuildSubResult> BuildS7StageKpiAsync(
  401. string metricCode, string batchId, DateTime now, S7MdpRefreshOption option,
  402. string triggerType, CancellationToken ct)
  403. {
  404. var (sql, emptyDenom) = metricCode switch
  405. {
  406. "S7_L2_001" => (
  407. """
  408. SELECT ROUND(AVG(TIMESTAMPDIFF(MINUTE,q.FAPPLYTIME,q.jywcsj)/1440),4) AS MetricValue,
  409. COUNT(*) AS RowCount
  410. FROM qms_fqcbj q
  411. WHERE q.tenant_id=@TenantId
  412. AND q.FAPPLYTIME IS NOT NULL AND q.jywcsj>=q.FAPPLYTIME
  413. AND UPPER(IFNULL(q.FINSPECTSTATUS,'')) IN ('检验完成','COMPLETED','CLOSED')
  414. """,
  415. "NO_COMPLETED_FQC"),
  416. "S7_L2_002" => (
  417. """
  418. SELECT ROUND(
  419. 100 * SUM(CASE WHEN q.jywcsj IS NOT NULL AND q.jywcsj<=w.DueDate THEN 1 ELSE 0 END)
  420. / NULLIF(COUNT(*),0),
  421. 4) AS MetricValue,
  422. COUNT(*) AS RowCount
  423. FROM qms_fqcbj q
  424. INNER JOIN WorkOrdMaster w
  425. ON w.tenant_id=q.tenant_id AND w.WorkOrd=q.sczld
  426. WHERE q.tenant_id=@TenantId AND w.DueDate IS NOT NULL
  427. """,
  428. "NO_FQC_REQUIRED_DATE"),
  429. "S7_L2_003" => (
  430. """
  431. SELECT ROUND(
  432. SUM(CASE WHEN q.jywcsj IS NOT NULL
  433. AND UPPER(IFNULL(q.FINSPECTSTATUS,'')) IN ('检验完成','COMPLETED','CLOSED')
  434. THEN 1 ELSE 0 END)
  435. / NULLIF(COUNT(DISTINCT NULLIF(TRIM(q.jyfzr),'')),0),
  436. 4) AS MetricValue,
  437. SUM(CASE WHEN q.jywcsj IS NOT NULL THEN 1 ELSE 0 END) AS RowCount
  438. FROM qms_fqcbj q
  439. WHERE q.tenant_id=@TenantId
  440. """,
  441. "NO_FQC_INSPECTOR"),
  442. "S7_L2_004" => (WarehouseTurnoverSql("FG_FQC_RELEASE", "FG_PROD_RECEIPT"), "NO_FINISHED_RECEIPT_COST"),
  443. "S7_L2_005" => (WarehouseCycleSql("FG_FQC_RELEASE", "FG_PUTAWAY"), "NO_FINISHED_PUTAWAY_CYCLE"),
  444. "S7_L2_006" => (WarehouseSatisfactionSql("FG_PUTAWAY"), "NO_FINISHED_PUTAWAY_REQUIRED_DATE"),
  445. "S7_L2_007" => (WarehouseEfficiencySql("FG_PUTAWAY"), "NO_FINISHED_PUTAWAY_OPERATOR"),
  446. "S7_L2_008" => (WarehouseTurnoverSql("FG_PUTAWAY", "FG_PROD_RECEIPT"), "NO_FINISHED_RECEIPT_COST"),
  447. "S7_L2_009" => (WarehouseCycleSql("FG_PUTAWAY", "FG_PICK"), "NO_FINISHED_PICK_CYCLE"),
  448. "S7_L2_010" => (WarehouseSatisfactionSql("FG_PICK"), "NO_FINISHED_PICK_REQUIRED_DATE"),
  449. "S7_L2_011" => (WarehouseEfficiencySql("FG_PICK"), "NO_FINISHED_PICK_OPERATOR"),
  450. "S7_L2_012" => (WarehouseTurnoverSql("FG_PICK", "FG_SHIP"), "NO_FINISHED_SHIPMENT_COST"),
  451. "S7_L2_013" => (WarehouseCycleSql("FG_SHIP", "FG_RECEIPT"), "NO_FINISHED_DELIVERY_CYCLE"),
  452. "S7_L2_014" => (
  453. """
  454. SELECT ROUND(100 * SUM(completed) / NULLIF(COUNT(*),0),4) AS MetricValue,
  455. COUNT(*) AS RowCount
  456. FROM (
  457. SELECT b.order_no, b.task_no, b.item_code,
  458. CASE WHEN SUM(IFNULL(d.qty_change,0))>=b.qty_planned THEN 1 ELSE 0 END AS completed
  459. FROM mdp_std_work_order_line b
  460. LEFT JOIN (
  461. SELECT tenant_id, ref_task_no, item_num, qty_change, DATE(approved_time) AS ship_date
  462. FROM mdp_std_inv_trans
  463. WHERE tenant_id=@TenantId
  464. AND biz_doc_type='SALES_SHIP' AND summary_flag=0 AND void_flag=0 AND approved_flag=1 AND (@sourceSystem='' OR source_system=@sourceSystem) AND (source_system<>'T8' OR domain=@sourceDomain)
  465. ) d ON d.tenant_id=b.tenant_id AND d.ref_task_no=b.task_no
  466. AND d.item_num=b.item_code AND d.ship_date<=b.plan_finish_date
  467. WHERE b.tenant_id=@TenantId
  468. AND b.doc_type='SALES_ORDER' AND b.void_flag=0 AND b.approved_flag=1
  469. GROUP BY b.order_no, b.task_no, b.item_code, b.qty_planned
  470. ) x
  471. """,
  472. "NO_SHIPMENT_NOTICE"),
  473. "S7_L2_015" => (WarehouseEfficiencySql("FG_SHIP"), "NO_FINISHED_SHIPMENT_OPERATOR"),
  474. "S7_L2_016" => (WarehouseTurnoverSql("FG_SHIP", "FG_RECEIPT"), "NO_SIGNED_RECEIPT_COST"),
  475. _ => throw new ArgumentOutOfRangeException(nameof(metricCode), metricCode, "不支持的 S7 阶段指标")
  476. };
  477. return DispatchS7StageKpiAsync(
  478. metricCode, sql, emptyDenom, batchId, now, option, triggerType, ct);
  479. }
  480. private static string WarehouseCycleSql(string fromStage, string toStage) =>
  481. $"""
  482. SELECT ROUND(AVG(TIMESTAMPDIFF(MINUTE,a.trans_time,b.trans_time)/1440),4) AS MetricValue,
  483. COUNT(*) AS RowCount
  484. FROM mdp_std_inv_trans a
  485. INNER JOIN mdp_std_inv_trans b
  486. ON b.tenant_id=a.tenant_id AND b.source_system=a.source_system
  487. AND b.item_num=a.item_num AND b.lot_serial=a.lot_serial
  488. AND b.trans_type='{toStage}'
  489. WHERE a.tenant_id=@TenantId AND a.trans_type='{fromStage}'
  490. AND (@sourceSystem='' OR a.source_system=@sourceSystem) AND (a.source_system<>'T8' OR a.domain=@sourceDomain)
  491. AND a.trans_time IS NOT NULL AND b.trans_time>=a.trans_time
  492. """;
  493. private static string WarehouseSatisfactionSql(string stage) =>
  494. $"""
  495. SELECT ROUND(100 * SUM(CASE WHEN trans_time<=eff_date THEN 1 ELSE 0 END)
  496. / NULLIF(COUNT(*),0),4) AS MetricValue,
  497. COUNT(*) AS RowCount
  498. FROM mdp_std_inv_trans
  499. WHERE tenant_id=@TenantId AND trans_type='{stage}'
  500. AND (@sourceSystem='' OR source_system=@sourceSystem) AND (source_system<>'T8' OR domain=@sourceDomain)
  501. AND trans_time IS NOT NULL AND eff_date IS NOT NULL
  502. """;
  503. private static string WarehouseEfficiencySql(string stage) =>
  504. $"""
  505. SELECT ROUND(COUNT(DISTINCT NULLIF(lot_serial,''))
  506. / NULLIF(COUNT(DISTINCT NULLIF(TRIM(create_user),'')),0),4) AS MetricValue,
  507. COUNT(*) AS RowCount
  508. FROM mdp_std_inv_trans
  509. WHERE tenant_id=@TenantId AND trans_type='{stage}'
  510. AND (@sourceSystem='' OR source_system=@sourceSystem) AND (source_system<>'T8' OR domain=@sourceDomain)
  511. """;
  512. private static string WarehouseTurnoverSql(string inventoryStage, string flowStage) =>
  513. $"""
  514. SELECT ROUND(
  515. 30 * SUM(CASE WHEN trans_type='{inventoryStage}'
  516. THEN IFNULL(end_balance,0)
  517. * IFNULL(CAST(NULLIF(dimension1,'') AS DECIMAL(18,4)),1)
  518. ELSE 0 END)
  519. / NULLIF(SUM(CASE WHEN trans_type='{flowStage}'
  520. THEN ABS(IFNULL(qty_change,0))
  521. * IFNULL(CAST(NULLIF(dimension1,'') AS DECIMAL(18,4)),1)
  522. ELSE 0 END),0),
  523. 4) AS MetricValue,
  524. SUM(CASE WHEN trans_type='{inventoryStage}' THEN 1 ELSE 0 END) AS RowCount
  525. FROM mdp_std_inv_trans
  526. WHERE tenant_id=@TenantId AND trans_type IN ('{inventoryStage}','{flowStage}')
  527. AND (@sourceSystem='' OR source_system=@sourceSystem) AND (source_system<>'T8' OR domain=@sourceDomain)
  528. """;
  529. private async Task<KpiBuildSubResult> DispatchS7StageKpiAsync(
  530. string metricCode, string sql, string emptyDenom, string batchId, DateTime now,
  531. S7MdpRefreshOption option, string triggerType, CancellationToken ct)
  532. {
  533. var row = await _db.Ado.SqlQuerySingleAsync<S7StageKpiRow>(
  534. sql,
  535. new SugarParameter("@TenantId", option.TargetTenantId),
  536. new SugarParameter("@sourceDomain", option.SourceZtid),
  537. new SugarParameter("@sourceSystem", ""));
  538. var value = row?.RowCount > 0 ? row.MetricValue : null;
  539. var dispatch = await _kpiCalcDispatcher.DispatchAsync(
  540. metricCode, ModuleCode, option.TargetTenantId, option.TargetFactoryId,
  541. option.BizDate, option.BizDate, option.BizDate, option.SourceZtid,
  542. batchId, triggerType, value, value.HasValue ? "OK" : emptyDenom, ct);
  543. var sub = new KpiBuildSubResult
  544. {
  545. T8Rows = row?.RowCount ?? 0,
  546. KpiRows = dispatch.ShouldUpsert
  547. ? await UpsertKpiValueAsync(
  548. metricCode, option.BizDate, dispatch.MetricValue, now, option,
  549. valueTable: L2ValueTable)
  550. : 0,
  551. DenominatorStatus = dispatch.DenominatorStatus
  552. };
  553. if (dispatch.ShouldUpsert)
  554. {
  555. try
  556. {
  557. await _dimensionRun.RunDimensionAsync(
  558. metricCode, ModuleCode, option.TargetTenantId, option.BizDate, batchId, triggerType, ct);
  559. }
  560. catch (Exception ex)
  561. {
  562. _logger.LogWarning(ex, "{MetricCode} 维度跑批异常(不影响汇总链路)", metricCode);
  563. }
  564. }
  565. return sub;
  566. }
  567. // ─────────────────────────────────────────────────────────────────────────
  568. /// <summary>S9_L1_001 质量退货率 = SALES_RETURN 数量 / SALES_SHIP 数量 × 1,000,000 PPM。</summary>
  569. private async Task<KpiBuildSubResult> BuildS9L1001QualityReturnRateAsync(
  570. string batchId, DateTime now, S7MdpRefreshOption option, string triggerType, CancellationToken ct)
  571. {
  572. var sub = new KpiBuildSubResult();
  573. var summary = await _db.Ado.SqlQuerySingleAsync<S7QualityReturnSummaryRow>(
  574. """
  575. SELECT
  576. SUM(CASE WHEN biz_doc_type='SALES_RETURN' THEN ABS(IFNULL(qty_change,0)) ELSE 0 END) AS return_qty,
  577. SUM(CASE WHEN biz_doc_type='SALES_SHIP' THEN ABS(IFNULL(qty_change,0)) ELSE 0 END) AS shipped_qty
  578. FROM mdp_std_inv_trans
  579. WHERE tenant_id=@TenantId
  580. AND summary_flag=0 AND void_flag=0 AND approved_flag=1 AND (@sourceSystem='' OR source_system=@sourceSystem) AND (source_system<>'T8' OR domain=@sourceDomain)
  581. AND biz_doc_type IN ('SALES_RETURN','SALES_SHIP')
  582. AND approved_time BETWEEN @PeriodStart AND @PeriodEnd
  583. """,
  584. new SugarParameter("@TenantId", option.TargetTenantId),
  585. new SugarParameter("@PeriodStart", option.MonthlyPeriodStart),
  586. new SugarParameter("@PeriodEnd", option.MonthlyPeriodEnd),
  587. new SugarParameter("@sourceDomain", option.SourceZtid),
  588. new SugarParameter("@sourceSystem", ""));
  589. sub.T8Rows = summary == null ? 0 : 1;
  590. decimal? ppm = summary?.shipped_qty > 0
  591. ? Math.Round((summary.return_qty ?? 0m) / summary.shipped_qty.Value * 1_000_000m, 4)
  592. : null;
  593. var denominatorStatus = summary?.shipped_qty > 0 ? "OK" : "NO_SHIPMENT";
  594. var dispatch = await _kpiCalcDispatcher.DispatchAsync(
  595. "S9_L1_001", "S9", option.TargetTenantId, option.TargetFactoryId,
  596. option.MonthlyPeriodEnd, option.MonthlyPeriodStart, option.MonthlyPeriodEnd, option.SourceZtid,
  597. batchId, triggerType, ppm, denominatorStatus, ct);
  598. sub.KpiRows = dispatch.ShouldUpsert
  599. ? await UpsertKpiValueAsync("S9_L1_001", option.MonthlyPeriodEnd, dispatch.MetricValue, now, option, "S9")
  600. : 0;
  601. sub.DenominatorStatus = dispatch.DenominatorStatus;
  602. return sub;
  603. }
  604. private async Task<int> UpsertKpiValueAsync(
  605. string metricCode,
  606. DateTime bizDate,
  607. decimal? metricValue,
  608. DateTime now,
  609. S7MdpRefreshOption option,
  610. string moduleCode = ModuleCode,
  611. string valueTable = "ado_s9_kpi_value_l1_day")
  612. {
  613. // 沿用 S3 UpsertS3KpiValueAsync 范式:先查现存行 → UPDATE;不存在 → SELECT MAX(id)+1 显式生成 id 后 INSERT。
  614. // ado_s9_kpi_value_l1_day.id 为手工分配主键(无 AUTO_INCREMENT),必须显式 set;
  615. // metric_value 允许 NULL(分母缺失不得伪装真实 0)。
  616. // FIX-2:截断时分秒(月度 KPI 入参可能为 YYYY-MM-DD 23:59:59),保证 SELECT WHERE biz_date=@BizDate 与 DB date 列匹配,避免重复 INSERT。
  617. // FIX-1:tenant_id/factory_id 取自 option,默认仍为 1300000000001/1,不破坏 Demo。
  618. bizDate = bizDate.Date;
  619. var snap = await _kpiTargetResolver.ResolveAsync(option.TargetTenantId, option.TargetFactoryId, metricCode, moduleCode, bizDate);
  620. var existingId = await _db.Ado.GetLongAsync(
  621. $"SELECT IFNULL((SELECT id FROM {valueTable} WHERE tenant_id=@TenantId AND factory_id=@FactoryId " +
  622. "AND module_code=@ModuleCode AND metric_code=@MetricCode AND biz_date=@BizDate AND is_deleted=0 " +
  623. "ORDER BY id LIMIT 1), 0)",
  624. new List<SugarParameter>
  625. {
  626. new("@TenantId", option.TargetTenantId),
  627. new("@FactoryId", option.TargetFactoryId),
  628. new("@ModuleCode", moduleCode),
  629. new("@MetricCode", metricCode),
  630. new("@BizDate", bizDate)
  631. });
  632. if (existingId > 0)
  633. {
  634. return await _db.Ado.ExecuteCommandAsync(
  635. $"UPDATE {valueTable} SET metric_value=@MetricValue, target_value=@TargetValue, " +
  636. "target_config_id=@TargetConfigId, target_source=@TargetSource, target_resolved_at=@TargetResolvedAt, " +
  637. "calc_time=@Now, update_time=@Now, is_deleted=0, is_active=1 WHERE id=@Id",
  638. new SugarParameter("@MetricValue", metricValue),
  639. new SugarParameter("@TargetValue", SmartOps.KpiTargetSnapshotSql.ValueOrDbNull(snap)),
  640. new SugarParameter("@TargetConfigId", SmartOps.KpiTargetSnapshotSql.ConfigIdOrDbNull(snap)),
  641. new SugarParameter("@TargetSource", SmartOps.KpiTargetSnapshotSql.SourceOrDbNull(snap)),
  642. new SugarParameter("@TargetResolvedAt", snap.ResolvedAt),
  643. new SugarParameter("@Now", now),
  644. new SugarParameter("@Id", existingId));
  645. }
  646. var nextId = Yitter.IdGenerator.YitIdHelper.NextId();
  647. return await _db.Ado.ExecuteCommandAsync($@"
  648. INSERT INTO {valueTable}
  649. (id, tenant_id, org_id, company_id, factory_id, status, biz_date,
  650. create_time, update_time, is_deleted, is_active,
  651. module_code, metric_code, metric_value, target_value, calc_time,
  652. target_config_id, target_source, target_resolved_at)
  653. VALUES
  654. (@Id, @TenantId, NULL, NULL, @FactoryId, NULL, @BizDate,
  655. @Now, @Now, 0, 1,
  656. @ModuleCode, @MetricCode, @MetricValue, @TargetValue, @Now,
  657. @TargetConfigId, @TargetSource, @TargetResolvedAt)",
  658. new SugarParameter("@Id", nextId),
  659. new SugarParameter("@TenantId", option.TargetTenantId),
  660. new SugarParameter("@FactoryId", option.TargetFactoryId),
  661. new SugarParameter("@BizDate", bizDate),
  662. new SugarParameter("@Now", now),
  663. new SugarParameter("@ModuleCode", moduleCode),
  664. new SugarParameter("@MetricCode", metricCode),
  665. new SugarParameter("@MetricValue", metricValue),
  666. new SugarParameter("@TargetValue", SmartOps.KpiTargetSnapshotSql.ValueOrDbNull(snap)),
  667. new SugarParameter("@TargetConfigId", SmartOps.KpiTargetSnapshotSql.ConfigIdOrDbNull(snap)),
  668. new SugarParameter("@TargetSource", SmartOps.KpiTargetSnapshotSql.SourceOrDbNull(snap)),
  669. new SugarParameter("@TargetResolvedAt", snap.ResolvedAt));
  670. }
  671. private async Task<long> InsertTransformRunLogAsync(string batchId, DateTime startedAt, string triggerType, S7MdpRefreshOption option)
  672. {
  673. await _db.Ado.ExecuteCommandAsync(@"
  674. INSERT INTO mdp_transform_run_log
  675. (tenant_id, job_code, job_name, trigger_type, batch_id, status, start_time, stage_rows, standard_rows, dwd_rows, create_time, update_time)
  676. VALUES
  677. (@TenantId, @JobCode, @JobName, @TriggerType, @BatchId, 'RUNNING', @StartTime, 0, 0, 0, @StartTime, @StartTime)",
  678. new SugarParameter("@TenantId", option.TargetTenantId),
  679. new SugarParameter("@JobCode", JobCode),
  680. new SugarParameter("@JobName", JobName),
  681. new SugarParameter("@TriggerType", triggerType),
  682. new SugarParameter("@BatchId", batchId),
  683. new SugarParameter("@StartTime", startedAt));
  684. return await _db.Ado.GetLongAsync(
  685. "SELECT id FROM mdp_transform_run_log WHERE batch_id=@BatchId ORDER BY id DESC LIMIT 1",
  686. new List<SugarParameter> { new("@BatchId", batchId) });
  687. }
  688. private async Task MarkTransformRunSuccessAsync(long runLogId, DateTime startedAt, S7MdpSyncTransformResult result)
  689. {
  690. var finishedAt = DateTime.Now;
  691. await _db.Ado.ExecuteCommandAsync(@"
  692. UPDATE mdp_transform_run_log
  693. SET status='SUCCESS', end_time=@EndTime, duration_ms=@DurationMs,
  694. stage_rows=@StageRows, standard_rows=@StandardRows, dwd_rows=@DwdRows,
  695. summary_json=@SummaryJson, update_time=CURRENT_TIMESTAMP
  696. WHERE id=@Id",
  697. new SugarParameter("@EndTime", finishedAt),
  698. new SugarParameter("@DurationMs", (int)(finishedAt - startedAt).TotalMilliseconds),
  699. new SugarParameter("@StageRows", result.StageRows),
  700. new SugarParameter("@StandardRows", result.StandardRows),
  701. new SugarParameter("@DwdRows", result.DwdRows),
  702. new SugarParameter("@SummaryJson", JsonSerializer.Serialize(new
  703. {
  704. batchId = result.BatchId,
  705. sourceZtid = result.SourceZtid,
  706. bizDate = result.BizDate.ToString("yyyy-MM-dd"),
  707. bizMonth = result.BizMonth,
  708. dwdRows = result.DwdRows,
  709. kpiRows = result.KpiRows,
  710. perKpiDwdRows = result.PerKpiDwdRows,
  711. perKpiKpiRows = result.PerKpiKpiRows,
  712. denominatorStatus = result.KpiDenominatorStatus
  713. })),
  714. new SugarParameter("@Id", runLogId));
  715. }
  716. private async Task MarkTransformRunFailedAsync(long runLogId, DateTime startedAt, string message, string batchId)
  717. {
  718. bool runLogUpdated = false;
  719. try
  720. {
  721. var finishedAt = DateTime.Now;
  722. await _db.Ado.ExecuteCommandAsync(@"
  723. UPDATE mdp_transform_run_log
  724. SET status='FAILED', end_time=@EndTime, duration_ms=@DurationMs,
  725. error_message=@ErrorMessage, update_time=CURRENT_TIMESTAMP
  726. WHERE id=@Id",
  727. new SugarParameter("@EndTime", finishedAt),
  728. new SugarParameter("@DurationMs", (int)(finishedAt - startedAt).TotalMilliseconds),
  729. new SugarParameter("@ErrorMessage", Truncate(message, 2000)),
  730. new SugarParameter("@Id", runLogId));
  731. runLogUpdated = true;
  732. }
  733. catch (Exception ex)
  734. {
  735. Console.Error.WriteLine($"[S7MdpSyncTransform] MarkTransformRunFailed write failed (runLogId={runLogId}): {ex.Message}");
  736. }
  737. // FAILURE-NOTIFICATION-1:写库 FAILED 成功后发通知给超级管理员;通知失败不影响主流程
  738. if (!runLogUpdated) return;
  739. try
  740. {
  741. await _sysNoticeService.AddNotice(new AddNoticeInput
  742. {
  743. Title = "S7 成品仓储 T8 KPI 跑批失败",
  744. Content = $"模块:S7 成品仓储\n批次ID:{batchId}\n失败时间:{DateTime.Now:yyyy-MM-dd HH:mm:ss}\n错误信息:{Truncate(message, 1000)}\n\n请查看 mdp_transform_run_log 获取完整错误与重试记录。",
  745. Type = NoticeTypeEnum.NOTICE,
  746. PublicTime = DateTime.Now,
  747. Status = NoticeStatusEnum.PUBLIC,
  748. PublicUserId = NoticeReceiverUserId,
  749. PublicUserName = NoticeReceiverUserName
  750. });
  751. }
  752. catch (Exception notifyEx)
  753. {
  754. _logger.LogError(notifyEx, "[S7MdpSyncTransform] SysNotice 发送失败 (runLogId={RunLogId}, batchId={BatchId})", runLogId, batchId);
  755. }
  756. }
  757. private static string NormalizeTriggerType(string s) =>
  758. string.IsNullOrWhiteSpace(s) ? "AUTO" : s.Trim().ToUpperInvariant();
  759. private static void NormalizeOption(S7MdpRefreshOption option)
  760. {
  761. var d = S7MdpRefreshOption.Default();
  762. if (option.TargetFactoryId <= 0) option.TargetFactoryId = d.TargetFactoryId;
  763. if (string.IsNullOrWhiteSpace(option.SourceZtid)) option.SourceZtid = d.SourceZtid;
  764. option.TargetTenantId = AidopSourceTenantMap.ResolveTenantId(option.SourceZtid, option.TargetTenantId);
  765. if (option.BizDate == default) option.BizDate = d.BizDate;
  766. if (string.IsNullOrWhiteSpace(option.BizMonth)) option.BizMonth = d.BizMonth;
  767. if (option.MonthlyPeriodStart == default) option.MonthlyPeriodStart = d.MonthlyPeriodStart;
  768. if (option.MonthlyPeriodEnd == default) option.MonthlyPeriodEnd = d.MonthlyPeriodEnd;
  769. }
  770. private static string Truncate(string s, int max) =>
  771. string.IsNullOrEmpty(s) ? "" : (s.Length <= max ? s : s.Substring(0, max));
  772. }
  773. // DTO ────────────────────────────────────────────────────────────────────────
  774. public sealed class S7MdpRefreshOption
  775. {
  776. public string SourceZtid { get; set; } = "pbxfxp";
  777. /// <summary>KPI/DWD 落库目标租户;≤0 时由 <see cref="AidopSourceTenantMap"/> 按 SourceZtid 解析。</summary>
  778. public long TargetTenantId { get; set; }
  779. /// <summary>KPI/DWD 落库目标工厂;默认 1。</summary>
  780. public long TargetFactoryId { get; set; } = 1L;
  781. public DateTime BizDate { get; set; }
  782. public string BizMonth { get; set; } = "";
  783. public DateTime MonthlyPeriodStart { get; set; }
  784. public DateTime MonthlyPeriodEnd { get; set; }
  785. public static S7MdpRefreshOption Default()
  786. {
  787. var today = DateTime.Today;
  788. var yesterday = today.AddDays(-1);
  789. var lastMonth = today.AddMonths(-1);
  790. var monthStart = new DateTime(lastMonth.Year, lastMonth.Month, 1);
  791. var monthEnd = monthStart.AddMonths(1).AddDays(-1);
  792. return new S7MdpRefreshOption
  793. {
  794. SourceZtid = "pbxfxp",
  795. TargetTenantId = 0,
  796. TargetFactoryId = 1L,
  797. BizDate = yesterday,
  798. BizMonth = lastMonth.ToString("yyyy-MM"),
  799. MonthlyPeriodStart = monthStart,
  800. MonthlyPeriodEnd = monthEnd.AddDays(1).AddSeconds(-1)
  801. };
  802. }
  803. }
  804. public sealed class S7MdpSyncTransformResult
  805. {
  806. public string BatchId { get; set; } = "";
  807. public long RunLogId { get; set; }
  808. public string TriggerType { get; set; } = "AUTO";
  809. public string SourceZtid { get; set; } = "";
  810. public long TargetTenantId { get; set; }
  811. public long TargetFactoryId { get; set; }
  812. public DateTime BizDate { get; set; }
  813. public string BizMonth { get; set; } = "";
  814. public DateTime MonthlyPeriodStart { get; set; }
  815. public DateTime MonthlyPeriodEnd { get; set; }
  816. public int StageRows { get; set; }
  817. public int StandardRows { get; set; }
  818. public int DwdRows { get; set; }
  819. public int KpiRows { get; set; }
  820. public Dictionary<string, int> PerKpiDwdRows { get; } = new();
  821. public Dictionary<string, int> PerKpiKpiRows { get; } = new();
  822. public List<string> KpiDenominatorStatus { get; } = new();
  823. public void MergeSub(string kpiCode, KpiBuildSubResult sub)
  824. {
  825. PerKpiDwdRows[kpiCode] = sub.DwdRows;
  826. PerKpiKpiRows[kpiCode] = sub.KpiRows;
  827. DwdRows += sub.DwdRows;
  828. KpiRows += sub.KpiRows;
  829. KpiDenominatorStatus.Add($"{kpiCode}:{sub.DenominatorStatus}");
  830. }
  831. }
  832. public sealed class KpiBuildSubResult
  833. {
  834. public int T8Rows { get; set; }
  835. public int DwdRows { get; set; }
  836. public int KpiRows { get; set; }
  837. public string DenominatorStatus { get; set; } = "OK";
  838. }
  839. // T8 result set 投影类型 ──────────────────────────────────────────────────────
  840. internal sealed class S7CycleRow
  841. {
  842. public string? order_no { get; set; }
  843. public int? cycle_days { get; set; }
  844. }
  845. internal sealed class S7FulfillmentRow
  846. {
  847. public string? order_no { get; set; }
  848. public int total_rows { get; set; }
  849. public int in_window_rows { get; set; }
  850. }
  851. internal sealed class S7ShipmentDetailRow
  852. {
  853. public string? task_no { get; set; }
  854. public string? item_code { get; set; }
  855. public string? approved_date { get; set; }
  856. public decimal? qty_change { get; set; }
  857. }
  858. internal sealed class S7PeNumRow
  859. {
  860. public int penum { get; set; }
  861. }
  862. internal sealed class S7QualityReturnSummaryRow
  863. {
  864. public decimal? return_qty { get; set; }
  865. public decimal? shipped_qty { get; set; }
  866. }
  867. internal sealed class S7StageKpiRow
  868. {
  869. public decimal? MetricValue { get; set; }
  870. public int RowCount { get; set; }
  871. }