using System.Text.Json; using Admin.NET.Plugin.AiDOP.DataPlatform; using Admin.NET.Plugin.AiDOP.DataPlatform.Executors; using Admin.NET.Plugin.AiDOP.Infrastructure; 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"; /// T8 报工源无 tenant_id;须经 解析。 private const string SourceZtid = "pbxfxp"; 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(); // T8 源表无 tenant_id 列;显式 tenantId≤0 时经账套映射解析真实租户。 var resolvedTenantId = AidopSourceTenantMap.ResolveTenantId(SourceZtid, tenantId); var code = string.IsNullOrWhiteSpace(entityCode) ? InboundEntityCode : entityCode.Trim(); var now = DateTime.Now; var pullCtx = new MdpPullContext { TenantId = resolvedTenantId, 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(resolvedTenantId, 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 字段兼容大小写) var sTenant = MdpJsonSql.TenantFromStg("s", "@tid"); var 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 {sTenant}, 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 <> '' AND {MdpJsonSql.TenantGuard(sTenant)} 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 MdpSchemaAligner.ExecuteAsync(_db, sql, new SugarParameter("@tid", tenantId), new SugarParameter("@batch", batchId), new SugarParameter("@now", now)); await MdpSchemaAligner.ExecuteAsync(_db, tenantId > 0 ? "UPDATE mdp_stg_s6_report SET process_status='DONE', update_time=NOW() WHERE process_status='PENDING' AND tenant_id = @tid" : "UPDATE mdp_stg_s6_report SET process_status='DONE', update_time=NOW() WHERE process_status='PENDING'", tenantId > 0 ? new SugarParameter("@tid", tenantId) : null!); return affected; } private async Task EnsureTablesAsync() { await MdpSchemaAligner.ExecuteAsync(_db, MdpSchemaDefinition.Ddl("mdp_stg_s6_report")); await MdpSchemaAligner.ExecuteAsync(_db, MdpSchemaDefinition.Ddl("mdp_std_s6_report")); } } 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; } }