AiDOPWarehouse/server/Plugins/Admin.NET.Plugin.AiDOP/Order/S1MdpSyncTransformService.cs

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

namespace Admin.NET.Plugin.AiDOP.Order;

/// <summary>
/// S1 首批 MDP 同步和标准化转换服务。
/// Phase2：源→stg 经统一执行器；std/DWD/KPI 仍读 stg。
/// </summary>
public class S1MdpSyncTransformService : ITransient
{
    private const string JobCode = "S1_MDP_SYNC_TRANSFORM";
    private readonly ISqlSugarClient _db;
    private readonly TransformRunLogFinalizer _runLogFinalizer;
    private readonly SmartOpsKpiAtomicBuildService _atomicBuild;
    private readonly MdpModuleStagingPuller _stagingPuller;
    private readonly IKpiTargetResolver _kpiTargetResolver;
    private readonly IS1MdpFullRunLock _fullRunLock;
    private readonly S9CompositeKpiWriter _s9Composite;
    private readonly MdpNeutralSourceGate _neutralGate;
    private readonly MdpTenantStdSourceResolver _stdSource;
    private readonly NativeNeutralProjectionService _nativeProjection;

    public S1MdpSyncTransformService(
        ISqlSugarClient db,
        SmartOpsKpiAtomicBuildService atomicBuild,
        MdpModuleStagingPuller stagingPuller,
        IKpiTargetResolver kpiTargetResolver,
        IS1MdpFullRunLock fullRunLock,
        TransformRunLogFinalizer runLogFinalizer,
        S9CompositeKpiWriter s9Composite,
        MdpNeutralSourceGate neutralGate,
        MdpTenantStdSourceResolver stdSource,
        NativeNeutralProjectionService nativeProjection)
    {
        _db = db;
        _runLogFinalizer = runLogFinalizer;
        _atomicBuild = atomicBuild;
        _stagingPuller = stagingPuller;
        _kpiTargetResolver = kpiTargetResolver;
        _fullRunLock = fullRunLock;
        _s9Composite = s9Composite;
        _neutralGate = neutralGate;
        _stdSource = stdSource;
        _nativeProjection = nativeProjection;
    }

    public async Task<S1MdpSyncTransformResult> RunFullAsync(
        S1MdpRunScope scope,
        CancellationToken cancellationToken = default,
        string triggerType = "AUTO",
        long? rebuildJobId = null,
        Func<S1MdpProgressUpdate, Task>? reportProgress = null)
    {
        scope = S1MdpRunScope.Create(scope.TenantId, scope.FactoryId);
        cancellationToken.ThrowIfCancellationRequested();
        var holderId = $"{triggerType}:{scope.ScopeKey}:{Environment.MachineName}:{Guid.NewGuid():N}";
        var lease = await _fullRunLock.TryAcquireAsync(scope.LockName, holderId, rebuildJobId, cancellationToken);
        if (lease == null)
            throw new S1MdpAlreadyRunningException();

        await using (lease)
        {
            using var heartbeatCts = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken);
            var heartbeat = KeepLeaseAliveAsync(lease, heartbeatCts.Token);
            var now = DateTime.Now;
            var batchId = $"S1_MDP_FULL_{scope.TenantId}_{scope.FactoryId}_{now:yyyyMMddHHmmss}";
            var runLogId = await InsertTransformRunLogAsync(scope, batchId, now, triggerType);
            var result = new S1MdpSyncTransformResult { BatchId = batchId, RunLogId = runLogId };

            try
            {
                await ReportAsync(reportProgress, new S1MdpProgressUpdate(S1MdpRebuildStage.Preparing, 0, 5, "准备运行环境"));
                await EnsureS1RuntimeObjectsAsync();

                await ReportAsync(reportProgress, new S1MdpProgressUpdate(S1MdpRebuildStage.Staging, 1, 10, "正在拉取源数据"));
                result.StageRows = await SyncStagingAsync(scope, batchId, now, cancellationToken, triggerType);
                await ReportAsync(reportProgress, new S1MdpProgressUpdate(S1MdpRebuildStage.Staging, 1, 35, "拉取源数据完成", result.StageRows, S1MdpRebuildStage.Staging));

                await ReportAsync(reportProgress, new S1MdpProgressUpdate(S1MdpRebuildStage.Standard, 2, 35, "正在标准化数据"));
                result.StandardRows = await TransformStandardAsync(scope, batchId, now, cancellationToken);
                await ReportAsync(reportProgress, new S1MdpProgressUpdate(S1MdpRebuildStage.Standard, 2, 50, "标准化数据完成", result.StandardRows, S1MdpRebuildStage.Standard));

                await ReportAsync(reportProgress, new S1MdpProgressUpdate(S1MdpRebuildStage.Dwd, 3, 50, "正在生成 DWD 明细"));
                result.DwdRows = await BuildDwdAsync(scope, batchId, now, result, cancellationToken);
                await ReportAsync(reportProgress, new S1MdpProgressUpdate(S1MdpRebuildStage.Dwd, 3, 65, "生成 DWD 明细完成", result.DwdRows, S1MdpRebuildStage.Dwd));

                await ReportAsync(reportProgress, new S1MdpProgressUpdate(S1MdpRebuildStage.Kpi, 4, 65, "正在重算 KPI"));
                result.KpiRows = await BuildS1KpiValuesAsync(scope, batchId, now, cancellationToken);
                await ReportAsync(reportProgress, new S1MdpProgressUpdate(S1MdpRebuildStage.Kpi, 4, 80, "重算 KPI 完成", result.KpiRows, S1MdpRebuildStage.Kpi));

                await ReportAsync(reportProgress, new S1MdpProgressUpdate(S1MdpRebuildStage.Atomic, 5, 82, "正在重建原子数据"));
                result.AtomicRows = await _atomicBuild.BuildOrderDeliveryDomainForAllDatesAsync(
                    scope.TenantId, scope.FactoryId, batchId, cancellationToken);
                await ReportAsync(reportProgress, new S1MdpProgressUpdate(S1MdpRebuildStage.Atomic, 5, 98, "重建原子数据完成", result.AtomicRows, S1MdpRebuildStage.Atomic));

                await ReportAsync(reportProgress, new S1MdpProgressUpdate(S1MdpRebuildStage.Finalizing, 5, 99, "正在收尾"));
                await _nativeProjection.ProjectAsync(scope.TenantId, batchId, cancellationToken);
                result.KpiRows += await _s9Composite.WriteRecentAsync(scope.TenantId, scope.FactoryId, now, cancellationToken);
                await MarkTransformRunSuccessAsync(runLogId, now, result);
                return result;
            }
            catch (Exception ex)
            {
                // 管理员取消由外层重算服务收口为 ABORTED，这里不能先写成 FAILED。
                if (ex is ModuleRebuildCancelledException)
                    throw;
                var summary = ex is OperationCanceledException
                    ? "HTTP 请求已取消或服务中断"
                    : ex.Message;
                // 宿主关停不是转换失败：交给 finally 收口为 ABORTED，不污染 FAILED 语义。
                // 上面那句 summary 之所以把两种取消写在一起，正是因为 token 分不开它们；
                // 判据只能是 ApplicationStopping。
                if (!_runLogFinalizer.IsHostStopping)
                    await MarkTransformRunFailedAsync(runLogId, now, summary);
                throw;
            }
            finally
            {
                await _runLogFinalizer.FinalizeIfHostStoppingAsync(runLogId, now);
                heartbeatCts.Cancel();
                try { await heartbeat; } catch (OperationCanceledException) { }
            }
        }
    }

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

    /// <summary>贴源已由 FILE 导入写入后，只跑标准层/DWD/KPI，不再 Pull。</summary>
    public async Task<S1MdpSyncTransformResult> RunPostStagingAsync(
        S1MdpRunScope scope,
        CancellationToken cancellationToken = default,
        string triggerType = "FILE_IMPORT")
    {
        scope = S1MdpRunScope.Create(scope.TenantId, scope.FactoryId);
        cancellationToken.ThrowIfCancellationRequested();
        var now = DateTime.Now;
        var batchId = $"S1_MDP_FILE_{scope.TenantId}_{scope.FactoryId}_{now:yyyyMMddHHmmss}";
        var runLogId = await InsertTransformRunLogAsync(scope, batchId, now, triggerType);
        var result = new S1MdpSyncTransformResult { BatchId = batchId, RunLogId = runLogId };
        try
        {
            await EnsureS1RuntimeObjectsAsync();
            result.StandardRows = await TransformStandardAsync(scope, batchId, now, cancellationToken);
            result.DwdRows = await BuildDwdAsync(scope, batchId, now, result, cancellationToken);
            result.KpiRows = await BuildS1KpiValuesAsync(scope, batchId, now, cancellationToken);
            result.AtomicRows = await _atomicBuild.BuildOrderDeliveryDomainForAllDatesAsync(
                scope.TenantId, scope.FactoryId, batchId, cancellationToken);
            await _nativeProjection.ProjectAsync(scope.TenantId, batchId, cancellationToken);
            result.KpiRows += await _s9Composite.WriteRecentAsync(scope.TenantId, scope.FactoryId, now, cancellationToken);
            await MarkTransformRunSuccessAsync(runLogId, now, result);
            return result;
        }
        catch (Exception ex)
        {
            if (ex is ModuleRebuildCancelledException)
                throw;
            // 宿主关停不是转换失败：交给 finally 收口为 ABORTED，不污染 FAILED 语义。
            if (!_runLogFinalizer.IsHostStopping)
                await MarkTransformRunFailedAsync(runLogId, now, ex.Message);
            throw;
        }
        finally
        {
            await _runLogFinalizer.FinalizeIfHostStoppingAsync(runLogId, now);
        }
    }

    private async Task EnsureS1RuntimeObjectsAsync()
    {
        await MdpSchemaAligner.ExecuteAsync(_db, MdpSchemaDefinition.Ddl("dwd_requirement_examine_detail"));

        await MdpSchemaAligner.ExecuteAsync(_db, 
            """
            INSERT INTO mdp_entity
            (tenant_id, source_id, entity_code, entity_name, entity_type, source_table_name, target_table_name, sync_mode, batch_size, status, remark, create_time, update_time)
            SELECT 0, s.id, 'S1_REQUIREMENT_EXAMINE_RESULT', 'S1需求核验结果主表', 'TABLE',
                   'b_examine_result', 'mdp_stg_so', 'FULL', 5000, 1, '需求核验结果主表，进入 S1 贴源层', CURRENT_TIMESTAMP, CURRENT_TIMESTAMP
            FROM mdp_source s
            WHERE s.tenant_id=0 AND s.source_code='AIDOPDEV_MYSQL'
            LIMIT 1
            ON DUPLICATE KEY UPDATE
                source_id=VALUES(source_id), entity_name=VALUES(entity_name), source_table_name=VALUES(source_table_name),
                target_table_name=VALUES(target_table_name), sync_mode=VALUES(sync_mode), status=VALUES(status),
                remark=VALUES(remark), update_time=CURRENT_TIMESTAMP;
            """);

        await MdpSchemaAligner.ExecuteAsync(_db, 
            """
            INSERT INTO mdp_entity
            (tenant_id, source_id, entity_code, entity_name, entity_type, source_table_name, target_table_name, sync_mode, batch_size, status, remark, create_time, update_time)
            SELECT 0, s.id, 'S1_REQUIREMENT_EXAMINE_DETAIL', 'S1需求核验BOM明细', 'TABLE',
                   'b_bom_child_examine', 'mdp_stg_so', 'FULL', 5000, 1, '需求核验BOM明细，进入 S1 贴源层和 DWD', CURRENT_TIMESTAMP, CURRENT_TIMESTAMP
            FROM mdp_source s
            WHERE s.tenant_id=0 AND s.source_code='AIDOPDEV_MYSQL'
            LIMIT 1
            ON DUPLICATE KEY UPDATE
                source_id=VALUES(source_id), entity_name=VALUES(entity_name), source_table_name=VALUES(source_table_name),
                target_table_name=VALUES(target_table_name), sync_mode=VALUES(sync_mode), status=VALUES(status),
                remark=VALUES(remark), update_time=CURRENT_TIMESTAMP;
            """);

        await MdpSchemaAligner.ExecuteAsync(_db, 
            """
            INSERT INTO mdp_entity
            (tenant_id, source_id, entity_code, entity_name, entity_type, source_table_name, target_table_name, sync_mode, incr_column, batch_size, status, remark, create_time, update_time)
            SELECT 0, s.id, v.entity_code, v.entity_name, 'TABLE', v.source_table_name, 'mdp_stg_so', v.sync_mode, v.incr_column, 5000, 1, v.remark, CURRENT_TIMESTAMP, CURRENT_TIMESTAMP
            FROM mdp_source s
            JOIN (
                SELECT 'S1_PRODUCT_DESIGN' AS entity_code, 'S1产品设计主表' AS entity_name, 'ado_product_design' AS source_table_name, 'INCR' AS sync_mode, 'UpdateTime' AS incr_column, '产品设计主表，进入订单标准层' AS remark
                UNION ALL SELECT 'S1_PRODUCT_DESIGN_BOM', 'S1产品设计BOM', 'ado_product_design_bom', 'FULL', NULL, '产品设计BOM，表无时间戳列，只能全量抽取'
                UNION ALL SELECT 'S1_PRODUCT_DESIGN_ROUTING', 'S1产品设计工艺路线', 'ado_product_design_routing', 'FULL', NULL, '产品设计工艺路线，表无时间戳列，只能全量抽取'
            ) v
            WHERE s.tenant_id=0 AND s.source_code='AIDOPDEV_MYSQL'
            ON DUPLICATE KEY UPDATE
                source_id=VALUES(source_id), entity_name=VALUES(entity_name), source_table_name=VALUES(source_table_name),
                target_table_name=VALUES(target_table_name), sync_mode=VALUES(sync_mode), incr_column=VALUES(incr_column),
                status=VALUES(status), remark=VALUES(remark), update_time=CURRENT_TIMESTAMP;
            """);
    }

    /// <summary>Phase2/3：源→stg 走执行器（DB/API）；遗留 SyncOneEntityAsync 不再调用。</summary>
    private async Task<int> SyncStagingAsync(S1MdpRunScope scope, string batchId, DateTime now, CancellationToken cancellationToken, string triggerType = "AUTO")
    {
        _ = now;
        var codes = S1MdpEntityConfig.All.Select(x => x.EntityCode);
        return await _stagingPuller.PullEntitiesAsync(
            codes, batchId, scope.TenantId, fullRefresh: ModuleRebuildTriggerType.ShouldPullFull(triggerType), taskCode: "S1_MDP_INBOUND",
            cancellationToken, scope.FactoryId, requireMatchingSourceTenant: true);
    }

    /// <summary>
    /// 双模式入站：仅抽 stg（可指定 DB 实体码或 *_API），再跑 std/DWD/KPI。
    /// </summary>
    public async Task<S1MdpSyncTransformResult> RunInboundAsync(
        S1MdpRunScope scope,
        IEnumerable<string>? entityCodes = null,
        bool fullRefresh = false,
        CancellationToken cancellationToken = default)
    {
        scope = S1MdpRunScope.Create(scope.TenantId, scope.FactoryId);
        cancellationToken.ThrowIfCancellationRequested();
        var now = DateTime.Now;
        var batchId = $"S1_MDP_IN_{scope.TenantId}_{scope.FactoryId}_{now:yyyyMMddHHmmss}";
        var runLogId = await InsertTransformRunLogAsync(scope, batchId, now, "INBOUND");
        var result = new S1MdpSyncTransformResult { BatchId = batchId, RunLogId = runLogId };
        try
        {
            await EnsureS1RuntimeObjectsAsync();
            var codes = entityCodes?.ToList() ?? S1MdpEntityConfig.All.Select(x => x.EntityCode).ToList();
            result.StageRows = await _stagingPuller.PullEntitiesAsync(
                codes, batchId, scope.TenantId, fullRefresh, "S1_MDP_INBOUND", cancellationToken,
                scope.FactoryId, requireMatchingSourceTenant: true);
            result.StandardRows = await TransformStandardAsync(scope, batchId, now, cancellationToken);
            result.DwdRows = await BuildDwdAsync(scope, batchId, now, result, cancellationToken);
            result.KpiRows = await BuildS1KpiValuesAsync(scope, batchId, now, cancellationToken);
            result.AtomicRows = await _atomicBuild.BuildOrderDeliveryDomainForAllDatesAsync(
                scope.TenantId, scope.FactoryId, batchId, cancellationToken);
            await _nativeProjection.ProjectAsync(scope.TenantId, batchId, cancellationToken);
            result.KpiRows += await _s9Composite.WriteRecentAsync(scope.TenantId, scope.FactoryId, now, cancellationToken);
            await MarkTransformRunSuccessAsync(runLogId, now, result);
            return result;
        }
        catch (Exception ex)
        {
            if (ex is ModuleRebuildCancelledException)
                throw;
            // 宿主关停不是转换失败：交给 finally 收口为 ABORTED，不污染 FAILED 语义。
            if (!_runLogFinalizer.IsHostStopping)
                await MarkTransformRunFailedAsync(runLogId, now, ex.Message);
            throw;
        }
        finally
        {
            await _runLogFinalizer.FinalizeIfHostStoppingAsync(runLogId, now);
        }
    }

    /// <summary>Phase3：已停用。源→stg 改由执行器路径；保留方法供紧急回滚对照，勿再调用。</summary>
    [Obsolete("Phase3: use MdpModuleStagingPuller / executors")]
    private async Task<int> SyncOneEntityAsync(S1MdpEntityConfig entity, string batchId, DateTime now, long tenantId)
    {
        if (tenantId <= 0) throw new InvalidOperationException("S1 同步日志必须指定有效 tenantId");
        var entityRow = await _db.Ado.SqlQuerySingleAsync<S1MdpEntityRow>(
            "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}");

        var columns = await _db.Ado.SqlQueryAsync<S1ColumnRow>(
            """
            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 = BuildOptionalColumnExpr(names, "tenant_id", "0");
        var factoryExpr = BuildOptionalColumnExpr(names, "factory_id", "NULL");
        var companyExpr = BuildOptionalColumnExpr(names, "company_id", "NULL");
        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(tenantId, entityRow.Id, entityRow.EntityName, batchId, rowsRead);
        var started = DateTime.Now;

        try
        {
            var affected = await MdpSchemaAligner.ExecuteAsync(_db, 
                $"""
                INSERT INTO `{entity.TargetTable}`
                (tenant_id, factory_id, company_id, source_system, source_table, source_row_id, source_biz_key, sync_batch_id, sync_time, process_status, raw_data)
                SELECT
                    {tenantExpr},
                    {factoryExpr},
                    {companyExpr},
                    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
                    tenant_id=VALUES(tenant_id),
                    factory_id=VALUES(factory_id),
                    company_id=VALUES(company_id),
                    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);
            throw;
        }
    }

    private async Task<int> TransformStandardAsync(S1MdpRunScope scope, string batchId, DateTime now, CancellationToken cancellationToken)
    {
        using var db = _db.CopyNew();
        db.Ado.CommandTimeOut = Math.Max(db.Ado.CommandTimeOut, 180);
        var total = 0;
        var registeredSource = (await _stdSource.GetSourceAsync(scope.TenantId, "SO", cancellationToken))?.SourceSystem;
        foreach (var command in BuildStandardCommands(scope, batchId, now))
        {
            cancellationToken.ThrowIfCancellationRequested();
            var gateSource = string.IsNullOrWhiteSpace(registeredSource)
                ? InferNeutralSource(command.Sql)
                : registeredSource;
            if (!await _neutralGate.AllowsAsync(scope.TenantId, "SO", gateSource))
                continue;
            total += await db.Ado.ExecuteCommandAsync(command.Sql, command.Parameters);
        }
        return total;
    }

    /// <summary>S1 标准 INSERT：发运相关表走 165；其余本库业务走 AIDOP_NATIVE。</summary>
    private static string InferNeutralSource(string sql)
    {
        if (sql.Contains("mdp_std_ship_trans", StringComparison.OrdinalIgnoreCase)
            || sql.Contains("ShippingPlan", StringComparison.Ordinal)
            || sql.Contains("ASNBOL", StringComparison.Ordinal))
            return "DOPDEMORQ_SQLSERVER";
        return MdpSourceIdentity.Native;
    }

    private async Task<int> BuildDwdAsync(
        S1MdpRunScope scope, string batchId, DateTime now,
        S1MdpSyncTransformResult result, CancellationToken cancellationToken)
    {
        using var db = _db.CopyNew();
        db.Ado.CommandTimeOut = Math.Max(db.Ado.CommandTimeOut, 180);
        var total = 0;
        foreach (var command in BuildDwdCommands(scope, batchId, now))
        {
            cancellationToken.ThrowIfCancellationRequested();
            total += await db.Ado.ExecuteCommandAsync(command.Sql, command.Parameters);
        }

        // DWD 阶段的最后一步：把本批次发布为「当前快照」。
        // 必须在所有 DWD 写入完成之后，否则读者会看到一个还没灌完的批次被标成 current。
        // 不计入 total —— total 语义是「本轮构建的 DWD 行数」，翻牌影响的行数属于发布动作，不是新建行。
        cancellationToken.ThrowIfCancellationRequested();
        var publishedRows = await PublishCurrentSnapshotAsync(db, scope, batchId);

        // 发布证据随运行日志一起落盘（见 MarkTransformRunSuccessAsync）。
        // 走到这里就意味着两条 DWD 写入与发布语句都已成功；若中途抛错，
        // 调用方的 catch 会写 FAILED，本字段保持 null —— 失败方向是保守的。
        result.Publication = new S1MdpSnapshotPublication
        {
            Table = "dwd_requirement_examine_detail",
            BatchId = batchId,
            FactoryId = scope.FactoryId,
            CurrentRows = publishedRows
        };

        return total;
    }

    /// <summary>
    /// 把 <paramref name="batchId"/> 原子地发布为该 (tenant, factory) 作用域下的当前快照。
    ///
    /// 【为什么必须是「一条」语句】
    /// BuildDwdAsync 全程没有事务（用的是 _db.CopyNew() 上的裸 ExecuteCommandAsync，没有 BeginTran）。
    /// 因此任何「两条语句」的写法都会在两条之间留下一个对并发读者可见的错误窗口：
    ///   · 先清后置（先把旧批次清 0，再把新批次置 1）→ 中间窗口「零个当前批次」，
    ///     读者拿到空结果，会被当成「本工单已齐套、无缺料」——缺料监控直接漏报。
    ///   · 先置后清（先把新批次置 1，再把旧批次清 0）→ 中间窗口「两个当前批次」，
    ///     读者按 is_current_flag=1 取数会同时拿到新旧两批，缺料数量凭空翻倍。
    ///   · 而且第二条语句一旦失败/进程被杀，上述错误窗口就不再是「瞬时」，而是永久残留。
    /// 单条 UPDATE 由 InnoDB 保证语句级原子性，不存在中间可见状态，也没有"第二条没跑成"的失败模式。
    ///
    /// 【为什么 factory_id 要 COALESCE 归一】
    /// 同一次 run（scope.FactoryId=1）产出的行，factory_id 既可能是 1 也可能是 NULL：
    /// 入库过滤用的是 COALESCE(NULLIF(d.factory_id,0),1)=@FactoryId，而落库值取的是
    /// COALESCE(NULLIF(d.factory_id,0), NULLIF(r.factory_id,0), NULLIF(so.factory_id,0))，三者全空时就是 NULL。
    /// 实测（aidopdev，只读）：1019060 行里 296189 行 factory_id IS NULL，
    /// 且 3439 个批次里有 594 个批次「同时含 factory_id=1 和 factory_id IS NULL」。
    /// 若作用域谓词直接写 factory_id=@FactoryId：
    ///   ① NULL 行永远不被 = 匹配（SQL 三值逻辑，NULL=1 得 NULL 而非 TRUE），旧批次的 NULL 行会一直挂着 is_current_flag=1；
    ///   ② 新批次也只有一半的行被置 1。
    /// 结果就是任务里点名要避免的「两个当前批次」永久共存。
    /// 故必须用 COALESCE(NULLIF(factory_id,0),1) 归一 —— 这也是本仓库通用的 factory 作用域写法。
    /// </summary>
    /// <summary>
    /// 发布当前快照，并返回本批次发布后的当前行数。
    /// <para>返回 0 是<b>合法结果</b>——源侧本轮没有数据时，成功发布的就是一个空快照。
    /// 调用方把这个数字记进运行日志，下游才能把「正常的空」与「没发布成」区分开。</para>
    /// </summary>
    private static async Task<int> PublishCurrentSnapshotAsync(ISqlSugarClient db, S1MdpRunScope scope, string batchId)
    {
        await db.Ado.ExecuteCommandAsync(
            """
            UPDATE dwd_requirement_examine_detail
               SET is_current_flag = CASE WHEN calc_batch_id=@BatchId THEN 1 ELSE 0 END
             WHERE tenant_id=@TenantId
               AND COALESCE(NULLIF(factory_id, 0), 1)=@FactoryId
               AND (calc_batch_id=@BatchId OR is_current_flag=1)
            """,
            new SugarParameter("@BatchId", batchId),
            new SugarParameter("@TenantId", scope.TenantId),
            new SugarParameter("@FactoryId", scope.FactoryId));

        // 复核发布结果而不是相信影响行数：ODKU 与 CASE 的影响行数都不等于「当前有几行」。
        return await db.Ado.GetIntAsync(
            """
            SELECT COUNT(*) FROM dwd_requirement_examine_detail
             WHERE tenant_id=@TenantId
               AND COALESCE(NULLIF(factory_id, 0), 1)=@FactoryId
               AND calc_batch_id=@BatchId
               AND is_current_flag=1
            """,
            new SugarParameter("@BatchId", batchId),
            new SugarParameter("@TenantId", scope.TenantId),
            new SugarParameter("@FactoryId", scope.FactoryId));
    }

    private async Task<int> BuildS1KpiValuesAsync(S1MdpRunScope scope, string batchId, DateTime now, CancellationToken cancellationToken)
    {
        var affected = 0;
        for (var dayOffset = 13; dayOffset >= 0; dayOffset--)
        {
            var statDate = now.Date.AddDays(-dayOffset);
            var rows = await CalculateS1KpiValuesAsync(scope, batchId, statDate);
            foreach (var row in rows)
            {
                cancellationToken.ThrowIfCancellationRequested();
                if (row.TenantId != scope.TenantId || row.FactoryId != scope.FactoryId)
                    continue;
                affected += await UpsertS1KpiValueAsync(row, statDate, now);
            }
        }

        return affected;
    }

    private async Task<List<S1KpiCalcRow>> CalculateS1KpiValuesAsync(S1MdpRunScope scope, string batchId, DateTime statDate)
    {
        return await _db.Ado.SqlQueryAsync<S1KpiCalcRow>(
            """
            SELECT tenant_id AS TenantId, COALESCE(NULLIF(factory_id, 0), 1) AS FactoryId,
                   'S1_L1_001' AS MetricCode,
                   ROUND(AVG(TIMESTAMPDIFF(HOUR, order_date, COALESCE(promised_delivery_date, capacity_date, material_ready_date, plan_delivery_date)) / 24), 4) AS MetricValue
            FROM mdp_std_so
            WHERE tenant_id=@TenantId AND COALESCE(NULLIF(factory_id, 0), 1)=@FactoryId
              AND order_date IS NOT NULL
              AND COALESCE(promised_delivery_date, capacity_date, material_ready_date, plan_delivery_date) IS NOT NULL
              AND COALESCE(promised_delivery_date, capacity_date, material_ready_date, plan_delivery_date) >= order_date
            GROUP BY tenant_id, COALESCE(NULLIF(factory_id, 0), 1)
            UNION ALL
            SELECT tenant_id AS TenantId, COALESCE(NULLIF(factory_id, 0), 1) AS FactoryId,
                   'S1_L1_002' AS MetricCode,
                   ROUND(100 * SUM(CASE WHEN TIMESTAMPDIFF(HOUR, order_date, COALESCE(promised_delivery_date, capacity_date, material_ready_date, plan_delivery_date)) <= 72 THEN 1 ELSE 0 END) / COUNT(1), 4) AS MetricValue
            FROM mdp_std_so
            WHERE tenant_id=@TenantId AND COALESCE(NULLIF(factory_id, 0), 1)=@FactoryId
              AND order_date IS NOT NULL
              AND COALESCE(promised_delivery_date, capacity_date, material_ready_date, plan_delivery_date) IS NOT NULL
            GROUP BY tenant_id, COALESCE(NULLIF(factory_id, 0), 1)
            UNION ALL
            SELECT tenant_id AS TenantId, COALESCE(NULLIF(factory_id, 0), 1) AS FactoryId,
                   'S1_L1_003' AS MetricCode,
                   ROUND(COUNT(1) / GREATEST(COUNT(DISTINCT NULLIF(planner_no, '')), 1), 4) AS MetricValue
            FROM mdp_std_so
            WHERE tenant_id=@TenantId AND COALESCE(NULLIF(factory_id, 0), 1)=@FactoryId
              AND order_date IS NOT NULL AND order_date <= @StatDate
            GROUP BY tenant_id, COALESCE(NULLIF(factory_id, 0), 1)
            UNION ALL
            SELECT tenant_id AS TenantId, COALESCE(NULLIF(factory_id, 0), 1) AS FactoryId,
                   'S1_L1_004' AS MetricCode,
                   ROUND(AVG(GREATEST(IFNULL(remaining_qty, 0), 0)) / GREATEST(SUM(IFNULL(shipped_qty, 0)), 1) * 30, 4) AS MetricValue
            FROM dwd_ship_trans
            WHERE calc_batch_id=@BatchId
              AND tenant_id=@TenantId AND COALESCE(NULLIF(factory_id, 0), 1)=@FactoryId
            GROUP BY tenant_id, COALESCE(NULLIF(factory_id, 0), 1)
            HAVING SUM(IFNULL(shipped_qty, 0)) > 0
            UNION ALL
            SELECT tenant_id AS TenantId, COALESCE(NULLIF(factory_id, 0), 1) AS FactoryId,
                   'S1_L2_010' AS MetricCode,
                   ROUND(AVG(TIMESTAMPDIFF(HOUR, order_date, COALESCE(promised_delivery_date, capacity_date, material_ready_date, plan_delivery_date)) / 24), 4) AS MetricValue
            FROM mdp_std_so
            WHERE tenant_id=@TenantId AND COALESCE(NULLIF(factory_id, 0), 1)=@FactoryId
              AND order_date IS NOT NULL
              AND COALESCE(promised_delivery_date, capacity_date, material_ready_date, plan_delivery_date) IS NOT NULL
              AND COALESCE(promised_delivery_date, capacity_date, material_ready_date, plan_delivery_date) >= order_date
            GROUP BY tenant_id, COALESCE(NULLIF(factory_id, 0), 1)
            UNION ALL
            SELECT tenant_id AS TenantId, COALESCE(NULLIF(factory_id, 0), 1) AS FactoryId,
                   'S1_L2_011' AS MetricCode,
                   ROUND(100 * SUM(CASE WHEN TIMESTAMPDIFF(HOUR, order_date, COALESCE(promised_delivery_date, capacity_date, material_ready_date, plan_delivery_date)) <= 72 THEN 1 ELSE 0 END) / COUNT(1), 4) AS MetricValue
            FROM mdp_std_so
            WHERE tenant_id=@TenantId AND COALESCE(NULLIF(factory_id, 0), 1)=@FactoryId
              AND order_date IS NOT NULL
              AND COALESCE(promised_delivery_date, capacity_date, material_ready_date, plan_delivery_date) IS NOT NULL
            GROUP BY tenant_id, COALESCE(NULLIF(factory_id, 0), 1)
            UNION ALL
            SELECT tenant_id AS TenantId, COALESCE(NULLIF(factory_id, 0), 1) AS FactoryId,
                   'S1_L2_012' AS MetricCode,
                   ROUND(COUNT(1) / GREATEST(COUNT(DISTINCT NULLIF(planner_no, '')), 1), 4) AS MetricValue
            FROM mdp_std_so
            WHERE tenant_id=@TenantId AND COALESCE(NULLIF(factory_id, 0), 1)=@FactoryId
              AND order_date IS NOT NULL AND order_date <= @StatDate
            GROUP BY tenant_id, COALESCE(NULLIF(factory_id, 0), 1)
            UNION ALL
            SELECT tenant_id AS TenantId, COALESCE(NULLIF(factory_id, 0), 1) AS FactoryId,
                   'S1_L2_013' AS MetricCode,
                   ROUND(100 * SUM(CASE WHEN planned_ship_qty > 0 THEN 1 ELSE 0 END) / COUNT(1), 4) AS MetricValue
            FROM dwd_ship_trans
            WHERE calc_batch_id=@BatchId
              AND tenant_id=@TenantId AND COALESCE(NULLIF(factory_id, 0), 1)=@FactoryId
            GROUP BY tenant_id, COALESCE(NULLIF(factory_id, 0), 1)
            UNION ALL
            SELECT tenant_id AS TenantId, COALESCE(NULLIF(factory_id, 0), 1) AS FactoryId,
                   'S1_L2_014' AS MetricCode,
                   ROUND(100 * SUM(CASE WHEN shipped_qty >= planned_ship_qty AND planned_ship_qty > 0 THEN 1 ELSE 0 END)
                         / GREATEST(SUM(CASE WHEN planned_ship_qty > 0 THEN 1 ELSE 0 END), 1), 4) AS MetricValue
            FROM dwd_ship_trans
            WHERE calc_batch_id=@BatchId
              AND tenant_id=@TenantId AND COALESCE(NULLIF(factory_id, 0), 1)=@FactoryId
            GROUP BY tenant_id, COALESCE(NULLIF(factory_id, 0), 1)
            HAVING SUM(CASE WHEN planned_ship_qty > 0 THEN 1 ELSE 0 END) > 0
            UNION ALL
            SELECT tenant_id AS TenantId, COALESCE(NULLIF(factory_id, 0), 1) AS FactoryId,
                   'S1_L2_015' AS MetricCode,
                   ROUND(100 * SUM(CASE WHEN linkage_status = 'LINKED' THEN 1 ELSE 0 END)
                         / GREATEST(SUM(CASE WHEN planned_ship_qty > 0 THEN 1 ELSE 0 END), 1), 4) AS MetricValue
            FROM dwd_ship_trans
            WHERE calc_batch_id=@BatchId
              AND tenant_id=@TenantId AND COALESCE(NULLIF(factory_id, 0), 1)=@FactoryId
            GROUP BY tenant_id, COALESCE(NULLIF(factory_id, 0), 1)
            HAVING SUM(CASE WHEN planned_ship_qty > 0 THEN 1 ELSE 0 END) > 0
            UNION ALL
            SELECT tenant_id AS TenantId, 1 AS FactoryId,
                   'S1_L2_001' AS MetricCode,
                   ROUND(AVG(TIMESTAMPDIFF(HOUR, create_time, update_time) / 24), 4) AS MetricValue
            FROM (
                SELECT tenant_id,
                       STR_TO_DATE(REPLACE(NULLIF(NULLIF(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.CreateTime')), 'null'), ''), 'T', ' '), '%Y-%m-%d %H:%i:%s.%f') AS create_time,
                       COALESCE(
                           STR_TO_DATE(REPLACE(NULLIF(NULLIF(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.UpdateTime')), 'null'), ''), 'T', ' '), '%Y-%m-%d %H:%i:%s.%f'),
                           STR_TO_DATE(REPLACE(NULLIF(NULLIF(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.CreateTime')), 'null'), ''), 'T', ' '), '%Y-%m-%d %H:%i:%s.%f')
                       ) AS update_time
                FROM mdp_stg_so
                WHERE tenant_id=@TenantId AND COALESCE(NULLIF(factory_id, 0), 1)=@FactoryId
                  AND source_table = 'ado_contract_review'
                  AND COALESCE(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.BillNo')),'') NOT LIKE 'UAT-PROC-%'
                  AND COALESCE(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.BillNo')),'') NOT LIKE 'UAT-DEL-%'
            ) t
            WHERE create_time IS NOT NULL AND update_time IS NOT NULL AND update_time >= create_time
            GROUP BY tenant_id
            UNION ALL
            SELECT tenant_id AS TenantId, 1 AS FactoryId,
                   'S1_L2_002' AS MetricCode,
                   ROUND(100 * SUM(CASE WHEN TIMESTAMPDIFF(HOUR, create_time, update_time) <= 72 THEN 1 ELSE 0 END) / COUNT(1), 4) AS MetricValue
            FROM (
                SELECT tenant_id,
                       STR_TO_DATE(REPLACE(NULLIF(NULLIF(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.CreateTime')), 'null'), ''), 'T', ' '), '%Y-%m-%d %H:%i:%s.%f') AS create_time,
                       COALESCE(
                           STR_TO_DATE(REPLACE(NULLIF(NULLIF(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.UpdateTime')), 'null'), ''), 'T', ' '), '%Y-%m-%d %H:%i:%s.%f'),
                           STR_TO_DATE(REPLACE(NULLIF(NULLIF(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.CreateTime')), 'null'), ''), 'T', ' '), '%Y-%m-%d %H:%i:%s.%f')
                       ) AS update_time
                FROM mdp_stg_so
                WHERE tenant_id=@TenantId AND COALESCE(NULLIF(factory_id, 0), 1)=@FactoryId
                  AND source_table = 'ado_contract_review'
                  AND COALESCE(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.BillNo')),'') NOT LIKE 'UAT-PROC-%'
                  AND COALESCE(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.BillNo')),'') NOT LIKE 'UAT-DEL-%'
            ) t
            WHERE create_time IS NOT NULL AND update_time IS NOT NULL AND update_time >= create_time
            GROUP BY tenant_id
            UNION ALL
            SELECT tenant_id AS TenantId, 1 AS FactoryId,
                   'S1_L2_003' AS MetricCode,
                   ROUND(COUNT(1) / GREATEST(COUNT(DISTINCT NULLIF(owner_account, '')), 1), 4) AS MetricValue
            FROM (
                SELECT tenant_id,
                       COALESCE(
                           JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.ResponsibleAccount')),
                           JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.CreateUser'))
                       ) AS owner_account
                FROM mdp_stg_so
                WHERE tenant_id=@TenantId AND COALESCE(NULLIF(factory_id, 0), 1)=@FactoryId
                  AND source_table = 'ado_contract_review'
                  AND COALESCE(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.BillNo')),'') NOT LIKE 'UAT-PROC-%'
                  AND COALESCE(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.BillNo')),'') NOT LIKE 'UAT-DEL-%'
            ) t
            GROUP BY tenant_id
            UNION ALL
            SELECT tenant_id AS TenantId, 1 AS FactoryId,
                   'S1_L2_007' AS MetricCode,
                   ROUND(AVG(TIMESTAMPDIFF(HOUR,create_time,update_time)/24),4) AS MetricValue
            FROM (
                SELECT tenant_id,
                       STR_TO_DATE(REPLACE(NULLIF(NULLIF(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.CreateTime')),'null'),''),'T',' '),'%Y-%m-%d %H:%i:%s.%f') AS create_time,
                       STR_TO_DATE(REPLACE(NULLIF(NULLIF(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.UpdateTime')),'null'),''),'T',' '),'%Y-%m-%d %H:%i:%s.%f') AS update_time
                FROM mdp_stg_so
                WHERE tenant_id=@TenantId AND COALESCE(NULLIF(factory_id,0),1)=@FactoryId
                  AND source_table='ado_contract_review'
                  AND COALESCE(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.BillNo')),'') LIKE 'UAT-PROC-%'
            ) t
            WHERE create_time IS NOT NULL AND update_time>=create_time
            GROUP BY tenant_id
            UNION ALL
            SELECT tenant_id AS TenantId, 1 AS FactoryId,
                   'S1_L2_008' AS MetricCode,
                   ROUND(100*SUM(CASE WHEN TIMESTAMPDIFF(HOUR,create_time,update_time)<=72 THEN 1 ELSE 0 END)
                         / NULLIF(COUNT(*),0),4) AS MetricValue
            FROM (
                SELECT tenant_id,
                       STR_TO_DATE(REPLACE(NULLIF(NULLIF(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.CreateTime')),'null'),''),'T',' '),'%Y-%m-%d %H:%i:%s.%f') AS create_time,
                       STR_TO_DATE(REPLACE(NULLIF(NULLIF(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.UpdateTime')),'null'),''),'T',' '),'%Y-%m-%d %H:%i:%s.%f') AS update_time
                FROM mdp_stg_so
                WHERE tenant_id=@TenantId AND COALESCE(NULLIF(factory_id,0),1)=@FactoryId
                  AND source_table='ado_contract_review'
                  AND COALESCE(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.BillNo')),'') LIKE 'UAT-PROC-%'
            ) t
            WHERE create_time IS NOT NULL AND update_time IS NOT NULL
            GROUP BY tenant_id
            UNION ALL
            SELECT tenant_id AS TenantId, 1 AS FactoryId,
                   'S1_L2_009' AS MetricCode,
                   ROUND(COUNT(*)/NULLIF(COUNT(DISTINCT NULLIF(owner_account,'')),0),4) AS MetricValue
            FROM (
                SELECT tenant_id,
                       COALESCE(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.ResponsibleAccount')),
                                JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.CreateUser'))) AS owner_account
                FROM mdp_stg_so
                WHERE tenant_id=@TenantId AND COALESCE(NULLIF(factory_id,0),1)=@FactoryId
                  AND source_table='ado_contract_review'
                  AND COALESCE(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.BillNo')),'') LIKE 'UAT-PROC-%'
            ) t
            GROUP BY tenant_id
            UNION ALL
            SELECT tenant_id AS TenantId, 1 AS FactoryId,
                   'S1_L2_004' AS MetricCode,
                   ROUND(AVG(cycle_hours / 24), 4) AS MetricValue
            FROM (
                SELECT tenant_id,
                       COALESCE(
                           CASE WHEN JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.DrawingDesignCycle')) REGEXP '^-?[0-9]+$'
                                THEN CAST(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.DrawingDesignCycle')) AS DECIMAL(18,6)) END,
                           TIMESTAMPDIFF(HOUR,
                               STR_TO_DATE(REPLACE(NULLIF(NULLIF(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.DrawingActualStart')), 'null'), ''), 'T', ' '), '%Y-%m-%d %H:%i:%s.%f'),
                               STR_TO_DATE(REPLACE(NULLIF(NULLIF(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.DrawingActualEnd')), 'null'), ''), 'T', ' '), '%Y-%m-%d %H:%i:%s.%f')
                           )
                       ) AS cycle_hours
                FROM mdp_stg_so
                WHERE tenant_id=@TenantId AND COALESCE(NULLIF(factory_id, 0), 1)=@FactoryId
                  AND source_table = 'ado_product_design'
            ) t
            WHERE cycle_hours IS NOT NULL AND cycle_hours >= 0
            GROUP BY tenant_id
            UNION ALL
            SELECT tenant_id AS TenantId, 1 AS FactoryId,
                   'S1_L2_005' AS MetricCode,
                   ROUND(100 * SUM(CASE WHEN actual_end <= plan_end THEN 1 ELSE 0 END) / COUNT(1), 4) AS MetricValue
            FROM (
                SELECT tenant_id,
                       STR_TO_DATE(REPLACE(NULLIF(NULLIF(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.DrawingPlanEnd')), 'null'), ''), 'T', ' '), '%Y-%m-%d %H:%i:%s.%f') AS plan_end,
                       STR_TO_DATE(REPLACE(NULLIF(NULLIF(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.DrawingActualEnd')), 'null'), ''), 'T', ' '), '%Y-%m-%d %H:%i:%s.%f') AS actual_end
                FROM mdp_stg_so
                WHERE tenant_id=@TenantId AND COALESCE(NULLIF(factory_id, 0), 1)=@FactoryId
                  AND source_table = 'ado_product_design'
            ) t
            WHERE plan_end IS NOT NULL AND actual_end IS NOT NULL
            GROUP BY tenant_id
            UNION ALL
            SELECT tenant_id AS TenantId, 1 AS FactoryId,
                   'S1_L2_006' AS MetricCode,
                   ROUND(COUNT(1) / GREATEST(COUNT(DISTINCT NULLIF(design_lead, '')), 1), 4) AS MetricValue
            FROM (
                SELECT tenant_id,
                       COALESCE(
                           JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.DesignLeadAccount')),
                           JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.DesignLeadName')),
                           JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.CreateUser'))
                       ) AS design_lead
                FROM mdp_stg_so
                WHERE tenant_id=@TenantId AND COALESCE(NULLIF(factory_id, 0), 1)=@FactoryId
                  AND source_table = 'ado_product_design'
            ) t
            GROUP BY tenant_id
            UNION ALL
            SELECT tenant_id AS TenantId, 1 AS FactoryId,
                   CASE stage_no
                       WHEN 1 THEN 'S1_L3_001'
                       WHEN 2 THEN 'S1_L3_002'
                       WHEN 3 THEN 'S1_L3_003'
                       WHEN 4 THEN 'S1_L3_004'
                       WHEN 5 THEN 'S1_L3_005'
                   END AS MetricCode,
                   ROUND(AVG(cycle_hours), 4) AS MetricValue
            FROM (
                SELECT tenant_id,
                       CASE WHEN JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.StageNo')) REGEXP '^-?[0-9]+$'
                            THEN CAST(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.StageNo')) AS SIGNED) END AS stage_no,
                       COALESCE(
                           TIMESTAMPDIFF(HOUR,
                               STR_TO_DATE(REPLACE(NULLIF(NULLIF(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.StartTime')), 'null'), ''), 'T', ' '), '%Y-%m-%d %H:%i:%s.%f'),
                               STR_TO_DATE(REPLACE(NULLIF(NULLIF(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.CompleteTime')), 'null'), ''), 'T', ' '), '%Y-%m-%d %H:%i:%s.%f')
                           ),
                           CASE WHEN JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.ActualDays')) REGEXP '^-?[0-9]+(\\.[0-9]+)?$'
                                THEN CAST(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.ActualDays')) AS DECIMAL(18,6)) * 24 END
                       ) AS cycle_hours
                FROM mdp_stg_so
                WHERE tenant_id=@TenantId AND COALESCE(NULLIF(factory_id, 0), 1)=@FactoryId
                  AND source_table = 'ado_contract_review_flow'
                  AND COALESCE(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.ReviewBillNo')),'') NOT LIKE 'UAT-PROC-%'
                  AND COALESCE(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.ReviewBillNo')),'') NOT LIKE 'UAT-DEL-%'
            ) t
            WHERE stage_no BETWEEN 1 AND 5
              AND cycle_hours IS NOT NULL AND cycle_hours >= 0
            GROUP BY tenant_id, stage_no
            UNION ALL
            SELECT tenant_id AS TenantId, 1 AS FactoryId,
                   CASE stage_no
                       WHEN 1 THEN 'S1_L3_101'
                       WHEN 2 THEN 'S1_L3_102'
                       WHEN 3 THEN 'S1_L3_103'
                       WHEN 4 THEN 'S1_L3_104'
                       WHEN 5 THEN 'S1_L3_105'
                   END AS MetricCode,
                   ROUND(AVG(cycle_hours), 4) AS MetricValue
            FROM (
                SELECT tenant_id,
                       CASE WHEN JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.StageNo')) REGEXP '^-?[0-9]+$'
                            THEN CAST(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.StageNo')) AS SIGNED) END AS stage_no,
                       COALESCE(
                           TIMESTAMPDIFF(HOUR,
                               STR_TO_DATE(REPLACE(NULLIF(NULLIF(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.StartTime')), 'null'), ''), 'T', ' '), '%Y-%m-%d %H:%i:%s.%f'),
                               STR_TO_DATE(REPLACE(NULLIF(NULLIF(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.CompleteTime')), 'null'), ''), 'T', ' '), '%Y-%m-%d %H:%i:%s.%f')
                           ),
                           CASE WHEN JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.ActualDays')) REGEXP '^-?[0-9]+(\\.[0-9]+)?$'
                                THEN CAST(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.ActualDays')) AS DECIMAL(18,6)) * 24 END
                       ) AS cycle_hours
                FROM mdp_stg_so
                WHERE tenant_id=@TenantId AND COALESCE(NULLIF(factory_id, 0), 1)=@FactoryId
                  AND source_table = 'ado_contract_review_flow'
                  AND COALESCE(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.ReviewBillNo')),'') NOT LIKE 'UAT-PROC-%'
                  AND COALESCE(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.ReviewBillNo')),'') NOT LIKE 'UAT-DEL-%'
            ) t
            WHERE stage_no BETWEEN 1 AND 5
              AND cycle_hours IS NOT NULL AND cycle_hours >= 0
            GROUP BY tenant_id, stage_no
            UNION ALL
            SELECT tenant_id AS TenantId, 1 AS FactoryId,
                   metric_code AS MetricCode,
                   ROUND(AVG(cycle_hours), 4) AS MetricValue
            FROM (
                SELECT tenant_id,
                       CASE COALESCE(NULLIF(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.DeptNo')), ''), JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.Department')))
                           WHEN 'LAW' THEN 'S1_L4_001'
                           WHEN 'PRE_SALES' THEN 'S1_L4_002'
                           WHEN 'MPS' THEN 'S1_L4_003'
                           WHEN 'TEST' THEN 'S1_L4_004'
                           WHEN '法律事务部' THEN 'S1_L4_001'
                           WHEN '技术售前组' THEN 'S1_L4_002'
                           WHEN '综合主计划' THEN 'S1_L4_003'
                           WHEN '试验站' THEN 'S1_L4_004'
                       END AS metric_code,
                       COALESCE(
                           TIMESTAMPDIFF(HOUR,
                               STR_TO_DATE(REPLACE(NULLIF(NULLIF(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.StartTime')), 'null'), ''), 'T', ' '), '%Y-%m-%d %H:%i:%s.%f'),
                               STR_TO_DATE(REPLACE(NULLIF(NULLIF(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.CompleteTime')), 'null'), ''), 'T', ' '), '%Y-%m-%d %H:%i:%s.%f')
                           ),
                           CASE WHEN JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.ActualDays')) REGEXP '^-?[0-9]+(\\.[0-9]+)?$'
                                THEN CAST(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.ActualDays')) AS DECIMAL(18,6)) * 24 END
                       ) AS cycle_hours
                FROM mdp_stg_so
                WHERE tenant_id=@TenantId AND COALESCE(NULLIF(factory_id, 0), 1)=@FactoryId
                  AND source_table = 'ado_contract_review_flow'
                  AND JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.StageNo')) = '1'
                  AND COALESCE(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.ReviewBillNo')),'') NOT LIKE 'UAT-PROC-%'
                  AND COALESCE(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.ReviewBillNo')),'') NOT LIKE 'UAT-DEL-%'
            ) t
            WHERE metric_code IS NOT NULL
              AND cycle_hours IS NOT NULL AND cycle_hours >= 0
            GROUP BY tenant_id, metric_code
            UNION ALL
            SELECT tenant_id AS TenantId, 1 AS FactoryId,
                   metric_code AS MetricCode,
                   ROUND(AVG(cycle_hours), 4) AS MetricValue
            FROM (
                SELECT tenant_id,
                       CASE COALESCE(NULLIF(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.DeptNo')), ''), JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.Department')))
                           WHEN 'LAW' THEN 'S1_L4_101'
                           WHEN 'PRE_SALES' THEN 'S1_L4_102'
                           WHEN 'MPS' THEN 'S1_L4_103'
                           WHEN 'TEST' THEN 'S1_L4_104'
                           WHEN '法律事务部' THEN 'S1_L4_101'
                           WHEN '技术售前组' THEN 'S1_L4_102'
                           WHEN '综合主计划' THEN 'S1_L4_103'
                           WHEN '试验站' THEN 'S1_L4_104'
                       END AS metric_code,
                       COALESCE(
                           TIMESTAMPDIFF(HOUR,
                               STR_TO_DATE(REPLACE(NULLIF(NULLIF(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.StartTime')), 'null'), ''), 'T', ' '), '%Y-%m-%d %H:%i:%s.%f'),
                               STR_TO_DATE(REPLACE(NULLIF(NULLIF(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.CompleteTime')), 'null'), ''), 'T', ' '), '%Y-%m-%d %H:%i:%s.%f')
                           ),
                           CASE WHEN JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.ActualDays')) REGEXP '^-?[0-9]+(\\.[0-9]+)?$'
                                THEN CAST(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.ActualDays')) AS DECIMAL(18,6)) * 24 END
                       ) AS cycle_hours
                FROM mdp_stg_so
                WHERE tenant_id=@TenantId AND COALESCE(NULLIF(factory_id, 0), 1)=@FactoryId
                  AND source_table = 'ado_contract_review_flow'
                  AND JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.StageNo')) = '1'
                  AND COALESCE(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.ReviewBillNo')),'') NOT LIKE 'UAT-PROC-%'
                  AND COALESCE(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.ReviewBillNo')),'') NOT LIKE 'UAT-DEL-%'
            ) t
            WHERE metric_code IS NOT NULL
              AND cycle_hours IS NOT NULL AND cycle_hours >= 0
            GROUP BY tenant_id, metric_code
            UNION ALL
            SELECT tenant_id AS TenantId, 1 AS FactoryId,
                   CONCAT('S1_L3_',branch_code,LPAD(stage_no,2,'0')) AS MetricCode,
                   ROUND(AVG(cycle_hours),4) AS MetricValue
            FROM (
                SELECT tenant_id,
                       CASE
                           WHEN COALESCE(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.ReviewBillNo')),'') LIKE 'UAT-PROC-%' THEN '3'
                           WHEN COALESCE(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.ReviewBillNo')),'') LIKE 'UAT-DEL-%' THEN '4'
                       END AS branch_code,
                       CAST(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.StageNo')) AS SIGNED) AS stage_no,
                       TIMESTAMPDIFF(
                           HOUR,
                           STR_TO_DATE(REPLACE(NULLIF(NULLIF(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.StartTime')),'null'),''),'T',' '),'%Y-%m-%d %H:%i:%s.%f'),
                           STR_TO_DATE(REPLACE(NULLIF(NULLIF(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.CompleteTime')),'null'),''),'T',' '),'%Y-%m-%d %H:%i:%s.%f')
                       ) AS cycle_hours
                FROM mdp_stg_so
                WHERE tenant_id=@TenantId AND COALESCE(NULLIF(factory_id,0),1)=@FactoryId
                  AND source_table='ado_contract_review_flow'
                  AND (COALESCE(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.ReviewBillNo')),'') LIKE 'UAT-PROC-%'
                       OR COALESCE(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.ReviewBillNo')),'') LIKE 'UAT-DEL-%')
            ) t
            WHERE branch_code IS NOT NULL AND stage_no BETWEEN 1 AND 5
              AND cycle_hours IS NOT NULL AND cycle_hours>=0
            GROUP BY tenant_id,branch_code,stage_no
            UNION ALL
            SELECT tenant_id AS TenantId, 1 AS FactoryId,
                   CONCAT('S1_L4_',branch_code,dept_code) AS MetricCode,
                   ROUND(AVG(cycle_hours),4) AS MetricValue
            FROM (
                SELECT tenant_id,
                       CASE
                           WHEN COALESCE(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.ReviewBillNo')),'') LIKE 'UAT-PROC-%' THEN '3'
                           WHEN COALESCE(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.ReviewBillNo')),'') LIKE 'UAT-DEL-%' THEN '4'
                       END AS branch_code,
                       CASE COALESCE(NULLIF(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.DeptNo')),''),
                                              JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.Department')))
                           WHEN 'LAW' THEN '01' WHEN '法律事务部' THEN '01'
                           WHEN 'PRE_SALES' THEN '02' WHEN '技术售前组' THEN '02'
                           WHEN 'MPS' THEN '03' WHEN '综合主计划' THEN '03'
                           WHEN 'TEST' THEN '04' WHEN '试验站' THEN '04'
                       END AS dept_code,
                       TIMESTAMPDIFF(
                           HOUR,
                           STR_TO_DATE(REPLACE(NULLIF(NULLIF(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.StartTime')),'null'),''),'T',' '),'%Y-%m-%d %H:%i:%s.%f'),
                           STR_TO_DATE(REPLACE(NULLIF(NULLIF(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.CompleteTime')),'null'),''),'T',' '),'%Y-%m-%d %H:%i:%s.%f')
                       ) AS cycle_hours
                FROM mdp_stg_so
                WHERE tenant_id=@TenantId AND COALESCE(NULLIF(factory_id,0),1)=@FactoryId
                  AND source_table='ado_contract_review_flow'
                  AND JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.StageNo'))='1'
                  AND (COALESCE(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.ReviewBillNo')),'') LIKE 'UAT-PROC-%'
                       OR COALESCE(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.ReviewBillNo')),'') LIKE 'UAT-DEL-%')
            ) t
            WHERE branch_code IS NOT NULL AND dept_code IS NOT NULL
              AND cycle_hours IS NOT NULL AND cycle_hours>=0
            GROUP BY tenant_id,branch_code,dept_code
            """,
            new SugarParameter("@BatchId", batchId),
            new SugarParameter("@StatDate", statDate),
            new SugarParameter("@TenantId", scope.TenantId),
            new SugarParameter("@FactoryId", scope.FactoryId));
    }

    private async Task<int> UpsertS1KpiValueAsync(S1KpiCalcRow row, DateTime statDate, DateTime now)
    {
        var meta = await _db.Ado.SqlQuerySingleAsync<S1KpiMetaRow>(
            """
            SELECT MetricLevel, Direction, YellowThreshold, RedThreshold
            FROM ado_smart_ops_kpi_master
            WHERE TenantId=@TenantId AND ModuleCode='S1' 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<S1KpiValueRow>(
            $"""
            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='S1'
              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<S1KpiValueRow>(
            $"""
            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='S1'
              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, "S1", 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 {Admin.NET.Plugin.AiDOP.SmartOps.KpiResultWrite.FromMetricSet}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='S1'
                  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, result_status, result_reason, 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,
             'S1', @MetricCode, {Admin.NET.Plugin.AiDOP.SmartOps.KpiResultWrite.FromMetricVals}@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 IEnumerable<S1MdpSqlCommand> BuildStandardCommands(S1MdpRunScope scope, string batchId, DateTime now)
    {
        yield return Cmd(
            """
            INSERT INTO mdp_std_so
            (tenant_id, factory_id, company_id, source_system, order_id, order_entry_id, order_no, order_line, order_type,
             customer_id, customer_no, customer_name, customer_order_no, country, item_code, item_name, item_spec,
             map_number, map_name, bom_number, unit, order_qty, delivered_notice_qty, delivered_qty, price, tax_price,
             amount, total_amount, order_date, customer_request_date, plan_delivery_date, promised_delivery_date,
             capacity_date, material_ready_date, planner_no, planner_name, order_status, review_status, review_stage,
             flow_state, progress, urgent, closed, deleted_flag, source_table, source_row_id, source_biz_key, sync_batch_id, sync_time)
            SELECT
                COALESCE(NULLIF(e.tenant_id, 0), NULLIF(h.tenant_id, 0),
                         CASE WHEN JSON_UNQUOTE(JSON_EXTRACT(e.raw_data,'$.tenant_id')) REGEXP '^[0-9]+$'
                              THEN CAST(JSON_UNQUOTE(JSON_EXTRACT(e.raw_data,'$.tenant_id')) AS SIGNED) END),
                COALESCE(NULLIF(NULLIF(e.factory_id, 0), e.tenant_id),
                         NULLIF(NULLIF(h.factory_id, 0), h.tenant_id)),
                NULLIF(e.company_id, 0),
                COALESCE(NULLIF(e.source_system,''), (SELECT r.source_system FROM mdp_source_table_registry r WHERE r.source_table = e.source_table LIMIT 1), ''),
                CASE WHEN JSON_UNQUOTE(JSON_EXTRACT(h.raw_data,'$.Id')) REGEXP '^-?[0-9]+$' THEN CAST(JSON_UNQUOTE(JSON_EXTRACT(h.raw_data,'$.Id')) AS SIGNED) END,
                CASE WHEN JSON_UNQUOTE(JSON_EXTRACT(e.raw_data,'$.Id')) REGEXP '^-?[0-9]+$' THEN CAST(JSON_UNQUOTE(JSON_EXTRACT(e.raw_data,'$.Id')) AS SIGNED) END,
                COALESCE(JSON_UNQUOTE(JSON_EXTRACT(e.raw_data,'$.bill_no')), JSON_UNQUOTE(JSON_EXTRACT(h.raw_data,'$.bill_no')), e.source_biz_key),
                CAST(COALESCE(JSON_UNQUOTE(JSON_EXTRACT(e.raw_data,'$.entry_seq')), JSON_UNQUOTE(JSON_EXTRACT(e.raw_data,'$.Id'))) AS CHAR),
                CAST(JSON_UNQUOTE(JSON_EXTRACT(h.raw_data,'$.order_type')) AS CHAR),
                CASE WHEN JSON_UNQUOTE(JSON_EXTRACT(h.raw_data,'$.custom_id')) REGEXP '^-?[0-9]+$' THEN CAST(JSON_UNQUOTE(JSON_EXTRACT(h.raw_data,'$.custom_id')) AS SIGNED) END,
                JSON_UNQUOTE(JSON_EXTRACT(h.raw_data,'$.custom_no')),
                JSON_UNQUOTE(JSON_EXTRACT(h.raw_data,'$.custom_name')),
                COALESCE(JSON_UNQUOTE(JSON_EXTRACT(e.raw_data,'$.custom_order_bill_no')), JSON_UNQUOTE(JSON_EXTRACT(h.raw_data,'$.bill_from'))),
                JSON_UNQUOTE(JSON_EXTRACT(h.raw_data,'$.country')),
                JSON_UNQUOTE(JSON_EXTRACT(e.raw_data,'$.item_number')),
                JSON_UNQUOTE(JSON_EXTRACT(e.raw_data,'$.item_name')),
                JSON_UNQUOTE(JSON_EXTRACT(e.raw_data,'$.specification')),
                JSON_UNQUOTE(JSON_EXTRACT(e.raw_data,'$.map_number')),
                JSON_UNQUOTE(JSON_EXTRACT(e.raw_data,'$.map_name')),
                JSON_UNQUOTE(JSON_EXTRACT(e.raw_data,'$.bom_number')),
                JSON_UNQUOTE(JSON_EXTRACT(e.raw_data,'$.unit')),
                CASE WHEN JSON_UNQUOTE(JSON_EXTRACT(e.raw_data,'$.qty')) REGEXP '^-?[0-9]+(\\.[0-9]+)?$' THEN CAST(JSON_UNQUOTE(JSON_EXTRACT(e.raw_data,'$.qty')) AS DECIMAL(18,6)) END,
                CASE WHEN JSON_UNQUOTE(JSON_EXTRACT(e.raw_data,'$.deliver_notice_count')) REGEXP '^-?[0-9]+(\\.[0-9]+)?$' THEN CAST(JSON_UNQUOTE(JSON_EXTRACT(e.raw_data,'$.deliver_notice_count')) AS DECIMAL(18,6)) END,
                CASE WHEN JSON_UNQUOTE(JSON_EXTRACT(e.raw_data,'$.deliver_count')) REGEXP '^-?[0-9]+(\\.[0-9]+)?$' THEN CAST(JSON_UNQUOTE(JSON_EXTRACT(e.raw_data,'$.deliver_count')) AS DECIMAL(18,6)) END,
                CASE WHEN JSON_UNQUOTE(JSON_EXTRACT(e.raw_data,'$.price')) REGEXP '^-?[0-9]+(\\.[0-9]+)?$' THEN CAST(JSON_UNQUOTE(JSON_EXTRACT(e.raw_data,'$.price')) AS DECIMAL(18,6)) END,
                CASE WHEN JSON_UNQUOTE(JSON_EXTRACT(e.raw_data,'$.tax_price')) REGEXP '^-?[0-9]+(\\.[0-9]+)?$' THEN CAST(JSON_UNQUOTE(JSON_EXTRACT(e.raw_data,'$.tax_price')) AS DECIMAL(18,6)) END,
                CASE WHEN JSON_UNQUOTE(JSON_EXTRACT(e.raw_data,'$.amount')) REGEXP '^-?[0-9]+(\\.[0-9]+)?$' THEN CAST(JSON_UNQUOTE(JSON_EXTRACT(e.raw_data,'$.amount')) AS DECIMAL(18,6)) END,
                CASE WHEN JSON_UNQUOTE(JSON_EXTRACT(e.raw_data,'$.total_amount')) REGEXP '^-?[0-9]+(\\.[0-9]+)?$' THEN CAST(JSON_UNQUOTE(JSON_EXTRACT(e.raw_data,'$.total_amount')) AS DECIMAL(18,6)) END,
                NULLIF(NULLIF(JSON_UNQUOTE(JSON_EXTRACT(h.raw_data,'$.date')), 'null'), ''),
                NULLIF(NULLIF(JSON_UNQUOTE(JSON_EXTRACT(h.raw_data,'$.rdate')), 'null'), ''),
                NULLIF(NULLIF(JSON_UNQUOTE(JSON_EXTRACT(e.raw_data,'$.plan_date')), 'null'), ''),
                NULLIF(NULLIF(JSON_UNQUOTE(JSON_EXTRACT(e.raw_data,'$.date')), 'null'), ''),
                NULLIF(NULLIF(JSON_UNQUOTE(JSON_EXTRACT(e.raw_data,'$.sys_capacity_date')), 'null'), ''),
                NULLIF(NULLIF(JSON_UNQUOTE(JSON_EXTRACT(e.raw_data,'$.sys_material_date')), 'null'), ''),
                JSON_UNQUOTE(JSON_EXTRACT(e.raw_data,'$.planner_no')),
                JSON_UNQUOTE(JSON_EXTRACT(e.raw_data,'$.planner_name')),
                CASE WHEN JSON_EXTRACT(h.raw_data,'$.closed') IN (1, true) THEN 'CLOSED' ELSE 'OPEN' END,
                COALESCE(JSON_UNQUOTE(JSON_EXTRACT(r.raw_data,'$.FlowStatus')), JSON_UNQUOTE(JSON_EXTRACT(h.raw_data,'$.flowstate'))),
                COALESCE(JSON_UNQUOTE(JSON_EXTRACT(r.raw_data,'$.CurrentDept')), JSON_UNQUOTE(JSON_EXTRACT(r.raw_data,'$.CurrentStage'))),
                JSON_UNQUOTE(JSON_EXTRACT(h.raw_data,'$.flowstate')),
                JSON_UNQUOTE(JSON_EXTRACT(e.raw_data,'$.progress')),
                CASE WHEN JSON_UNQUOTE(JSON_EXTRACT(COALESCE(e.raw_data, h.raw_data),'$.urgent')) REGEXP '^-?[0-9]+$' THEN CAST(JSON_UNQUOTE(JSON_EXTRACT(COALESCE(e.raw_data, h.raw_data),'$.urgent')) AS SIGNED) END,
                CASE WHEN JSON_UNQUOTE(JSON_EXTRACT(h.raw_data,'$.closed')) REGEXP '^-?[0-9]+$' THEN CAST(JSON_UNQUOTE(JSON_EXTRACT(h.raw_data,'$.closed')) AS SIGNED) END,
                COALESCE(CASE WHEN JSON_UNQUOTE(JSON_EXTRACT(COALESCE(e.raw_data, h.raw_data),'$.IsDeleted')) REGEXP '^-?[0-9]+$' THEN CAST(JSON_UNQUOTE(JSON_EXTRACT(COALESCE(e.raw_data, h.raw_data),'$.IsDeleted')) AS SIGNED) END, 0),
                e.source_table,
                e.source_row_id,
                e.source_biz_key,
                @BatchId,
                @Now
            FROM mdp_stg_so e
            LEFT JOIN mdp_stg_so h ON h.source_table='crm_seorder'
                AND h.tenant_id = e.tenant_id
                AND h.sync_batch_id = (SELECT lb.sync_batch_id FROM mdp_stg_so lb
                                       WHERE lb.source_table='crm_seorder' AND lb.tenant_id=@TenantId
                                       ORDER BY lb.sync_time DESC, lb.id DESC LIMIT 1)
                AND JSON_UNQUOTE(JSON_EXTRACT(h.raw_data,'$.Id')) = JSON_UNQUOTE(JSON_EXTRACT(e.raw_data,'$.seorder_id'))
            LEFT JOIN mdp_stg_so r ON r.source_table='ado_contract_review'
                AND r.tenant_id = e.tenant_id
                AND JSON_UNQUOTE(JSON_EXTRACT(r.raw_data,'$.BillNo')) = COALESCE(JSON_UNQUOTE(JSON_EXTRACT(e.raw_data,'$.contract_no')), JSON_UNQUOTE(JSON_EXTRACT(e.raw_data,'$.bill_no')), JSON_UNQUOTE(JSON_EXTRACT(h.raw_data,'$.bill_no')))
            WHERE e.source_table='crm_seorderentry'
              AND e.tenant_id=@TenantId
              AND COALESCE(NULLIF(e.factory_id, 0), 1)=@FactoryId
              -- ── 只取「最新一次 sync_batch_id」，使 mdp_std_so 成为源的真实镜像 ────────────────
              -- 贴源层是纯 upsert、从不淘汰：源侧硬删除的订单行会永远留在 mdp_stg_so 里，
              -- 而本 INSERT 原先不带任何批次条件、每轮重读整张贴源表，于是把这些孤儿一并
              -- 物化进标准层。实测租户 797403760988229：贴源 414 行，源侧实际只剩 28 行，
              -- 其余 386 行中 361 行是源侧已删除的孤儿、25 行是 biz_key 分隔符从 ':' 改成 '#'
              -- 之后被弃用的旧格式重复行（1.0.259 改的 biz_key_expr）。
              --
              -- 这些孤儿不是只影响 S8：mdp_std_so 同时驱动 dwd_ship_trans（每条孤儿凭空
              -- 造出一条早已过期、必然判 DELAYED 的发运行）与 S1/S7/S9 的多项 KPI。
              --
              -- 该等价关系（不在最新批次 = 源侧已删除）成立的前提是该 entity 每轮都全量重读。
              -- 本批已把 S1_SEORDER / S1_SEORDER_ENTRY 的 sync_mode 由 INCR 改为 FULL
              -- （见 UpdateScripts/1.0.521.sql）：MdpSyncWindowResolver 对 FULL 实体会强制
              -- ctx.FullRefresh=true，从而在**任何**调用路径上都跳过增量水位（含
              -- RunInboundAsync 与 MdpHotWatchService 这两条 fullRefresh=false 的路径）。
              -- ⚠️ 若哪天把它们改回 INCREMENTAL，本过滤会把「未变更的存量行」误当孤儿丢掉，
              --    届时必须回到贴源层整体替换的正解，不能只改这里。
              AND e.sync_batch_id = (SELECT lb.sync_batch_id FROM mdp_stg_so lb
                                     WHERE lb.source_table='crm_seorderentry' AND lb.tenant_id=@TenantId
                                     ORDER BY lb.sync_time DESC, lb.id DESC LIMIT 1)
              AND IFNULL(COALESCE(JSON_UNQUOTE(JSON_EXTRACT(e.raw_data,'$.bill_no')), JSON_UNQUOTE(JSON_EXTRACT(h.raw_data,'$.bill_no'))), '') <> ''
              AND COALESCE(NULLIF(e.tenant_id, 0), NULLIF(h.tenant_id, 0),
                         CASE WHEN JSON_UNQUOTE(JSON_EXTRACT(e.raw_data,'$.tenant_id')) REGEXP '^[0-9]+$'
                              THEN CAST(JSON_UNQUOTE(JSON_EXTRACT(e.raw_data,'$.tenant_id')) AS SIGNED) END) IS NOT NULL
            ON DUPLICATE KEY UPDATE
                factory_id=VALUES(factory_id), customer_no=VALUES(customer_no), customer_name=VALUES(customer_name), item_code=VALUES(item_code),
                item_name=VALUES(item_name), item_spec=VALUES(item_spec), order_qty=VALUES(order_qty),
                delivered_notice_qty=VALUES(delivered_notice_qty), delivered_qty=VALUES(delivered_qty),
                plan_delivery_date=VALUES(plan_delivery_date), promised_delivery_date=VALUES(promised_delivery_date),
                capacity_date=VALUES(capacity_date), material_ready_date=VALUES(material_ready_date),
                order_status=VALUES(order_status), review_status=VALUES(review_status), review_stage=VALUES(review_stage),
                flow_state=VALUES(flow_state), progress=VALUES(progress), closed=VALUES(closed), deleted_flag=VALUES(deleted_flag),
                sync_batch_id=VALUES(sync_batch_id), sync_time=VALUES(sync_time), update_time=CURRENT_TIMESTAMP
            """, scope, batchId, now);

        yield return Cmd(
            """
            INSERT INTO mdp_std_ship_trans
            (tenant_id, factory_id, company_id, source_system, trans_type, plan_id, plan_no, plan_line, order_id, order_entry_id,
             order_no, order_line, customer_no, customer_name, country, item_code, item_name, item_spec, qty, plan_qty,
             weight, volume, order_date, plan_ship_date, shipping_site, shipping_address, consignee, telephone,
             status, confirm_status, source_table, source_row_id, source_biz_key, sync_batch_id, sync_time)
            SELECT
                COALESCE(NULLIF(d.tenant_id, 0),
                         CASE WHEN JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.tenant_id')) REGEXP '^[0-9]+$'
                              THEN CAST(JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.tenant_id')) AS SIGNED) END),
                NULLIF(d.factory_id, 0),
                NULLIF(d.company_id, 0),
                COALESCE(NULLIF(d.source_system,''), (SELECT r.source_system FROM mdp_source_table_registry r WHERE r.source_table = d.source_table LIMIT 1), ''),
                'SHIP_PLAN',
                CASE WHEN JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.plan_id')) REGEXP '^-?[0-9]+$' THEN CAST(JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.plan_id')) AS SIGNED) END,
                COALESCE(JSON_UNQUOTE(JSON_EXTRACT(m.raw_data,'$.LotSerial')), CAST(JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.plan_id')) AS CHAR)),
                CAST(JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.RecID')) AS CHAR),
                CASE WHEN JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.seorder_id')) REGEXP '^-?[0-9]+$' THEN CAST(JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.seorder_id')) AS SIGNED) END,
                CASE WHEN JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.sentry_id')) REGEXP '^-?[0-9]+$' THEN CAST(JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.sentry_id')) AS SIGNED) END,
                COALESCE(JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.bill_no')), JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.OrdNbr'))),
                CAST(JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.sentry_id')) AS CHAR),
                JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.CustomNo')),
                JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.CustomName')),
                JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.Country')),
                JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.ItemNum')),
                JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.ItemName')),
                JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.Specification')),
                CASE WHEN JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.Qty')) REGEXP '^-?[0-9]+(\\.[0-9]+)?$' THEN CAST(JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.Qty')) AS DECIMAL(18,6)) END,
                CASE WHEN JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.Qty')) REGEXP '^-?[0-9]+(\\.[0-9]+)?$' THEN CAST(JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.Qty')) AS DECIMAL(18,6)) END,
                CASE WHEN JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.Weight')) REGEXP '^-?[0-9]+(\\.[0-9]+)?$' THEN CAST(JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.Weight')) AS DECIMAL(18,6)) END,
                CASE WHEN JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.Volume')) REGEXP '^-?[0-9]+(\\.[0-9]+)?$' THEN CAST(JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.Volume')) AS DECIMAL(18,6)) END,
                NULLIF(NULLIF(JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.OrdDate')), 'null'), ''),
                NULLIF(NULLIF(COALESCE(JSON_UNQUOTE(JSON_EXTRACT(m.raw_data,'$.ShippingDate')), JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.CreateTime'))), 'null'), ''),
                JSON_UNQUOTE(JSON_EXTRACT(m.raw_data,'$.ShippingSite')),
                JSON_UNQUOTE(JSON_EXTRACT(m.raw_data,'$.ShippingAddress')),
                JSON_UNQUOTE(JSON_EXTRACT(m.raw_data,'$.Consignee')),
                JSON_UNQUOTE(JSON_EXTRACT(m.raw_data,'$.Telephone')),
                COALESCE(JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.Status')), JSON_UNQUOTE(JSON_EXTRACT(m.raw_data,'$.Status'))),
                CAST(COALESCE(JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.IsConfirm')), JSON_UNQUOTE(JSON_EXTRACT(m.raw_data,'$.IsConfirm'))) AS CHAR),
                d.source_table,
                d.source_row_id,
                d.source_biz_key,
                @BatchId,
                @Now
            FROM mdp_stg_ship_trans d
            LEFT JOIN mdp_stg_ship_trans m ON m.source_table='ShippingPlan'
                AND m.tenant_id = d.tenant_id
                AND JSON_UNQUOTE(JSON_EXTRACT(m.raw_data,'$.RecID')) = JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.plan_id'))
            WHERE d.source_table='ShippingPlanDetail'
              AND d.tenant_id=@TenantId
              AND COALESCE(NULLIF(d.factory_id, 0), 1)=@FactoryId
              AND IFNULL(COALESCE(JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.bill_no')), JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.OrdNbr'))), '') <> ''
              AND COALESCE(NULLIF(d.tenant_id, 0),
                         CASE WHEN JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.tenant_id')) REGEXP '^[0-9]+$'
                              THEN CAST(JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.tenant_id')) AS SIGNED) END) IS NOT NULL
            ON DUPLICATE KEY UPDATE
                factory_id=VALUES(factory_id), plan_no=VALUES(plan_no), order_no=VALUES(order_no), customer_no=VALUES(customer_no), customer_name=VALUES(customer_name),
                item_code=VALUES(item_code), item_name=VALUES(item_name), qty=VALUES(qty), plan_qty=VALUES(plan_qty),
                plan_ship_date=VALUES(plan_ship_date), shipping_site=VALUES(shipping_site), shipping_address=VALUES(shipping_address),
                status=VALUES(status), confirm_status=VALUES(confirm_status), sync_batch_id=VALUES(sync_batch_id),
                sync_time=VALUES(sync_time), update_time=CURRENT_TIMESTAMP
            """, scope, batchId, now);

        yield return Cmd(
            """
            INSERT INTO mdp_std_ship_trans
            (tenant_id, factory_id, source_system, trans_type, shipper_rec_id, shipper_no, shipper_line, order_no, order_line,
             customer_no, item_code, item_name, qty_to_ship, picking_qty, real_qty, gross_weight, net_weight, volume,
             plan_ship_date, actual_ship_date, site, status, confirm_status, source_table, source_row_id, source_biz_key, sync_batch_id, sync_time)
            SELECT
                COALESCE(NULLIF(d.tenant_id, 0),
                         CASE WHEN JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.tenant_id')) REGEXP '^[0-9]+$'
                              THEN CAST(JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.tenant_id')) AS SIGNED) END),
                NULLIF(d.factory_id, 0),
                COALESCE(NULLIF(d.source_system,''), (SELECT r.source_system FROM mdp_source_table_registry r WHERE r.source_table = d.source_table LIMIT 1), ''),
                'ASN_SHIPPER',
                CASE WHEN JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.ASNBOLShipperRecID')) REGEXP '^-?[0-9]+$' THEN CAST(JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.ASNBOLShipperRecID')) AS SIGNED) END,
                COALESCE(JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.Id')), JSON_UNQUOTE(JSON_EXTRACT(m.raw_data,'$.Id'))),
                CAST(JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.Line')) AS CHAR),
                COALESCE(JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.OrdNbr')), JSON_UNQUOTE(JSON_EXTRACT(m.raw_data,'$.OrdNbr'))),
                CAST(JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.OrdLine')) AS CHAR),
                JSON_UNQUOTE(JSON_EXTRACT(m.raw_data,'$.SoldTo')),
                JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.ContainerItem')),
                JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.Descr')),
                CASE WHEN JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.QtyToShip')) REGEXP '^-?[0-9]+(\\.[0-9]+)?$' THEN CAST(JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.QtyToShip')) AS DECIMAL(18,6)) END,
                CASE WHEN JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.PickingQty')) REGEXP '^-?[0-9]+(\\.[0-9]+)?$' THEN CAST(JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.PickingQty')) AS DECIMAL(18,6)) END,
                CASE WHEN JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.RealQty')) REGEXP '^-?[0-9]+(\\.[0-9]+)?$' THEN CAST(JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.RealQty')) AS DECIMAL(18,6)) END,
                CASE WHEN JSON_UNQUOTE(JSON_EXTRACT(m.raw_data,'$.GrossWeight')) REGEXP '^-?[0-9]+(\\.[0-9]+)?$' THEN CAST(JSON_UNQUOTE(JSON_EXTRACT(m.raw_data,'$.GrossWeight')) AS DECIMAL(18,6)) END,
                CASE WHEN JSON_UNQUOTE(JSON_EXTRACT(m.raw_data,'$.NetWeight')) REGEXP '^-?[0-9]+(\\.[0-9]+)?$' THEN CAST(JSON_UNQUOTE(JSON_EXTRACT(m.raw_data,'$.NetWeight')) AS DECIMAL(18,6)) END,
                CASE WHEN COALESCE(JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.Volume')), JSON_UNQUOTE(JSON_EXTRACT(m.raw_data,'$.Volume'))) REGEXP '^-?[0-9]+(\\.[0-9]+)?$' THEN CAST(COALESCE(JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.Volume')), JSON_UNQUOTE(JSON_EXTRACT(m.raw_data,'$.Volume'))) AS DECIMAL(18,6)) END,
                NULLIF(NULLIF(COALESCE(JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.ShipDate')), JSON_UNQUOTE(JSON_EXTRACT(m.raw_data,'$.ShipDate'))), 'null'), ''),
                NULLIF(NULLIF(COALESCE(JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.ShipDate')), JSON_UNQUOTE(JSON_EXTRACT(m.raw_data,'$.ShipDate'))), 'null'), ''),
                COALESCE(JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.Site')), JSON_UNQUOTE(JSON_EXTRACT(m.raw_data,'$.Site'))),
                COALESCE(JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.Status')), JSON_UNQUOTE(JSON_EXTRACT(m.raw_data,'$.Status'))),
                CAST(COALESCE(JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.IsConfirm')), JSON_UNQUOTE(JSON_EXTRACT(m.raw_data,'$.IsConfirm'))) AS CHAR),
                d.source_table,
                d.source_row_id,
                d.source_biz_key,
                @BatchId,
                @Now
            FROM mdp_stg_ship_trans d
            LEFT JOIN mdp_stg_ship_trans m ON m.source_table='ASNBOLShipperMaster'
                AND m.tenant_id = d.tenant_id
                AND JSON_UNQUOTE(JSON_EXTRACT(m.raw_data,'$.RecID')) = JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.ASNBOLShipperRecID'))
            WHERE d.source_table='ASNBOLShipperDetail'
              AND d.tenant_id=@TenantId
              AND COALESCE(NULLIF(d.factory_id, 0), 1)=@FactoryId
              AND IFNULL(COALESCE(JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.Id')), JSON_UNQUOTE(JSON_EXTRACT(m.raw_data,'$.Id'))), '') <> ''
              AND COALESCE(NULLIF(d.tenant_id, 0),
                         CASE WHEN JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.tenant_id')) REGEXP '^[0-9]+$'
                              THEN CAST(JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.tenant_id')) AS SIGNED) END) IS NOT NULL
            ON DUPLICATE KEY UPDATE
                factory_id=VALUES(factory_id), shipper_no=VALUES(shipper_no), order_no=VALUES(order_no), order_line=VALUES(order_line),
                customer_no=VALUES(customer_no), item_code=VALUES(item_code), item_name=VALUES(item_name),
                qty_to_ship=VALUES(qty_to_ship), picking_qty=VALUES(picking_qty), real_qty=VALUES(real_qty),
                actual_ship_date=VALUES(actual_ship_date), site=VALUES(site), status=VALUES(status),
                confirm_status=VALUES(confirm_status), sync_batch_id=VALUES(sync_batch_id), sync_time=VALUES(sync_time),
                update_time=CURRENT_TIMESTAMP
            """, scope, batchId, now);

        yield return Cmd(
            """
            INSERT INTO mdp_std_ship_trans
            (tenant_id, factory_id, source_system, trans_type, order_no, customer_no, item_code, item_name, qty,
             plan_ship_date, status, linkage_status, source_table, source_row_id, source_biz_key, sync_batch_id, sync_time)
            SELECT
                COALESCE(NULLIF(d.tenant_id, 0),
                         CASE WHEN JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.tenant_id')) REGEXP '^[0-9]+$'
                              THEN CAST(JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.tenant_id')) AS SIGNED) END),
                NULLIF(d.factory_id, 0),
                COALESCE(NULLIF(d.source_system,''), (SELECT r.source_system FROM mdp_source_table_registry r WHERE r.source_table = d.source_table LIMIT 1), ''),
                'LINKAGE_PLAN',
                JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.bill_no')),
                JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.custom_no')),
                JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.item_number')),
                JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.Descr')),
                CASE WHEN JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.qty')) REGEXP '^-?[0-9]+(\\.[0-9]+)?$' THEN CAST(JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.qty')) AS DECIMAL(18,6)) END,
                NULLIF(NULLIF(COALESCE(JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.sys_capacity_date')), JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.fystarttime'))), 'null'), ''),
                JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.type')),
                CASE
                    WHEN JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.isuse')) = '1' THEN 'LINKED'
                    ELSE 'INACTIVE'
                END,
                d.source_table,
                d.source_row_id,
                d.source_biz_key,
                @BatchId,
                @Now
            FROM mdp_stg_ship_trans d
            WHERE d.source_table='LinkagePlan'
              AND d.tenant_id=@TenantId
              AND COALESCE(NULLIF(d.factory_id, 0), 1)=@FactoryId
              AND IFNULL(JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.bill_no')), '') <> ''
              AND COALESCE(NULLIF(d.tenant_id, 0),
                         CASE WHEN JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.tenant_id')) REGEXP '^[0-9]+$'
                              THEN CAST(JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.tenant_id')) AS SIGNED) END) IS NOT NULL
            ON DUPLICATE KEY UPDATE
                factory_id=VALUES(factory_id), order_no=VALUES(order_no), customer_no=VALUES(customer_no), item_code=VALUES(item_code),
                item_name=VALUES(item_name), qty=VALUES(qty), plan_ship_date=VALUES(plan_ship_date),
                status=VALUES(status), linkage_status=VALUES(linkage_status), sync_batch_id=VALUES(sync_batch_id),
                sync_time=VALUES(sync_time), update_time=CURRENT_TIMESTAMP
            """, scope, batchId, now);
    }

    private IEnumerable<S1MdpSqlCommand> BuildDwdCommands(S1MdpRunScope scope, string batchId, DateTime now)
    {
        yield return Cmd(
            """
            INSERT INTO dwd_requirement_examine_detail
            (tenant_id, factory_id, stat_date, row_id, parent_row_id, examine_id, order_entry_id, bill_no, morder_no,
             num, item_number, item_name, bom_number, model, kitting_time, item_type, erp_cls_name, qty, wastage,
             need_count, sqty, use_qty, self_lack_qty, lack_qty, mo_qty, make_qty, purchase_qty, purchase_occupy_qty,
             satisfy_time, have_ic_subs, substitute_code, create_time, source_system, sync_batch_id, calc_batch_id, calc_time,
             bom_level, material_role, is_current_flag)
            SELECT
                COALESCE(NULLIF(d.tenant_id, 0), NULLIF(r.tenant_id, 0), NULLIF(so.tenant_id, 0),
                         CASE WHEN JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.tenant_id')) REGEXP '^[0-9]+$'
                              THEN CAST(JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.tenant_id')) AS SIGNED) END),
                COALESCE(NULLIF(d.factory_id, 0), NULLIF(r.factory_id, 0), NULLIF(so.factory_id, 0)),
                @StatDate,
                CAST(d.source_row_id AS SIGNED),
                CASE
                    WHEN JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.num')) = '1' THEN NULL
                    ELSE p.parent_row_id
                END,
                CASE WHEN JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.examine_id')) REGEXP '^-?[0-9]+$'
                     THEN CAST(JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.examine_id')) AS SIGNED) END,
                CASE WHEN JSON_UNQUOTE(JSON_EXTRACT(r.raw_data,'$.sentry_id')) REGEXP '^-?[0-9]+$'
                     THEN CAST(JSON_UNQUOTE(JSON_EXTRACT(r.raw_data,'$.sentry_id')) AS SIGNED) END,
                COALESCE(so.order_no, JSON_UNQUOTE(JSON_EXTRACT(r.raw_data,'$.bill_no'))),
                JSON_UNQUOTE(JSON_EXTRACT(r.raw_data,'$.morder_no')),
                CAST(JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.num')) AS CHAR),
                JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.item_number')),
                JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.item_name')),
                JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.bom_number')),
                JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.model')),
                STR_TO_DATE(REPLACE(NULLIF(NULLIF(JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.kitting_time')), 'null'), ''), 'T', ' '), '%Y-%m-%d %H:%i:%s.%f'),
                CASE WHEN JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.type')) = '1' THEN '替代件' ELSE '标准件' END,
                COALESCE(
                    JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.erp_cls_name')),
                    CASE JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.erp_cls'))
                        WHEN '0' THEN '配置类'
                        WHEN '1' THEN '自制'
                        WHEN '2' THEN '委外加工'
                        WHEN '3' THEN '外购'
                        WHEN '4' THEN '虚拟件'
                    END
                ),
                CASE WHEN JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.qty')) REGEXP '^-?[0-9]+(\\.[0-9]+)?$' THEN CAST(JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.qty')) AS DECIMAL(18,4)) END,
                CASE WHEN JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.wastage')) REGEXP '^-?[0-9]+(\\.[0-9]+)?$' THEN CAST(JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.wastage')) AS DECIMAL(18,4)) END,
                CASE WHEN JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.needCount')) REGEXP '^-?[0-9]+(\\.[0-9]+)?$' THEN CAST(JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.needCount')) AS DECIMAL(18,4)) END,
                CASE WHEN JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.sqty')) REGEXP '^-?[0-9]+(\\.[0-9]+)?$' THEN CAST(JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.sqty')) AS DECIMAL(18,4)) END,
                CASE WHEN JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.use_qty')) REGEXP '^-?[0-9]+(\\.[0-9]+)?$' THEN CAST(JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.use_qty')) AS DECIMAL(18,4)) END,
                CASE WHEN JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.self_lack_qty')) REGEXP '^-?[0-9]+(\\.[0-9]+)?$' THEN CAST(JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.self_lack_qty')) AS DECIMAL(18,4)) END,
                CASE WHEN JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.lack_qty')) REGEXP '^-?[0-9]+(\\.[0-9]+)?$' THEN CAST(JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.lack_qty')) AS DECIMAL(18,4)) END,
                CASE WHEN JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.mo_qty')) REGEXP '^-?[0-9]+(\\.[0-9]+)?$' THEN CAST(JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.mo_qty')) AS DECIMAL(18,4)) END,
                CASE
                    WHEN JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.make_qty')) = '0' AND JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.erp_cls')) = '1'
                        THEN CASE WHEN JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.lack_qty')) REGEXP '^-?[0-9]+(\\.[0-9]+)?$' THEN CAST(JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.lack_qty')) AS DECIMAL(18,4)) END
                    WHEN JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.make_qty')) REGEXP '^-?[0-9]+(\\.[0-9]+)?$'
                        THEN CAST(JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.make_qty')) AS DECIMAL(18,4))
                END,
                CASE WHEN JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.purchase_qty')) REGEXP '^-?[0-9]+(\\.[0-9]+)?$' THEN CAST(JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.purchase_qty')) AS DECIMAL(18,4)) END,
                CASE WHEN JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.purchase_occupy_qty')) REGEXP '^-?[0-9]+(\\.[0-9]+)?$' THEN CAST(JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.purchase_occupy_qty')) AS DECIMAL(18,4)) END,
                STR_TO_DATE(REPLACE(NULLIF(NULLIF(JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.satisfy_time')), 'null'), ''), 'T', ' '), '%Y-%m-%d %H:%i:%s.%f'),
                CASE WHEN JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.haveicsubs')) = '1' THEN '是' ELSE '否' END,
                JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.substitute_code')),
                STR_TO_DATE(REPLACE(NULLIF(NULLIF(JSON_UNQUOTE(JSON_EXTRACT(r.raw_data,'$.create_time')), 'null'), ''), 'T', ' '), '%Y-%m-%d %H:%i:%s.%f'),
                COALESCE(NULLIF(d.source_system,''), (SELECT r.source_system FROM mdp_source_table_registry r WHERE r.source_table = d.source_table LIMIT 1), ''),
                d.sync_batch_id,
                @BatchId,
                @Now,
                -- bom_level：仅当 level 是纯数字时才物化，脏值留 NULL（不猜层级）。
                CASE WHEN JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.level')) REGEXP '^[0-9]+$'
                     THEN CAST(JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.level')) AS SIGNED) END,
                -- material_role：把「层级」翻译成「角色」，让下游（含 S8）永远不必知道 level 的存在。
                -- 失败方向必须保守：只有 level 精确等于 '1' 才判成品根；NULL / 脏值 / 空串一律落 INPUT_MATERIAL，
                -- 绝不产生 NULL、绝不把行丢掉——漏判一个投入料会漏报缺料，误判一个成品根只会少算一行。
                -- （SQL 三值逻辑：level 为 NULL 时 `= '1'` 得 NULL，CASE 自然走 ELSE，符合上述保守方向。）
                CASE WHEN JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.level')) = '1'
                     THEN 'FINISHED_GOOD_ROOT' ELSE 'INPUT_MATERIAL' END,
                -- is_current_flag 一律先写 0，由本阶段最后一步的原子发布语句统一翻牌（见 PublishCurrentSnapshotAsync）。
                0
            FROM mdp_stg_so d
            INNER JOIN mdp_stg_so r ON r.source_table='b_examine_result'
                AND r.tenant_id = d.tenant_id
                AND r.source_row_id = JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.examine_id'))
            LEFT JOIN mdp_std_so so ON so.tenant_id = COALESCE(r.tenant_id, d.tenant_id)
                AND COALESCE(NULLIF(so.factory_id, 0), 1)=@FactoryId
                AND so.order_entry_id = CASE WHEN JSON_UNQUOTE(JSON_EXTRACT(r.raw_data,'$.sentry_id')) REGEXP '^-?[0-9]+$'
                                             THEN CAST(JSON_UNQUOTE(JSON_EXTRACT(r.raw_data,'$.sentry_id')) AS SIGNED) END
            LEFT JOIN (
                SELECT examine_id, MIN(source_row_id) AS parent_row_id
                FROM (
                    SELECT JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.examine_id')) AS examine_id, CAST(source_row_id AS SIGNED) AS source_row_id
                    FROM mdp_stg_so
                    WHERE source_table='b_bom_child_examine'
                      AND tenant_id=@TenantId
                      AND COALESCE(NULLIF(factory_id, 0), 1)=@FactoryId
                      AND JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.num')) = '1'
                      AND JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.is_use')) IN ('1', 'true', 'True', 'base64:type16:AQ==')
                      AND sync_batch_id = (SELECT lb.sync_batch_id FROM mdp_stg_so lb
                                           WHERE lb.source_table='b_bom_child_examine' AND lb.tenant_id=@TenantId
                                           ORDER BY lb.sync_time DESC, lb.id DESC LIMIT 1)
                ) x
                GROUP BY examine_id
            ) p ON p.examine_id = JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.examine_id'))
            WHERE d.source_table='b_bom_child_examine'
              AND d.tenant_id=@TenantId
              AND COALESCE(NULLIF(d.factory_id, 0), 1)=@FactoryId
              AND JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.is_use')) IN ('1', 'true', 'True', 'base64:type16:AQ==')
              -- ── B5 表头软删过滤 ────────────────────────────────────────────────────────────
              -- b_examine_result 是软删表：核验单被作废后仍留在源里，只把 IsDeleted 置真。
              -- 之前这里完全没有版本过滤，等于把已作废核验单的缺料当成现行缺料输出。
              -- 实测（aidopdev，只读）：贴源 5969 行表头里 5851 行 IsDeleted 为真，仅 118 行存活
              -- ——即旧 DWD 有约 98% 建立在已作废表头上。
              -- bit(1) 列经贴源后有两种编码并存（'1'/'0' 与 base64:type16:AQ==/AA==），两种都必须认。
              -- 这里用「存活白名单」而不是「已删黑名单」，与上面 is_use 的写法保持同一方向；
              -- 实测该列在贴源层恰好只有这 4 种取值、无 NULL、无其它编码，故白名单当前不会误杀任何行。
              AND JSON_UNQUOTE(JSON_EXTRACT(r.raw_data,'$.IsDeleted')) IN ('0', 'false', 'False', 'base64:type16:AA==')
              -- ── B6 FULL=镜像 的 DWD 侧权宜实现（STOPGAP）────────────────────────────────────
              -- 正解应当是贴源层按 (tenant_id, factory_id, source_table) 整体替换，但贴源写入由
              -- DataPlatform/Executors/MdpStagingWriter.cs + MdpDbPullExecutor.cs 通用实现，
              -- 且 mdp_stg_so 的唯一键 uk_source_key=(source_system, source_table, source_biz_key)
              -- 连 tenant_id 都不含 —— 改成整体替换必须同时改那两个文件和该唯一键，均不在本次可改文件范围内。
              -- 故在 DWD 侧兜底：只取每个 (tenant_id, source_table) 的「最新一次 sync_batch_id」。
              -- 该等价关系成立的前提是 sync_mode='FULL'（实测两个 entity 均为 FULL）：
              -- FULL 每轮重读整张源表，仍存在的行其 sync_batch_id 必被刷新到最新批次；
              -- 留着旧 sync_batch_id 的行 = 上一轮全量里已经查不到 = 源侧已删除的孤儿。
              -- 实测孤儿量很大：贴源 b_bom_child_examine 共 32105 行，最新批次只有 15272 行。
              -- ⚠️ 若哪天该 entity 被改成 INCREMENTAL，本过滤会错杀未变更的存量行，届时必须回到贴源层正解。
              AND d.sync_batch_id = (SELECT lb.sync_batch_id FROM mdp_stg_so lb
                                     WHERE lb.source_table='b_bom_child_examine' AND lb.tenant_id=@TenantId
                                     ORDER BY lb.sync_time DESC, lb.id DESC LIMIT 1)
              AND r.sync_batch_id = (SELECT lb.sync_batch_id FROM mdp_stg_so lb
                                     WHERE lb.source_table='b_examine_result' AND lb.tenant_id=@TenantId
                                     ORDER BY lb.sync_time DESC, lb.id DESC LIMIT 1)
              AND COALESCE(NULLIF(d.tenant_id, 0), NULLIF(r.tenant_id, 0), NULLIF(so.tenant_id, 0),
                         CASE WHEN JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.tenant_id')) REGEXP '^[0-9]+$'
                              THEN CAST(JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.tenant_id')) AS SIGNED) END) IS NOT NULL
            ON DUPLICATE KEY UPDATE
                parent_row_id=VALUES(parent_row_id), order_entry_id=VALUES(order_entry_id), bill_no=VALUES(bill_no),
                morder_no=VALUES(morder_no), num=VALUES(num), item_number=VALUES(item_number), item_name=VALUES(item_name),
                bom_number=VALUES(bom_number), model=VALUES(model), kitting_time=VALUES(kitting_time),
                item_type=VALUES(item_type), erp_cls_name=VALUES(erp_cls_name), qty=VALUES(qty), wastage=VALUES(wastage),
                need_count=VALUES(need_count), sqty=VALUES(sqty), use_qty=VALUES(use_qty), self_lack_qty=VALUES(self_lack_qty),
                lack_qty=VALUES(lack_qty), mo_qty=VALUES(mo_qty), make_qty=VALUES(make_qty), purchase_qty=VALUES(purchase_qty),
                purchase_occupy_qty=VALUES(purchase_occupy_qty), satisfy_time=VALUES(satisfy_time), have_ic_subs=VALUES(have_ic_subs),
                substitute_code=VALUES(substitute_code), create_time=VALUES(create_time), sync_batch_id=VALUES(sync_batch_id),
                calc_time=VALUES(calc_time), update_time=CURRENT_TIMESTAMP,
                bom_level=VALUES(bom_level), material_role=VALUES(material_role)
                -- 故意不写 is_current_flag：该列的唯一写者是本阶段最后一步的原子发布语句。
                -- 若在这里也回写，同批次重跑会把已发布状态打回 0，制造瞬时「无当前快照」窗口。
            """, scope, batchId, now);

        yield return Cmd(
            """
            INSERT INTO dwd_ship_trans
            (tenant_id, factory_id, company_id, stat_date, order_id, order_entry_id, order_no, order_line, customer_no,
             customer_name, country, item_code, item_name, item_spec, order_qty, planned_ship_qty, shipped_qty,
             remaining_qty, order_date, customer_request_date, plan_delivery_date, promised_delivery_date,
             plan_ship_date, actual_ship_date, review_status, order_status, delivery_status, linkage_status, risk_level,
             source_system, source_table, source_row_id, source_biz_key, sync_batch_id, calc_batch_id, calc_time)
            SELECT
                so.tenant_id,
                so.factory_id,
                so.company_id,
                @StatDate,
                so.order_id,
                so.order_entry_id,
                so.order_no,
                IFNULL(so.order_line, ''),
                so.customer_no,
                so.customer_name,
                so.country,
                IFNULL(so.item_code, ''),
                so.item_name,
                so.item_spec,
                IFNULL(so.order_qty, 0),
                IFNULL(p.plan_qty, 0),
                IFNULL(a.real_qty, 0),
                GREATEST(IFNULL(so.order_qty, 0) - IFNULL(a.real_qty, 0), 0),
                so.order_date,
                so.customer_request_date,
                so.plan_delivery_date,
                so.promised_delivery_date,
                p.plan_ship_date,
                a.actual_ship_date,
                so.review_status,
                so.order_status,
                CASE
                    WHEN IFNULL(so.order_qty, 0) > 0 AND IFNULL(a.real_qty, 0) >= IFNULL(so.order_qty, 0) THEN 'COMPLETED'
                    WHEN COALESCE(p.plan_ship_date, so.promised_delivery_date, so.plan_delivery_date) < @Now THEN 'DELAYED'
                    WHEN IFNULL(p.plan_qty, 0) > 0 THEN 'PLANNED'
                    ELSE 'OPEN'
                END,
                l.linkage_status,
                CASE
                    WHEN COALESCE(p.plan_ship_date, so.promised_delivery_date, so.plan_delivery_date) < @Now
                         AND IFNULL(a.real_qty, 0) < IFNULL(so.order_qty, 0) THEN 'HIGH'
                    WHEN IFNULL(a.real_qty, 0) < IFNULL(so.order_qty, 0) THEN 'MEDIUM'
                    ELSE 'LOW'
                END,
                COALESCE(NULLIF(so.source_system,''), (SELECT r.source_system FROM mdp_source_table_registry r WHERE r.source_table = so.source_table LIMIT 1), ''),
                so.source_table,
                so.source_row_id,
                so.source_biz_key,
                so.sync_batch_id,
                @BatchId,
                @Now
            FROM mdp_std_so so
            LEFT JOIN (
                SELECT tenant_id, order_no, order_entry_id, IFNULL(order_line, '') AS order_line, IFNULL(item_code, '') AS item_code,
                       SUM(IFNULL(plan_qty, IFNULL(qty, 0))) AS plan_qty,
                       MIN(plan_ship_date) AS plan_ship_date
                FROM mdp_std_ship_trans
                WHERE trans_type='SHIP_PLAN'
                  AND tenant_id=@TenantId
                  AND COALESCE(NULLIF(factory_id, 0), 1)=@FactoryId
                GROUP BY tenant_id, order_no, order_entry_id, IFNULL(order_line, ''), IFNULL(item_code, '')
            ) p ON so.tenant_id=p.tenant_id
                AND so.order_no=p.order_no
                AND IFNULL(so.item_code, '')=p.item_code
                AND (
                    (p.order_entry_id IS NOT NULL AND so.order_entry_id=p.order_entry_id)
                    OR (p.order_entry_id IS NULL AND IFNULL(so.order_line, '')=p.order_line)
                )
            LEFT JOIN (
                SELECT tenant_id, order_no, IFNULL(order_line, '') AS order_line, IFNULL(item_code, '') AS item_code,
                       SUM(IFNULL(real_qty, IFNULL(qty_to_ship, 0))) AS real_qty,
                       MAX(actual_ship_date) AS actual_ship_date
                FROM mdp_std_ship_trans
                WHERE trans_type='ASN_SHIPPER'
                  AND tenant_id=@TenantId
                  AND COALESCE(NULLIF(factory_id, 0), 1)=@FactoryId
                GROUP BY tenant_id, order_no, IFNULL(order_line, ''), IFNULL(item_code, '')
            ) a ON so.tenant_id=a.tenant_id AND so.order_no=a.order_no AND IFNULL(so.order_line, '')=a.order_line AND IFNULL(so.item_code, '')=a.item_code
            LEFT JOIN (
                SELECT tenant_id, order_no, item_code, MAX(linkage_status) AS linkage_status
                FROM mdp_std_ship_trans
                WHERE IFNULL(linkage_status, '') <> ''
                  AND tenant_id=@TenantId
                  AND COALESCE(NULLIF(factory_id, 0), 1)=@FactoryId
                GROUP BY tenant_id, order_no, item_code
            ) l ON so.tenant_id=l.tenant_id AND so.order_no=l.order_no AND IFNULL(so.item_code, '')=IFNULL(l.item_code, '')
            WHERE so.tenant_id=@TenantId
              AND COALESCE(NULLIF(so.factory_id, 0), 1)=@FactoryId
              AND IFNULL(so.order_no, '') <> ''
            ON DUPLICATE KEY UPDATE
                factory_id=VALUES(factory_id), customer_no=VALUES(customer_no), customer_name=VALUES(customer_name), item_name=VALUES(item_name),
                order_qty=VALUES(order_qty), planned_ship_qty=VALUES(planned_ship_qty), shipped_qty=VALUES(shipped_qty),
                remaining_qty=VALUES(remaining_qty), plan_ship_date=VALUES(plan_ship_date), actual_ship_date=VALUES(actual_ship_date),
                review_status=VALUES(review_status), order_status=VALUES(order_status), delivery_status=VALUES(delivery_status),
                linkage_status=VALUES(linkage_status), risk_level=VALUES(risk_level), sync_batch_id=VALUES(sync_batch_id),
                calc_batch_id=VALUES(calc_batch_id), calc_time=VALUES(calc_time), update_time=CURRENT_TIMESTAMP
            """, scope, batchId, now);
    }

    private async Task<long> InsertSyncLogAsync(long tenantId, long entityId, string entityName, string batchId, int rowsRead)
    {
        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", 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 sync_end=NOW(), duration_ms=@DurationMs, rows_insert=@RowsInsert, rows_update=0, rows_skip=0, rows_error=0, status='SUCCESS'
            WHERE id=@Id
            """,
            new SugarParameter("@DurationMs", (int)(DateTime.Now - started).TotalMilliseconds),
            new SugarParameter("@RowsInsert", affected),
            new SugarParameter("@Id", logId));
    }

    private async Task MarkSyncLogFailedAsync(long logId, DateTime started, string message)
    {
        try
        {
            await MdpSchemaAligner.ExecuteAsync(_db, 
                """
                UPDATE mdp_sync_log
                SET sync_end=NOW(), duration_ms=@DurationMs, rows_error=1, status='FAILED', error_msg=@ErrorMsg
                WHERE id=@Id
                """,
                new SugarParameter("@DurationMs", (int)(DateTime.Now - started).TotalMilliseconds),
                new SugarParameter("@ErrorMsg", Truncate(message, 1000)),
                new SugarParameter("@Id", logId));
        }
        catch (Exception ex)
        {
            // 写库自身失败兜底：避免再抛掩盖原异常；遗留 RUNNING 行可由运维手动清理
            Console.Error.WriteLine($"[S1MdpSyncTransform] MarkSyncLogFailed write failed (syncLogId={logId}): {ex.Message}");
        }
    }

    private async Task<long> InsertTransformRunLogAsync(S1MdpRunScope scope, 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, 'S1 MDP同步与标准化转换', @TriggerType, @BatchId, 'RUNNING', @StartTime)
            """,
            new SugarParameter("@TenantId", scope.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, S1MdpSyncTransformResult 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", ToDurationMs(finishedAt - startedAt)),
            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;
            var db = _db.CopyNew();
            await db.Ado.ExecuteCommandAsync(
                """
                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", ToDurationMs(finishedAt - startedAt)),
                new SugarParameter("@ErrorMessage", Truncate(message, 2000)),
                new SugarParameter("@Id", runLogId));
        }
        catch (Exception ex)
        {
            // 写库自身失败兜底（典型场景：远端 MySQL 瞬断导致 MarkFailed 自身也连不上）：
            // 避免再抛二次异常掩盖原错；遗留 RUNNING 行可由运维手动清理。
            Console.Error.WriteLine($"[S1MdpSyncTransform] MarkTransformRunFailed write failed (runLogId={runLogId}): {ex.Message}");
        }
    }

    private static S1MdpSqlCommand Cmd(string sql, S1MdpRunScope scope, string batchId, DateTime now)
    {
        return new S1MdpSqlCommand(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 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 BuildOptionalColumnExpr(IReadOnlyCollection<string> columns, string expected, string fallback)
    {
        return columns.Any(u => string.Equals(u, expected, StringComparison.OrdinalIgnoreCase))
            ? $"s.`{FindColumn(columns, expected)}`"
            : fallback;
    }

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

    private static async Task KeepLeaseAliveAsync(IS1MdpFullRunLease lease, CancellationToken ct)
    {
        while (!ct.IsCancellationRequested)
        {
            try
            {
                await Task.Delay(TimeSpan.FromSeconds(30), ct);
                await lease.HeartbeatAsync(CancellationToken.None);
            }
            catch (OperationCanceledException)
            {
                return;
            }
        }
    }

    private static int ToDurationMs(TimeSpan elapsed)
    {
        var ms = elapsed.TotalMilliseconds;
        if (double.IsNaN(ms) || ms <= 0) return 0;
        return ms >= int.MaxValue ? int.MaxValue : (int)ms;
    }

    private static string NormalizeTriggerType(string? triggerType)
    {
        return string.IsNullOrWhiteSpace(triggerType) ? "AUTO" : triggerType.Trim().ToUpperInvariant();
    }

    /// <summary>
    /// 运行摘要。改用 JsonSerializer 而非字符串插值：batchId 今天不可能含引号，
    /// 但拼 JSON 是会被后来者继承的脆弱写法，且本次要加的 publish 是嵌套对象。
    /// </summary>
    private static string BuildRunSummaryJson(S1MdpSyncTransformResult result)
    {
        return System.Text.Json.JsonSerializer.Serialize(new
        {
            batchId = result.BatchId,
            stageRows = result.StageRows,
            standardRows = result.StandardRows,
            dwdRows = result.DwdRows,
            kpiRows = result.KpiRows,
            publish = result.Publication is null
                ? null
                : new
                {
                    table = result.Publication.Table,
                    batchId = result.Publication.BatchId,
                    factoryId = result.Publication.FactoryId,
                    currentRows = result.Publication.CurrentRows
                }
        });
    }

    private static string ResolveKpiValueTable(int metricLevel)
    {
        return 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 DefaultS1Target(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 Truncate(string? raw, int maxLength)
    {
        if (string.IsNullOrEmpty(raw)) return string.Empty;
        return raw.Length <= maxLength ? raw : raw[..maxLength];
    }

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

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

    private sealed class S1KpiCalcRow
    {
        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 S1KpiMetaRow
    {
        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 S1KpiValueRow
    {
        public long Id { get; set; }
        public decimal? MetricValue { get; set; }
        public decimal? TargetValue { get; set; }
    }
}

public sealed class S1MdpSyncTransformResult
{
    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; }
    public int AtomicRows { get; set; }

    /// <summary>
    /// 当前快照发布证据。为 null 表示本轮没走到发布这一步。
    ///
    /// <para><b>为什么需要它</b>：发布语句在本轮零行时命中 0 行，在库里不留任何痕迹，
    /// 于是「源侧本轮确实没有数据」与「发布压根没跑成」在 DWD 里字面等同。
    /// 下游的 Authority 健康判定因此只能保守地判 UNKNOWN、拦下恢复 ——
    /// 哪怕这是一个完全正常的空快照。</para>
    ///
    /// <para>把发布结果记进运行日志，就把这个隐式事实变成了可观测事实：
    /// 「本轮成功发布了批次 X，其中当前行数为 N（N 可以是 0）」。</para>
    /// </summary>
    public S1MdpSnapshotPublication? Publication { get; set; }
}

/// <summary>
/// 一次当前快照发布的结果。与 <c>status='SUCCESS'</c> 写在同一条 UPDATE 里，
/// 因此不存在「成功了但没有发布证据」的中间态。
/// </summary>
public sealed class S1MdpSnapshotPublication
{
    /// <summary>被发布为当前快照的表。</summary>
    public string Table { get; set; } = string.Empty;

    /// <summary>本轮批次号。读者据此确认「当前快照」出自哪一次运行。</summary>
    public string BatchId { get; set; } = string.Empty;

    /// <summary>发布作用域的工厂号。运行日志表本身没有工厂列，只能由这里承载。</summary>
    public long FactoryId { get; set; }

    /// <summary>发布后该批次的当前行数。<b>0 是合法值</b>，表示成功发布了一个空快照。</summary>
    public int CurrentRows { get; set; }
}

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

internal sealed record S1MdpEntityConfig(
    string EntityCode,
    string SourceTable,
    string TargetTable,
    string SourceRowIdExpression,
    string SourceBizKeyExpression)
{
    public static readonly IReadOnlyList<S1MdpEntityConfig> All = new List<S1MdpEntityConfig>
    {
        new("S1_SEORDER", "crm_seorder", "mdp_stg_so", "Id", "COALESCE(s.`bill_no`, CAST(s.`Id` AS CHAR))"),
        new("S1_SEORDER_ENTRY", "crm_seorderentry", "mdp_stg_so", "Id", "CONCAT(IFNULL(s.`bill_no`,''), ':', IFNULL(s.`entry_seq`, CAST(s.`Id` AS CHAR)))"),
        new("S1_SEORDER_CHANGE", "crm_seorder_change", "mdp_stg_so", "Id", "CONCAT(IFNULL(s.`bill_no`,''), ':', CAST(s.`Id` AS CHAR))"),
        new("S1_CONTRACT_REVIEW", "ado_contract_review", "mdp_stg_so", "RecID", "COALESCE(s.`BillNo`, CAST(s.`RecID` AS CHAR))"),
        new("S1_CONTRACT_REVIEW_FLOW", "ado_contract_review_flow", "mdp_stg_so", "RecID", "CONCAT(IFNULL(s.`ReviewBillNo`,''), ':', IFNULL(s.`StageNo`,''), ':', CAST(s.`RecID` AS CHAR))"),
        new("S1_PRODUCT_DESIGN", "ado_product_design", "mdp_stg_so", "Id", "COALESCE(s.`BillNo`, CAST(s.`Id` AS CHAR))"),
        new("S1_PRODUCT_DESIGN_BOM", "ado_product_design_bom", "mdp_stg_so", "Id", "CONCAT(CAST(s.`ProductDesignId` AS CHAR), ':', CAST(s.`Id` AS CHAR))"),
        new("S1_PRODUCT_DESIGN_ROUTING", "ado_product_design_routing", "mdp_stg_so", "Id", "CONCAT(CAST(s.`ProductDesignId` AS CHAR), ':', CAST(s.`Id` AS CHAR))"),
        new("S1_REQUIREMENT_EXAMINE_RESULT", "b_examine_result", "mdp_stg_so", "Id", "CONCAT(IFNULL(s.`bill_no`,''), ':', IFNULL(s.`morder_no`,''), ':', CAST(s.`Id` AS CHAR))"),
        new("S1_REQUIREMENT_EXAMINE_DETAIL", "b_bom_child_examine", "mdp_stg_so", "Id", "CONCAT(IFNULL(s.`examine_id`,''), ':', IFNULL(s.`item_number`,''), ':', CAST(s.`Id` AS CHAR))"),
        new("S1_SHIPPING_PLAN", "ShippingPlan", "mdp_stg_ship_trans", "RecID", "COALESCE(s.`LotSerial`, CAST(s.`RecID` AS CHAR))"),
        new("S1_SHIPPING_PLAN_DETAIL", "ShippingPlanDetail", "mdp_stg_ship_trans", "RecID", "CONCAT(IFNULL(s.`plan_id`,''), ':', IFNULL(s.`OrdNbr`,''), ':', CAST(s.`RecID` AS CHAR))"),
        new("S1_ASN_SHIPPER_MASTER", "ASNBOLShipperMaster", "mdp_stg_ship_trans", "RecID", "COALESCE(s.`Id`, CONCAT(IFNULL(s.`OrdNbr`,''), ':', CAST(s.`RecID` AS CHAR)))"),
        new("S1_ASN_SHIPPER_DETAIL", "ASNBOLShipperDetail", "mdp_stg_ship_trans", "RecID", "CONCAT(IFNULL(s.`Id`,''), ':', IFNULL(s.`Line`, CAST(s.`RecID` AS CHAR)))"),
        new("S1_LINKAGE_PLAN", "LinkagePlan", "mdp_stg_ship_trans", "id", "CONCAT(IFNULL(s.`bill_no`,''), ':', IFNULL(s.`item_number`,''), ':', CAST(s.`id` AS CHAR))")
    };
}