S3MdpSyncTransformService.cs 62 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950951952953954955956957958959960961962963964965966967968969970971972973974975976977978979980981982983984985986987988989990991992993994995996997998999100010011002100310041005100610071008100910101011101210131014101510161017101810191020102110221023
  1. namespace Admin.NET.Plugin.AiDOP.Supply;
  2. /// <summary>
  3. /// S3 首批 MDP 同步和标准化转换服务。
  4. /// </summary>
  5. public class S3MdpSyncTransformService : ITransient
  6. {
  7. private readonly ISqlSugarClient _db;
  8. public S3MdpSyncTransformService(ISqlSugarClient db)
  9. {
  10. _db = db;
  11. }
  12. public async Task<S3MdpSyncTransformResult> RunFullAsync(CancellationToken cancellationToken = default, string triggerType = "AUTO")
  13. {
  14. cancellationToken.ThrowIfCancellationRequested();
  15. var now = DateTime.Now;
  16. var batchId = $"S3_MDP_FULL_{now:yyyyMMddHHmmss}";
  17. var runLogId = await InsertTransformRunLogAsync(batchId, now, triggerType);
  18. var result = new S3MdpSyncTransformResult { BatchId = batchId, RunLogId = runLogId };
  19. try
  20. {
  21. result.StageRows = await SyncStagingAsync(batchId, now, cancellationToken);
  22. result.StandardRows = await TransformStandardAsync(batchId, now, cancellationToken);
  23. result.DwdRows = await BuildDwdAsync(batchId, now, cancellationToken);
  24. result.KpiRows = await BuildS3KpiValuesAsync(now, cancellationToken);
  25. await MarkTransformRunSuccessAsync(runLogId, now, result);
  26. return result;
  27. }
  28. catch (Exception ex)
  29. {
  30. await MarkTransformRunFailedAsync(runLogId, now, ex.Message);
  31. throw;
  32. }
  33. }
  34. public async Task<S3MaterialRefreshResult> RefreshMaterialReadinessAsync(string? batchId = null, CancellationToken cancellationToken = default)
  35. {
  36. cancellationToken.ThrowIfCancellationRequested();
  37. var now = DateTime.Now;
  38. batchId ??= $"S3_MATERIAL_{now:yyyyMMddHHmmss}";
  39. _db.Ado.BeginTran();
  40. try
  41. {
  42. var stdCount = await _db.Ado.GetIntAsync("SELECT COUNT(1) FROM mdp_std_material_readiness WHERE IFNULL(work_order, '') <> ''");
  43. var readinessCount = await UpsertMaterialReadinessDwdAsync(batchId, now);
  44. await DeleteCurrentShortageAsync(now.Date);
  45. var shortageCount = await InsertMaterialShortageAsync(batchId, now);
  46. _db.Ado.CommitTran();
  47. return new S3MaterialRefreshResult
  48. {
  49. BatchId = batchId,
  50. StdCount = stdCount,
  51. ReadinessCount = readinessCount,
  52. ShortageCount = shortageCount
  53. };
  54. }
  55. catch
  56. {
  57. _db.Ado.RollbackTran();
  58. throw;
  59. }
  60. }
  61. private async Task<int> SyncStagingAsync(string batchId, DateTime now, CancellationToken cancellationToken)
  62. {
  63. var total = 0;
  64. foreach (var entity in S3MdpEntityConfig.All)
  65. {
  66. cancellationToken.ThrowIfCancellationRequested();
  67. total += await SyncOneEntityAsync(entity, batchId, now);
  68. }
  69. return total;
  70. }
  71. private async Task<int> SyncOneEntityAsync(S3MdpEntityConfig entity, string batchId, DateTime now)
  72. {
  73. var entityRow = await _db.Ado.SqlQuerySingleAsync<S3MdpEntityRow>(
  74. "SELECT id AS Id, entity_name AS EntityName FROM mdp_entity WHERE tenant_id=0 AND entity_code=@EntityCode LIMIT 1",
  75. new SugarParameter("@EntityCode", entity.EntityCode));
  76. if (entityRow == null) throw Oops.Oh($"未找到 MDP 实体配置:{entity.EntityCode}");
  77. var columns = await _db.Ado.SqlQueryAsync<S3ColumnRow>(
  78. """
  79. SELECT COLUMN_NAME AS ColumnName
  80. FROM information_schema.COLUMNS
  81. WHERE TABLE_SCHEMA=DATABASE() AND TABLE_NAME=@TableName
  82. ORDER BY ORDINAL_POSITION
  83. """,
  84. new SugarParameter("@TableName", entity.SourceTable));
  85. if (columns.Count == 0) throw Oops.Oh($"未找到源表:{entity.SourceTable}");
  86. var names = columns.Select(u => u.ColumnName).ToList();
  87. var tenantExpr = names.Any(u => string.Equals(u, "tenant_id", StringComparison.OrdinalIgnoreCase))
  88. ? $"IFNULL(s.`{FindColumn(names, "tenant_id")}`,0)"
  89. : "0";
  90. var sourceRowExpr = names.Any(u => string.Equals(u, entity.SourceRowIdExpression, StringComparison.OrdinalIgnoreCase))
  91. ? $"s.`{FindColumn(names, entity.SourceRowIdExpression)}`"
  92. : entity.SourceRowIdExpression;
  93. var rawDataExpr = BuildJsonObjectExpression(names);
  94. var rowsRead = await _db.Ado.GetIntAsync($"SELECT COUNT(1) FROM `{entity.SourceTable}`");
  95. var logId = await InsertSyncLogAsync(entityRow.Id, entityRow.EntityName, batchId, rowsRead);
  96. var started = DateTime.Now;
  97. try
  98. {
  99. var affected = await _db.Ado.ExecuteCommandAsync(
  100. $"""
  101. INSERT INTO `{entity.TargetTable}`
  102. (tenant_id, source_system, source_table, source_row_id, source_biz_key, sync_batch_id, sync_time, process_status, raw_data)
  103. SELECT
  104. {tenantExpr},
  105. 'AIDOP',
  106. @SourceTable,
  107. CAST({sourceRowExpr} AS CHAR),
  108. CAST(COALESCE({entity.SourceBizKeyExpression}, CAST({sourceRowExpr} AS CHAR)) AS CHAR),
  109. @BatchId,
  110. @Now,
  111. 'PENDING',
  112. {rawDataExpr}
  113. FROM `{entity.SourceTable}` s
  114. ON DUPLICATE KEY UPDATE
  115. source_row_id=VALUES(source_row_id),
  116. sync_batch_id=VALUES(sync_batch_id),
  117. sync_time=VALUES(sync_time),
  118. process_status=VALUES(process_status),
  119. raw_data=VALUES(raw_data),
  120. update_time=CURRENT_TIMESTAMP
  121. """,
  122. new SugarParameter("@SourceTable", entity.SourceTable),
  123. new SugarParameter("@BatchId", batchId),
  124. new SugarParameter("@Now", now));
  125. await MarkSyncLogSuccessAsync(logId, started, affected);
  126. return rowsRead;
  127. }
  128. catch (Exception ex)
  129. {
  130. await MarkSyncLogFailedAsync(logId, started, ex.Message);
  131. throw;
  132. }
  133. }
  134. private async Task<int> TransformStandardAsync(string batchId, DateTime now, CancellationToken cancellationToken)
  135. {
  136. var total = 0;
  137. foreach (var command in BuildStandardCommands(batchId, now))
  138. {
  139. cancellationToken.ThrowIfCancellationRequested();
  140. total += await _db.Ado.ExecuteCommandAsync(command.Sql, command.Parameters);
  141. }
  142. return total;
  143. }
  144. private async Task<int> BuildDwdAsync(string batchId, DateTime now, CancellationToken cancellationToken)
  145. {
  146. var total = 0;
  147. foreach (var command in BuildDwdCommands(batchId, now))
  148. {
  149. cancellationToken.ThrowIfCancellationRequested();
  150. total += await _db.Ado.ExecuteCommandAsync(command.Sql, command.Parameters);
  151. }
  152. total += await UpsertMaterialReadinessDwdAsync(batchId, now);
  153. await DeleteCurrentShortageAsync(now.Date);
  154. total += await InsertMaterialShortageAsync(batchId, now);
  155. return total;
  156. }
  157. private async Task<int> BuildS3KpiValuesAsync(DateTime now, CancellationToken cancellationToken)
  158. {
  159. var statDate = now.Date;
  160. var rows = await CalculateS3KpiValuesAsync(statDate);
  161. var affected = 0;
  162. foreach (var row in rows)
  163. {
  164. cancellationToken.ThrowIfCancellationRequested();
  165. affected += await UpsertS3KpiValueAsync(row, statDate, now);
  166. }
  167. return affected;
  168. }
  169. private async Task<List<S3KpiCalcRow>> CalculateS3KpiValuesAsync(DateTime statDate)
  170. {
  171. return await _db.Ado.SqlQueryAsync<S3KpiCalcRow>(
  172. """
  173. SELECT tenant_id AS TenantId, 1 AS FactoryId, 'S3_L2_004' AS MetricCode,
  174. ROUND(AVG(TIMESTAMPDIFF(HOUR, request_date, submit_date) / 24), 4) AS MetricValue
  175. FROM mdp_std_delivery_schedule
  176. WHERE request_date IS NOT NULL AND submit_date IS NOT NULL AND submit_date >= request_date
  177. GROUP BY tenant_id
  178. UNION ALL
  179. SELECT tenant_id AS TenantId, 1 AS FactoryId, 'S3_L2_005' AS MetricCode,
  180. ROUND(100 * SUM(CASE WHEN IFNULL(schedule_qty,0) >= IFNULL(order_qty,0) AND IFNULL(order_qty,0) > 0 THEN 1 ELSE 0 END) / COUNT(1), 4) AS MetricValue
  181. FROM dwd_supplier_delivery
  182. WHERE stat_date=@StatDate AND IFNULL(order_qty,0) > 0
  183. GROUP BY tenant_id
  184. UNION ALL
  185. SELECT tenant_id AS TenantId, 1 AS FactoryId, 'S3_L1_001' AS MetricCode,
  186. ROUND(AVG(TIMESTAMPDIFF(HOUR, request_date, submit_date) / 24), 4) AS MetricValue
  187. FROM mdp_std_delivery_schedule
  188. WHERE request_date IS NOT NULL AND submit_date IS NOT NULL AND submit_date >= request_date
  189. GROUP BY tenant_id
  190. UNION ALL
  191. SELECT tenant_id AS TenantId, 1 AS FactoryId, 'S3_L1_002' AS MetricCode,
  192. ROUND(100 * SUM(CASE WHEN IFNULL(schedule_qty,0) >= IFNULL(order_qty,0) AND IFNULL(order_qty,0) > 0 THEN 1 ELSE 0 END) / COUNT(1), 4) AS MetricValue
  193. FROM dwd_supplier_delivery
  194. WHERE stat_date=@StatDate AND IFNULL(order_qty,0) > 0
  195. GROUP BY tenant_id
  196. """,
  197. new SugarParameter("@StatDate", statDate));
  198. }
  199. private async Task<int> UpsertS3KpiValueAsync(S3KpiCalcRow row, DateTime statDate, DateTime now)
  200. {
  201. var meta = await _db.Ado.SqlQuerySingleAsync<S3KpiMetaRow>(
  202. """
  203. SELECT MetricLevel, Direction, YellowThreshold, RedThreshold
  204. FROM ado_smart_ops_kpi_master
  205. WHERE TenantId=@TenantId AND ModuleCode='S3' AND MetricCode=@MetricCode AND IsEnabled=1
  206. LIMIT 1
  207. """,
  208. new SugarParameter("@TenantId", row.TenantId),
  209. new SugarParameter("@MetricCode", row.MetricCode));
  210. if (meta == null || row.MetricValue == null)
  211. return 0;
  212. var table = ResolveKpiValueTable(meta.MetricLevel);
  213. var current = await _db.Ado.SqlQuerySingleAsync<S3KpiValueRow>(
  214. $"""
  215. SELECT id AS Id, metric_value AS MetricValue, target_value AS TargetValue
  216. FROM {table}
  217. WHERE tenant_id=@TenantId AND factory_id=@FactoryId AND module_code='S3'
  218. AND metric_code=@MetricCode AND biz_date=@BizDate AND is_deleted=0
  219. ORDER BY id
  220. LIMIT 1
  221. """,
  222. new SugarParameter("@TenantId", row.TenantId),
  223. new SugarParameter("@FactoryId", row.FactoryId),
  224. new SugarParameter("@MetricCode", row.MetricCode),
  225. new SugarParameter("@BizDate", statDate));
  226. var prior = await _db.Ado.SqlQuerySingleAsync<S3KpiValueRow>(
  227. $"""
  228. SELECT id AS Id, metric_value AS MetricValue, target_value AS TargetValue
  229. FROM {table}
  230. WHERE tenant_id=@TenantId AND factory_id=@FactoryId AND module_code='S3'
  231. AND metric_code=@MetricCode AND biz_date<@BizDate AND is_deleted=0
  232. ORDER BY biz_date DESC, id DESC
  233. LIMIT 1
  234. """,
  235. new SugarParameter("@TenantId", row.TenantId),
  236. new SugarParameter("@FactoryId", row.FactoryId),
  237. new SugarParameter("@MetricCode", row.MetricCode),
  238. new SugarParameter("@BizDate", statDate));
  239. var actual = Math.Round(row.MetricValue.Value, 4);
  240. var target = current?.TargetValue ?? prior?.TargetValue ?? DefaultS3Target(row.MetricCode);
  241. var status = ResolveKpiStatus(actual, target, meta.Direction, meta.YellowThreshold, meta.RedThreshold);
  242. var trend = ResolveTrendFlag(actual, prior?.MetricValue);
  243. if (current != null)
  244. {
  245. return await _db.Ado.ExecuteCommandAsync(
  246. $"""
  247. UPDATE {table}
  248. SET metric_value=@MetricValue, target_value=@TargetValue, status_color=@StatusColor, trend_flag=@TrendFlag,
  249. is_active=1, status='ACTIVE', calc_time=@CalcTime, update_time=@CalcTime
  250. WHERE tenant_id=@TenantId AND factory_id=@FactoryId AND module_code='S3'
  251. AND metric_code=@MetricCode AND biz_date=@BizDate AND is_deleted=0
  252. """,
  253. new SugarParameter("@MetricValue", actual),
  254. new SugarParameter("@TargetValue", target),
  255. new SugarParameter("@StatusColor", status),
  256. new SugarParameter("@TrendFlag", trend),
  257. new SugarParameter("@CalcTime", now),
  258. new SugarParameter("@TenantId", row.TenantId),
  259. new SugarParameter("@FactoryId", row.FactoryId),
  260. new SugarParameter("@MetricCode", row.MetricCode),
  261. new SugarParameter("@BizDate", statDate));
  262. }
  263. var nextId = await _db.Ado.GetLongAsync($"SELECT COALESCE(MAX(id), 0) + 1 FROM {table}");
  264. return await _db.Ado.ExecuteCommandAsync(
  265. $"""
  266. INSERT INTO {table}
  267. (id, tenant_id, factory_id, status, biz_date, create_time, update_time, is_deleted, is_active,
  268. module_code, metric_code, metric_value, target_value, status_color, trend_flag, calc_time)
  269. VALUES
  270. (@Id, @TenantId, @FactoryId, 'ACTIVE', @BizDate, @CalcTime, @CalcTime, 0, 1,
  271. 'S3', @MetricCode, @MetricValue, @TargetValue, @StatusColor, @TrendFlag, @CalcTime)
  272. """,
  273. new SugarParameter("@Id", nextId),
  274. new SugarParameter("@TenantId", row.TenantId),
  275. new SugarParameter("@FactoryId", row.FactoryId),
  276. new SugarParameter("@BizDate", statDate),
  277. new SugarParameter("@CalcTime", now),
  278. new SugarParameter("@MetricCode", row.MetricCode),
  279. new SugarParameter("@MetricValue", actual),
  280. new SugarParameter("@TargetValue", target),
  281. new SugarParameter("@StatusColor", status),
  282. new SugarParameter("@TrendFlag", trend));
  283. }
  284. private IEnumerable<S3MdpSqlCommand> BuildStandardCommands(string batchId, DateTime now)
  285. {
  286. yield return Cmd(
  287. """
  288. INSERT INTO mdp_std_supplier
  289. (tenant_id, factory_id, company_id, source_system, supplier_code, supplier_name, supplier_type, status, contact, address, currency_type, source_biz_key, sync_batch_id, sync_time)
  290. SELECT tenant_id, factory_id, company_id, 'AIDOP',
  291. JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.Supp')),
  292. JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.SortName')),
  293. JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.Type')),
  294. CASE WHEN JSON_EXTRACT(raw_data,'$.IsActive') IN (1, true) THEN 'ACTIVE' ELSE 'INACTIVE' END,
  295. JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.contact')),
  296. JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.address')),
  297. JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.Curr')),
  298. source_biz_key, @BatchId, @Now
  299. FROM mdp_stg_supplier
  300. WHERE source_table='SuppMaster' AND IFNULL(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.Supp')), '') <> ''
  301. ON DUPLICATE KEY UPDATE supplier_name=VALUES(supplier_name), supplier_type=VALUES(supplier_type), status=VALUES(status),
  302. contact=VALUES(contact), address=VALUES(address), currency_type=VALUES(currency_type), sync_batch_id=VALUES(sync_batch_id),
  303. sync_time=VALUES(sync_time), update_time=CURRENT_TIMESTAMP
  304. """, batchId, now);
  305. yield return Cmd(
  306. """
  307. INSERT INTO mdp_std_item
  308. (tenant_id, factory_id, company_id, source_system, item_code, item_name, model, unit, item_type, status, source_biz_key, sync_batch_id, sync_time)
  309. SELECT tenant_id, factory_id, company_id, 'AIDOP',
  310. COALESCE(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.ItemNum')), JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.number'))),
  311. COALESCE(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.Descr')), JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.name')), JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.icitem_name'))),
  312. COALESCE(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.Drawing')), JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.model'))),
  313. COALESCE(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.UM')), JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.unit'))),
  314. COALESCE(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.ItemType')), JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.item_type'))),
  315. COALESCE(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.Status')), JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.is_active'))),
  316. source_biz_key, @BatchId, @Now
  317. FROM mdp_stg_item
  318. WHERE COALESCE(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.ItemNum')), JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.number')), '') <> ''
  319. ON DUPLICATE KEY UPDATE item_name=VALUES(item_name), model=VALUES(model), unit=VALUES(unit), item_type=VALUES(item_type),
  320. status=VALUES(status), sync_batch_id=VALUES(sync_batch_id), sync_time=VALUES(sync_time), update_time=CURRENT_TIMESTAMP
  321. """, batchId, now);
  322. yield return Cmd(
  323. """
  324. INSERT INTO mdp_std_supplier_item
  325. (tenant_id, factory_id, company_id, source_system, item_code, supplier_code, supplier_name, supplier_type, quota_rate, lead_time, min_qty, packaging_qty, price, currency_type, effective_date, expire_date, status, source_biz_key, sync_batch_id, sync_time)
  326. SELECT tenant_id, factory_id, company_id, 'AIDOP',
  327. IFNULL(COALESCE(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.number')), JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.icitem_number')), JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.icitem_name'))), ''),
  328. JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.supplier_number')),
  329. JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.supplier_name')),
  330. JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.supplier_type')),
  331. CASE WHEN JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.quota_rate')) REGEXP '^-?[0-9]+(\\.[0-9]+)?$' THEN CAST(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.quota_rate')) AS DECIMAL(18,6)) END,
  332. CASE WHEN JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.lead_time')) REGEXP '^-?[0-9]+(\\.[0-9]+)?$' THEN CAST(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.lead_time')) AS DECIMAL(18,6)) END,
  333. CASE WHEN JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.qty_min')) REGEXP '^-?[0-9]+(\\.[0-9]+)?$' THEN CAST(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.qty_min')) AS DECIMAL(18,6)) END,
  334. CASE WHEN JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.packaging_qty')) REGEXP '^-?[0-9]+(\\.[0-9]+)?$' THEN CAST(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.packaging_qty')) AS DECIMAL(18,6)) END,
  335. COALESCE(
  336. CASE WHEN JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.netpurchase_price')) REGEXP '^-?[0-9]+(\\.[0-9]+)?$' THEN CAST(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.netpurchase_price')) AS DECIMAL(18,6)) END,
  337. CASE WHEN JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.order_price')) REGEXP '^-?[0-9]+(\\.[0-9]+)?$' THEN CAST(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.order_price')) AS DECIMAL(18,6)) END
  338. ),
  339. JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.currency_type')),
  340. NULLIF(NULLIF(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.effective_date')), 'null'), ''),
  341. NULLIF(NULLIF(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.expiring_date')), 'null'), ''),
  342. JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.is_active')),
  343. source_biz_key, @BatchId, @Now
  344. FROM mdp_stg_source_list
  345. WHERE source_table='srm_purchase' AND IFNULL(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.supplier_number')), '') <> ''
  346. AND IFNULL(COALESCE(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.number')), JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.icitem_number')), JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.icitem_name'))), '') <> ''
  347. ON DUPLICATE KEY UPDATE supplier_name=VALUES(supplier_name), quota_rate=VALUES(quota_rate), lead_time=VALUES(lead_time),
  348. min_qty=VALUES(min_qty), packaging_qty=VALUES(packaging_qty), price=VALUES(price), sync_batch_id=VALUES(sync_batch_id),
  349. sync_time=VALUES(sync_time), update_time=CURRENT_TIMESTAMP
  350. """, batchId, now);
  351. yield return Cmd(
  352. """
  353. INSERT INTO mdp_std_supply_demand
  354. (tenant_id, factory_id, company_id, source_system, demand_no, demand_line, demand_type, item_code, item_name, required_qty, required_date, supplier_code, status, source_biz_key, sync_batch_id, sync_time)
  355. SELECT tenant_id, factory_id, company_id, 'AIDOP',
  356. COALESCE(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.number')), source_row_id),
  357. COALESCE(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.line')), JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.Line'))),
  358. source_table,
  359. IFNULL(COALESCE(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.itemnum')), JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.item_number')), JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.ItemNum'))), ''),
  360. IFNULL(COALESCE(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.itemname')), JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.item_name')), JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.Descr'))), ''),
  361. COALESCE(
  362. CASE WHEN JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.qty')) REGEXP '^-?[0-9]+(\\.[0-9]+)?$' THEN CAST(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.qty')) AS DECIMAL(18,6)) END,
  363. CASE WHEN JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.required_qty')) REGEXP '^-?[0-9]+(\\.[0-9]+)?$' THEN CAST(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.required_qty')) AS DECIMAL(18,6)) END,
  364. CASE WHEN JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.shortqty')) REGEXP '^-?[0-9]+(\\.[0-9]+)?$' THEN CAST(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.shortqty')) AS DECIMAL(18,6)) END,
  365. 0),
  366. COALESCE(NULLIF(NULLIF(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.required_date')), 'null'), ''), NULLIF(NULLIF(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.needdate')), 'null'), ''), NULLIF(NULLIF(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.NeedDate')), 'null'), '')),
  367. JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.supplier_number')),
  368. COALESCE(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.status')), JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.Status'))),
  369. source_biz_key, @BatchId, @Now
  370. FROM mdp_stg_supply_demand
  371. WHERE IFNULL(source_biz_key,'') <> ''
  372. ON DUPLICATE KEY UPDATE item_code=VALUES(item_code), item_name=VALUES(item_name), required_qty=VALUES(required_qty),
  373. required_date=VALUES(required_date), supplier_code=VALUES(supplier_code), status=VALUES(status), sync_batch_id=VALUES(sync_batch_id),
  374. sync_time=VALUES(sync_time), update_time=CURRENT_TIMESTAMP
  375. """, batchId, now);
  376. yield return Cmd(
  377. """
  378. INSERT INTO mdp_std_purchase_request
  379. (tenant_id, factory_id, company_id, source_system, pr_no, pr_line, item_code, item_name, supplier_code, request_qty, request_date, send_date, arrive_date, status, source_biz_key, sync_batch_id, sync_time)
  380. SELECT tenant_id, factory_id, company_id, 'AIDOP',
  381. COALESCE(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.pr_billno')), source_row_id),
  382. JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.line_no')),
  383. IFNULL(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.item_number')), ''),
  384. IFNULL(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.item_name')), ''),
  385. JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.supplier_number')),
  386. COALESCE(
  387. CASE WHEN JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.qty')) REGEXP '^-?[0-9]+(\\.[0-9]+)?$' THEN CAST(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.qty')) AS DECIMAL(18,6)) END,
  388. CASE WHEN JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.request_qty')) REGEXP '^-?[0-9]+(\\.[0-9]+)?$' THEN CAST(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.request_qty')) AS DECIMAL(18,6)) END,
  389. 0),
  390. NULLIF(NULLIF(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.create_time')), 'null'), ''),
  391. NULLIF(NULLIF(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.send_date')), 'null'), ''),
  392. NULLIF(NULLIF(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.arrive_date')), 'null'), ''),
  393. JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.status')),
  394. source_biz_key, @BatchId, @Now
  395. FROM mdp_stg_supply_demand
  396. WHERE source_table='srm_pr_main'
  397. ON DUPLICATE KEY UPDATE item_code=VALUES(item_code), item_name=VALUES(item_name), supplier_code=VALUES(supplier_code),
  398. request_qty=VALUES(request_qty), status=VALUES(status), sync_batch_id=VALUES(sync_batch_id), sync_time=VALUES(sync_time),
  399. update_time=CURRENT_TIMESTAMP
  400. """, batchId, now);
  401. yield return Cmd(
  402. """
  403. INSERT INTO mdp_std_purchase_order
  404. (tenant_id, factory_id, company_id, source_system, po_no, po_line, po_type, supplier_code, item_code, item_name, order_qty, received_qty, returned_qty, due_date, need_date, order_date, status, buyer, work_order, source_biz_key, sync_batch_id, sync_time)
  405. SELECT d.tenant_id, d.factory_id, d.company_id, 'AIDOP',
  406. JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.PurOrd')),
  407. CAST(JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.Line')) AS CHAR),
  408. JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.Potype')),
  409. JSON_UNQUOTE(JSON_EXTRACT(m.raw_data,'$.Supp')),
  410. IFNULL(JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.ItemNum')), ''),
  411. IFNULL(COALESCE(JSON_UNQUOTE(JSON_EXTRACT(i.raw_data,'$.Descr')), JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.Descr'))), ''),
  412. COALESCE(CASE WHEN JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.QtyOrded')) REGEXP '^-?[0-9]+(\\.[0-9]+)?$' THEN CAST(JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.QtyOrded')) AS DECIMAL(18,6)) END, 0),
  413. COALESCE(
  414. CASE WHEN JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.QtyReceived')) REGEXP '^-?[0-9]+(\\.[0-9]+)?$' THEN CAST(JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.QtyReceived')) AS DECIMAL(18,6)) END,
  415. CASE WHEN JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.RctQty')) REGEXP '^-?[0-9]+(\\.[0-9]+)?$' THEN CAST(JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.RctQty')) AS DECIMAL(18,6)) END,
  416. 0),
  417. COALESCE(CASE WHEN JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.QtyReturned')) REGEXP '^-?[0-9]+(\\.[0-9]+)?$' THEN CAST(JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.QtyReturned')) AS DECIMAL(18,6)) END, 0),
  418. NULLIF(NULLIF(JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.DueDate')), 'null'), ''),
  419. NULLIF(NULLIF(JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.NeedDate')), 'null'), ''),
  420. NULLIF(NULLIF(JSON_UNQUOTE(JSON_EXTRACT(m.raw_data,'$.OrdDate')), 'null'), ''),
  421. JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.Status')),
  422. JSON_UNQUOTE(JSON_EXTRACT(m.raw_data,'$.Buyer')),
  423. JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.WorkOrd')),
  424. d.source_biz_key, @BatchId, @Now
  425. FROM mdp_stg_purchase_order d
  426. LEFT JOIN mdp_stg_purchase_order m ON m.source_table='PurOrdMaster'
  427. AND JSON_UNQUOTE(JSON_EXTRACT(m.raw_data,'$.PurOrd')) = JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.PurOrd'))
  428. LEFT JOIN mdp_stg_item i ON i.source_table='ItemMaster'
  429. AND JSON_UNQUOTE(JSON_EXTRACT(i.raw_data,'$.ItemNum')) = JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.ItemNum'))
  430. WHERE d.source_table='PurOrdDetail' AND IFNULL(JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.PurOrd')), '') <> ''
  431. ON DUPLICATE KEY UPDATE po_type=VALUES(po_type), supplier_code=VALUES(supplier_code), item_code=VALUES(item_code),
  432. item_name=VALUES(item_name), order_qty=VALUES(order_qty), received_qty=VALUES(received_qty), returned_qty=VALUES(returned_qty),
  433. due_date=VALUES(due_date), need_date=VALUES(need_date), order_date=VALUES(order_date), status=VALUES(status),
  434. buyer=VALUES(buyer), work_order=VALUES(work_order), sync_batch_id=VALUES(sync_batch_id), sync_time=VALUES(sync_time),
  435. update_time=CURRENT_TIMESTAMP
  436. """, batchId, now);
  437. yield return Cmd(
  438. """
  439. INSERT INTO mdp_std_delivery_schedule
  440. (tenant_id, source_system, delivery_plan_no, po_no, po_line, item_code, supplier_code, supplier_name, schedule_qty, sent_qty, rest_qty, return_qty, request_date, need_date, submit_date, last_sent_date, status, source_biz_key, sync_batch_id, sync_time)
  441. SELECT tenant_id, 'AIDOP',
  442. JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.dsnum')),
  443. JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.ponumber')),
  444. CAST(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.poline')) AS CHAR),
  445. IFNULL(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.itemnum')), ''),
  446. JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.suppliercode')),
  447. JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.supplier')),
  448. COALESCE(CASE WHEN JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.schedqty')) REGEXP '^-?[0-9]+(\\.[0-9]+)?$' THEN CAST(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.schedqty')) AS DECIMAL(18,6)) END,0),
  449. COALESCE(CASE WHEN JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.sentqty')) REGEXP '^-?[0-9]+(\\.[0-9]+)?$' THEN CAST(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.sentqty')) AS DECIMAL(18,6)) END,0),
  450. COALESCE(CASE WHEN JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.restqty')) REGEXP '^-?[0-9]+(\\.[0-9]+)?$' THEN CAST(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.restqty')) AS DECIMAL(18,6)) END,0),
  451. COALESCE(CASE WHEN JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.returnqty')) REGEXP '^-?[0-9]+(\\.[0-9]+)?$' THEN CAST(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.returnqty')) AS DECIMAL(18,6)) END,0),
  452. NULLIF(NULLIF(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.requestdate')), 'null'), ''),
  453. NULLIF(NULLIF(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.needdate')), 'null'), ''),
  454. NULLIF(NULLIF(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.submitdate')), 'null'), ''),
  455. NULLIF(NULLIF(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.lastsentdate')), 'null'), ''),
  456. JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.status')),
  457. source_biz_key, @BatchId, @Now
  458. FROM mdp_stg_delivery
  459. WHERE source_table='srm_polist_ds' AND IFNULL(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.dsnum')), '') <> ''
  460. ON DUPLICATE KEY UPDATE po_no=VALUES(po_no), po_line=VALUES(po_line), item_code=VALUES(item_code), supplier_code=VALUES(supplier_code),
  461. supplier_name=VALUES(supplier_name), schedule_qty=VALUES(schedule_qty), sent_qty=VALUES(sent_qty), rest_qty=VALUES(rest_qty),
  462. return_qty=VALUES(return_qty), need_date=VALUES(need_date), status=VALUES(status), sync_batch_id=VALUES(sync_batch_id),
  463. sync_time=VALUES(sync_time), update_time=CURRENT_TIMESTAMP
  464. """, batchId, now);
  465. yield return Cmd(
  466. """
  467. INSERT INTO mdp_std_delivery_result
  468. (tenant_id, source_system, delivery_no, delivery_line, delivery_plan_no, po_no, po_line, item_code, delivery_qty, receipt_qty, return_qty, receipt_date, qc_status, event_time, source_biz_key, sync_batch_id, sync_time)
  469. SELECT tenant_id, 'AIDOP',
  470. COALESCE(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.shddh')), source_row_id),
  471. CAST(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.id')) AS CHAR),
  472. JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.dsnum')),
  473. JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.po_bill')),
  474. CAST(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.po_billline')) AS CHAR),
  475. IFNULL(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.itemnum')), ''),
  476. COALESCE(CASE WHEN JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.sh_delivery_quantity')) REGEXP '^-?[0-9]+(\\.[0-9]+)?$' THEN CAST(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.sh_delivery_quantity')) AS DECIMAL(18,6)) END,0),
  477. 0,
  478. 0,
  479. NULLIF(NULLIF(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.receipt_date')), 'null'), ''),
  480. JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.qc_status')),
  481. COALESCE(NULLIF(NULLIF(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.updatetime')), 'null'), ''), NULLIF(NULLIF(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.createtime')), 'null'), '')),
  482. source_biz_key, @BatchId, @Now
  483. FROM mdp_stg_delivery
  484. WHERE source_table IN ('scm_shd','scm_shdzb')
  485. ON DUPLICATE KEY UPDATE delivery_qty=VALUES(delivery_qty), receipt_qty=VALUES(receipt_qty), return_qty=VALUES(return_qty),
  486. receipt_date=VALUES(receipt_date), event_time=VALUES(event_time), sync_batch_id=VALUES(sync_batch_id), sync_time=VALUES(sync_time),
  487. update_time=CURRENT_TIMESTAMP
  488. """, batchId, now);
  489. yield return Cmd(
  490. """
  491. INSERT INTO mdp_std_material_readiness
  492. (tenant_id, source_system, work_order, op_code, item_code, component_item_code, required_qty, issued_qty, received_qty, available_qty, in_transit_qty, incoming_qty, shortage_qty, ready_status, need_date, supplier_code, source_biz_key, sync_batch_id, sync_time)
  493. SELECT tenant_id, 'AIDOP',
  494. JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.WorkOrd')),
  495. CAST(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.Op')) AS CHAR),
  496. IFNULL(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.PMBOM')), ''),
  497. COALESCE(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.ItemNum')), source_row_id),
  498. COALESCE(
  499. CASE WHEN JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.QtyRequired')) REGEXP '^-?[0-9]+(\\.[0-9]+)?$' THEN CAST(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.QtyRequired')) AS DECIMAL(18,6)) END,
  500. CASE WHEN JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.QtyReq')) REGEXP '^-?[0-9]+(\\.[0-9]+)?$' THEN CAST(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.QtyReq')) AS DECIMAL(18,6)) END,
  501. 0),
  502. COALESCE(CASE WHEN JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.QtyIssued')) REGEXP '^-?[0-9]+(\\.[0-9]+)?$' THEN CAST(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.QtyIssued')) AS DECIMAL(18,6)) END, 0),
  503. COALESCE(CASE WHEN JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.QtyReceived')) REGEXP '^-?[0-9]+(\\.[0-9]+)?$' THEN CAST(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.QtyReceived')) AS DECIMAL(18,6)) END, 0),
  504. 0, 0, 0,
  505. 0,
  506. 'READY',
  507. NULLIF(NULLIF(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.DueDate')), 'null'), ''),
  508. NULL,
  509. source_biz_key, @BatchId, @Now
  510. FROM mdp_stg_work_order_material
  511. WHERE source_table='WorkOrdDetail' AND IFNULL(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.WorkOrd')), '') <> ''
  512. ON DUPLICATE KEY UPDATE required_qty=VALUES(required_qty), received_qty=VALUES(received_qty), need_date=VALUES(need_date),
  513. sync_batch_id=VALUES(sync_batch_id), sync_time=VALUES(sync_time), update_time=CURRENT_TIMESTAMP
  514. """, batchId, now);
  515. yield return Cmd(
  516. """
  517. INSERT INTO mdp_std_process_outsource_order
  518. (tenant_id, source_system, work_order, op_code, routing_code, supplier_code, po_no, po_line, order_qty, completed_qty, due_date, status, source_biz_key, sync_batch_id, sync_time)
  519. SELECT tenant_id, 'AIDOP',
  520. COALESCE(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.WorkOrd')), ''),
  521. CAST(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.Op')) AS CHAR),
  522. COALESCE(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.RoutingCode')), JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.RouteCode'))),
  523. JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.SupplierCode')),
  524. NULL, NULL,
  525. COALESCE(CASE WHEN JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.PackingQty')) REGEXP '^-?[0-9]+(\\.[0-9]+)?$' THEN CAST(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.PackingQty')) AS DECIMAL(18,6)) END, 0),
  526. 0,
  527. NULLIF(NULLIF(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.UpdatedAt')), 'null'), ''),
  528. CASE WHEN JSON_EXTRACT(raw_data,'$.IsEnabled') IN (1, true) THEN 'OPEN' ELSE 'DISABLED' END,
  529. source_biz_key, @BatchId, @Now
  530. FROM mdp_stg_work_order_material
  531. WHERE source_table='RoutingOpDetail'
  532. ON DUPLICATE KEY UPDATE routing_code=VALUES(routing_code), supplier_code=VALUES(supplier_code), order_qty=VALUES(order_qty),
  533. due_date=VALUES(due_date), status=VALUES(status), sync_batch_id=VALUES(sync_batch_id), sync_time=VALUES(sync_time),
  534. update_time=CURRENT_TIMESTAMP
  535. """, batchId, now);
  536. }
  537. private IEnumerable<S3MdpSqlCommand> BuildDwdCommands(string batchId, DateTime now)
  538. {
  539. yield return Cmd(
  540. """
  541. INSERT INTO dwd_supply_demand
  542. (tenant_id, stat_date, demand_no, demand_line, demand_type, item_code, item_name, supplier_code, required_qty, fulfilled_qty, shortage_qty, required_date, demand_status, source_system, sync_batch_id, calc_time)
  543. SELECT tenant_id, @StatDate, demand_no, IFNULL(demand_line,''), demand_type, IFNULL(item_code,''), IFNULL(item_name,''), supplier_code,
  544. IFNULL(required_qty,0), 0, IFNULL(required_qty,0), required_date, status, source_system, @BatchId, @Now
  545. FROM mdp_std_supply_demand
  546. WHERE IFNULL(demand_no,'') <> ''
  547. ON DUPLICATE KEY UPDATE item_code=VALUES(item_code), item_name=VALUES(item_name), supplier_code=VALUES(supplier_code),
  548. required_qty=VALUES(required_qty), fulfilled_qty=VALUES(fulfilled_qty), shortage_qty=VALUES(shortage_qty),
  549. demand_status=VALUES(demand_status), sync_batch_id=VALUES(sync_batch_id), calc_time=VALUES(calc_time), update_time=CURRENT_TIMESTAMP
  550. """, batchId, now);
  551. yield return Cmd(
  552. """
  553. INSERT INTO dwd_supplier_delivery
  554. (tenant_id, stat_date, po_no, po_line, po_type, supplier_code, supplier_name, item_code, item_name, order_qty, schedule_qty, delivery_qty, receipt_qty, return_qty, remaining_qty, due_date, need_date, last_delivery_date, last_receipt_date, delivery_status, risk_level, source_system, sync_batch_id, calc_time)
  555. SELECT po.tenant_id, @StatDate, po.po_no, po.po_line, po.po_type, po.supplier_code, IFNULL(s.supplier_name, ds.supplier_name),
  556. IFNULL(po.item_code,''), IFNULL(po.item_name,''), IFNULL(po.order_qty,0), IFNULL(ds.schedule_qty,0), IFNULL(dr.delivery_qty,0),
  557. IFNULL(po.received_qty,0) + IFNULL(dr.receipt_qty,0), IFNULL(po.returned_qty,0) + IFNULL(dr.return_qty,0),
  558. GREATEST(IFNULL(po.order_qty,0) - IFNULL(po.received_qty,0) - IFNULL(dr.receipt_qty,0) - IFNULL(po.returned_qty,0), 0),
  559. po.due_date, COALESCE(ds.need_date, po.need_date), dr.event_time, dr.receipt_date,
  560. CASE
  561. WHEN IFNULL(po.order_qty,0) <= IFNULL(po.received_qty,0) + IFNULL(dr.receipt_qty,0) THEN 'COMPLETED'
  562. WHEN COALESCE(ds.need_date, po.need_date, po.due_date) < @Now THEN 'DELAYED'
  563. ELSE 'OPEN'
  564. END,
  565. CASE
  566. WHEN COALESCE(ds.need_date, po.need_date, po.due_date) < @Now
  567. AND IFNULL(po.order_qty,0) > IFNULL(po.received_qty,0) + IFNULL(dr.receipt_qty,0) THEN 'HIGH'
  568. WHEN IFNULL(po.order_qty,0) > IFNULL(po.received_qty,0) + IFNULL(dr.receipt_qty,0) THEN 'MEDIUM'
  569. ELSE 'LOW'
  570. END,
  571. 'AIDOP', @BatchId, @Now
  572. FROM mdp_std_purchase_order po
  573. LEFT JOIN (
  574. SELECT tenant_id, po_no, po_line, MIN(supplier_name) AS supplier_name, SUM(IFNULL(schedule_qty,0)) AS schedule_qty, MIN(need_date) AS need_date
  575. FROM mdp_std_delivery_schedule
  576. WHERE IFNULL(po_no,'') <> '' AND IFNULL(po_line,'') <> ''
  577. GROUP BY tenant_id, po_no, po_line
  578. ) ds ON po.tenant_id=ds.tenant_id AND po.po_no=ds.po_no AND po.po_line=ds.po_line
  579. LEFT JOIN (
  580. SELECT tenant_id, po_no, po_line, SUM(IFNULL(delivery_qty,0)) AS delivery_qty, SUM(IFNULL(receipt_qty,0)) AS receipt_qty,
  581. SUM(IFNULL(return_qty,0)) AS return_qty, MAX(event_time) AS event_time, MAX(receipt_date) AS receipt_date
  582. FROM mdp_std_delivery_result
  583. WHERE IFNULL(po_no,'') <> '' AND IFNULL(po_line,'') <> ''
  584. GROUP BY tenant_id, po_no, po_line
  585. ) dr ON po.tenant_id=dr.tenant_id AND po.po_no=dr.po_no AND po.po_line=dr.po_line
  586. LEFT JOIN (
  587. SELECT tenant_id, supplier_code, MAX(supplier_name) AS supplier_name
  588. FROM mdp_std_supplier
  589. WHERE IFNULL(supplier_code,'') <> ''
  590. GROUP BY tenant_id, supplier_code
  591. ) s ON po.tenant_id=s.tenant_id AND po.supplier_code=s.supplier_code
  592. WHERE IFNULL(po.po_no,'') <> '' AND IFNULL(po.po_line,'') <> ''
  593. ON DUPLICATE KEY UPDATE supplier_name=VALUES(supplier_name), item_code=VALUES(item_code), item_name=VALUES(item_name),
  594. order_qty=VALUES(order_qty), schedule_qty=VALUES(schedule_qty), delivery_qty=VALUES(delivery_qty), receipt_qty=VALUES(receipt_qty),
  595. return_qty=VALUES(return_qty), remaining_qty=VALUES(remaining_qty), delivery_status=VALUES(delivery_status),
  596. risk_level=VALUES(risk_level), sync_batch_id=VALUES(sync_batch_id), calc_time=VALUES(calc_time), update_time=CURRENT_TIMESTAMP
  597. """, batchId, now);
  598. yield return Cmd("DELETE FROM dwd_supplier_risk WHERE stat_date=@StatDate", batchId, now);
  599. yield return Cmd(
  600. """
  601. INSERT INTO dwd_supplier_risk
  602. (tenant_id, stat_date, supplier_code, supplier_name, item_code, risk_type, risk_level, risk_count, risk_qty, risk_reason, source_system, calc_batch_id, calc_time)
  603. SELECT tenant_id, stat_date, IFNULL(supplier_code,''), supplier_name, IFNULL(item_code,''),
  604. 'DELIVERY_DELAY', risk_level, COUNT(1), SUM(IFNULL(remaining_qty,0)),
  605. '供应交付存在延期或未完成风险', 'AIDOP', @BatchId, @Now
  606. FROM dwd_supplier_delivery
  607. WHERE stat_date=@StatDate AND risk_level IN ('HIGH','MEDIUM') AND IFNULL(supplier_code,'') <> ''
  608. GROUP BY tenant_id, stat_date, supplier_code, supplier_name, item_code, risk_level
  609. ON DUPLICATE KEY UPDATE risk_level=VALUES(risk_level), risk_count=VALUES(risk_count), risk_qty=VALUES(risk_qty),
  610. risk_reason=VALUES(risk_reason), calc_batch_id=VALUES(calc_batch_id), calc_time=VALUES(calc_time), update_time=CURRENT_TIMESTAMP
  611. """, batchId, now);
  612. yield return Cmd(
  613. """
  614. INSERT INTO dwd_process_outsource_delivery
  615. (tenant_id, stat_date, work_order, op_code, routing_code, supplier_code, supplier_name, po_no, po_line, order_qty, completed_qty, remaining_qty, due_date, delivery_status, risk_level, source_system, calc_batch_id, calc_time)
  616. SELECT o.tenant_id, @StatDate, o.work_order, o.op_code, o.routing_code, o.supplier_code, s.supplier_name, o.po_no, IFNULL(o.po_line,''),
  617. IFNULL(o.order_qty,0), IFNULL(o.completed_qty,0), GREATEST(IFNULL(o.order_qty,0)-IFNULL(o.completed_qty,0),0),
  618. o.due_date,
  619. CASE WHEN IFNULL(o.completed_qty,0) >= IFNULL(o.order_qty,0) AND IFNULL(o.order_qty,0) > 0 THEN 'COMPLETED'
  620. WHEN o.due_date < @Now THEN 'DELAYED' ELSE IFNULL(o.status,'OPEN') END,
  621. CASE WHEN o.due_date < @Now AND IFNULL(o.completed_qty,0) < IFNULL(o.order_qty,0) THEN 'HIGH'
  622. WHEN IFNULL(o.completed_qty,0) < IFNULL(o.order_qty,0) THEN 'MEDIUM' ELSE 'LOW' END,
  623. 'AIDOP', @BatchId, @Now
  624. FROM mdp_std_process_outsource_order o
  625. LEFT JOIN (
  626. SELECT tenant_id, supplier_code, MAX(supplier_name) AS supplier_name
  627. FROM mdp_std_supplier
  628. WHERE IFNULL(supplier_code,'') <> ''
  629. GROUP BY tenant_id, supplier_code
  630. ) s ON o.tenant_id=s.tenant_id AND o.supplier_code=s.supplier_code
  631. WHERE IFNULL(o.work_order,'') <> '' OR IFNULL(o.routing_code,'') <> ''
  632. ON DUPLICATE KEY UPDATE supplier_code=VALUES(supplier_code), supplier_name=VALUES(supplier_name), order_qty=VALUES(order_qty),
  633. completed_qty=VALUES(completed_qty), remaining_qty=VALUES(remaining_qty), due_date=VALUES(due_date),
  634. delivery_status=VALUES(delivery_status), risk_level=VALUES(risk_level), calc_batch_id=VALUES(calc_batch_id),
  635. calc_time=VALUES(calc_time), update_time=CURRENT_TIMESTAMP
  636. """, batchId, now);
  637. }
  638. private async Task<int> UpsertMaterialReadinessDwdAsync(string batchId, DateTime now)
  639. {
  640. return await _db.Ado.ExecuteCommandAsync(
  641. """
  642. INSERT INTO dwd_material_readiness
  643. (tenant_id, stat_date, work_order, op_code, parent_item_code, component_item_code, component_item_name, required_qty, cumulative_required_qty, stock_available_qty, qc_pending_qty, in_transit_qty, delivery_reply_qty, available_qty, shortage_qty, ready_status, supplier_code, supplier_name, need_date, calc_batch_id, calc_time)
  644. SELECT s.tenant_id, @StatDate, s.work_order, s.op_code, s.item_code, s.component_item_code, IFNULL(i.item_name, ''),
  645. IFNULL(s.required_qty, 0), IFNULL(s.required_qty, 0), IFNULL(s.available_qty, 0), 0, IFNULL(s.in_transit_qty, 0),
  646. IFNULL(s.incoming_qty, 0), IFNULL(s.available_qty, 0), IFNULL(s.shortage_qty, 0),
  647. CASE WHEN IFNULL(s.shortage_qty, 0) > 0 THEN 'SHORTAGE' ELSE 'READY' END,
  648. s.supplier_code, IFNULL(sp.supplier_name, ''), s.need_date, @BatchId, @Now
  649. FROM mdp_std_material_readiness s
  650. LEFT JOIN mdp_std_item i ON s.tenant_id=i.tenant_id AND s.component_item_code=i.item_code
  651. LEFT JOIN mdp_std_supplier sp ON s.tenant_id=sp.tenant_id AND s.supplier_code=sp.supplier_code
  652. WHERE IFNULL(s.work_order, '') <> ''
  653. ON DUPLICATE KEY UPDATE component_item_name=VALUES(component_item_name), required_qty=VALUES(required_qty),
  654. cumulative_required_qty=VALUES(cumulative_required_qty), stock_available_qty=VALUES(stock_available_qty),
  655. qc_pending_qty=VALUES(qc_pending_qty), in_transit_qty=VALUES(in_transit_qty), delivery_reply_qty=VALUES(delivery_reply_qty),
  656. available_qty=VALUES(available_qty), shortage_qty=VALUES(shortage_qty), ready_status=VALUES(ready_status),
  657. supplier_code=VALUES(supplier_code), supplier_name=VALUES(supplier_name), need_date=VALUES(need_date),
  658. calc_batch_id=VALUES(calc_batch_id), calc_time=VALUES(calc_time), update_time=CURRENT_TIMESTAMP
  659. """,
  660. new SugarParameter("@StatDate", now.Date),
  661. new SugarParameter("@BatchId", batchId),
  662. new SugarParameter("@Now", now));
  663. }
  664. private async Task<int> DeleteCurrentShortageAsync(DateTime statDate)
  665. {
  666. return await _db.Ado.ExecuteCommandAsync(
  667. """
  668. DELETE s
  669. FROM dwd_material_shortage s
  670. JOIN (
  671. SELECT DISTINCT work_order
  672. FROM mdp_std_material_readiness
  673. WHERE IFNULL(work_order, '') <> ''
  674. ) w ON s.work_order = w.work_order
  675. WHERE s.stat_date = @StatDate
  676. """,
  677. new SugarParameter("@StatDate", statDate));
  678. }
  679. private async Task<int> InsertMaterialShortageAsync(string batchId, DateTime now)
  680. {
  681. return await _db.Ado.ExecuteCommandAsync(
  682. """
  683. INSERT INTO dwd_material_shortage
  684. (tenant_id, stat_date, work_order, op_code, component_item_code, shortage_qty, shortage_reason, expected_supply_date, supplier_code, related_po_no, risk_level, calc_batch_id, calc_time)
  685. SELECT s.tenant_id, @StatDate, s.work_order, s.op_code, s.component_item_code, IFNULL(s.shortage_qty, 0),
  686. '标准层缺料数量大于 0', s.need_date, s.supplier_code, NULL,
  687. CASE WHEN IFNULL(s.shortage_qty, 0) > IFNULL(s.required_qty, 0) THEN 'HIGH'
  688. WHEN IFNULL(s.shortage_qty, 0) > 0 THEN 'MEDIUM' ELSE 'LOW' END,
  689. @BatchId, @Now
  690. FROM mdp_std_material_readiness s
  691. WHERE IFNULL(s.work_order, '') <> '' AND IFNULL(s.shortage_qty, 0) > 0
  692. """,
  693. new SugarParameter("@StatDate", now.Date),
  694. new SugarParameter("@BatchId", batchId),
  695. new SugarParameter("@Now", now));
  696. }
  697. private async Task<long> InsertSyncLogAsync(long entityId, string entityName, string batchId, int rowsRead)
  698. {
  699. await _db.Ado.ExecuteCommandAsync(
  700. """
  701. INSERT INTO mdp_sync_log
  702. (tenant_id, entity_id, source_code, entity_name, sync_batch_id, sync_type, trigger_type, sync_start, rows_read, status)
  703. VALUES (0, @EntityId, 'AIDOPDEV_MYSQL', @EntityName, @BatchId, 'FULL', 'AUTO', NOW(), @RowsRead, 'RUNNING')
  704. """,
  705. new SugarParameter("@EntityId", entityId),
  706. new SugarParameter("@EntityName", entityName),
  707. new SugarParameter("@BatchId", batchId),
  708. new SugarParameter("@RowsRead", rowsRead));
  709. return await _db.Ado.GetLongAsync(
  710. "SELECT id FROM mdp_sync_log WHERE sync_batch_id=@BatchId AND entity_id=@EntityId ORDER BY id DESC LIMIT 1",
  711. new List<SugarParameter>
  712. {
  713. new("@BatchId", batchId),
  714. new("@EntityId", entityId)
  715. });
  716. }
  717. private async Task MarkSyncLogSuccessAsync(long logId, DateTime started, int affected)
  718. {
  719. await _db.Ado.ExecuteCommandAsync(
  720. """
  721. UPDATE mdp_sync_log
  722. SET sync_end=NOW(), duration_ms=@DurationMs, rows_insert=@RowsInsert, rows_update=0, rows_skip=0, rows_error=0, status='SUCCESS'
  723. WHERE id=@Id
  724. """,
  725. new SugarParameter("@DurationMs", (int)(DateTime.Now - started).TotalMilliseconds),
  726. new SugarParameter("@RowsInsert", affected),
  727. new SugarParameter("@Id", logId));
  728. }
  729. private async Task MarkSyncLogFailedAsync(long logId, DateTime started, string message)
  730. {
  731. try
  732. {
  733. await _db.Ado.ExecuteCommandAsync(
  734. """
  735. UPDATE mdp_sync_log
  736. SET sync_end=NOW(), duration_ms=@DurationMs, rows_error=1, status='FAILED', error_msg=@ErrorMsg
  737. WHERE id=@Id
  738. """,
  739. new SugarParameter("@DurationMs", (int)(DateTime.Now - started).TotalMilliseconds),
  740. new SugarParameter("@ErrorMsg", message.Length <= 1000 ? message : message[..1000]),
  741. new SugarParameter("@Id", logId));
  742. }
  743. catch (Exception ex)
  744. {
  745. // 写库自身失败兜底:避免再抛掩盖原异常;遗留 RUNNING 行可由运维手动清理
  746. Console.Error.WriteLine($"[S3MdpSyncTransform] MarkSyncLogFailed write failed (syncLogId={logId}): {ex.Message}");
  747. }
  748. }
  749. private async Task<long> InsertTransformRunLogAsync(string batchId, DateTime startedAt, string triggerType)
  750. {
  751. await _db.Ado.ExecuteCommandAsync(
  752. """
  753. INSERT INTO mdp_transform_run_log
  754. (tenant_id, job_code, job_name, trigger_type, batch_id, status, start_time)
  755. VALUES (0, 'S3_MDP_SYNC_TRANSFORM', 'S3 MDP同步与标准化转换', @TriggerType, @BatchId, 'RUNNING', @StartTime)
  756. """,
  757. new SugarParameter("@TriggerType", NormalizeTriggerType(triggerType)),
  758. new SugarParameter("@BatchId", batchId),
  759. new SugarParameter("@StartTime", startedAt));
  760. return await _db.Ado.GetLongAsync(
  761. "SELECT id FROM mdp_transform_run_log WHERE batch_id=@BatchId ORDER BY id DESC LIMIT 1",
  762. new List<SugarParameter> { new("@BatchId", batchId) });
  763. }
  764. private async Task MarkTransformRunSuccessAsync(long runLogId, DateTime startedAt, S3MdpSyncTransformResult result)
  765. {
  766. var finishedAt = DateTime.Now;
  767. await _db.Ado.ExecuteCommandAsync(
  768. """
  769. UPDATE mdp_transform_run_log
  770. SET status='SUCCESS', end_time=@EndTime, duration_ms=@DurationMs,
  771. stage_rows=@StageRows, standard_rows=@StandardRows, dwd_rows=@DwdRows,
  772. summary_json=@SummaryJson, update_time=CURRENT_TIMESTAMP
  773. WHERE id=@Id
  774. """,
  775. new SugarParameter("@EndTime", finishedAt),
  776. new SugarParameter("@DurationMs", (int)(finishedAt - startedAt).TotalMilliseconds),
  777. new SugarParameter("@StageRows", result.StageRows),
  778. new SugarParameter("@StandardRows", result.StandardRows),
  779. new SugarParameter("@DwdRows", result.DwdRows),
  780. new SugarParameter("@SummaryJson", BuildRunSummaryJson(result)),
  781. new SugarParameter("@Id", runLogId));
  782. }
  783. private async Task MarkTransformRunFailedAsync(long runLogId, DateTime startedAt, string message)
  784. {
  785. try
  786. {
  787. var finishedAt = DateTime.Now;
  788. await _db.Ado.ExecuteCommandAsync(
  789. """
  790. UPDATE mdp_transform_run_log
  791. SET status='FAILED', end_time=@EndTime, duration_ms=@DurationMs,
  792. error_message=@ErrorMessage, update_time=CURRENT_TIMESTAMP
  793. WHERE id=@Id
  794. """,
  795. new SugarParameter("@EndTime", finishedAt),
  796. new SugarParameter("@DurationMs", (int)(finishedAt - startedAt).TotalMilliseconds),
  797. new SugarParameter("@ErrorMessage", Truncate(message, 2000)),
  798. new SugarParameter("@Id", runLogId));
  799. }
  800. catch (Exception ex)
  801. {
  802. // 写库自身失败兜底(典型场景:远端 MySQL 瞬断导致 MarkFailed 自身也连不上)
  803. Console.Error.WriteLine($"[S3MdpSyncTransform] MarkTransformRunFailed write failed (runLogId={runLogId}): {ex.Message}");
  804. }
  805. }
  806. private static S3MdpSqlCommand Cmd(string sql, string batchId, DateTime now)
  807. {
  808. return new S3MdpSqlCommand(sql, new[]
  809. {
  810. new SugarParameter("@BatchId", batchId),
  811. new SugarParameter("@Now", now),
  812. new SugarParameter("@StatDate", now.Date)
  813. });
  814. }
  815. private static string BuildJsonObjectExpression(IEnumerable<string> columns)
  816. {
  817. var parts = columns.SelectMany(c => new[] { $"'{c.Replace("'", "''")}'", $"s.`{c}`" });
  818. return $"JSON_OBJECT({string.Join(",", parts)})";
  819. }
  820. private static string FindColumn(IEnumerable<string> columns, string expected)
  821. {
  822. return columns.First(u => string.Equals(u, expected, StringComparison.OrdinalIgnoreCase));
  823. }
  824. private static string NormalizeTriggerType(string? triggerType)
  825. {
  826. return string.IsNullOrWhiteSpace(triggerType) ? "AUTO" : triggerType.Trim().ToUpperInvariant();
  827. }
  828. private static string BuildRunSummaryJson(S3MdpSyncTransformResult result)
  829. {
  830. return $$"""{"batchId":"{{result.BatchId}}","stageRows":{{result.StageRows}},"standardRows":{{result.StandardRows}},"dwdRows":{{result.DwdRows}},"kpiRows":{{result.KpiRows}}}""";
  831. }
  832. private static string ResolveKpiValueTable(int metricLevel)
  833. {
  834. return metricLevel switch
  835. {
  836. 1 => "ado_s9_kpi_value_l1_day",
  837. 2 => "ado_s9_kpi_value_l2_day",
  838. 3 => "ado_s9_kpi_value_l3_day",
  839. 4 => "ado_s9_kpi_value_l4_day",
  840. _ => "ado_s9_kpi_value_l2_day"
  841. };
  842. }
  843. private static decimal DefaultS3Target(string metricCode)
  844. {
  845. return metricCode switch
  846. {
  847. "S3_L1_001" => 15m,
  848. "S3_L1_002" => 95m,
  849. "S3_L2_004" => 10.5m,
  850. "S3_L2_005" => 96.72m,
  851. _ => 0m
  852. };
  853. }
  854. private static string ResolveKpiStatus(decimal actual, decimal target, string? direction, decimal? yellowThreshold, decimal? redThreshold)
  855. {
  856. if (target <= 0) return "gray";
  857. var ratio = actual / target * 100m;
  858. if (string.Equals(direction, "lower_is_better", StringComparison.OrdinalIgnoreCase))
  859. {
  860. if (actual <= target) return "green";
  861. if (ratio <= (yellowThreshold ?? 110m)) return "yellow";
  862. return ratio >= (redThreshold ?? 120m) ? "red" : "yellow";
  863. }
  864. if (actual >= target) return "green";
  865. if (ratio >= (yellowThreshold ?? 95m)) return "yellow";
  866. return ratio <= (redThreshold ?? 80m) ? "red" : "yellow";
  867. }
  868. private static string ResolveTrendFlag(decimal actual, decimal? previous)
  869. {
  870. if (previous == null) return "flat";
  871. if (actual > previous.Value) return "up";
  872. if (actual < previous.Value) return "down";
  873. return "flat";
  874. }
  875. private static string Truncate(string? raw, int maxLength)
  876. {
  877. if (string.IsNullOrEmpty(raw)) return string.Empty;
  878. return raw.Length <= maxLength ? raw : raw[..maxLength];
  879. }
  880. private sealed class S3ColumnRow
  881. {
  882. public string ColumnName { get; set; } = string.Empty;
  883. }
  884. private sealed class S3MdpEntityRow
  885. {
  886. public long Id { get; set; }
  887. public string EntityName { get; set; } = string.Empty;
  888. }
  889. private sealed class S3KpiCalcRow
  890. {
  891. public long TenantId { get; set; }
  892. public long FactoryId { get; set; }
  893. public string MetricCode { get; set; } = string.Empty;
  894. public decimal? MetricValue { get; set; }
  895. }
  896. private sealed class S3KpiMetaRow
  897. {
  898. public int MetricLevel { get; set; }
  899. public string Direction { get; set; } = "higher_is_better";
  900. public decimal? YellowThreshold { get; set; }
  901. public decimal? RedThreshold { get; set; }
  902. }
  903. private sealed class S3KpiValueRow
  904. {
  905. public long Id { get; set; }
  906. public decimal? MetricValue { get; set; }
  907. public decimal? TargetValue { get; set; }
  908. }
  909. }
  910. public sealed class S3MdpSyncTransformResult
  911. {
  912. public long RunLogId { get; set; }
  913. public string BatchId { get; set; } = string.Empty;
  914. public int StageRows { get; set; }
  915. public int StandardRows { get; set; }
  916. public int DwdRows { get; set; }
  917. public int KpiRows { get; set; }
  918. }
  919. public sealed class S3MaterialRefreshResult
  920. {
  921. public string BatchId { get; set; } = string.Empty;
  922. public int StdCount { get; set; }
  923. public int ReadinessCount { get; set; }
  924. public int ShortageCount { get; set; }
  925. }
  926. internal sealed record S3MdpSqlCommand(string Sql, SugarParameter[] Parameters);
  927. internal sealed record S3MdpEntityConfig(
  928. string EntityCode,
  929. string SourceTable,
  930. string TargetTable,
  931. string SourceRowIdExpression,
  932. string SourceBizKeyExpression)
  933. {
  934. public static readonly IReadOnlyList<S3MdpEntityConfig> All = new List<S3MdpEntityConfig>
  935. {
  936. new("S3_SUPPLIER", "SuppMaster", "mdp_stg_supplier", "RecID", "CONCAT(IFNULL(s.`Domain`,''), ':', IFNULL(s.`Supp`,''))"),
  937. new("S3_CONSIGNEE_SUPPLIER", "ConsigneeAddressMaster", "mdp_stg_supplier", "RecID", "CONCAT(IFNULL(s.`Domain`,''), ':', IFNULL(s.`Address`,''))"),
  938. new("S3_ITEM_ERP", "ItemMaster", "mdp_stg_item", "RecID", "CONCAT(IFNULL(s.`Domain`,''), ':', IFNULL(s.`ItemNum`,''))"),
  939. new("S3_ITEM_NEW", "ic_item", "mdp_stg_item", "Id", "CONCAT(IFNULL(s.`tenant_id`,0), ':', IFNULL(s.`number`,''))"),
  940. new("S3_SOURCE_LIST", "srm_purchase", "mdp_stg_source_list", "Id", "CONCAT(IFNULL(s.`number`,''), ':', IFNULL(s.`supplier_number`,''))"),
  941. new("S3_SUPPLY_DEMAND", "ic_demandschedule", "mdp_stg_supply_demand", "Id", "CAST(s.`Id` AS CHAR)"),
  942. new("S3_PURCHASE_REQUEST", "srm_pr_main", "mdp_stg_supply_demand", "Id", "COALESCE(s.`pr_billno`, CAST(s.`Id` AS CHAR))"),
  943. new("S3_PURCHASE_ORDER_MASTER", "PurOrdMaster", "mdp_stg_purchase_order", "RecID", "CONCAT(IFNULL(s.`Domain`,''), ':', IFNULL(s.`PurOrd`,''))"),
  944. new("S3_PURCHASE_ORDER_DETAIL", "PurOrdDetail", "mdp_stg_purchase_order", "RecID", "CONCAT(IFNULL(s.`Domain`,''), ':', IFNULL(s.`PurOrd`,''), ':', IFNULL(s.`Line`,''))"),
  945. new("S3_DELIVERY_PLAN", "srm_polist_ds", "mdp_stg_delivery", "Id", "s.`dsnum`"),
  946. new("S3_SHIPPER_MASTER", "scm_shd", "mdp_stg_delivery", "id", "COALESCE(s.`shddh`, CAST(s.`id` AS CHAR))"),
  947. new("S3_SHIPPER_DETAIL", "scm_shdzb", "mdp_stg_delivery", "id", "CONCAT(IFNULL(s.`glid`,''), ':', IFNULL(s.`id`,''))"),
  948. new("S3_RECEIPT_MASTER", "PurOrdRctMaster", "mdp_stg_receipt", "RecID", "CONCAT(IFNULL(s.`Domain`,''), ':', IFNULL(s.`Receiver`,''))"),
  949. new("S3_RECEIPT_DETAIL", "PurOrdRctDetail", "mdp_stg_receipt", "RecID", "CONCAT(IFNULL(s.`Domain`,''), ':', IFNULL(s.`Receiver`,''), ':', IFNULL(s.`Line`,''))"),
  950. new("S3_WORK_ORDER_MASTER", "WorkOrdMaster", "mdp_stg_work_order_material", "RecID", "CONCAT(IFNULL(s.`Domain`,''), ':', IFNULL(s.`WorkOrd`,''))"),
  951. new("S3_WORK_ORDER_DETAIL", "WorkOrdDetail", "mdp_stg_work_order_material", "RecID", "CONCAT(IFNULL(s.`Domain`,''), ':', IFNULL(s.`WorkOrd`,''), ':', IFNULL(s.`Op`,''), ':', IFNULL(s.`ItemNum`,''))"),
  952. new("S3_WORK_ORDER_ROUTING", "WorkOrdRouting", "mdp_stg_work_order_material", "RecID", "CONCAT(IFNULL(s.`Domain`,''), ':', IFNULL(s.`WorkOrd`,''), ':', IFNULL(s.`OP`,''))"),
  953. new("S3_ROUTING_OUTSOURCE", "RoutingOpDetail", "mdp_stg_work_order_material", "RecID", "CONCAT(IFNULL(s.`Domain`,''), ':', IFNULL(s.`RoutingCode`,''), ':', IFNULL(s.`Op`,''), ':', IFNULL(s.`SupplierCode`,''))"),
  954. new("S3_INVENTORY", "InvMaster", "mdp_stg_work_order_material", "RecID", "CONCAT(IFNULL(s.`Domain`,''), ':', IFNULL(s.`ItemNum`,''), ':', IFNULL(s.`Location`,''))")
  955. };
  956. }