ProductionIssueMdpSyncService.cs 11 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212
  1. namespace Admin.NET.Plugin.AiDOP.MaterialWarehouse;
  2. /// <summary>
  3. /// S5 生产领料单 数据中台只读同步转换服务(DOP 内部,头级单表,独立于 S5MdpSyncTransformService 的 KPI 管线)。
  4. ///
  5. /// 源:aidopdev NbrMaster(m),业务类型 Type='SM';维表 DepartmentMaster(d) LEFT JOIN(Domain+Department)。
  6. /// 链路:NbrMaster(SM) 只读 → mdp_std_production_issue(typed)。
  7. /// 不经 stg、不建明细、不建 dwd(只读列表);SM=领料单业务类型有 WorkOrderPickBillService 代码实证(工单下达 Type=SM)。
  8. ///
  9. /// 约束:
  10. /// - 只读源表,仅写 mdp_std_production_issue;绝不 INSERT/UPDATE/DELETE NbrMaster;不碰 WorkOrderPickBillService 写路径。
  11. /// - 头过滤 m.Type='SM' AND m.IsActive=1(领料无 IsReturn 语义)。
  12. /// - status_desc 用旧系统领料 SQL 口径:PretreatmentState 非空取其值,Status='C'→'已下架','A'→'备料中',否则空。
  13. /// - 领料人 issue_user 取 m.User1(列表 SQL 口径),原样落库。
  14. /// - trans_type 展示:TransType='Z61'→'补料',否则'正常';trans_type_text:TransType='PrevProcess'→'需要前处理',否则空(旧系统口径)。
  15. /// - 不做 danjia/jiage/库存事务/状态流转;SM 为 0 行时转换成功完成、处理数为 0,不报错。
  16. /// </summary>
  17. public class ProductionIssueMdpSyncService : ITransient
  18. {
  19. private const string JobCode = "S5_PRODUCTION_ISSUE_MDP_SYNC";
  20. private readonly ISqlSugarClient _db;
  21. public ProductionIssueMdpSyncService(ISqlSugarClient db)
  22. {
  23. _db = db;
  24. }
  25. /// <summary>全量同步转换:源(SM 头) → 标准层(头)。</summary>
  26. public async Task<ProductionIssueMdpSyncResult> RunFullAsync(CancellationToken cancellationToken = default, string triggerType = "AUTO")
  27. {
  28. cancellationToken.ThrowIfCancellationRequested();
  29. await EnsureTablesAsync();
  30. var now = DateTime.Now;
  31. var batchId = $"S5_PROD_ISSUE_FULL_{now:yyyyMMddHHmmss}";
  32. var runLogId = await InsertRunLogAsync(batchId, now, triggerType);
  33. var result = new ProductionIssueMdpSyncResult { BatchId = batchId, RunLogId = runLogId };
  34. try
  35. {
  36. result.HeadRows = await TransformHeadStandardAsync(batchId, now);
  37. await MarkRunSuccessAsync(runLogId, now, result);
  38. return result;
  39. }
  40. catch (Exception ex)
  41. {
  42. await MarkRunFailedAsync(runLogId, now, ex.Message);
  43. throw;
  44. }
  45. }
  46. /// <summary>防御式建表(与 UpdateScripts DDL 同构,幂等)。</summary>
  47. private async Task EnsureTablesAsync()
  48. {
  49. await _db.Ado.ExecuteCommandAsync(
  50. """
  51. CREATE TABLE IF NOT EXISTS mdp_std_production_issue (
  52. id BIGINT AUTO_INCREMENT PRIMARY KEY,
  53. tenant_id BIGINT NOT NULL DEFAULT 0,
  54. factory_id BIGINT NULL DEFAULT 1,
  55. source_system VARCHAR(50) NOT NULL DEFAULT 'AIDOP',
  56. domain VARCHAR(80) NOT NULL,
  57. rec_id INT NOT NULL,
  58. nbr VARCHAR(24) NULL,
  59. issue_date DATETIME NULL,
  60. status VARCHAR(8) NULL,
  61. status_desc VARCHAR(20) NULL,
  62. work_ord VARCHAR(64) NULL,
  63. department VARCHAR(8) NULL,
  64. department_desc VARCHAR(255) NULL,
  65. qty_ord DECIMAL(18,5) NULL DEFAULT 0,
  66. prod_line VARCHAR(8) NULL,
  67. applicant_name VARCHAR(12) NULL,
  68. issue_user VARCHAR(255) NULL,
  69. user1 TEXT NULL,
  70. remark VARCHAR(200) NULL,
  71. create_user VARCHAR(24) NULL,
  72. source_create_time DATETIME NULL,
  73. pretreatment_state VARCHAR(50) NULL,
  74. trans_type VARCHAR(24) NULL,
  75. trans_type_text VARCHAR(50) NULL,
  76. eff_date DATETIME NULL,
  77. address VARCHAR(120) NULL,
  78. source_biz_key VARCHAR(200) NULL,
  79. sync_batch_id VARCHAR(100) NOT NULL,
  80. sync_time DATETIME NOT NULL,
  81. create_time DATETIME DEFAULT CURRENT_TIMESTAMP,
  82. update_time DATETIME DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
  83. UNIQUE KEY uk_mdp_std_prod_issue (tenant_id, domain, rec_id),
  84. KEY idx_mdp_std_prod_issue_nbr (tenant_id, nbr),
  85. KEY idx_mdp_std_prod_issue_date (tenant_id, issue_date),
  86. KEY idx_mdp_std_prod_issue_status (tenant_id, status),
  87. KEY idx_mdp_std_prod_issue_workord (tenant_id, work_ord),
  88. KEY idx_mdp_std_prod_issue_prodline (tenant_id, prod_line),
  89. KEY idx_mdp_std_prod_issue_dept (tenant_id, department)
  90. ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci COMMENT='S5生产领料单头标准层'
  91. """);
  92. }
  93. /// <summary>
  94. /// 标准化头:NbrMaster(SM) + DepartmentMaster -> mdp_std_production_issue。返回处理头行数。
  95. /// SQL 由旧系统领料头 SQL 翻译(SQL Server -&gt; MySQL):IsNull-&gt;IFNULL、upper() 保留、
  96. /// rtrim(a+' '+b)-&gt;TRIM(CONCAT(...))、with(nolock) 去除、convert(varchar,EffDate,23)-&gt; DATE(EffDate)。
  97. /// </summary>
  98. private async Task<int> TransformHeadStandardAsync(string batchId, DateTime now)
  99. {
  100. var rows = await _db.Ado.GetIntAsync(
  101. "SELECT COUNT(1) FROM NbrMaster m WHERE m.Type='SM' AND m.IsActive=1");
  102. await _db.Ado.ExecuteCommandAsync(
  103. """
  104. INSERT INTO mdp_std_production_issue
  105. (tenant_id, factory_id, source_system, domain, rec_id, nbr, issue_date, status, status_desc,
  106. work_ord, department, department_desc, qty_ord, prod_line, applicant_name, issue_user,
  107. user1, remark, create_user, source_create_time, pretreatment_state, trans_type, trans_type_text,
  108. eff_date, address, source_biz_key, sync_batch_id, sync_time)
  109. SELECT
  110. IFNULL(m.tenant_id, 0), 1, 'AIDOP', m.Domain, m.RecID, m.Nbr, m.Date,
  111. UPPER(IFNULL(m.Status, '')),
  112. CASE
  113. WHEN IFNULL(m.PretreatmentState,'')<>'' THEN m.PretreatmentState
  114. WHEN UPPER(IFNULL(m.Status,''))='C' THEN '已下架'
  115. WHEN UPPER(IFNULL(m.Status,''))='A' THEN '备料中'
  116. ELSE ''
  117. END,
  118. m.WorkOrd, m.Department, TRIM(CONCAT(m.Department, ' ', IFNULL(d.Descr, ''))),
  119. m.QtyOrd, m.ProdLine, m.Name, CAST(m.User1 AS CHAR),
  120. CAST(m.User1 AS CHAR), m.Remark, m.CreateUser, m.CreateTime,
  121. IFNULL(m.PretreatmentState, ''),
  122. CASE WHEN m.TransType='Z61' THEN '补料' ELSE '正常' END,
  123. CASE WHEN m.TransType='PrevProcess' THEN '需要前处理' ELSE '' END,
  124. m.EffDate, m.Address,
  125. CONCAT(m.Domain, ':', m.RecID), @BatchId, @Now
  126. FROM NbrMaster m
  127. LEFT JOIN DepartmentMaster d ON d.Domain = m.Domain AND d.Department = m.Department
  128. WHERE m.Type='SM' AND m.IsActive=1
  129. ON DUPLICATE KEY UPDATE
  130. factory_id=VALUES(factory_id), nbr=VALUES(nbr), issue_date=VALUES(issue_date),
  131. status=VALUES(status), status_desc=VALUES(status_desc), work_ord=VALUES(work_ord),
  132. department=VALUES(department), department_desc=VALUES(department_desc), qty_ord=VALUES(qty_ord),
  133. prod_line=VALUES(prod_line), applicant_name=VALUES(applicant_name), issue_user=VALUES(issue_user),
  134. user1=VALUES(user1), remark=VALUES(remark), create_user=VALUES(create_user),
  135. source_create_time=VALUES(source_create_time), pretreatment_state=VALUES(pretreatment_state),
  136. trans_type=VALUES(trans_type), trans_type_text=VALUES(trans_type_text),
  137. eff_date=VALUES(eff_date), address=VALUES(address),
  138. sync_batch_id=VALUES(sync_batch_id), sync_time=VALUES(sync_time), update_time=CURRENT_TIMESTAMP
  139. """,
  140. new SugarParameter("@BatchId", batchId),
  141. new SugarParameter("@Now", now));
  142. return rows;
  143. }
  144. private async Task<long> InsertRunLogAsync(string batchId, DateTime startedAt, string triggerType)
  145. {
  146. await _db.Ado.ExecuteCommandAsync(
  147. """
  148. INSERT INTO mdp_transform_run_log
  149. (tenant_id, job_code, job_name, trigger_type, batch_id, status, start_time)
  150. VALUES (0, @JobCode, 'S5生产领料单MDP同步与标准化转换', @TriggerType, @BatchId, 'RUNNING', @StartTime)
  151. """,
  152. new SugarParameter("@JobCode", JobCode),
  153. new SugarParameter("@TriggerType", NormalizeTriggerType(triggerType)),
  154. new SugarParameter("@BatchId", batchId),
  155. new SugarParameter("@StartTime", startedAt));
  156. return await _db.Ado.GetLongAsync(
  157. "SELECT id FROM mdp_transform_run_log WHERE batch_id=@BatchId ORDER BY id DESC LIMIT 1",
  158. new List<SugarParameter> { new("@BatchId", batchId) });
  159. }
  160. private async Task MarkRunSuccessAsync(long runLogId, DateTime startedAt, ProductionIssueMdpSyncResult result)
  161. {
  162. var finishedAt = DateTime.Now;
  163. await _db.Ado.ExecuteCommandAsync(
  164. """
  165. UPDATE mdp_transform_run_log
  166. SET status='SUCCESS', end_time=@EndTime, duration_ms=@DurationMs,
  167. stage_rows=0, standard_rows=@StandardRows, dwd_rows=0, update_time=CURRENT_TIMESTAMP
  168. WHERE id=@Id
  169. """,
  170. new SugarParameter("@EndTime", finishedAt),
  171. new SugarParameter("@DurationMs", (int)(finishedAt - startedAt).TotalMilliseconds),
  172. new SugarParameter("@StandardRows", result.HeadRows),
  173. new SugarParameter("@Id", runLogId));
  174. }
  175. private async Task MarkRunFailedAsync(long runLogId, DateTime startedAt, string message)
  176. {
  177. var finishedAt = DateTime.Now;
  178. await _db.Ado.ExecuteCommandAsync(
  179. """
  180. UPDATE mdp_transform_run_log
  181. SET status='FAILED', end_time=@EndTime, duration_ms=@DurationMs,
  182. error_message=@ErrorMessage, update_time=CURRENT_TIMESTAMP
  183. WHERE id=@Id
  184. """,
  185. new SugarParameter("@EndTime", finishedAt),
  186. new SugarParameter("@DurationMs", (int)(finishedAt - startedAt).TotalMilliseconds),
  187. new SugarParameter("@ErrorMessage", message.Length > 2000 ? message[..2000] : message),
  188. new SugarParameter("@Id", runLogId));
  189. }
  190. private static string NormalizeTriggerType(string? triggerType)
  191. => string.IsNullOrWhiteSpace(triggerType) ? "AUTO" : triggerType.Trim().ToUpperInvariant();
  192. }
  193. /// <summary>生产领料单 MDP 同步转换结果。</summary>
  194. public sealed class ProductionIssueMdpSyncResult
  195. {
  196. public long RunLogId { get; set; }
  197. public string BatchId { get; set; } = string.Empty;
  198. public int HeadRows { get; set; }
  199. }