S7MdpSyncTransformService.cs 28 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602
  1. using Admin.NET.Core.Service;
  2. using Admin.NET.Plugin.AiDOP.Infrastructure;
  3. using Microsoft.Extensions.Logging;
  4. using System.Text.Json;
  5. namespace Admin.NET.Plugin.AiDOP.FinishedWarehouse;
  6. /// <summary>
  7. /// S7 成品仓储 — KPI 计算与刷新转换服务。
  8. /// 双模式:读本地标准层 mdp_std_t8_kc_*(由 T8BaseInboundMdpSyncService 从 T8 贴源→标准),不再直连 T8。
  9. /// 计算逻辑沿用方老师 v5.4 KPI J 列口径(等价改写为 MySQL 读 std)。
  10. /// 包含 KPI:S7_L1_001 订单发货周期 / S7_L1_002 订单发货满足率 / S7_L1_003 成品仓储人效。
  11. /// </summary>
  12. public class S7MdpSyncTransformService : ITransient
  13. {
  14. private readonly ISqlSugarClient _db;
  15. private readonly SysNoticeService _sysNoticeService;
  16. private readonly ILogger<S7MdpSyncTransformService> _logger;
  17. private readonly SmartOps.KpiCalcDispatcher _kpiCalcDispatcher;
  18. private const string JobCode = "S7_MDP_SYNC_TRANSFORM";
  19. private const string JobName = "S7 成品仓储 MDP 同步与转换";
  20. private const string ModuleCode = "S7";
  21. // FAILURE-NOTIFICATION-1:超级管理员 superAdmin.NET(AccountType=999)
  22. private const long NoticeReceiverUserId = 1300000000101L;
  23. private const string NoticeReceiverUserName = "超级管理员";
  24. public S7MdpSyncTransformService(
  25. ISqlSugarClient db,
  26. SysNoticeService sysNoticeService,
  27. ILogger<S7MdpSyncTransformService> logger,
  28. SmartOps.KpiCalcDispatcher kpiCalcDispatcher)
  29. {
  30. _db = db;
  31. _sysNoticeService = sysNoticeService;
  32. _logger = logger;
  33. _kpiCalcDispatcher = kpiCalcDispatcher;
  34. }
  35. public async Task<S7MdpSyncTransformResult> RunFullAsync(
  36. CancellationToken cancellationToken = default,
  37. string triggerType = "AUTO",
  38. S7MdpRefreshOption? option = null)
  39. {
  40. cancellationToken.ThrowIfCancellationRequested();
  41. option ??= S7MdpRefreshOption.Default();
  42. NormalizeOption(option);
  43. var now = DateTime.Now;
  44. var batchId = $"S7_MDP_FULL_{now:yyyyMMddHHmmss}";
  45. var normalizedTrigger = NormalizeTriggerType(triggerType);
  46. var runLogId = await InsertTransformRunLogAsync(batchId, now, normalizedTrigger, option);
  47. var result = new S7MdpSyncTransformResult
  48. {
  49. BatchId = batchId,
  50. RunLogId = runLogId,
  51. TriggerType = normalizedTrigger,
  52. SourceZtid = option.SourceZtid,
  53. TargetTenantId = option.TargetTenantId,
  54. TargetFactoryId = option.TargetFactoryId,
  55. BizDate = option.BizDate,
  56. BizMonth = option.BizMonth,
  57. MonthlyPeriodStart = option.MonthlyPeriodStart,
  58. MonthlyPeriodEnd = option.MonthlyPeriodEnd
  59. };
  60. try
  61. {
  62. result.StageRows = 0;
  63. result.StandardRows = 0;
  64. var sub25 = await BuildS7L1001OrderShipmentCycleAsync(batchId, now, option, normalizedTrigger, cancellationToken);
  65. result.MergeSub("S7_L1_001", sub25);
  66. var sub26 = await BuildS7L1002OrderShipmentFulfillmentAsync(batchId, now, option, normalizedTrigger, cancellationToken);
  67. result.MergeSub("S7_L1_002", sub26);
  68. var sub27 = await BuildS7L1003FinishedWarehouseEfficiencyAsync(batchId, now, option, normalizedTrigger, cancellationToken);
  69. result.MergeSub("S7_L1_003", sub27);
  70. await MarkTransformRunSuccessAsync(runLogId, now, result);
  71. return result;
  72. }
  73. catch (Exception ex)
  74. {
  75. await MarkTransformRunFailedAsync(runLogId, now, ex.Message, batchId);
  76. throw;
  77. }
  78. }
  79. // ─────────────────────────────────────────────────────────────────────────
  80. /// <summary>S7_L1_001 订单发货周期 = 最晚发货日期 - 最早 FQC 报检日期(5 表 JOIN)。</summary>
  81. private async Task<KpiBuildSubResult> BuildS7L1001OrderShipmentCycleAsync(
  82. string batchId, DateTime now, S7MdpRefreshOption option, string triggerType, CancellationToken ct)
  83. {
  84. var sub = new KpiBuildSubResult();
  85. // 双模式:读本地标准层 mdp_std_t8_*(源 identity Id→src_id);datediff(day,a,b)→DATEDIFF(b,a),IsNull→IFNULL。
  86. const string sql = @"
  87. select noid as noid,
  88. datediff(max(shtime), min(shdate)) as scts
  89. from (
  90. select a.noid as noid, b.code as code,
  91. ifnull(c.shdate, b.addtime) as shdate,
  92. (case when b.gdyn=1 then b.gdtime else d.shtime end) as shtime,
  93. (case when b.gdyn=1 or b.sl<=b.slzx then 1 else 0 end) as wczt
  94. from mdp_std_t8_kc_dd_head a
  95. left join mdp_std_t8_kc_dd_list b on a.src_id=b.idid
  96. left join (
  97. select min(b.src_id) as id, b.lynoid as lynoid, b.code as code,
  98. max(a.shtime) as shtime, sum(b.slzx) as slzx
  99. from mdp_std_t8_kc_tz_head a
  100. inner join mdp_std_t8_kc_tz_list b on a.src_id=b.idid
  101. where a.ztid=@ztid and a.lbs='销售出库' and a.hzyn=0 and a.zfyn=0 and a.shyn=1
  102. group by b.lynoid, b.code
  103. ) d on b.rwnoid=d.lynoid and b.code=d.code
  104. left join mdp_std_t8_kc_zj_list c on a.ztid=c.ztid and c.lyid=b.src_id and c.zjyn=1
  105. where a.ztid=@ztid and a.lbs='销售订单' and a.zf=0 and a.shyn=1
  106. ) n
  107. group by noid
  108. having min(wczt)=1";
  109. var rows = await _db.Ado.SqlQueryAsync<S7CycleRow>(sql, new[] { new SugarParameter("@ztid", option.SourceZtid) });
  110. sub.T8Rows = rows.Count;
  111. var dwdAffected = 0;
  112. var cycleList = new List<int>();
  113. foreach (var r in rows)
  114. {
  115. ct.ThrowIfCancellationRequested();
  116. if (string.IsNullOrEmpty(r.noid)) continue;
  117. if (r.scts.HasValue) cycleList.Add(r.scts.Value);
  118. dwdAffected += await _db.Ado.ExecuteCommandAsync(@"
  119. INSERT INTO dwd_t8_order_shipment_cycle
  120. (tenant_id, factory_id, biz_date, source_ztid, order_no, cycle_days, batch_id, create_time)
  121. VALUES
  122. (@tenantId, @factoryId, @bizDate, @ztid, @orderNo, @cycleDays, @batchId, @now)
  123. ON DUPLICATE KEY UPDATE
  124. cycle_days=VALUES(cycle_days),
  125. batch_id=VALUES(batch_id), update_time=@now",
  126. new SugarParameter("@tenantId", option.TargetTenantId),
  127. new SugarParameter("@factoryId", option.TargetFactoryId),
  128. new SugarParameter("@bizDate", option.BizDate),
  129. new SugarParameter("@ztid", option.SourceZtid),
  130. new SugarParameter("@orderNo", r.noid),
  131. new SugarParameter("@cycleDays", r.scts),
  132. new SugarParameter("@batchId", batchId),
  133. new SugarParameter("@now", now));
  134. }
  135. sub.DwdRows = dwdAffected;
  136. // 数据准备(dwd 逐单 cycle_days)已完成。最终聚合交分发器(日 KPI:period 复用 BizDate,SQL 按 biz_date 圈选)。
  137. decimal? legacyValue = cycleList.Count > 0
  138. ? Math.Round((decimal)cycleList.Average(), 4)
  139. : null;
  140. var legacyDenom = cycleList.Count > 0 ? "OK" : "NO_COMPLETED_ORDER";
  141. var dispatch = await _kpiCalcDispatcher.DispatchAsync(
  142. "S7_L1_001", ModuleCode, option.TargetTenantId, option.TargetFactoryId,
  143. option.BizDate, option.BizDate, option.BizDate, option.SourceZtid,
  144. batchId, triggerType, legacyValue, legacyDenom, ct);
  145. sub.KpiRows = dispatch.ShouldUpsert
  146. ? await UpsertKpiValueAsync("S7_L1_001", option.BizDate, dispatch.MetricValue, now, option)
  147. : 0;
  148. sub.DenominatorStatus = dispatch.DenominatorStatus;
  149. return sub;
  150. }
  151. /// <summary>S7_L1_002 订单发货满足率 = (交期前发货行数 / 该订单总行数) × 100%。</summary>
  152. private async Task<KpiBuildSubResult> BuildS7L1002OrderShipmentFulfillmentAsync(
  153. string batchId, DateTime now, S7MdpRefreshOption option, string triggerType, CancellationToken ct)
  154. {
  155. var sub = new KpiBuildSubResult();
  156. // 双模式:读标准层;convert(varchar(10),shtime,23)→date(shtime)。
  157. const string sql = @"
  158. select noid as noid,
  159. count(noid) as total_rows,
  160. sum(wczt) as in_window_rows
  161. from (
  162. select a.noid as noid, b.rwnoid as rwnoid, b.code as code,
  163. (case when sum(d.slzx)>=b.sl then 1 else 0 end) as wczt
  164. from mdp_std_t8_kc_dd_head a
  165. left join mdp_std_t8_kc_dd_list b on a.src_id=b.idid
  166. left join (
  167. select b.lynoid as lynoid, b.code as code,
  168. date(a.shtime) as shtime, b.slzx as slzx
  169. from mdp_std_t8_kc_tz_head a
  170. inner join mdp_std_t8_kc_tz_list b on a.src_id=b.idid
  171. where a.ztid=@ztid and a.lbs='销售出库' and a.hzyn=0 and a.zfyn=0 and a.shyn=1
  172. ) d on b.rwnoid=d.lynoid and b.code=d.code and d.shtime<=b.jhdate
  173. where a.ztid=@ztid and a.lbs='销售订单' and a.zf=0 and a.shyn=1
  174. group by a.noid, b.rwnoid, b.code, b.sl
  175. ) n
  176. group by noid";
  177. var rows = await _db.Ado.SqlQueryAsync<S7FulfillmentRow>(sql, new[] { new SugarParameter("@ztid", option.SourceZtid) });
  178. sub.T8Rows = rows.Count;
  179. var dwdAffected = 0;
  180. var rateList = new List<decimal>();
  181. foreach (var r in rows)
  182. {
  183. ct.ThrowIfCancellationRequested();
  184. if (string.IsNullOrEmpty(r.noid)) continue;
  185. decimal? rate = (r.total_rows > 0)
  186. ? Math.Round((decimal)r.in_window_rows / r.total_rows * 100m, 4)
  187. : null;
  188. if (rate.HasValue) rateList.Add(rate.Value);
  189. dwdAffected += await _db.Ado.ExecuteCommandAsync(@"
  190. INSERT INTO dwd_t8_order_shipment_fulfillment
  191. (tenant_id, factory_id, biz_date, source_ztid, order_no,
  192. total_rows, in_window_rows, fulfillment_rate, batch_id, create_time)
  193. VALUES
  194. (@tenantId, @factoryId, @bizDate, @ztid, @orderNo,
  195. @total, @inWindow, @rate, @batchId, @now)
  196. ON DUPLICATE KEY UPDATE
  197. total_rows=VALUES(total_rows), in_window_rows=VALUES(in_window_rows),
  198. fulfillment_rate=VALUES(fulfillment_rate),
  199. batch_id=VALUES(batch_id), update_time=@now",
  200. new SugarParameter("@tenantId", option.TargetTenantId),
  201. new SugarParameter("@factoryId", option.TargetFactoryId),
  202. new SugarParameter("@bizDate", option.BizDate),
  203. new SugarParameter("@ztid", option.SourceZtid),
  204. new SugarParameter("@orderNo", r.noid),
  205. new SugarParameter("@total", r.total_rows),
  206. new SugarParameter("@inWindow", r.in_window_rows),
  207. new SugarParameter("@rate", rate),
  208. new SugarParameter("@batchId", batchId),
  209. new SugarParameter("@now", now));
  210. }
  211. sub.DwdRows = dwdAffected;
  212. // 数据准备(dwd 逐单 fulfillment_rate,已 ×100)已完成。最终聚合交分发器(CONFIG_SQL 直接 AVG 该列不再 ×100)。
  213. decimal? legacyValue = rateList.Count > 0
  214. ? Math.Round(rateList.Average(), 4)
  215. : null;
  216. var legacyDenom = rateList.Count > 0 ? "OK" : "NO_VALID_ORDER";
  217. var dispatch = await _kpiCalcDispatcher.DispatchAsync(
  218. "S7_L1_002", ModuleCode, option.TargetTenantId, option.TargetFactoryId,
  219. option.BizDate, option.BizDate, option.BizDate, option.SourceZtid,
  220. batchId, triggerType, legacyValue, legacyDenom, ct);
  221. sub.KpiRows = dispatch.ShouldUpsert
  222. ? await UpsertKpiValueAsync("S7_L1_002", option.BizDate, dispatch.MetricValue, now, option)
  223. : 0;
  224. sub.DenominatorStatus = dispatch.DenominatorStatus;
  225. return sub;
  226. }
  227. /// <summary>S7_L1_003 成品仓储人效 = SUM(slzx where lbs=销售出库) / count(gw=仓管)。</summary>
  228. private async Task<KpiBuildSubResult> BuildS7L1003FinishedWarehouseEfficiencyAsync(
  229. string batchId, DateTime now, S7MdpRefreshOption option, string triggerType, CancellationToken ct)
  230. {
  231. var sub = new KpiBuildSubResult();
  232. const string sqlNumer = @"
  233. select b.lynoid as lynoid, b.code as code,
  234. date(a.shtime) as shtime, b.slzx as slzx
  235. from mdp_std_t8_kc_tz_head a
  236. inner join mdp_std_t8_kc_tz_list b on a.src_id=b.idid
  237. where a.ztid=@ztid and a.lbs='销售出库' and a.hzyn=0 and a.zfyn=0 and a.shyn=1
  238. and date(a.shtime) between @startDateText and @endDateText";
  239. const string sqlDenom = @"
  240. select count(*) as penum
  241. from mdp_std_t8_sys_pelist
  242. where ztid=@ztid and zzzt='在职' and gw='仓管'";
  243. var pNumer = new[]
  244. {
  245. new SugarParameter("@ztid", option.SourceZtid),
  246. new SugarParameter("@startDateText", option.MonthlyPeriodStart.ToString("yyyy-MM-dd")),
  247. new SugarParameter("@endDateText", option.MonthlyPeriodEnd.ToString("yyyy-MM-dd"))
  248. };
  249. var pDenom = new[] { new SugarParameter("@ztid", option.SourceZtid) };
  250. var numerRows = await _db.Ado.SqlQueryAsync<S7ShipmentDetailRow>(sqlNumer, pNumer);
  251. var denomRows = await _db.Ado.SqlQueryAsync<S7PeNumRow>(sqlDenom, pDenom);
  252. sub.T8Rows = numerRows.Count + denomRows.Count;
  253. decimal? shipmentQty = numerRows.Sum(r => r.slzx ?? 0m);
  254. if (numerRows.Count == 0) shipmentQty = null;
  255. int? headcount = denomRows.FirstOrDefault()?.penum;
  256. decimal? efficiency = null;
  257. string denomStatus;
  258. if (!headcount.HasValue || headcount.Value <= 0)
  259. denomStatus = "NO_HEADCOUNT";
  260. else if (!shipmentQty.HasValue)
  261. denomStatus = "NO_NUMERATOR";
  262. else
  263. {
  264. efficiency = Math.Round(shipmentQty.Value / headcount.Value, 4);
  265. denomStatus = "OK";
  266. }
  267. sub.DenominatorStatus = denomStatus;
  268. var dwdAffected = await _db.Ado.ExecuteCommandAsync(@"
  269. INSERT INTO dwd_t8_finished_warehouse_efficiency
  270. (tenant_id, factory_id, biz_month, source_ztid, period_start, period_end,
  271. shipment_qty, warehouse_headcount, efficiency, denominator_status, batch_id, create_time)
  272. VALUES
  273. (@tenantId, @factoryId, @bizMonth, @ztid, @periodStart, @periodEnd,
  274. @shipmentQty, @headcount, @efficiency, @denomStatus, @batchId, @now)
  275. ON DUPLICATE KEY UPDATE
  276. period_start=VALUES(period_start), period_end=VALUES(period_end),
  277. shipment_qty=VALUES(shipment_qty), warehouse_headcount=VALUES(warehouse_headcount),
  278. efficiency=VALUES(efficiency), denominator_status=VALUES(denominator_status),
  279. batch_id=VALUES(batch_id), update_time=@now",
  280. new SugarParameter("@tenantId", option.TargetTenantId),
  281. new SugarParameter("@factoryId", option.TargetFactoryId),
  282. new SugarParameter("@bizMonth", option.BizMonth),
  283. new SugarParameter("@ztid", option.SourceZtid),
  284. new SugarParameter("@periodStart", option.MonthlyPeriodStart),
  285. new SugarParameter("@periodEnd", option.MonthlyPeriodEnd),
  286. new SugarParameter("@shipmentQty", shipmentQty),
  287. new SugarParameter("@headcount", headcount),
  288. new SugarParameter("@efficiency", efficiency),
  289. new SugarParameter("@denomStatus", denomStatus),
  290. new SugarParameter("@batchId", batchId),
  291. new SugarParameter("@now", now));
  292. sub.DwdRows = dwdAffected;
  293. // 月度 KPI 最终聚合交分发器;bizDate=月末、period=当月窗口;
  294. // legacyValue=efficiency、legacyDenom 保留 NO_HEADCOUNT/NO_NUMERATOR(CONFIG_SQL 下塌缩为 NO_DATA)。
  295. var dispatch = await _kpiCalcDispatcher.DispatchAsync(
  296. "S7_L1_003", ModuleCode, option.TargetTenantId, option.TargetFactoryId,
  297. option.MonthlyPeriodEnd, option.MonthlyPeriodStart, option.MonthlyPeriodEnd, option.SourceZtid,
  298. batchId, triggerType, efficiency, denomStatus, ct);
  299. sub.KpiRows = dispatch.ShouldUpsert
  300. ? await UpsertKpiValueAsync("S7_L1_003", option.MonthlyPeriodEnd, dispatch.MetricValue, now, option)
  301. : 0;
  302. sub.DenominatorStatus = dispatch.DenominatorStatus;
  303. return sub;
  304. }
  305. // ─────────────────────────────────────────────────────────────────────────
  306. private async Task<int> UpsertKpiValueAsync(string metricCode, DateTime bizDate, decimal? metricValue, DateTime now, S7MdpRefreshOption option)
  307. {
  308. // 沿用 S3 UpsertS3KpiValueAsync 范式:先查现存行 → UPDATE;不存在 → SELECT MAX(id)+1 显式生成 id 后 INSERT。
  309. // ado_s9_kpi_value_l1_day.id 为手工分配主键(无 AUTO_INCREMENT),必须显式 set;
  310. // metric_value 允许 NULL(分母缺失不得伪装真实 0)。
  311. // FIX-2:截断时分秒(月度 KPI 入参可能为 YYYY-MM-DD 23:59:59),保证 SELECT WHERE biz_date=@BizDate 与 DB date 列匹配,避免重复 INSERT。
  312. // FIX-1:tenant_id/factory_id 取自 option,默认仍为 1300000000001/1,不破坏 Demo。
  313. bizDate = bizDate.Date;
  314. var existingId = await _db.Ado.GetLongAsync(
  315. "SELECT IFNULL((SELECT id FROM ado_s9_kpi_value_l1_day WHERE tenant_id=@TenantId AND factory_id=@FactoryId " +
  316. "AND module_code=@ModuleCode AND metric_code=@MetricCode AND biz_date=@BizDate AND is_deleted=0 " +
  317. "ORDER BY id LIMIT 1), 0)",
  318. new List<SugarParameter>
  319. {
  320. new("@TenantId", option.TargetTenantId),
  321. new("@FactoryId", option.TargetFactoryId),
  322. new("@ModuleCode", ModuleCode),
  323. new("@MetricCode", metricCode),
  324. new("@BizDate", bizDate)
  325. });
  326. if (existingId > 0)
  327. {
  328. return await _db.Ado.ExecuteCommandAsync(
  329. "UPDATE ado_s9_kpi_value_l1_day SET metric_value=@MetricValue, calc_time=@Now, " +
  330. "update_time=@Now, is_deleted=0, is_active=1 WHERE id=@Id",
  331. new SugarParameter("@MetricValue", metricValue),
  332. new SugarParameter("@Now", now),
  333. new SugarParameter("@Id", existingId));
  334. }
  335. var nextId = await _db.Ado.GetLongAsync(
  336. "SELECT COALESCE(MAX(id), 0) + 1 FROM ado_s9_kpi_value_l1_day");
  337. return await _db.Ado.ExecuteCommandAsync(@"
  338. INSERT INTO ado_s9_kpi_value_l1_day
  339. (id, tenant_id, org_id, company_id, factory_id, status, biz_date,
  340. create_time, update_time, is_deleted, is_active,
  341. module_code, metric_code, metric_value, calc_time)
  342. VALUES
  343. (@Id, @TenantId, NULL, NULL, @FactoryId, NULL, @BizDate,
  344. @Now, @Now, 0, 1,
  345. @ModuleCode, @MetricCode, @MetricValue, @Now)",
  346. new SugarParameter("@Id", nextId),
  347. new SugarParameter("@TenantId", option.TargetTenantId),
  348. new SugarParameter("@FactoryId", option.TargetFactoryId),
  349. new SugarParameter("@BizDate", bizDate),
  350. new SugarParameter("@Now", now),
  351. new SugarParameter("@ModuleCode", ModuleCode),
  352. new SugarParameter("@MetricCode", metricCode),
  353. new SugarParameter("@MetricValue", metricValue));
  354. }
  355. private async Task<long> InsertTransformRunLogAsync(string batchId, DateTime startedAt, string triggerType, S7MdpRefreshOption option)
  356. {
  357. await _db.Ado.ExecuteCommandAsync(@"
  358. INSERT INTO mdp_transform_run_log
  359. (tenant_id, job_code, job_name, trigger_type, batch_id, status, start_time, stage_rows, standard_rows, dwd_rows, create_time, update_time)
  360. VALUES
  361. (@TenantId, @JobCode, @JobName, @TriggerType, @BatchId, 'RUNNING', @StartTime, 0, 0, 0, @StartTime, @StartTime)",
  362. new SugarParameter("@TenantId", option.TargetTenantId),
  363. new SugarParameter("@JobCode", JobCode),
  364. new SugarParameter("@JobName", JobName),
  365. new SugarParameter("@TriggerType", triggerType),
  366. new SugarParameter("@BatchId", batchId),
  367. new SugarParameter("@StartTime", startedAt));
  368. return await _db.Ado.GetLongAsync(
  369. "SELECT id FROM mdp_transform_run_log WHERE batch_id=@BatchId ORDER BY id DESC LIMIT 1",
  370. new List<SugarParameter> { new("@BatchId", batchId) });
  371. }
  372. private async Task MarkTransformRunSuccessAsync(long runLogId, DateTime startedAt, S7MdpSyncTransformResult result)
  373. {
  374. var finishedAt = DateTime.Now;
  375. await _db.Ado.ExecuteCommandAsync(@"
  376. UPDATE mdp_transform_run_log
  377. SET status='SUCCESS', end_time=@EndTime, duration_ms=@DurationMs,
  378. stage_rows=@StageRows, standard_rows=@StandardRows, dwd_rows=@DwdRows,
  379. summary_json=@SummaryJson, update_time=CURRENT_TIMESTAMP
  380. WHERE id=@Id",
  381. new SugarParameter("@EndTime", finishedAt),
  382. new SugarParameter("@DurationMs", (int)(finishedAt - startedAt).TotalMilliseconds),
  383. new SugarParameter("@StageRows", result.StageRows),
  384. new SugarParameter("@StandardRows", result.StandardRows),
  385. new SugarParameter("@DwdRows", result.DwdRows),
  386. new SugarParameter("@SummaryJson", JsonSerializer.Serialize(new
  387. {
  388. batchId = result.BatchId,
  389. sourceZtid = result.SourceZtid,
  390. bizDate = result.BizDate.ToString("yyyy-MM-dd"),
  391. bizMonth = result.BizMonth,
  392. dwdRows = result.DwdRows,
  393. kpiRows = result.KpiRows,
  394. perKpiDwdRows = result.PerKpiDwdRows,
  395. perKpiKpiRows = result.PerKpiKpiRows,
  396. denominatorStatus = result.KpiDenominatorStatus
  397. })),
  398. new SugarParameter("@Id", runLogId));
  399. }
  400. private async Task MarkTransformRunFailedAsync(long runLogId, DateTime startedAt, string message, string batchId)
  401. {
  402. bool runLogUpdated = false;
  403. try
  404. {
  405. var finishedAt = DateTime.Now;
  406. await _db.Ado.ExecuteCommandAsync(@"
  407. UPDATE mdp_transform_run_log
  408. SET status='FAILED', end_time=@EndTime, duration_ms=@DurationMs,
  409. error_message=@ErrorMessage, update_time=CURRENT_TIMESTAMP
  410. WHERE id=@Id",
  411. new SugarParameter("@EndTime", finishedAt),
  412. new SugarParameter("@DurationMs", (int)(finishedAt - startedAt).TotalMilliseconds),
  413. new SugarParameter("@ErrorMessage", Truncate(message, 2000)),
  414. new SugarParameter("@Id", runLogId));
  415. runLogUpdated = true;
  416. }
  417. catch (Exception ex)
  418. {
  419. Console.Error.WriteLine($"[S7MdpSyncTransform] MarkTransformRunFailed write failed (runLogId={runLogId}): {ex.Message}");
  420. }
  421. // FAILURE-NOTIFICATION-1:写库 FAILED 成功后发通知给超级管理员;通知失败不影响主流程
  422. if (!runLogUpdated) return;
  423. try
  424. {
  425. await _sysNoticeService.AddNotice(new AddNoticeInput
  426. {
  427. Title = "S7 成品仓储 T8 KPI 跑批失败",
  428. Content = $"模块:S7 成品仓储\n批次ID:{batchId}\n失败时间:{DateTime.Now:yyyy-MM-dd HH:mm:ss}\n错误信息:{Truncate(message, 1000)}\n\n请查看 mdp_transform_run_log 获取完整错误与重试记录。",
  429. Type = NoticeTypeEnum.NOTICE,
  430. PublicTime = DateTime.Now,
  431. Status = NoticeStatusEnum.PUBLIC,
  432. PublicUserId = NoticeReceiverUserId,
  433. PublicUserName = NoticeReceiverUserName
  434. });
  435. }
  436. catch (Exception notifyEx)
  437. {
  438. _logger.LogError(notifyEx, "[S7MdpSyncTransform] SysNotice 发送失败 (runLogId={RunLogId}, batchId={BatchId})", runLogId, batchId);
  439. }
  440. }
  441. private static string NormalizeTriggerType(string s) =>
  442. string.IsNullOrWhiteSpace(s) ? "AUTO" : s.Trim().ToUpperInvariant();
  443. private static void NormalizeOption(S7MdpRefreshOption option)
  444. {
  445. var d = S7MdpRefreshOption.Default();
  446. if (option.TargetFactoryId <= 0) option.TargetFactoryId = d.TargetFactoryId;
  447. if (string.IsNullOrWhiteSpace(option.SourceZtid)) option.SourceZtid = d.SourceZtid;
  448. option.TargetTenantId = AidopSourceTenantMap.ResolveTenantId(option.SourceZtid, option.TargetTenantId);
  449. if (option.BizDate == default) option.BizDate = d.BizDate;
  450. if (string.IsNullOrWhiteSpace(option.BizMonth)) option.BizMonth = d.BizMonth;
  451. if (option.MonthlyPeriodStart == default) option.MonthlyPeriodStart = d.MonthlyPeriodStart;
  452. if (option.MonthlyPeriodEnd == default) option.MonthlyPeriodEnd = d.MonthlyPeriodEnd;
  453. }
  454. private static string Truncate(string s, int max) =>
  455. string.IsNullOrEmpty(s) ? "" : (s.Length <= max ? s : s.Substring(0, max));
  456. }
  457. // DTO ────────────────────────────────────────────────────────────────────────
  458. public sealed class S7MdpRefreshOption
  459. {
  460. public string SourceZtid { get; set; } = "pbxfxp";
  461. /// <summary>KPI/DWD 落库目标租户;≤0 时由 <see cref="AidopSourceTenantMap"/> 按 SourceZtid 解析。</summary>
  462. public long TargetTenantId { get; set; }
  463. /// <summary>KPI/DWD 落库目标工厂;默认 1。</summary>
  464. public long TargetFactoryId { get; set; } = 1L;
  465. public DateTime BizDate { get; set; }
  466. public string BizMonth { get; set; } = "";
  467. public DateTime MonthlyPeriodStart { get; set; }
  468. public DateTime MonthlyPeriodEnd { get; set; }
  469. public static S7MdpRefreshOption Default()
  470. {
  471. var today = DateTime.Today;
  472. var yesterday = today.AddDays(-1);
  473. var lastMonth = today.AddMonths(-1);
  474. var monthStart = new DateTime(lastMonth.Year, lastMonth.Month, 1);
  475. var monthEnd = monthStart.AddMonths(1).AddDays(-1);
  476. return new S7MdpRefreshOption
  477. {
  478. SourceZtid = "pbxfxp",
  479. TargetTenantId = 0,
  480. TargetFactoryId = 1L,
  481. BizDate = yesterday,
  482. BizMonth = lastMonth.ToString("yyyy-MM"),
  483. MonthlyPeriodStart = monthStart,
  484. MonthlyPeriodEnd = monthEnd.AddDays(1).AddSeconds(-1)
  485. };
  486. }
  487. }
  488. public sealed class S7MdpSyncTransformResult
  489. {
  490. public string BatchId { get; set; } = "";
  491. public long RunLogId { get; set; }
  492. public string TriggerType { get; set; } = "AUTO";
  493. public string SourceZtid { get; set; } = "";
  494. public long TargetTenantId { get; set; }
  495. public long TargetFactoryId { get; set; }
  496. public DateTime BizDate { get; set; }
  497. public string BizMonth { get; set; } = "";
  498. public DateTime MonthlyPeriodStart { get; set; }
  499. public DateTime MonthlyPeriodEnd { get; set; }
  500. public int StageRows { get; set; }
  501. public int StandardRows { get; set; }
  502. public int DwdRows { get; set; }
  503. public int KpiRows { get; set; }
  504. public Dictionary<string, int> PerKpiDwdRows { get; } = new();
  505. public Dictionary<string, int> PerKpiKpiRows { get; } = new();
  506. public List<string> KpiDenominatorStatus { get; } = new();
  507. public void MergeSub(string kpiCode, KpiBuildSubResult sub)
  508. {
  509. PerKpiDwdRows[kpiCode] = sub.DwdRows;
  510. PerKpiKpiRows[kpiCode] = sub.KpiRows;
  511. DwdRows += sub.DwdRows;
  512. KpiRows += sub.KpiRows;
  513. KpiDenominatorStatus.Add($"{kpiCode}:{sub.DenominatorStatus}");
  514. }
  515. }
  516. public sealed class KpiBuildSubResult
  517. {
  518. public int T8Rows { get; set; }
  519. public int DwdRows { get; set; }
  520. public int KpiRows { get; set; }
  521. public string DenominatorStatus { get; set; } = "OK";
  522. }
  523. // T8 result set 投影类型 ──────────────────────────────────────────────────────
  524. internal sealed class S7CycleRow
  525. {
  526. public string? noid { get; set; }
  527. public int? scts { get; set; }
  528. }
  529. internal sealed class S7FulfillmentRow
  530. {
  531. public string? noid { get; set; }
  532. public int total_rows { get; set; }
  533. public int in_window_rows { get; set; }
  534. }
  535. internal sealed class S7ShipmentDetailRow
  536. {
  537. public string? lynoid { get; set; }
  538. public string? code { get; set; }
  539. public string? shtime { get; set; }
  540. public decimal? slzx { get; set; }
  541. }
  542. internal sealed class S7PeNumRow
  543. {
  544. public int penum { get; set; }
  545. }