using System.Text.Json; using Admin.NET.Plugin.AiDOP.DataPlatform.Executors; namespace Admin.NET.Plugin.AiDOP.Manufacturing; /// /// S6 报工双模式入站:执行器 → mdp_stg_s6_report → mdp_std_s6_report。 /// 源优先 T8 Cj_Bg_Head_Rep(实体 S6_REPORT);API 对偶 S6_REPORT_API。 /// public class ReportWorkMdpSyncService : ITransient { private const string InboundEntityCode = "S6_REPORT"; private readonly ISqlSugarClient _db; private readonly MdpSourcePullDispatcher _pullDispatcher; public ReportWorkMdpSyncService(ISqlSugarClient db, MdpSourcePullDispatcher pullDispatcher) { _db = db; _pullDispatcher = pullDispatcher; } public async Task RunInboundAsync( long tenantId = 0, bool fullRefresh = false, string? entityCode = null, CancellationToken cancellationToken = default) { cancellationToken.ThrowIfCancellationRequested(); await EnsureTablesAsync(); var code = string.IsNullOrWhiteSpace(entityCode) ? InboundEntityCode : entityCode.Trim(); var now = DateTime.Now; var pullCtx = new MdpPullContext { TenantId = tenantId, FullRefresh = fullRefresh, TaskCode = "S6_REPORT_INBOUND", BatchId = $"S6_RPT_IN_{now:yyyyMMddHHmmss}" }; var pull = await _pullDispatcher.PullByEntityCodeAsync(code, pullCtx, cancellationToken); var transformBatch = $"{pullCtx.BatchId}_STD"; var stdRows = await TransformStandardAsync(tenantId, transformBatch, now); return new ReportWorkInboundResult { PullBatchId = pullCtx.BatchId, RowsPulled = pull.RowsWritten, RowsWrittenStg = pull.RowsWritten, TransformBatchId = transformBatch, StandardRows = stdRows }; } private async Task TransformStandardAsync(long tenantId, string batchId, DateTime now) { // 将 PENDING stg 投影到 std(JSON 字段兼容大小写) const string sql = """ INSERT INTO mdp_std_s6_report (tenant_id, factory_id, source_system, work_order_no, report_date, report_qty, ztid, source_row_id, source_biz_key, sync_batch_id, sync_time) SELECT IFNULL(s.tenant_id, @tid), 1, IFNULL(NULLIF(s.source_system,''), 'T8'), IFNULL(NULLIF(JSON_UNQUOTE(JSON_EXTRACT(s.raw_data, '$.noid')), 'null'), IFNULL(NULLIF(JSON_UNQUOTE(JSON_EXTRACT(s.raw_data, '$.NOID')), 'null'), s.source_biz_key)), COALESCE( STR_TO_DATE(NULLIF(NULLIF(JSON_UNQUOTE(JSON_EXTRACT(s.raw_data, '$.kgdate')), 'null'), ''), '%Y-%m-%d %H:%i:%s'), STR_TO_DATE(NULLIF(NULLIF(JSON_UNQUOTE(JSON_EXTRACT(s.raw_data, '$.KGDATE')), 'null'), ''), '%Y-%m-%d %H:%i:%s'), NULL), CAST(NULLIF(NULLIF(JSON_UNQUOTE(JSON_EXTRACT(s.raw_data, '$.sl')), 'null'), '') AS DECIMAL(18,6)), NULLIF(JSON_UNQUOTE(JSON_EXTRACT(s.raw_data, '$.ztid')), 'null'), IFNULL(NULLIF(s.source_row_id,''), s.source_biz_key), s.source_biz_key, @batch, @now FROM mdp_stg_s6_report s WHERE s.process_status = 'PENDING' AND s.source_biz_key IS NOT NULL AND s.source_biz_key <> '' ON DUPLICATE KEY UPDATE work_order_no = VALUES(work_order_no), report_date = VALUES(report_date), report_qty = VALUES(report_qty), ztid = VALUES(ztid), source_row_id = VALUES(source_row_id), sync_batch_id = VALUES(sync_batch_id), sync_time = VALUES(sync_time), update_time = CURRENT_TIMESTAMP """; var affected = await _db.Ado.ExecuteCommandAsync(sql, new SugarParameter("@tid", tenantId), new SugarParameter("@batch", batchId), new SugarParameter("@now", now)); await _db.Ado.ExecuteCommandAsync( "UPDATE mdp_stg_s6_report SET process_status='DONE', update_time=NOW() WHERE process_status='PENDING'"); return affected; } private async Task EnsureTablesAsync() { await _db.Ado.ExecuteCommandAsync(""" CREATE TABLE IF NOT EXISTS mdp_stg_s6_report ( id bigint NOT NULL AUTO_INCREMENT, tenant_id bigint NOT NULL DEFAULT 0, source_system varchar(50) DEFAULT NULL, source_table varchar(200) DEFAULT NULL, source_row_id varchar(200) DEFAULT NULL, source_biz_key varchar(300) DEFAULT NULL, raw_data json DEFAULT NULL, sync_batch_id varchar(100) DEFAULT NULL, create_time datetime NOT NULL DEFAULT CURRENT_TIMESTAMP, sync_time datetime DEFAULT CURRENT_TIMESTAMP, process_status varchar(20) NOT NULL DEFAULT 'PENDING', process_message varchar(500) DEFAULT NULL, update_time datetime DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, PRIMARY KEY (id), UNIQUE KEY uk_source_key (source_system, source_table, source_biz_key) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 """); await _db.Ado.ExecuteCommandAsync(""" CREATE TABLE IF NOT EXISTS mdp_std_s6_report ( id bigint NOT NULL AUTO_INCREMENT, tenant_id bigint NOT NULL DEFAULT 0, factory_id bigint DEFAULT 1, source_system varchar(50) NOT NULL DEFAULT 'T8', work_order_no varchar(100) NOT NULL, report_date datetime DEFAULT NULL, report_qty decimal(18,6) DEFAULT NULL, ztid varchar(50) DEFAULT NULL, source_row_id varchar(100) NOT NULL, source_biz_key varchar(200) NOT NULL, sync_batch_id varchar(100) NOT NULL, sync_time datetime NOT NULL, update_time datetime DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, PRIMARY KEY (id), UNIQUE KEY uk_mdp_std_s6_report (tenant_id, source_system, source_biz_key) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 """); } } public class ReportWorkInboundResult { public string PullBatchId { get; set; } = string.Empty; public int RowsPulled { get; set; } public int RowsWrittenStg { get; set; } public string TransformBatchId { get; set; } = string.Empty; public int StandardRows { get; set; } }