ProductionReceiptMdpSyncService.cs 12 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223
  1. namespace Admin.NET.Plugin.AiDOP.MaterialWarehouse;
  2. /// <summary>
  3. /// S5 生产入库单 数据中台只读同步转换服务(DOP 内部,扁平单表,独立于 S5MdpSyncTransformService 的 KPI 管线)。
  4. ///
  5. /// 源:aidopdev NbrMaster(n) + NbrDetail(d),业务类型 Type='WOI'(生产入库);
  6. /// 扁平 LIST join:n.Domain=d.Domain AND n.Nbr=d.Nbr(一行=一条 NbrDetail 明细)。
  7. /// 维表 DepartmentMaster(p) / ItemMaster(i) / LocationMaster(lt,lf) 全 LEFT JOIN。
  8. /// LocationMaster 无 Domain 列(DOP 重建:domain_code/location/descr),按 tenant_id+location join。
  9. /// 链路:NbrMaster/NbrDetail(WOI) 只读 → mdp_std_production_receipt(typed 扁平)。
  10. /// 不经 stg、不建 dwd;不采用生产退料单 head-detail 双表逻辑;不碰 S2 写路径(S2 不写 WOI)。
  11. ///
  12. /// 约束:
  13. /// - 只读源表,仅写 mdp_std_production_receipt;绝不 INSERT/UPDATE/DELETE NbrMaster/NbrDetail。
  14. /// - 过滤 n.Type='WOI' AND n.IsActive=1。
  15. /// - status_desc 本批暂 = raw status(UPPER);remark=NbrMaster.Remark(列表显示名“入库类型”待确认)。
  16. /// - qty_rec=NbrDetail.QtyRec(列表用);qty_to/location_from 保留落库、本批不展示。
  17. /// - WOI 为 0 行时转换成功完成、处理数为 0,不报错。
  18. /// </summary>
  19. public class ProductionReceiptMdpSyncService : ITransient
  20. {
  21. private const string JobCode = "S5_PRODUCTION_RECEIPT_MDP_SYNC";
  22. private readonly ISqlSugarClient _db;
  23. public ProductionReceiptMdpSyncService(ISqlSugarClient db)
  24. {
  25. _db = db;
  26. }
  27. /// <summary>全量同步转换:源(WOI 扁平) → mdp_std_production_receipt。</summary>
  28. public async Task<ProductionReceiptMdpSyncResult> RunFullAsync(CancellationToken cancellationToken = default, string triggerType = "AUTO")
  29. {
  30. cancellationToken.ThrowIfCancellationRequested();
  31. await EnsureTablesAsync();
  32. var now = DateTime.Now;
  33. var batchId = $"S5_PROD_RCPT_FULL_{now:yyyyMMddHHmmss}";
  34. var runLogId = await InsertRunLogAsync(batchId, now, triggerType);
  35. var result = new ProductionReceiptMdpSyncResult { BatchId = batchId, RunLogId = runLogId };
  36. try
  37. {
  38. result.Rows = await TransformStandardAsync(batchId, now);
  39. await MarkRunSuccessAsync(runLogId, now, result);
  40. return result;
  41. }
  42. catch (Exception ex)
  43. {
  44. await MarkRunFailedAsync(runLogId, now, ex.Message);
  45. throw;
  46. }
  47. }
  48. /// <summary>防御式建表(与 UpdateScripts/1.0.214.sql 同构,幂等)。</summary>
  49. private async Task EnsureTablesAsync()
  50. {
  51. await _db.Ado.ExecuteCommandAsync(
  52. """
  53. CREATE TABLE IF NOT EXISTS mdp_std_production_receipt (
  54. id BIGINT AUTO_INCREMENT PRIMARY KEY,
  55. tenant_id BIGINT NOT NULL DEFAULT 0,
  56. factory_id BIGINT NULL DEFAULT 1,
  57. source_system VARCHAR(50) NOT NULL DEFAULT 'AIDOP',
  58. domain VARCHAR(80) NOT NULL,
  59. master_rec_id INT NULL,
  60. detail_rec_id INT NOT NULL,
  61. nbr VARCHAR(24) NULL,
  62. line SMALLINT NOT NULL DEFAULT 0,
  63. receipt_date DATETIME NULL,
  64. status VARCHAR(8) NULL,
  65. status_desc VARCHAR(20) NULL,
  66. remark VARCHAR(200) NULL,
  67. prod_line VARCHAR(8) NULL,
  68. work_ord VARCHAR(64) NULL,
  69. erp_work_ord VARCHAR(60) NULL,
  70. department VARCHAR(8) NULL,
  71. department_desc VARCHAR(255) NULL,
  72. applicant_name VARCHAR(12) NULL,
  73. item_num VARCHAR(24) NULL,
  74. item_name VARCHAR(200) NULL,
  75. item_spec VARCHAR(200) NULL,
  76. um VARCHAR(8) NULL,
  77. location_to VARCHAR(8) NULL,
  78. location_to_desc VARCHAR(255) NULL,
  79. lot_serial VARCHAR(120) NULL,
  80. qty_rec DECIMAL(18,5) NULL DEFAULT 0,
  81. qty_to DECIMAL(18,5) NULL DEFAULT 0,
  82. location_from VARCHAR(8) NULL,
  83. location_from_desc VARCHAR(255) NULL,
  84. ord_nbr VARCHAR(48) NULL,
  85. source_biz_key VARCHAR(200) NULL,
  86. sync_batch_id VARCHAR(100) NOT NULL,
  87. sync_time DATETIME NOT NULL,
  88. create_time DATETIME DEFAULT CURRENT_TIMESTAMP,
  89. update_time DATETIME DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
  90. UNIQUE KEY uk_mdp_std_prod_receipt (tenant_id, domain, detail_rec_id),
  91. KEY idx_mdp_std_prod_rcpt_nbr (tenant_id, nbr),
  92. KEY idx_mdp_std_prod_rcpt_date (tenant_id, receipt_date),
  93. KEY idx_mdp_std_prod_rcpt_workord (tenant_id, work_ord),
  94. KEY idx_mdp_std_prod_rcpt_erp (tenant_id, erp_work_ord),
  95. KEY idx_mdp_std_prod_rcpt_lot (tenant_id, lot_serial),
  96. KEY idx_mdp_std_prod_rcpt_item (tenant_id, item_num),
  97. KEY idx_mdp_std_prod_rcpt_locto (tenant_id, location_to)
  98. ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci COMMENT='S5生产入库单标准层(扁平,一行=一条明细)'
  99. """);
  100. }
  101. /// <summary>
  102. /// 标准化:NbrMaster(WOI) + NbrDetail + 维表 -> mdp_std_production_receipt(扁平,一行=一条明细)。返回处理行数。
  103. /// SQL 由旧系统扁平 LIST SQL 翻译(SQL Server -&gt; MySQL):with(nolock) 去除、IsNull-&gt;IFNULL、
  104. /// rtrim(a+' '+b)-&gt;TRIM(CONCAT(...))、row_number() 不落库(分页在读 API 侧)。
  105. /// LocationMaster 无 Domain 列,按 tenant_id+location join(生产退料单 1R 教训)。
  106. /// </summary>
  107. private async Task<int> TransformStandardAsync(string batchId, DateTime now)
  108. {
  109. var rows = await _db.Ado.GetIntAsync(
  110. """
  111. SELECT COUNT(1)
  112. FROM NbrMaster n
  113. LEFT JOIN NbrDetail d ON n.Domain = d.Domain AND n.Nbr = d.Nbr
  114. WHERE n.Type='WOI' AND n.IsActive=1 AND d.RecID IS NOT NULL
  115. """);
  116. await _db.Ado.ExecuteCommandAsync(
  117. """
  118. INSERT INTO mdp_std_production_receipt
  119. (tenant_id, factory_id, source_system, domain, master_rec_id, detail_rec_id, nbr, line,
  120. receipt_date, status, status_desc, remark, prod_line, work_ord, erp_work_ord,
  121. department, department_desc, applicant_name, item_num, item_name, item_spec, um,
  122. location_to, location_to_desc, lot_serial, qty_rec, qty_to, location_from, location_from_desc,
  123. ord_nbr, source_biz_key, sync_batch_id, sync_time)
  124. SELECT
  125. IFNULL(n.tenant_id, 0), 1, 'AIDOP', n.Domain, n.RecID, d.RecID, n.Nbr, d.Line,
  126. n.Date, UPPER(IFNULL(n.Status, '')), UPPER(IFNULL(n.Status, '')), n.Remark,
  127. d.ProdLine, d.WorkOrd, d.Address,
  128. n.Department, TRIM(CONCAT(n.Department, ' ', IFNULL(p.Descr, ''))), n.Name,
  129. d.ItemNum, i.Descr, i.Descr1, i.UM,
  130. d.LocationTo, lt.descr, CAST(d.LotSerial AS CHAR), d.QtyRec, d.QtyTo,
  131. d.LocationFrom, lf.descr, d.OrdNbr,
  132. CONCAT(n.Domain, ':', d.RecID), @BatchId, @Now
  133. FROM NbrMaster n
  134. LEFT JOIN NbrDetail d ON n.Domain = d.Domain AND n.Nbr = d.Nbr
  135. LEFT JOIN DepartmentMaster p ON p.Domain = n.Domain AND p.Department = n.Department
  136. LEFT JOIN ItemMaster i ON i.Domain = d.Domain AND i.ItemNum = d.ItemNum
  137. LEFT JOIN LocationMaster lt ON lt.tenant_id = IFNULL(d.tenant_id, 0) AND lt.location = d.LocationTo
  138. LEFT JOIN LocationMaster lf ON lf.tenant_id = IFNULL(d.tenant_id, 0) AND lf.location = d.LocationFrom
  139. WHERE n.Type='WOI' AND n.IsActive=1 AND d.RecID IS NOT NULL
  140. ON DUPLICATE KEY UPDATE
  141. factory_id=VALUES(factory_id), master_rec_id=VALUES(master_rec_id), nbr=VALUES(nbr), line=VALUES(line),
  142. receipt_date=VALUES(receipt_date), status=VALUES(status), status_desc=VALUES(status_desc), remark=VALUES(remark),
  143. prod_line=VALUES(prod_line), work_ord=VALUES(work_ord), erp_work_ord=VALUES(erp_work_ord),
  144. department=VALUES(department), department_desc=VALUES(department_desc), applicant_name=VALUES(applicant_name),
  145. item_num=VALUES(item_num), item_name=VALUES(item_name), item_spec=VALUES(item_spec), um=VALUES(um),
  146. location_to=VALUES(location_to), location_to_desc=VALUES(location_to_desc), lot_serial=VALUES(lot_serial),
  147. qty_rec=VALUES(qty_rec), qty_to=VALUES(qty_to), location_from=VALUES(location_from),
  148. location_from_desc=VALUES(location_from_desc), ord_nbr=VALUES(ord_nbr),
  149. sync_batch_id=VALUES(sync_batch_id), sync_time=VALUES(sync_time), update_time=CURRENT_TIMESTAMP
  150. """,
  151. new SugarParameter("@BatchId", batchId),
  152. new SugarParameter("@Now", now));
  153. return rows;
  154. }
  155. private async Task<long> InsertRunLogAsync(string batchId, DateTime startedAt, string triggerType)
  156. {
  157. await _db.Ado.ExecuteCommandAsync(
  158. """
  159. INSERT INTO mdp_transform_run_log
  160. (tenant_id, job_code, job_name, trigger_type, batch_id, status, start_time)
  161. VALUES (0, @JobCode, 'S5生产入库单MDP同步与标准化转换', @TriggerType, @BatchId, 'RUNNING', @StartTime)
  162. """,
  163. new SugarParameter("@JobCode", JobCode),
  164. new SugarParameter("@TriggerType", NormalizeTriggerType(triggerType)),
  165. new SugarParameter("@BatchId", batchId),
  166. new SugarParameter("@StartTime", startedAt));
  167. return await _db.Ado.GetLongAsync(
  168. "SELECT id FROM mdp_transform_run_log WHERE batch_id=@BatchId ORDER BY id DESC LIMIT 1",
  169. new List<SugarParameter> { new("@BatchId", batchId) });
  170. }
  171. private async Task MarkRunSuccessAsync(long runLogId, DateTime startedAt, ProductionReceiptMdpSyncResult result)
  172. {
  173. var finishedAt = DateTime.Now;
  174. await _db.Ado.ExecuteCommandAsync(
  175. """
  176. UPDATE mdp_transform_run_log
  177. SET status='SUCCESS', end_time=@EndTime, duration_ms=@DurationMs,
  178. stage_rows=0, standard_rows=@StandardRows, dwd_rows=0, update_time=CURRENT_TIMESTAMP
  179. WHERE id=@Id
  180. """,
  181. new SugarParameter("@EndTime", finishedAt),
  182. new SugarParameter("@DurationMs", (int)(finishedAt - startedAt).TotalMilliseconds),
  183. new SugarParameter("@StandardRows", result.Rows),
  184. new SugarParameter("@Id", runLogId));
  185. }
  186. private async Task MarkRunFailedAsync(long runLogId, DateTime startedAt, string message)
  187. {
  188. var finishedAt = DateTime.Now;
  189. await _db.Ado.ExecuteCommandAsync(
  190. """
  191. UPDATE mdp_transform_run_log
  192. SET status='FAILED', end_time=@EndTime, duration_ms=@DurationMs,
  193. error_message=@ErrorMessage, update_time=CURRENT_TIMESTAMP
  194. WHERE id=@Id
  195. """,
  196. new SugarParameter("@EndTime", finishedAt),
  197. new SugarParameter("@DurationMs", (int)(finishedAt - startedAt).TotalMilliseconds),
  198. new SugarParameter("@ErrorMessage", message.Length > 2000 ? message[..2000] : message),
  199. new SugarParameter("@Id", runLogId));
  200. }
  201. private static string NormalizeTriggerType(string? triggerType)
  202. => string.IsNullOrWhiteSpace(triggerType) ? "AUTO" : triggerType.Trim().ToUpperInvariant();
  203. }
  204. /// <summary>生产入库单 MDP 同步转换结果。</summary>
  205. public sealed class ProductionReceiptMdpSyncResult
  206. {
  207. public long RunLogId { get; set; }
  208. public string BatchId { get; set; } = string.Empty;
  209. public int Rows { get; set; }
  210. }