AiDOPWarehouse/server/Plugins/Admin.NET.Plugin.AiDOP/ProcurementExecution/S4MdpSyncTransformService.cs

using Admin.NET.Plugin.AiDOP.DataPlatform.Executors;
using Admin.NET.Plugin.AiDOP.DataPlatform.MdpRebuild;
using Admin.NET.Plugin.AiDOP.Infrastructure;
using Admin.NET.Plugin.AiDOP.SmartOps;

namespace Admin.NET.Plugin.AiDOP.ProcurementExecution;

/// <summary>
/// S4 采购执行 MDP 转换：S4 专属 STG/STD + 消费 S3 共享 DWD，写入 dwd_s4_purchase_execution / dwd_po_trans 与 S4 KPI。
/// Phase2：源→stg 经统一执行器；std/DWD/KPI 仍读 stg。
/// </summary>
public class S4MdpSyncTransformService : ITransient
{
    private const string JobCode = "S4_MDP_SYNC_TRANSFORM";

    private readonly ISqlSugarClient _db;
    private readonly TransformRunLogFinalizer _runLogFinalizer;
    private readonly MdpModuleStagingPuller _stagingPuller;
    private readonly IKpiTargetResolver _kpiTargetResolver;
    private MdpRebuildScope _runScope = null!;

    public S4MdpSyncTransformService(ISqlSugarClient db, MdpModuleStagingPuller stagingPuller, IKpiTargetResolver kpiTargetResolver, TransformRunLogFinalizer runLogFinalizer)
    {
        _db = db;
        _runLogFinalizer = runLogFinalizer;
        _stagingPuller = stagingPuller;
        _kpiTargetResolver = kpiTargetResolver;
    }

    public Task<S4MdpSyncTransformResult> RunFullAsync(CancellationToken cancellationToken = default, string triggerType = "AUTO") =>
        throw new InvalidOperationException("S4 全量重算必须指定租户/工厂作用域，请走 ModuleRebuildService.Enqueue");

    public async Task<S4MdpSyncTransformResult> RunFullAsync(
        MdpRebuildScope scope,
        CancellationToken cancellationToken = default,
        string triggerType = "AUTO",
        long? rebuildJobId = null,
        Func<ModuleProgressUpdate, Task>? reportProgress = null)
    {
        _ = rebuildJobId;
        _runScope = MdpRebuildScope.Create("S4", scope.TenantId, scope.FactoryId);
        cancellationToken.ThrowIfCancellationRequested();
        await EnsureS4TablesAsync();
        var now = DateTime.Now;
        var batchId = $"S4_MDP_FULL_{_runScope.TenantId}_{_runScope.FactoryId}_{now:yyyyMMddHHmmss}";
        var runLogId = await InsertTransformRunLogAsync(batchId, now, triggerType);
        var result = new S4MdpSyncTransformResult { BatchId = batchId, RunLogId = runLogId };

        try
        {
            await ReportAsync(reportProgress, new ModuleProgressUpdate(ModuleRebuildStages.Preparing, 0, 5, "准备运行环境"));
            await ReportAsync(reportProgress, new ModuleProgressUpdate(ModuleRebuildStages.Staging, 1, 12, "正在拉取源数据"));
            result.StageRows = await SyncStagingAsync(batchId, now, cancellationToken);
            await ReportAsync(reportProgress, new ModuleProgressUpdate(ModuleRebuildStages.Staging, 1, 35, "拉取源数据完成", result.StageRows, ModuleRebuildStages.Staging));

            await ReportAsync(reportProgress, new ModuleProgressUpdate(ModuleRebuildStages.Standard, 2, 40, "正在标准化数据"));
            result.StandardRows = await TransformStandardAsync(batchId, now, cancellationToken);
            await ReportAsync(reportProgress, new ModuleProgressUpdate(ModuleRebuildStages.Standard, 2, 55, "标准化数据完成", result.StandardRows, ModuleRebuildStages.Standard));

            await ReportAsync(reportProgress, new ModuleProgressUpdate(ModuleRebuildStages.Dwd, 3, 60, "正在生成 DWD 明细"));
            result.DwdRows = await BuildDwdAsync(batchId, now, cancellationToken);
            await ReportAsync(reportProgress, new ModuleProgressUpdate(ModuleRebuildStages.Dwd, 3, 75, "生成 DWD 明细完成", result.DwdRows, ModuleRebuildStages.Dwd));

            await ReportAsync(reportProgress, new ModuleProgressUpdate(ModuleRebuildStages.Kpi, 4, 80, "正在重算 KPI"));
            result.KpiRows = await BuildS4KpiValuesAsync(now, cancellationToken);
            await ReportAsync(reportProgress, new ModuleProgressUpdate(ModuleRebuildStages.Kpi, 4, 96, "重算 KPI 完成", result.KpiRows, ModuleRebuildStages.Kpi));

            await ReportAsync(reportProgress, new ModuleProgressUpdate(ModuleRebuildStages.Finalizing, 4, 99, "正在收尾"));
            await MarkTransformRunSuccessAsync(runLogId, now, result);
            return result;
        }
        catch (Exception ex)
        {
            // 宿主关停不是转换失败：交给 finally 收口为 ABORTED，不污染 FAILED 语义。
            if (!_runLogFinalizer.IsHostStopping)
                await MarkTransformRunFailedAsync(runLogId, now, ex.Message);
            throw;
        }
        finally
        {
            await _runLogFinalizer.FinalizeIfHostStoppingAsync(runLogId, now);
        }
    }

    private static async Task ReportAsync(Func<ModuleProgressUpdate, Task>? report, ModuleProgressUpdate update)
    {
        if (report == null) return;
        await report(update);
    }

    private async Task EnsureS4TablesAsync()
    {
        await MdpSchemaAligner.ExecuteAsync(_db, MdpSchemaDefinition.Ddl("mdp_stg_s4_iqc"));
        await MdpSchemaAligner.ExecuteAsync(_db, MdpSchemaDefinition.Ddl("mdp_stg_s4_shipment"));
        await MdpSchemaAligner.ExecuteAsync(_db, MdpSchemaDefinition.Ddl("mdp_stg_s4_return"));
        await MdpSchemaAligner.ExecuteAsync(_db, MdpSchemaDefinition.Ddl("mdp_stg_s4_shortage"));
        await MdpSchemaAligner.ExecuteAsync(_db, MdpSchemaDefinition.Ddl("mdp_std_s4_iqc"));
        await MdpSchemaAligner.ExecuteAsync(_db, MdpSchemaDefinition.Ddl("mdp_std_s4_shipment"));
        await MdpSchemaAligner.ExecuteAsync(_db, MdpSchemaDefinition.Ddl("mdp_std_s4_return"));
        await MdpSchemaAligner.ExecuteAsync(_db, MdpSchemaDefinition.Ddl("mdp_std_s4_shortage"));
        await MdpSchemaAligner.ExecuteAsync(_db, MdpSchemaDefinition.Ddl("dwd_s4_purchase_execution"));
    }

    private async Task<int> SyncStagingAsync(string batchId, DateTime now, CancellationToken cancellationToken)
    {
        _ = now;
        EnsureRunScope();
        var total = await _stagingPuller.PullEntitiesAsync(
            S4MdpEntityConfig.All.Select(x => x.EntityCode),
            batchId, _runScope.TenantId, fullRefresh: true, taskCode: "S4_MDP_INBOUND", cancellationToken,
            factoryId: _runScope.FactoryId, requireMatchingSourceTenant: true);
        total += await CountSharedStagingRowsAsync();
        return total;
    }

    public async Task<S4MdpSyncTransformResult> RunInboundAsync(
        IEnumerable<string>? entityCodes = null,
        long tenantId = 0,
        bool fullRefresh = false,
        CancellationToken cancellationToken = default)
    {
        cancellationToken.ThrowIfCancellationRequested();
        if (tenantId <= 0)
            throw new InvalidOperationException("S4 inbound 必须指定有效 tenantId");
        _runScope = MdpRebuildScope.Create("S4", tenantId, 1);
        await EnsureS4TablesAsync();
        var now = DateTime.Now;
        var batchId = $"S4_MDP_IN_{_runScope.TenantId}_{now:yyyyMMddHHmmss}";
        var runLogId = await InsertTransformRunLogAsync(batchId, now, "INBOUND");
        var result = new S4MdpSyncTransformResult { BatchId = batchId, RunLogId = runLogId };
        try
        {
            var codes = entityCodes?.ToList() ?? S4MdpEntityConfig.All.Select(x => x.EntityCode).ToList();
            result.StageRows = await _stagingPuller.PullEntitiesAsync(
                codes, batchId, tenantId, fullRefresh, "S4_MDP_INBOUND", cancellationToken,
                factoryId: _runScope.FactoryId);
            result.StageRows += await CountSharedStagingRowsAsync();
            result.StandardRows = await TransformStandardAsync(batchId, now, cancellationToken);
            result.DwdRows = await BuildDwdAsync(batchId, now, cancellationToken);
            result.KpiRows = await BuildS4KpiValuesAsync(now, cancellationToken);
            await MarkTransformRunSuccessAsync(runLogId, now, result);
            return result;
        }
        catch (Exception ex)
        {
            // 宿主关停不是转换失败：交给 finally 收口为 ABORTED，不污染 FAILED 语义。
            if (!_runLogFinalizer.IsHostStopping)
                await MarkTransformRunFailedAsync(runLogId, now, ex.Message);
            throw;
        }
        finally
        {
            await _runLogFinalizer.FinalizeIfHostStoppingAsync(runLogId, now);
        }
    }

    private async Task<int> CountSharedStagingRowsAsync()
    {
        try
        {
            return await _db.Ado.GetIntAsync(
                """
                SELECT IFNULL(SUM(cnt), 0) FROM (
                    SELECT COUNT(1) AS cnt FROM mdp_stg_purchase_order
                    UNION ALL SELECT COUNT(1) FROM mdp_stg_delivery
                    UNION ALL SELECT COUNT(1) FROM mdp_stg_receipt
                ) t
                """);
        }
        catch
        {
            return 0;
        }
    }

    [Obsolete("Phase3: use MdpModuleStagingPuller / executors")]
    private async Task<int> SyncOneEntityAsync(S4MdpEntityConfig entity, string batchId, DateTime now)
    {
        var entityRow = await _db.Ado.SqlQuerySingleAsync<S4MdpEntityRow>(
            "SELECT id AS Id, entity_name AS EntityName FROM mdp_entity WHERE tenant_id=0 AND entity_code=@EntityCode LIMIT 1",
            new SugarParameter("@EntityCode", entity.EntityCode));
        if (entityRow == null)
            throw Oops.Oh($"未找到 MDP 实体配置：{entity.EntityCode}，请先执行 1.0.154.sql 或启动迁移。");

        var tableExists = await _db.Ado.GetIntAsync(
            """
            SELECT COUNT(1) FROM information_schema.TABLES
            WHERE TABLE_SCHEMA=DATABASE() AND TABLE_NAME=@TableName
            """,
            new SugarParameter("@TableName", entity.SourceTable));
        if (tableExists == 0)
        {
            if (entity.Optional)
                return 0;
            throw Oops.Oh($"未找到 S4 源表：{entity.SourceTable}（实体 {entity.EntityCode}）");
        }

        var columns = await _db.Ado.SqlQueryAsync<S4ColumnRow>(
            """
            SELECT COLUMN_NAME AS ColumnName
            FROM information_schema.COLUMNS
            WHERE TABLE_SCHEMA=DATABASE() AND TABLE_NAME=@TableName
            ORDER BY ORDINAL_POSITION
            """,
            new SugarParameter("@TableName", entity.SourceTable));
        if (columns.Count == 0)
            throw Oops.Oh($"未找到源表字段：{entity.SourceTable}");

        var names = columns.Select(u => u.ColumnName).ToList();
        var tenantExpr = names.Any(u => string.Equals(u, "tenant_id", StringComparison.OrdinalIgnoreCase))
            ? $"IFNULL(s.`{FindColumn(names, "tenant_id")}`,0)"
            : "0";
        var sourceRowExpr = names.Any(u => string.Equals(u, entity.SourceRowIdExpression, StringComparison.OrdinalIgnoreCase))
            ? $"s.`{FindColumn(names, entity.SourceRowIdExpression)}`"
            : entity.SourceRowIdExpression;
        var rawDataExpr = BuildJsonObjectExpression(names);

        var rowsRead = await _db.Ado.GetIntAsync($"SELECT COUNT(1) FROM `{entity.SourceTable}`");
        var logId = await InsertSyncLogAsync(entityRow.Id, entityRow.EntityName, batchId, rowsRead);
        var started = DateTime.Now;

        try
        {
            var affected = await MdpSchemaAligner.ExecuteAsync(_db, 
                $"""
                INSERT INTO `{entity.TargetTable}`
                (tenant_id, source_system, source_table, source_row_id, source_biz_key, sync_batch_id, sync_time, process_status, raw_data)
                SELECT
                    {tenantExpr},
                    COALESCE((SELECT r.source_system FROM mdp_source_table_registry r WHERE r.source_table = @SourceTable LIMIT 1), NULLIF('',''), ''),
                    @SourceTable,
                    CAST({sourceRowExpr} AS CHAR),
                    CAST(COALESCE({entity.SourceBizKeyExpression}, CAST({sourceRowExpr} AS CHAR)) AS CHAR),
                    @BatchId,
                    @Now,
                    'PENDING',
                    {rawDataExpr}
                FROM `{entity.SourceTable}` s
                ON DUPLICATE KEY UPDATE
                    source_row_id=VALUES(source_row_id),
                    sync_batch_id=VALUES(sync_batch_id),
                    sync_time=VALUES(sync_time),
                    process_status=VALUES(process_status),
                    raw_data=VALUES(raw_data),
                    update_time=CURRENT_TIMESTAMP
                """,
                new SugarParameter("@SourceTable", entity.SourceTable),
                new SugarParameter("@BatchId", batchId),
                new SugarParameter("@Now", now));

            await MarkSyncLogSuccessAsync(logId, started, affected);
            return rowsRead;
        }
        catch (Exception ex)
        {
            await MarkSyncLogFailedAsync(logId, started, ex.Message);
            if (entity.Optional) return 0;
            throw;
        }
    }

    private async Task<int> TransformStandardAsync(string batchId, DateTime now, CancellationToken cancellationToken)
    {
        var total = 0;
        foreach (var command in BuildStandardCommands(batchId, now))
        {
            cancellationToken.ThrowIfCancellationRequested();
            try
            {
                total += await MdpSchemaAligner.ExecuteAsync(_db, command.Sql, command.Parameters);
            }
            catch
            {
                // 标准层表可能尚未有贴源数据，跳过单条失败
            }
        }
        total += await CountSharedStandardRowsAsync();
        return total;
    }

    private async Task<int> CountSharedStandardRowsAsync()
    {
        try
        {
            return await _db.Ado.GetIntAsync(
                """
                SELECT IFNULL(SUM(cnt), 0) FROM (
                    SELECT COUNT(1) AS cnt FROM mdp_std_purchase_order
                    UNION ALL SELECT COUNT(1) FROM mdp_std_delivery_schedule
                    UNION ALL SELECT COUNT(1) FROM mdp_std_delivery_result
                ) t
                """);
        }
        catch
        {
            return 0;
        }
    }

    private IEnumerable<S4MdpSqlCommand> BuildStandardCommands(string batchId, DateTime now)
    {
        yield return Cmd(
            """
            INSERT INTO mdp_std_s4_iqc
            (tenant_id, factory_id, source_system, po_no, po_line, supplier_code, item_code, receipt_qty, sample_qty, defect_qty, qc_result, receipt_date, source_biz_key, sync_batch_id, sync_time)
            SELECT tenant_id, COALESCE(NULLIF(factory_id,0),1), COALESCE((SELECT r.source_system FROM mdp_source_table_registry r WHERE r.source_table = source_table LIMIT 1), NULLIF(source_system,''), ''),
                   COALESCE(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.PurOrd')), JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.OrdNbr'))),
                   CAST(COALESCE(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.OrdLine')), JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.Line'))) AS CHAR),
                   IFNULL(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.Supp')), ''),
                   IFNULL(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.ItemNum')), ''),
                   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),
                   COALESCE(CASE WHEN COALESCE(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.SampleQty')), JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.QcQty'))) REGEXP '^-?[0-9]+(\\.[0-9]+)?$' THEN CAST(COALESCE(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.SampleQty')), JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.QcQty'))) AS DECIMAL(18,6)) END, 0),
                   COALESCE(CASE WHEN COALESCE(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.RejectQty')), JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.QtyReturn'))) REGEXP '^-?[0-9]+(\\.[0-9]+)?$' THEN CAST(COALESCE(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.RejectQty')), JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.QtyReturn'))) AS DECIMAL(18,6)) END, 0),
                   CASE WHEN COALESCE(CASE WHEN COALESCE(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.RejectQty')), JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.QtyReturn'))) REGEXP '^-?[0-9]+(\\.[0-9]+)?$' THEN CAST(COALESCE(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.RejectQty')), JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.QtyReturn'))) AS DECIMAL(18,6)) END, 0) > 0 THEN 'FAIL' ELSE 'PASS' END,
                   NULLIF(NULLIF(COALESCE(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.RcptDate')), JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.RctDate'))), 'null'), ''),
                   source_biz_key, @BatchId, @Now
            FROM mdp_stg_s4_iqc
            WHERE source_table='PurOrdRctDetail'
            ON DUPLICATE KEY UPDATE receipt_qty=VALUES(receipt_qty), sample_qty=VALUES(sample_qty), defect_qty=VALUES(defect_qty),
                qc_result=VALUES(qc_result), receipt_date=VALUES(receipt_date), sync_batch_id=VALUES(sync_batch_id), sync_time=VALUES(sync_time), update_time=CURRENT_TIMESTAMP
            """, batchId, now);

        yield return Cmd(
            """
            INSERT INTO mdp_std_s4_shipment
            (tenant_id, factory_id, source_system, shipment_no, po_no, po_line, supplier_code, item_code, ship_qty, ship_date, source_biz_key, sync_batch_id, sync_time)
            SELECT tenant_id, COALESCE(NULLIF(factory_id,0),1), COALESCE((SELECT r.source_system FROM mdp_source_table_registry r WHERE r.source_table = source_table LIMIT 1), NULLIF(source_system,''), ''),
                   COALESCE(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.shddh')), source_row_id),
                   JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.po_bill')),
                   CAST(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.po_billline')) AS CHAR),
                   IFNULL(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.suppliercode')), ''),
                   IFNULL(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.itemnum')), ''),
                   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),
                   NULLIF(NULLIF(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.updatetime')), 'null'), ''),
                   source_biz_key, @BatchId, @Now
            FROM mdp_stg_s4_shipment
            WHERE source_table='scm_shdzb'
            ON DUPLICATE KEY UPDATE ship_qty=VALUES(ship_qty), ship_date=VALUES(ship_date), sync_batch_id=VALUES(sync_batch_id), sync_time=VALUES(sync_time), update_time=CURRENT_TIMESTAMP
            """, batchId, now);

        yield return Cmd(
            """
            INSERT INTO mdp_std_s4_return
            (tenant_id, factory_id, source_system, po_no, po_line, supplier_code, item_code, return_qty, return_reason, return_status, source_biz_key, sync_batch_id, sync_time)
            SELECT tenant_id, COALESCE(NULLIF(factory_id,0),1), COALESCE((SELECT r.source_system FROM mdp_source_table_registry r WHERE r.source_table = source_table LIMIT 1), NULLIF(source_system,''), ''),
                   JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.ponumber')),
                   CAST(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.poline')) AS CHAR),
                   IFNULL(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.suppliercode')), ''),
                   IFNULL(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.itemnum')), ''),
                   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),
                   JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.remark')),
                   JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.status')),
                   source_biz_key, @BatchId, @Now
            FROM mdp_stg_s4_return
            WHERE source_table='srm_polist_ds'
              AND 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) > 0
            ON DUPLICATE KEY UPDATE return_qty=VALUES(return_qty), return_reason=VALUES(return_reason), return_status=VALUES(return_status),
                sync_batch_id=VALUES(sync_batch_id), sync_time=VALUES(sync_time), update_time=CURRENT_TIMESTAMP
            """, batchId, now);

        yield return Cmd(
            """
            INSERT INTO mdp_std_s4_shortage
            (tenant_id, factory_id, source_system, work_order, supplier_code, item_code, shortage_qty, risk_level, need_date, source_biz_key, sync_batch_id, sync_time)
            SELECT tenant_id, COALESCE(NULLIF(factory_id,0),1), COALESCE((SELECT r.source_system FROM mdp_source_table_registry r WHERE r.source_table = source_table LIMIT 1), NULLIF(source_system,''), ''),
                   JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.work_order')),
                   IFNULL(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.supplier_code')), ''),
                   IFNULL(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.component_item_code')), ''),
                   COALESCE(CASE WHEN JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.shortage_qty')) REGEXP '^-?[0-9]+(\\.[0-9]+)?$' THEN CAST(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.shortage_qty')) AS DECIMAL(18,6)) END, 0),
                   IFNULL(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.risk_level')), 'MEDIUM'),
                   NULLIF(NULLIF(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.expected_supply_date')), 'null'), ''),
                   source_biz_key, @BatchId, @Now
            FROM mdp_stg_s4_shortage
            WHERE source_table='dwd_material_shortage'
              AND COALESCE(CASE WHEN JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.shortage_qty')) REGEXP '^-?[0-9]+(\\.[0-9]+)?$' THEN CAST(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.shortage_qty')) AS DECIMAL(18,6)) END, 0) > 0
            ON DUPLICATE KEY UPDATE shortage_qty=VALUES(shortage_qty), risk_level=VALUES(risk_level), need_date=VALUES(need_date),
                sync_batch_id=VALUES(sync_batch_id), sync_time=VALUES(sync_time), update_time=CURRENT_TIMESTAMP
            """, batchId, now);
    }

    private async Task<int> BuildDwdAsync(string batchId, DateTime now, CancellationToken cancellationToken)
    {
        cancellationToken.ThrowIfCancellationRequested();
        var scope = RequireScope();
        string Scoped(string sql) => MdpSqlScope.InjectTenantFactory(sql);
        SugarParameter[] ScopedParameters(params SugarParameter[] parameters) =>
        [
            .. parameters,
            new SugarParameter("@TenantId", scope.TenantId),
            new SugarParameter("@FactoryId", scope.FactoryId)
        ];
        var statDate = now.Date;
        var total = 0;

        try
        {
            await MdpSchemaAligner.ExecuteAsync(_db, 
                Scoped("DELETE FROM dwd_s4_purchase_execution WHERE stat_date=@StatDate"),
                ScopedParameters(new SugarParameter("@StatDate", statDate)));

            total += await MdpSchemaAligner.ExecuteAsync(_db, 
                Scoped("""
                INSERT INTO dwd_s4_purchase_execution
                (tenant_id, factory_id, stat_date, po_no, po_line, supplier_code, item_code,
                 order_qty, delivery_qty, received_qty, returned_qty, shortage_qty,
                 due_date, actual_arrival_date, risk_level, source_system, sync_batch_id, sync_time)
                SELECT d.tenant_id, d.factory_id, @StatDate,
                       d.po_no, d.po_line, IFNULL(d.supplier_code,''), IFNULL(d.item_code,''),
                       IFNULL(d.order_qty,0), IFNULL(d.delivery_qty,0), IFNULL(d.receipt_qty,0),
                       IFNULL(ret.return_qty,0), IFNULL(sh.shortage_qty,0),
                       DATE(d.due_date), DATE(d.last_receipt_date),
                       CASE WHEN d.risk_level IN ('HIGH','MEDIUM','LOW') THEN LOWER(d.risk_level)
                            WHEN IFNULL(d.remaining_qty,0) > 0 THEN 'high'
                            WHEN IFNULL(d.receipt_qty,0) >= IFNULL(d.order_qty,0) AND IFNULL(d.order_qty,0) > 0 THEN 'low'
                            ELSE 'medium' END,
                       'AIDOP_NATIVE', @BatchId, @Now
                FROM dwd_supplier_delivery d
                LEFT JOIN (
                    SELECT tenant_id, factory_id, po_no, po_line, SUM(IFNULL(return_qty,0)) AS return_qty
                    FROM mdp_std_s4_return
                    WHERE IFNULL(po_no,'') <> ''
                    GROUP BY tenant_id, factory_id, po_no, po_line
                ) ret ON d.tenant_id=ret.tenant_id AND d.factory_id=ret.factory_id AND d.po_no=ret.po_no AND d.po_line=ret.po_line
                LEFT JOIN (
                    SELECT tenant_id, factory_id, supplier_code, item_code, SUM(IFNULL(shortage_qty,0)) AS shortage_qty
                    FROM mdp_std_s4_shortage
                    GROUP BY tenant_id, factory_id, supplier_code, item_code
                ) sh ON d.tenant_id=sh.tenant_id AND d.factory_id=sh.factory_id AND IFNULL(d.supplier_code,'')=IFNULL(sh.supplier_code,'') AND IFNULL(d.item_code,'')=IFNULL(sh.item_code,'')
                WHERE d.stat_date=@StatDate AND IFNULL(d.po_no,'') <> ''
                ON DUPLICATE KEY UPDATE
                    order_qty=VALUES(order_qty), delivery_qty=VALUES(delivery_qty), received_qty=VALUES(received_qty),
                    returned_qty=VALUES(returned_qty), shortage_qty=VALUES(shortage_qty),
                    due_date=VALUES(due_date), actual_arrival_date=VALUES(actual_arrival_date),
                    risk_level=VALUES(risk_level), sync_batch_id=VALUES(sync_batch_id), sync_time=VALUES(sync_time), update_time=CURRENT_TIMESTAMP
                """),
                ScopedParameters(
                    new SugarParameter("@StatDate", statDate),
                    new SugarParameter("@BatchId", batchId),
                    new SugarParameter("@Now", now)));
        }
        catch
        {
            // dwd_supplier_delivery 可能尚未由 S3 生成
        }

        try
        {
            await MdpSchemaAligner.ExecuteAsync(_db, 
                Scoped("DELETE FROM dwd_po_trans WHERE trans_date=@StatDate"),
                ScopedParameters(new SugarParameter("@StatDate", statDate)));

            total += await MdpSchemaAligner.ExecuteAsync(_db, 
                Scoped("""
                INSERT INTO dwd_po_trans
                (tenant_id, factory_id, po_no, po_line, supplier_code, item_code, order_qty, received_qty,
                 returned_qty, shortage_qty, due_date, actual_arrival_date, risk_level,
                 trans_date, source_system, sync_batch_id, sync_time)
                SELECT tenant_id, factory_id, po_no, po_line, supplier_code, item_code,
                       order_qty, received_qty, returned_qty, shortage_qty,
                       due_date, actual_arrival_date, risk_level,
                       stat_date, source_system, sync_batch_id, sync_time
                FROM dwd_s4_purchase_execution
                WHERE stat_date=@StatDate
                """),
                ScopedParameters(new SugarParameter("@StatDate", statDate)));
        }
        catch
        {
            try
            {
                total += await MdpSchemaAligner.ExecuteAsync(_db, 
                    Scoped("""
                    INSERT INTO dwd_po_trans
                    (tenant_id, factory_id, po_no, supplier_code, item_code, order_qty, received_qty, trans_date, source_system, sync_time)
                    SELECT po.tenant_id, po.factory_id, po.po_no, IFNULL(po.supplier_code,''), IFNULL(po.item_code,''),
                           IFNULL(po.order_qty,0), IFNULL(po.received_qty,0), @StatDate, 'AIDOP_NATIVE', @Now
                    FROM mdp_std_purchase_order po
                    WHERE IFNULL(po.po_no,'') <> ''
                    """),
                    ScopedParameters(
                        new SugarParameter("@StatDate", statDate),
                        new SugarParameter("@Now", now)));
            }
            catch
            {
                // ignore
            }
        }

        try
        {
            total += await MdpSchemaAligner.ExecuteAsync(_db, 
                Scoped("""
                INSERT INTO dwd_qc_trans
                (tenant_id, factory_id, item_code, supplier_code, batch_no, sample_qty, defect_qty, result, trans_date, source_system, sync_time)
                SELECT tenant_id, factory_id, IFNULL(item_code,''), IFNULL(supplier_code,''), source_biz_key,
                       CAST(IFNULL(sample_qty,0) AS SIGNED), CAST(IFNULL(defect_qty,0) AS SIGNED),
                       CASE WHEN qc_result='FAIL' THEN 'FAIL' WHEN qc_result='CONCESSION' THEN 'CONCESSION' ELSE 'PASS' END,
                       @StatDate, 'AIDOP_NATIVE', @Now
                FROM mdp_std_s4_iqc
                WHERE IFNULL(item_code,'') <> ''
                ON DUPLICATE KEY UPDATE sample_qty=VALUES(sample_qty), defect_qty=VALUES(defect_qty), result=VALUES(result), sync_time=VALUES(sync_time)
                """),
                ScopedParameters(
                    new SugarParameter("@StatDate", statDate),
                    new SugarParameter("@Now", now)));
        }
        catch
        {
            // dwd_qc_trans 可能无唯一键，忽略
        }

        return total;
    }

    private async Task<int> BuildS4KpiValuesAsync(DateTime now, CancellationToken cancellationToken)
    {
        var affected = 0;
        for (var dayOffset = 13; dayOffset >= 0; dayOffset--)
        {
            var statDate = now.Date.AddDays(-dayOffset);
            var rows = await CalculateS4KpiValuesAsync(statDate);
            foreach (var row in rows)
            {
                cancellationToken.ThrowIfCancellationRequested();
                affected += await UpsertS4KpiValueAsync(row, statDate, now);
            }
        }
        return affected;
    }

    private async Task<List<S4KpiCalcRow>> CalculateS4KpiValuesAsync(DateTime statDate)
    {
        var scope = RequireScope();
        try
        {
            return await _db.Ado.SqlQueryAsync<S4KpiCalcRow>(
                MdpSqlScope.InjectTenantFactory(
                """
                SELECT ds.tenant_id AS TenantId, ds.factory_id AS FactoryId, 'S4_L1_001' AS MetricCode,
                       ROUND(AVG(TIMESTAMPDIFF(DAY, ds.request_date, r.receipt_date)), 4) AS MetricValue
                FROM mdp_std_delivery_schedule ds
                JOIN (
                    SELECT tenant_id, factory_id, po_no, po_line, item_code, MAX(receipt_date) AS receipt_date
                    FROM (
                        SELECT tenant_id, factory_id, po_no, po_line, item_code, receipt_date
                        FROM mdp_std_delivery_result
                        WHERE receipt_date IS NOT NULL
                        UNION ALL
                        SELECT tenant_id, factory_id, po_no, po_line, item_code, receipt_date
                        FROM mdp_std_s4_iqc
                        WHERE receipt_date IS NOT NULL
                    ) receipt_events
                    GROUP BY tenant_id, factory_id, po_no, po_line, item_code
                ) r ON r.tenant_id=ds.tenant_id AND r.factory_id=ds.factory_id
                   AND r.po_no=ds.po_no AND r.po_line=ds.po_line AND r.item_code=ds.item_code
                WHERE ds.request_date IS NOT NULL AND r.receipt_date >= ds.request_date
                GROUP BY ds.tenant_id, ds.factory_id
                HAVING MetricValue IS NOT NULL
                UNION ALL
                SELECT tenant_id AS TenantId, factory_id AS FactoryId, 'S4_L1_002' AS MetricCode,
                       ROUND(100 * SUM(CASE WHEN IFNULL(d.receipt_qty,0) >= IFNULL(d.order_qty,0) AND IFNULL(d.order_qty,0) > 0 THEN 1
                            WHEN d.delivery_status='COMPLETED' THEN 1 ELSE 0 END) / NULLIF(COUNT(1), 0), 4) AS MetricValue
                FROM dwd_supplier_delivery d
                WHERE d.stat_date=@StatDate AND IFNULL(d.order_qty,0) > 0
                GROUP BY tenant_id, factory_id
                UNION ALL
                SELECT tenant_id AS TenantId, factory_id AS FactoryId, 'S4_L1_003' AS MetricCode,
                       ROUND(SUM(IFNULL(d.receipt_qty,0)) / GREATEST(COUNT(DISTINCT NULLIF(d.supplier_code,'')), 1), 4) AS MetricValue
                FROM dwd_supplier_delivery d
                WHERE d.stat_date=@StatDate
                GROUP BY tenant_id, factory_id
                UNION ALL
                SELECT tenant_id AS TenantId, factory_id AS FactoryId, 'S4_L1_004' AS MetricCode,
                       ROUND(AVG(CASE WHEN IFNULL(d.order_qty,0) > 0
                            THEN (IFNULL(d.remaining_qty,0) / d.order_qty) * 30 END), 4) AS MetricValue
                FROM dwd_supplier_delivery d
                WHERE d.stat_date=@StatDate AND IFNULL(d.order_qty,0) > 0
                GROUP BY tenant_id, factory_id
                UNION ALL
                SELECT ds.tenant_id AS TenantId, ds.factory_id AS FactoryId, 'S4_L2_001' AS MetricCode,
                       ROUND(AVG(TIMESTAMPDIFF(DAY, ds.request_date, r.receipt_date)), 4) AS MetricValue
                FROM mdp_std_delivery_schedule ds
                JOIN (
                    SELECT tenant_id, factory_id, po_no, po_line, item_code, MAX(receipt_date) AS receipt_date
                    FROM (
                        SELECT tenant_id, factory_id, po_no, po_line, item_code, receipt_date
                        FROM mdp_std_delivery_result
                        WHERE receipt_date IS NOT NULL
                        UNION ALL
                        SELECT tenant_id, factory_id, po_no, po_line, item_code, receipt_date
                        FROM mdp_std_s4_iqc
                        WHERE receipt_date IS NOT NULL
                    ) receipt_events
                    GROUP BY tenant_id, factory_id, po_no, po_line, item_code
                ) r ON r.tenant_id=ds.tenant_id AND r.factory_id=ds.factory_id
                   AND r.po_no=ds.po_no AND r.po_line=ds.po_line AND r.item_code=ds.item_code
                WHERE ds.request_date IS NOT NULL AND r.receipt_date >= ds.request_date
                GROUP BY ds.tenant_id, ds.factory_id
                HAVING MetricValue IS NOT NULL
                UNION ALL
                SELECT tenant_id AS TenantId, factory_id AS FactoryId, 'S4_L2_002' AS MetricCode,
                       ROUND(100 * SUM(CASE WHEN IFNULL(d.receipt_qty,0) >= IFNULL(d.order_qty,0) AND IFNULL(d.order_qty,0) > 0 THEN 1
                            WHEN d.delivery_status='COMPLETED' THEN 1 ELSE 0 END) / NULLIF(COUNT(1), 0), 4) AS MetricValue
                FROM dwd_supplier_delivery d
                WHERE d.stat_date=@StatDate AND IFNULL(d.order_qty,0) > 0
                GROUP BY tenant_id, factory_id
                UNION ALL
                SELECT tenant_id AS TenantId, factory_id AS FactoryId, 'S4_L2_003' AS MetricCode,
                       ROUND(SUM(IFNULL(pe.received_qty,0)) / GREATEST(COUNT(DISTINCT NULLIF(pe.item_code,'')), 1), 4) AS MetricValue
                FROM dwd_s4_purchase_execution pe
                WHERE pe.stat_date=@StatDate
                GROUP BY tenant_id, factory_id
                UNION ALL
                SELECT tenant_id AS TenantId, factory_id AS FactoryId, 'S4_L2_004' AS MetricCode,
                       ROUND(AVG(CASE WHEN IFNULL(pe.order_qty,0) > 0
                            THEN ((IFNULL(pe.order_qty,0) - IFNULL(pe.received_qty,0)) / pe.order_qty) * 30 END), 4) AS MetricValue
                FROM dwd_s4_purchase_execution pe
                WHERE pe.stat_date=@StatDate AND IFNULL(pe.order_qty,0) > 0
                GROUP BY tenant_id, factory_id
                UNION ALL
                SELECT ds.tenant_id AS TenantId, ds.factory_id AS FactoryId, 'S4_L3_001' AS MetricCode,
                       ROUND(AVG(TIMESTAMPDIFF(DAY, ds.request_date, r.receipt_date)), 4) AS MetricValue
                FROM mdp_std_delivery_schedule ds
                JOIN (
                    SELECT tenant_id, factory_id, po_no, po_line, item_code, MAX(receipt_date) AS receipt_date
                    FROM (
                        SELECT tenant_id, factory_id, po_no, po_line, item_code, receipt_date
                        FROM mdp_std_delivery_result
                        WHERE receipt_date IS NOT NULL
                        UNION ALL
                        SELECT tenant_id, factory_id, po_no, po_line, item_code, receipt_date
                        FROM mdp_std_s4_iqc
                        WHERE receipt_date IS NOT NULL
                    ) receipt_events
                    GROUP BY tenant_id, factory_id, po_no, po_line, item_code
                ) r ON r.tenant_id=ds.tenant_id AND r.factory_id=ds.factory_id
                   AND r.po_no=ds.po_no AND r.po_line=ds.po_line AND r.item_code=ds.item_code
                WHERE ds.request_date IS NOT NULL AND r.receipt_date >= ds.request_date
                  AND IFNULL(ds.supplier_code,'') <> ''
                GROUP BY ds.tenant_id, ds.factory_id
                HAVING MetricValue IS NOT NULL
                UNION ALL
                SELECT tenant_id AS TenantId, factory_id AS FactoryId, 'S4_L3_002' AS MetricCode,
                       ROUND(100 * SUM(CASE WHEN IFNULL(d.receipt_qty,0) >= IFNULL(d.order_qty,0) AND IFNULL(d.order_qty,0) > 0 THEN 1
                            WHEN d.delivery_status='COMPLETED' THEN 1 ELSE 0 END) / NULLIF(COUNT(1), 0), 4) AS MetricValue
                FROM dwd_supplier_delivery d
                WHERE d.stat_date=@StatDate AND IFNULL(d.supplier_code,'') <> '' AND IFNULL(d.order_qty,0) > 0
                GROUP BY tenant_id, factory_id
                UNION ALL
                SELECT tenant_id AS TenantId, factory_id AS FactoryId, 'S4_L3_003' AS MetricCode,
                       ROUND(COUNT(DISTINCT NULLIF(pe.supplier_code,'')) / GREATEST(COUNT(DISTINCT NULLIF(pe.po_no,'')), 1), 4) AS MetricValue
                FROM dwd_s4_purchase_execution pe
                WHERE pe.stat_date=@StatDate AND IFNULL(pe.supplier_code,'') <> ''
                GROUP BY tenant_id, factory_id
                UNION ALL
                SELECT tenant_id AS TenantId, factory_id AS FactoryId, 'S4_L3_004' AS MetricCode,
                       ROUND(AVG(CASE WHEN IFNULL(pe.order_qty,0) > 0 AND IFNULL(pe.supplier_code,'') <> ''
                            THEN ((IFNULL(pe.order_qty,0) - IFNULL(pe.received_qty,0)) / pe.order_qty) * 30 END), 4) AS MetricValue
                FROM dwd_s4_purchase_execution pe
                WHERE pe.stat_date=@StatDate AND IFNULL(pe.order_qty,0) > 0 AND IFNULL(pe.supplier_code,'') <> ''
                GROUP BY tenant_id, factory_id
                """),
                new SugarParameter("@StatDate", statDate),
                new SugarParameter("@TenantId", scope.TenantId),
                new SugarParameter("@FactoryId", scope.FactoryId));
        }
        catch
        {
            return new List<S4KpiCalcRow>();
        }
    }

    private async Task<int> UpsertS4KpiValueAsync(S4KpiCalcRow row, DateTime statDate, DateTime now)
    {
        var meta = await _db.Ado.SqlQuerySingleAsync<S4KpiMetaRow>(
            """
            SELECT MetricLevel, Direction, YellowThreshold, RedThreshold
            FROM ado_smart_ops_kpi_master
            WHERE TenantId=@TenantId AND ModuleCode='S4' AND MetricCode=@MetricCode AND IsEnabled=1
            LIMIT 1
            """,
            new SugarParameter("@TenantId", row.TenantId),
            new SugarParameter("@MetricCode", row.MetricCode));
        if (meta == null || row.MetricValue == null)
            return 0;

        var table = ResolveKpiValueTable(meta.MetricLevel);
        var current = await _db.Ado.SqlQuerySingleAsync<S4KpiValueRow>(
            $"""
            SELECT id AS Id, metric_value AS MetricValue, target_value AS TargetValue
            FROM {table}
            WHERE tenant_id=@TenantId AND factory_id=@FactoryId AND module_code='S4'
              AND metric_code=@MetricCode AND biz_date=@BizDate AND is_deleted=0
            ORDER BY id
            LIMIT 1
            """,
            new SugarParameter("@TenantId", row.TenantId),
            new SugarParameter("@FactoryId", row.FactoryId),
            new SugarParameter("@MetricCode", row.MetricCode),
            new SugarParameter("@BizDate", statDate));
        var prior = await _db.Ado.SqlQuerySingleAsync<S4KpiValueRow>(
            $"""
            SELECT id AS Id, metric_value AS MetricValue, target_value AS TargetValue
            FROM {table}
            WHERE tenant_id=@TenantId AND factory_id=@FactoryId AND module_code='S4'
              AND metric_code=@MetricCode AND biz_date<@BizDate AND is_deleted=0
            ORDER BY biz_date DESC, id DESC
            LIMIT 1
            """,
            new SugarParameter("@TenantId", row.TenantId),
            new SugarParameter("@FactoryId", row.FactoryId),
            new SugarParameter("@MetricCode", row.MetricCode),
            new SugarParameter("@BizDate", statDate));

        var actual = Math.Round(row.MetricValue.Value, 4);
        var snap = await _kpiTargetResolver.ResolveAsync(row.TenantId, row.FactoryId, row.MetricCode, "S4", statDate);
        var target = snap.TargetValue;
        var status = AidopS4KpiMerge.AchievementLevel(actual, target, meta.Direction, meta.YellowThreshold, meta.RedThreshold);
        var trend = ResolveTrendFlag(actual, prior?.MetricValue);

        if (current != null)
        {
            return await MdpSchemaAligner.ExecuteAsync(_db, 
                $"""
                UPDATE {table}
                SET metric_value=@MetricValue, target_value=@TargetValue, status_color=@StatusColor, trend_flag=@TrendFlag,
                    target_config_id=@TargetConfigId, target_source=@TargetSource, target_resolved_at=@TargetResolvedAt,
                    is_active=1, status='ACTIVE', calc_time=@CalcTime, update_time=@CalcTime
                WHERE tenant_id=@TenantId AND factory_id=@FactoryId AND module_code='S4'
                  AND metric_code=@MetricCode AND biz_date=@BizDate AND is_deleted=0
                """,
                new SugarParameter("@MetricValue", actual),
                new SugarParameter("@TargetValue", KpiTargetSnapshotSql.ValueOrDbNull(snap)),
                new SugarParameter("@StatusColor", status),
                new SugarParameter("@TrendFlag", trend),
                new SugarParameter("@TargetConfigId", KpiTargetSnapshotSql.ConfigIdOrDbNull(snap)),
                new SugarParameter("@TargetSource", KpiTargetSnapshotSql.SourceOrDbNull(snap)),
                new SugarParameter("@TargetResolvedAt", snap.ResolvedAt),
                new SugarParameter("@CalcTime", now),
                new SugarParameter("@TenantId", row.TenantId),
                new SugarParameter("@FactoryId", row.FactoryId),
                new SugarParameter("@MetricCode", row.MetricCode),
                new SugarParameter("@BizDate", statDate));
        }

        var nextId = Yitter.IdGenerator.YitIdHelper.NextId();
        return await MdpSchemaAligner.ExecuteAsync(_db, 
            $"""
            INSERT INTO {table}
            (id, tenant_id, factory_id, status, biz_date, create_time, update_time, is_deleted, is_active,
             module_code, metric_code, metric_value, target_value, status_color, trend_flag, calc_time,
             target_config_id, target_source, target_resolved_at)
            VALUES
            (@Id, @TenantId, @FactoryId, 'ACTIVE', @BizDate, @CalcTime, @CalcTime, 0, 1,
             'S4', @MetricCode, @MetricValue, @TargetValue, @StatusColor, @TrendFlag, @CalcTime,
             @TargetConfigId, @TargetSource, @TargetResolvedAt)
            """,
            new SugarParameter("@Id", nextId),
            new SugarParameter("@TenantId", row.TenantId),
            new SugarParameter("@FactoryId", row.FactoryId),
            new SugarParameter("@BizDate", statDate),
            new SugarParameter("@CalcTime", now),
            new SugarParameter("@MetricCode", row.MetricCode),
            new SugarParameter("@MetricValue", actual),
            new SugarParameter("@TargetValue", KpiTargetSnapshotSql.ValueOrDbNull(snap)),
            new SugarParameter("@StatusColor", status),
            new SugarParameter("@TrendFlag", trend),
            new SugarParameter("@TargetConfigId", KpiTargetSnapshotSql.ConfigIdOrDbNull(snap)),
            new SugarParameter("@TargetSource", KpiTargetSnapshotSql.SourceOrDbNull(snap)),
            new SugarParameter("@TargetResolvedAt", snap.ResolvedAt));
    }

    private async Task<long> InsertSyncLogAsync(long entityId, string entityName, string batchId, int rowsRead)
    {
        EnsureRunScope();
        await MdpSchemaAligner.ExecuteAsync(_db, 
            """
            INSERT INTO mdp_sync_log
            (tenant_id, entity_id, source_code, entity_name, sync_batch_id, sync_type, trigger_type, sync_start, rows_read, status)
            VALUES (@TenantId, @EntityId, 'AIDOPDEV_MYSQL', @EntityName, @BatchId, 'FULL', 'AUTO', NOW(), @RowsRead, 'RUNNING')
            """,
            new SugarParameter("@TenantId", _runScope.TenantId),
            new SugarParameter("@EntityId", entityId),
            new SugarParameter("@EntityName", entityName),
            new SugarParameter("@BatchId", batchId),
            new SugarParameter("@RowsRead", rowsRead));
        return await _db.Ado.GetLongAsync(
            "SELECT id FROM mdp_sync_log WHERE sync_batch_id=@BatchId AND entity_id=@EntityId ORDER BY id DESC LIMIT 1",
            new List<SugarParameter> { new("@BatchId", batchId), new("@EntityId", entityId) });
    }

    private async Task MarkSyncLogSuccessAsync(long logId, DateTime started, int affected)
    {
        await MdpSchemaAligner.ExecuteAsync(_db, 
            """
            UPDATE mdp_sync_log
            SET status='SUCCESS', sync_end=NOW(), duration_ms=@DurationMs, rows_written=@RowsWritten
            WHERE id=@Id
            """,
            new SugarParameter("@DurationMs", (int)(DateTime.Now - started).TotalMilliseconds),
            new SugarParameter("@RowsWritten", affected),
            new SugarParameter("@Id", logId));
    }

    private async Task MarkSyncLogFailedAsync(long logId, DateTime started, string message)
    {
        await MdpSchemaAligner.ExecuteAsync(_db, 
            """
            UPDATE mdp_sync_log
            SET status='FAILED', sync_end=NOW(), duration_ms=@DurationMs, error_message=@ErrorMessage
            WHERE id=@Id
            """,
            new SugarParameter("@DurationMs", (int)(DateTime.Now - started).TotalMilliseconds),
            new SugarParameter("@ErrorMessage", Truncate(message, 2000)),
            new SugarParameter("@Id", logId));
    }

    private async Task<long> InsertTransformRunLogAsync(string batchId, DateTime startedAt, string triggerType)
    {
        await MdpSchemaAligner.ExecuteAsync(_db, 
            """
            INSERT INTO mdp_transform_run_log
            (tenant_id, job_code, job_name, trigger_type, batch_id, status, start_time)
            VALUES (@TenantId, @JobCode, 'S4 MDP同步与标准化转换', @TriggerType, @BatchId, 'RUNNING', @StartTime)
            """,
            new SugarParameter("@TenantId", RequireScope().TenantId),
            new SugarParameter("@JobCode", JobCode),
            new SugarParameter("@TriggerType", NormalizeTriggerType(triggerType)),
            new SugarParameter("@BatchId", batchId),
            new SugarParameter("@StartTime", startedAt));
        return await _db.Ado.GetLongAsync(
            "SELECT id FROM mdp_transform_run_log WHERE batch_id=@BatchId ORDER BY id DESC LIMIT 1",
            new List<SugarParameter> { new("@BatchId", batchId) });
    }

    private async Task MarkTransformRunSuccessAsync(long runLogId, DateTime startedAt, S4MdpSyncTransformResult result)
    {
        var finishedAt = DateTime.Now;
        await MdpSchemaAligner.ExecuteAsync(_db, 
            """
            UPDATE mdp_transform_run_log
            SET status='SUCCESS', end_time=@EndTime, duration_ms=@DurationMs,
                stage_rows=@StageRows, standard_rows=@StandardRows, dwd_rows=@DwdRows,
                summary_json=@SummaryJson, update_time=CURRENT_TIMESTAMP
            WHERE id=@Id
            """,
            new SugarParameter("@EndTime", finishedAt),
            new SugarParameter("@DurationMs", (int)(finishedAt - startedAt).TotalMilliseconds),
            new SugarParameter("@StageRows", result.StageRows),
            new SugarParameter("@StandardRows", result.StandardRows),
            new SugarParameter("@DwdRows", result.DwdRows),
            new SugarParameter("@SummaryJson", BuildRunSummaryJson(result)),
            new SugarParameter("@Id", runLogId));
    }

    private async Task MarkTransformRunFailedAsync(long runLogId, DateTime startedAt, string message)
    {
        try
        {
            var finishedAt = DateTime.Now;
            await MdpSchemaAligner.ExecuteAsync(_db, 
                """
                UPDATE mdp_transform_run_log
                SET status='FAILED', end_time=@EndTime, duration_ms=@DurationMs,
                    error_message=@ErrorMessage, update_time=CURRENT_TIMESTAMP
                WHERE id=@Id
                """,
                new SugarParameter("@EndTime", finishedAt),
                new SugarParameter("@DurationMs", (int)(finishedAt - startedAt).TotalMilliseconds),
                new SugarParameter("@ErrorMessage", Truncate(message, 2000)),
                new SugarParameter("@Id", runLogId));
        }
        catch (Exception ex)
        {
            Console.Error.WriteLine($"[S4MdpSyncTransform] MarkTransformRunFailed write failed (runLogId={runLogId}): {ex.Message}");
        }
    }

    private S4MdpSqlCommand Cmd(string sql, string batchId, DateTime now)
    {
        var scope = RequireScope();
        return new S4MdpSqlCommand(MdpSqlScope.InjectTenantFactory(sql), new[]
        {
            new SugarParameter("@BatchId", batchId),
            new SugarParameter("@Now", now),
            new SugarParameter("@StatDate", now.Date),
            new SugarParameter("@TenantId", scope.TenantId),
            new SugarParameter("@FactoryId", scope.FactoryId)
        });
    }

    private MdpRebuildScope RequireScope()
    {
        EnsureRunScope();
        return _runScope;
    }

    private void EnsureRunScope()
    {
        if (_runScope == null)
            throw new InvalidOperationException("S4 全量重算必须指定有效 tenantId/factoryId");
    }

    private static string BuildJsonObjectExpression(IEnumerable<string> columns)
    {
        var parts = columns.SelectMany(c => new[] { $"'{c.Replace("'", "''")}'", $"s.`{c}`" });
        return $"JSON_OBJECT({string.Join(",", parts)})";
    }

    private static string FindColumn(IEnumerable<string> columns, string expected) =>
        columns.First(u => string.Equals(u, expected, StringComparison.OrdinalIgnoreCase));

    private static string BuildRunSummaryJson(S4MdpSyncTransformResult result) =>
        $$"""{"batchId":"{{result.BatchId}}","stageRows":{{result.StageRows}},"standardRows":{{result.StandardRows}},"dwdRows":{{result.DwdRows}},"kpiRows":{{result.KpiRows}}}""";

    private static string ResolveKpiValueTable(int metricLevel) => metricLevel switch
    {
        1 => "ado_s9_kpi_value_l1_day",
        2 => "ado_s9_kpi_value_l2_day",
        3 => "ado_s9_kpi_value_l3_day",
        4 => "ado_s9_kpi_value_l4_day",
        _ => "ado_s9_kpi_value_l2_day"
    };

    private static decimal DefaultS4Target(string metricCode) => LegacyKpiCodeTargets.GetOrZero(metricCode);

    private static string ResolveKpiStatus(decimal actual, decimal target, string? direction, decimal? yellowThreshold, decimal? redThreshold)
    {
        if (target <= 0) return "gray";
        var ratio = actual / target * 100m;
        if (string.Equals(direction, "lower_is_better", StringComparison.OrdinalIgnoreCase))
        {
            if (actual <= target) return "green";
            if (ratio <= (yellowThreshold ?? 110m)) return "yellow";
            return ratio >= (redThreshold ?? 120m) ? "red" : "yellow";
        }
        if (actual >= target) return "green";
        if (ratio >= (yellowThreshold ?? 95m)) return "yellow";
        return ratio <= (redThreshold ?? 80m) ? "red" : "yellow";
    }

    private static string ResolveTrendFlag(decimal actual, decimal? previous)
    {
        if (previous == null) return "flat";
        if (actual > previous.Value) return "up";
        if (actual < previous.Value) return "down";
        return "flat";
    }

    private static string NormalizeTriggerType(string? triggerType)
        => string.IsNullOrWhiteSpace(triggerType) ? "AUTO" : triggerType.Trim().ToUpperInvariant();

    private static string Truncate(string? raw, int maxLength)
    {
        if (string.IsNullOrEmpty(raw)) return string.Empty;
        return raw.Length <= maxLength ? raw : raw[..maxLength];
    }

    private sealed class S4ColumnRow
    {
        public string ColumnName { get; set; } = string.Empty;
    }

    private sealed class S4MdpEntityRow
    {
        public long Id { get; set; }
        public string EntityName { get; set; } = string.Empty;
    }

    private sealed class S4KpiCalcRow
    {
        public long TenantId { get; set; }
        public long FactoryId { get; set; }
        public string MetricCode { get; set; } = string.Empty;
        public decimal? MetricValue { get; set; }
    }

    private sealed class S4KpiMetaRow
    {
        public int MetricLevel { get; set; }
        public string Direction { get; set; } = "higher_is_better";
        public decimal? YellowThreshold { get; set; }
        public decimal? RedThreshold { get; set; }
    }

    private sealed class S4KpiValueRow
    {
        public long Id { get; set; }
        public decimal? MetricValue { get; set; }
        public decimal? TargetValue { get; set; }
    }
}

public sealed class S4MdpSyncTransformResult
{
    public long RunLogId { get; set; }
    public string BatchId { get; set; } = string.Empty;
    public int StageRows { get; set; }
    public int StandardRows { get; set; }
    public int DwdRows { get; set; }
    public int KpiRows { get; set; }
}

internal sealed record S4MdpSqlCommand(string Sql, SugarParameter[] Parameters);

internal sealed record S4MdpEntityConfig(
    string EntityCode,
    string SourceTable,
    string TargetTable,
    string SourceRowIdExpression,
    string SourceBizKeyExpression,
    bool Optional = false)
{
    public static readonly IReadOnlyList<S4MdpEntityConfig> All = new List<S4MdpEntityConfig>
    {
        new("S4_IQC_RECEIPT", "PurOrdRctDetail", "mdp_stg_s4_iqc", "RecID", "CONCAT(IFNULL(s.`Domain`,''), ':', IFNULL(s.`Receiver`,''), ':', IFNULL(s.`Line`,''))"),
        new("S4_SHIPMENT_EXEC", "scm_shdzb", "mdp_stg_s4_shipment", "id", "CONCAT(IFNULL(s.`glid`,''), ':', IFNULL(s.`id`,''))"),
        new("S4_RETURN_EXEC", "srm_polist_ds", "mdp_stg_s4_return", "Id", "s.`dsnum`"),
        new("S4_SHORTAGE_EXEC", "dwd_material_shortage", "mdp_stg_s4_shortage", "id", "CONCAT(IFNULL(s.`work_order`,''), ':', IFNULL(s.`component_item_code`,''))", Optional: true)
    };
}