AiDOPWarehouse/server/Plugins/Admin.NET.Plugin.AiDOP/Production/S2MdpSyncTransformService.cs

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

namespace Admin.NET.Plugin.AiDOP.Production;

/// <summary>
/// S2 生产排程 MDP 同步、标准化、DWD 与 KPI 计算服务。
/// Phase2：源→stg 经统一执行器；std/DWD/KPI 仍读 stg。
/// </summary>
public class S2MdpSyncTransformService : ITransient
{
    private const string JobCode = "S2_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 S9CompositeKpiWriter _s9Composite;
    private MdpRebuildScope _runScope = null!;

    public S2MdpSyncTransformService(
        ISqlSugarClient db,
        SmartOpsKpiAtomicBuildService atomicBuild,
        MdpModuleStagingPuller stagingPuller,
        IKpiTargetResolver kpiTargetResolver,
        TransformRunLogFinalizer runLogFinalizer,
        S9CompositeKpiWriter s9Composite)
    {
        _db = db;
        _runLogFinalizer = runLogFinalizer;
        _atomicBuild = atomicBuild;
        _stagingPuller = stagingPuller;
        _kpiTargetResolver = kpiTargetResolver;
        _s9Composite = s9Composite;
    }

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

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

        try
        {
            await ReportAsync(reportProgress, new ModuleProgressUpdate(ModuleRebuildStages.Preparing, 0, 5, "准备运行环境"));
            await EnsureS2RuntimeObjectsAsync();

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

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

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

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

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

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

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

    private async Task EnsureS2RuntimeObjectsAsync()
    {
        foreach (var table in new[]
                 {
                     "mdp_stg_schedule",
                     "mdp_std_work_order_schedule",
                     "mdp_std_operation_schedule",
                     "dwd_order_schedule_trans"
                 })
            await MdpSchemaAligner.ExecuteAsync(_db, MdpSchemaDefinition.Ddl(table));
        foreach (var sql in S2MdpDdl.SqlBlocks)
        {
            await MdpSchemaAligner.ExecuteAsync(_db, sql);
        }
    }

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

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

    [Obsolete("Phase3: use MdpModuleStagingPuller / executors")]
    private async Task<int> SyncOneEntityAsync(S2MdpEntityConfig entity, string batchId, DateTime now)
    {
        var entityRow = await _db.Ado.SqlQuerySingleAsync<S2MdpEntityRow>(
            "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<S2ColumnRow>(
            """
            SELECT COLUMN_NAME AS ColumnName
            FROM information_schema.COLUMNS
            WHERE TABLE_SCHEMA=DATABASE() AND TABLE_NAME=@TableName
            ORDER BY ORDINAL_POSITION
            """,
            new SugarParameter("@TableName", entity.SourceTable));
        if (columns.Count == 0) throw Oops.Oh($"未找到源表：{entity.SourceTable}");

        var names = columns.Select(u => u.ColumnName).ToList();
        var tenantExpr = names.Any(u => string.Equals(u, "tenant_id", StringComparison.OrdinalIgnoreCase))
            ? $"IFNULL(s.`{FindColumn(names, "tenant_id")}`,0)"
            : "0";
        var factoryExpr = names.Any(u => string.Equals(u, "factory_id", StringComparison.OrdinalIgnoreCase))
            ? $"s.`{FindColumn(names, "factory_id")}`"
            : "NULL";
        var domainExpr = names.Any(u => string.Equals(u, "Domain", StringComparison.OrdinalIgnoreCase))
            ? $"s.`{FindColumn(names, "Domain")}`"
            : "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(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, source_system, source_table, source_row_id, source_biz_key, sync_batch_id, sync_time, process_status, raw_data)
                SELECT
                    {tenantExpr},
                    COALESCE({factoryExpr}, {domainExpr}),
                    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),
                    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(string batchId, DateTime now, CancellationToken cancellationToken)
    {
        var total = 0;
        foreach (var command in BuildStandardCommands(batchId, now))
        {
            cancellationToken.ThrowIfCancellationRequested();
            total += await MdpSchemaAligner.ExecuteAsync(_db, command.Sql, command.Parameters);
        }
        return total;
    }

    private async Task<int> BuildDwdAsync(string batchId, DateTime now, CancellationToken cancellationToken)
    {
        var total = 0;
        foreach (var command in BuildDwdCommands(batchId, now))
        {
            cancellationToken.ThrowIfCancellationRequested();
            total += await MdpSchemaAligner.ExecuteAsync(_db, command.Sql, command.Parameters);
        }
        return total;
    }

    private async Task<int> BuildS2KpiValuesAsync(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 CalculateS2KpiValuesAsync(batchId, statDate);
            foreach (var row in rows)
            {
                cancellationToken.ThrowIfCancellationRequested();
                affected += await UpsertS2KpiValueAsync(row, statDate, now);
            }
        }
        return affected;
    }

    private IEnumerable<S2MdpSqlCommand> BuildStandardCommands(string batchId, DateTime now)
    {
        yield return Cmd(
            """
            INSERT INTO mdp_std_work_order_schedule
            (tenant_id, factory_id, source_system, doc_type, work_order, sales_order_no, sales_order_entry_id, item_code, item_name, site_code, status,
             priority, urgent_flag, qty_ordered, qty_completed, order_date, due_date, release_date, prod_line,
             specification, lot_serial, drawing_no, project, work_order_type, labor_variance,
             source_biz_key, sync_batch_id, sync_time)
            SELECT s.tenant_id,
                   CASE WHEN s.factory_id REGEXP '^[0-9]+$' THEN CAST(s.factory_id AS UNSIGNED) ELSE 1 END,
                   COALESCE((SELECT r.source_system FROM mdp_source_table_registry r WHERE r.source_table = source_table LIMIT 1), NULLIF(source_system,''), ''),
                   'PROD_TASK',
                   JSON_UNQUOTE(JSON_EXTRACT(s.raw_data,'$.WorkOrd')),
                   JSON_UNQUOTE(JSON_EXTRACT(s.raw_data,'$.SalesJob')),
                   -- 订单行 ID 桥：WorkOrdMaster.BusinessID，>0 时恒等于 crm_seorderentry.Id
                   -- （全库实测 28/28 命中同租户订单行、0 孤儿）。源值 0 表示「无关联订单行」，
                   -- 必须归一为 NULL —— 0 不是合法订单行 Id，落 0 会在桥上造出假关联。
                   -- sales_order_no(=SalesJob) 只是文本，禁止用它做归因。
                   NULLIF(CASE WHEN JSON_UNQUOTE(JSON_EXTRACT(s.raw_data,'$.BusinessID')) REGEXP '^[0-9]+$'
                               THEN CAST(JSON_UNQUOTE(JSON_EXTRACT(s.raw_data,'$.BusinessID')) AS SIGNED) END, 0),
                   JSON_UNQUOTE(JSON_EXTRACT(s.raw_data,'$.ItemNum')),
                   JSON_UNQUOTE(JSON_EXTRACT(s.raw_data,'$.ItemName')),
                   JSON_UNQUOTE(JSON_EXTRACT(s.raw_data,'$.Site')),
                   JSON_UNQUOTE(JSON_EXTRACT(s.raw_data,'$.Status')),
                   CASE WHEN JSON_UNQUOTE(JSON_EXTRACT(s.raw_data,'$.Priority')) REGEXP '^-?[0-9]+(\\.[0-9]+)?$' THEN CAST(JSON_UNQUOTE(JSON_EXTRACT(s.raw_data,'$.Priority')) AS DECIMAL(18,6)) END,
                   CASE WHEN JSON_UNQUOTE(JSON_EXTRACT(s.raw_data,'$.Urgent')) IN ('1','true','True') THEN 1 ELSE 0 END,
                   CASE WHEN JSON_UNQUOTE(JSON_EXTRACT(s.raw_data,'$.QtyOrded')) REGEXP '^-?[0-9]+(\\.[0-9]+)?$' THEN CAST(JSON_UNQUOTE(JSON_EXTRACT(s.raw_data,'$.QtyOrded')) AS DECIMAL(18,6)) ELSE 0 END,
                   CASE WHEN JSON_UNQUOTE(JSON_EXTRACT(s.raw_data,'$.QtyCompleted')) REGEXP '^-?[0-9]+(\\.[0-9]+)?$' THEN CAST(JSON_UNQUOTE(JSON_EXTRACT(s.raw_data,'$.QtyCompleted')) AS DECIMAL(18,6)) ELSE 0 END,
                   NULLIF(NULLIF(JSON_UNQUOTE(JSON_EXTRACT(s.raw_data,'$.OrdDate')), 'null'), ''),
                   NULLIF(NULLIF(JSON_UNQUOTE(JSON_EXTRACT(s.raw_data,'$.DueDate')), 'null'), ''),
                   NULLIF(NULLIF(JSON_UNQUOTE(JSON_EXTRACT(s.raw_data,'$.ReleaseDate')), 'null'), ''),
                   JSON_UNQUOTE(JSON_EXTRACT(s.raw_data,'$.ProdLine')),
                   -- Specification（型号）= ItemMaster.Descr1 中台 enrich（(Domain,ItemNum) 去重取一，实测同键 Descr1 无冲突；
                   -- 严格按 Domain+ItemNum 关联，不做 ItemNum-only 去重，避免跨工厂串型号）：
                   im.Descr1,
                   NULLIF(NULLIF(JSON_UNQUOTE(JSON_EXTRACT(s.raw_data,'$.LotSerial')), 'null'), ''),
                   NULLIF(NULLIF(JSON_UNQUOTE(JSON_EXTRACT(s.raw_data,'$.Drawing')), 'null'), ''),
                   NULLIF(NULLIF(JSON_UNQUOTE(JSON_EXTRACT(s.raw_data,'$.Project')), 'null'), ''),
                   NULLIF(NULLIF(JSON_UNQUOTE(JSON_EXTRACT(s.raw_data,'$.WoTyped')), 'null'), ''),
                   CASE WHEN JSON_UNQUOTE(JSON_EXTRACT(s.raw_data,'$.LbrVar')) REGEXP '^-?[0-9]+(\\.[0-9]+)?$' THEN CAST(JSON_UNQUOTE(JSON_EXTRACT(s.raw_data,'$.LbrVar')) AS DECIMAL(18,6)) END,
                   s.source_biz_key, @BatchId, @Now
            FROM mdp_stg_schedule s
            LEFT JOIN (
                SELECT `Domain`, ItemNum, MIN(Descr1) AS Descr1
                FROM ItemMaster
                GROUP BY `Domain`, ItemNum
            ) im ON im.`Domain` = JSON_UNQUOTE(JSON_EXTRACT(s.raw_data,'$.Domain'))
                AND im.ItemNum = JSON_UNQUOTE(JSON_EXTRACT(s.raw_data,'$.ItemNum'))
            WHERE s.source_table='WorkOrdMaster' AND IFNULL(JSON_UNQUOTE(JSON_EXTRACT(s.raw_data,'$.WorkOrd')), '') <> ''
            ON DUPLICATE KEY UPDATE
                sales_order_no=VALUES(sales_order_no), sales_order_entry_id=VALUES(sales_order_entry_id),
                item_code=VALUES(item_code), item_name=VALUES(item_name),
                site_code=VALUES(site_code), status=VALUES(status), priority=VALUES(priority), urgent_flag=VALUES(urgent_flag),
                qty_ordered=VALUES(qty_ordered), qty_completed=VALUES(qty_completed), order_date=VALUES(order_date),
                due_date=VALUES(due_date), release_date=VALUES(release_date), prod_line=VALUES(prod_line),
                specification=VALUES(specification), lot_serial=VALUES(lot_serial), drawing_no=VALUES(drawing_no),
                project=VALUES(project), work_order_type=VALUES(work_order_type), labor_variance=VALUES(labor_variance),
                sync_batch_id=VALUES(sync_batch_id), sync_time=VALUES(sync_time), update_time=CURRENT_TIMESTAMP
            """, batchId, now);

        yield return Cmd(
            """
            INSERT INTO mdp_std_operation_schedule
            (tenant_id, factory_id, source_system, work_order, op_no, work_center, line_code, item_code,
             plan_date, prod_date, start_time, end_time, ord_qty, comp_qty, run_crew, employee, status,
             source_biz_key, sync_batch_id, sync_time)
            SELECT tenant_id,
                   CASE WHEN factory_id REGEXP '^[0-9]+$' THEN CAST(factory_id AS UNSIGNED) ELSE 1 END,
                   COALESCE((SELECT r.source_system FROM mdp_source_table_registry r WHERE r.source_table = source_table LIMIT 1), NULLIF(source_system,''), ''),
                   JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.WorkOrds')),
                   JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.Op')),
                   JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.WorkCtr')),
                   JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.Line')),
                   JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.ItemNum')),
                   NULLIF(NULLIF(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.PlanDate')), 'null'), ''),
                   NULLIF(NULLIF(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.ProdDate')), 'null'), ''),
                   NULLIF(NULLIF(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.StartTime')), 'null'), ''),
                   NULLIF(NULLIF(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.EndTime')), 'null'), ''),
                   CASE WHEN JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.OrdQty')) REGEXP '^-?[0-9]+(\\.[0-9]+)?$' THEN CAST(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.OrdQty')) AS DECIMAL(18,6)) ELSE 0 END,
                   CASE WHEN JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.CompQty')) REGEXP '^-?[0-9]+(\\.[0-9]+)?$' THEN CAST(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.CompQty')) AS DECIMAL(18,6)) ELSE 0 END,
                   CASE WHEN JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.RunCrew')) REGEXP '^-?[0-9]+(\\.[0-9]+)?$' THEN CAST(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.RunCrew')) AS DECIMAL(18,6)) END,
                   JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.Employee')),
                   -- 日计划下达状态：'Y'=已下达(可执行)，空=未下达。原样投影 PeriodSequenceDet.Status。
                   -- 它只是【状态佐证】—— Stage-2 End 的时间 Authority 一律取
                   -- S2_DAILY_PLAN_RELEASE SUCCESS 的 end_time，绝不能用 update_time 当下达时间
                   -- （update_time 会被后续任意一次 UPDATE 覆盖）。
                   JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.Status')),
                   source_biz_key, @BatchId, @Now
            FROM mdp_stg_schedule
            WHERE source_table='PeriodSequenceDet' AND IFNULL(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.WorkOrds')), '') <> ''
            ON DUPLICATE KEY UPDATE
                work_center=VALUES(work_center), line_code=VALUES(line_code), item_code=VALUES(item_code),
                plan_date=VALUES(plan_date), prod_date=VALUES(prod_date), start_time=VALUES(start_time), end_time=VALUES(end_time),
                ord_qty=VALUES(ord_qty), comp_qty=VALUES(comp_qty), run_crew=VALUES(run_crew), employee=VALUES(employee),
                status=VALUES(status),
                sync_batch_id=VALUES(sync_batch_id), sync_time=VALUES(sync_time), update_time=CURRENT_TIMESTAMP
            """, batchId, now);

        // ⚠️ 同表的另一个来源：ScheduleResultOpMaster —— 它没有「日计划下达」这个概念。
        //    因此既不投影 status，ODKU 里也【刻意不列】status，
        //    否则会把 PeriodSequenceDet 分支刚投影进来的下达状态覆盖成 NULL。
        //    （两来源 source_biz_key 格式不同，uk=(tenant_id,source_biz_key) 实测不碰撞；此为纵深防御。）
        yield return Cmd(
            """
            INSERT INTO mdp_std_operation_schedule
            (tenant_id, factory_id, source_system, work_order, op_no, work_center, line_code, item_code,
             plan_date, prod_date, start_time, end_time, ord_qty, comp_qty, run_crew, employee,
             source_biz_key, sync_batch_id, sync_time)
            SELECT tenant_id,
                   CASE WHEN factory_id REGEXP '^[0-9]+$' THEN CAST(factory_id AS UNSIGNED) ELSE 1 END,
                   COALESCE((SELECT r.source_system FROM mdp_source_table_registry r WHERE r.source_table = source_table LIMIT 1), NULLIF(source_system,''), ''),
                   JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.WorkOrd')),
                   JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.Op')),
                   JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.WorkCtr')),
                   JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.Line')),
                   JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.ItemNum')),
                   NULLIF(NULLIF(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.WorkDate')), 'null'), ''),
                   NULLIF(NULLIF(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.WorkDate')), 'null'), ''),
                   NULLIF(NULLIF(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.WorkStartTime')), 'null'), ''),
                   NULLIF(NULLIF(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.WorkEndTime')), 'null'), ''),
                   CASE WHEN JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.WorkQty')) REGEXP '^-?[0-9]+(\\.[0-9]+)?$' THEN CAST(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.WorkQty')) AS DECIMAL(18,6)) ELSE 0 END,
                   CASE WHEN JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.WorkQty')) REGEXP '^-?[0-9]+(\\.[0-9]+)?$' THEN CAST(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.WorkQty')) AS DECIMAL(18,6)) ELSE 0 END,
                   CASE WHEN JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.AssignedPersonnelCount')) REGEXP '^-?[0-9]+(\\.[0-9]+)?$' THEN CAST(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.AssignedPersonnelCount')) AS DECIMAL(18,6)) END,
                   JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.AssignedEmployeeID')),
                   source_biz_key, @BatchId, @Now
            FROM mdp_stg_schedule
            WHERE source_table='ScheduleResultOpMaster' AND IFNULL(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.WorkOrd')), '') <> ''
            ON DUPLICATE KEY UPDATE
                work_center=VALUES(work_center), line_code=VALUES(line_code), item_code=VALUES(item_code),
                plan_date=VALUES(plan_date), prod_date=VALUES(prod_date), start_time=VALUES(start_time), end_time=VALUES(end_time),
                ord_qty=VALUES(ord_qty), comp_qty=VALUES(comp_qty), run_crew=VALUES(run_crew), employee=VALUES(employee),
                sync_batch_id=VALUES(sync_batch_id), sync_time=VALUES(sync_time), update_time=CURRENT_TIMESTAMP
            """, batchId, now);
    }

    private IEnumerable<S2MdpSqlCommand> BuildDwdCommands(string batchId, DateTime now)
    {
        yield return Cmd(
            """
            INSERT INTO dwd_order_schedule_trans
            (tenant_id, factory_id, stat_date, doc_type, work_order, sales_order_no, item_code, item_name, site_code, prod_line,
             status, urgent_flag, qty_ordered, qty_completed, order_date, due_date, release_date, first_plan_date,
             last_plan_date, first_start_time, last_end_time, operation_count, scheduled_qty, completed_op_qty,
             schedule_cycle_days, schedule_satisfaction_flag, wip_qty, resource_person_count, calc_batch_id, calc_time)
            SELECT w.tenant_id, COALESCE(w.factory_id, 1), @StatDate, w.doc_type,
                   w.work_order, w.sales_order_no, w.item_code, w.item_name, w.site_code, w.prod_line,
                   w.status, w.urgent_flag, w.qty_ordered, w.qty_completed, w.order_date, w.due_date, w.release_date,
                   MIN(o.plan_date), MAX(o.plan_date), MIN(o.start_time), MAX(o.end_time),
                   COUNT(o.id), SUM(IFNULL(o.ord_qty, 0)), SUM(IFNULL(o.comp_qty, 0)),
                   CASE
                       WHEN COALESCE(MIN(o.plan_date), MAX(o.plan_date)) IS NOT NULL
                            AND COALESCE(w.release_date, w.order_date) IS NOT NULL
                       THEN TIMESTAMPDIFF(HOUR, COALESCE(w.release_date, w.order_date), COALESCE(MAX(o.plan_date), MIN(o.plan_date))) / 24
                   END,
                   CASE
                       WHEN w.due_date IS NOT NULL AND COALESCE(MAX(o.plan_date), MIN(o.plan_date)) IS NOT NULL
                            AND DATE(COALESCE(MAX(o.plan_date), MIN(o.plan_date))) <= DATE(w.due_date) THEN 1
                       ELSE 0
                   END,
                   GREATEST(IFNULL(w.qty_ordered, 0) - IFNULL(w.qty_completed, 0), 0),
                   SUM(IFNULL(o.run_crew, 0)),
                   @BatchId, @Now
            FROM mdp_std_work_order_schedule w
            LEFT JOIN mdp_std_operation_schedule o
              ON o.tenant_id=w.tenant_id AND IFNULL(o.work_order,'')=IFNULL(w.work_order,'')
            WHERE IFNULL(w.work_order, '') <> ''
            GROUP BY w.tenant_id, COALESCE(w.factory_id, 1), w.doc_type, w.work_order, w.sales_order_no, w.item_code, w.item_name,
                     w.site_code, w.prod_line, w.status, w.urgent_flag, w.qty_ordered, w.qty_completed,
                     w.order_date, w.due_date, w.release_date
            ON DUPLICATE KEY UPDATE
                sales_order_no=VALUES(sales_order_no), item_code=VALUES(item_code), item_name=VALUES(item_name),
                site_code=VALUES(site_code), prod_line=VALUES(prod_line), status=VALUES(status), urgent_flag=VALUES(urgent_flag),
                qty_ordered=VALUES(qty_ordered), qty_completed=VALUES(qty_completed), order_date=VALUES(order_date),
                due_date=VALUES(due_date), release_date=VALUES(release_date), first_plan_date=VALUES(first_plan_date),
                last_plan_date=VALUES(last_plan_date), first_start_time=VALUES(first_start_time), last_end_time=VALUES(last_end_time),
                operation_count=VALUES(operation_count), scheduled_qty=VALUES(scheduled_qty), completed_op_qty=VALUES(completed_op_qty),
                schedule_cycle_days=VALUES(schedule_cycle_days), schedule_satisfaction_flag=VALUES(schedule_satisfaction_flag),
                wip_qty=VALUES(wip_qty), resource_person_count=VALUES(resource_person_count), calc_time=VALUES(calc_time),
                update_time=CURRENT_TIMESTAMP
            """, batchId, now);
    }

    private async Task<List<S2KpiCalcRow>> CalculateS2KpiValuesAsync(string batchId, DateTime statDate)
    {
        var scope = RequireScope();
        return await _db.Ado.SqlQueryAsync<S2KpiCalcRow>(MdpSqlScope.InjectTenantFactory(
            """
            SELECT tenant_id AS TenantId, factory_id AS FactoryId, 'S2_L1_001' AS MetricCode,
                   ROUND(AVG(schedule_cycle_days), 4) AS MetricValue
            FROM dwd_order_schedule_trans
            WHERE calc_batch_id=@BatchId AND doc_type='PROD_TASK' AND schedule_cycle_days IS NOT NULL AND schedule_cycle_days >= 0
            GROUP BY tenant_id, factory_id
            UNION ALL
            SELECT tenant_id AS TenantId, factory_id AS FactoryId, 'S2_L1_002' AS MetricCode,
                   ROUND(100 * SUM(schedule_satisfaction_flag) / NULLIF(COUNT(1), 0), 4) AS MetricValue
            FROM dwd_order_schedule_trans
            WHERE calc_batch_id=@BatchId AND doc_type='PROD_TASK'
            GROUP BY tenant_id, factory_id
            UNION ALL
            SELECT tenant_id AS TenantId, factory_id AS FactoryId, 'S2_L1_003' AS MetricCode,
                   ROUND(COUNT(1) / NULLIF(SUM(CASE WHEN IFNULL(resource_person_count, 0) > 0 THEN resource_person_count ELSE 1 END), 0), 4) AS MetricValue
            FROM dwd_order_schedule_trans
            WHERE calc_batch_id=@BatchId AND doc_type='PROD_TASK'
            GROUP BY tenant_id, factory_id
            UNION ALL
            SELECT tenant_id AS TenantId, factory_id AS FactoryId, 'S2_L1_004' AS MetricCode,
                   ROUND(SUM(wip_qty) / NULLIF(SUM(CASE WHEN qty_completed > 0 THEN qty_completed ELSE completed_op_qty END), 0) * 30, 4) AS MetricValue
            FROM dwd_order_schedule_trans
            WHERE calc_batch_id=@BatchId AND doc_type='PROD_TASK'
            GROUP BY tenant_id, factory_id
            UNION ALL
            SELECT tenant_id AS TenantId, factory_id AS FactoryId, 'S2_L2_001' AS MetricCode,
                   ROUND(AVG(schedule_cycle_days), 4) AS MetricValue
            FROM dwd_order_schedule_trans
            WHERE calc_batch_id=@BatchId AND doc_type='PROD_TASK' AND schedule_cycle_days IS NOT NULL AND schedule_cycle_days >= 0
            GROUP BY tenant_id, factory_id
            UNION ALL
            SELECT tenant_id AS TenantId, factory_id AS FactoryId, 'S2_L2_002' AS MetricCode,
                   ROUND(100 * SUM(schedule_satisfaction_flag) / NULLIF(COUNT(1), 0), 4) AS MetricValue
            FROM dwd_order_schedule_trans
            WHERE calc_batch_id=@BatchId AND doc_type='PROD_TASK'
            GROUP BY tenant_id, factory_id
            UNION ALL
            SELECT tenant_id AS TenantId, factory_id AS FactoryId, 'S2_L2_003' AS MetricCode,
                   ROUND(COUNT(1) / NULLIF(SUM(CASE WHEN IFNULL(resource_person_count, 0) > 0 THEN resource_person_count ELSE 1 END), 0), 4) AS MetricValue
            FROM dwd_order_schedule_trans
            WHERE calc_batch_id=@BatchId AND doc_type='PROD_TASK'
            GROUP BY tenant_id, factory_id
            UNION ALL
            SELECT tenant_id AS TenantId, factory_id AS FactoryId, 'S2_L3_001' AS MetricCode,
                   ROUND(AVG(
                       CASE
                           WHEN start_time IS NOT NULL AND end_time IS NOT NULL AND end_time >= start_time
                           THEN TIMESTAMPDIFF(HOUR, start_time, end_time) / 24
                           WHEN plan_date IS NOT NULL AND prod_date IS NOT NULL
                           THEN ABS(TIMESTAMPDIFF(DAY, plan_date, prod_date))
                       END
                   ), 4) AS MetricValue
            FROM mdp_std_operation_schedule
            WHERE IFNULL(work_order, '') <> ''
            GROUP BY tenant_id, factory_id
            UNION ALL
            SELECT tenant_id AS TenantId, factory_id AS FactoryId, 'S2_L3_002' AS MetricCode,
                   ROUND(100 * SUM(CASE WHEN plan_date IS NOT NULL AND (prod_date IS NULL OR prod_date >= plan_date) THEN 1 ELSE 0 END) / NULLIF(COUNT(1), 0), 4) AS MetricValue
            FROM mdp_std_operation_schedule
            WHERE IFNULL(work_order, '') <> ''
            GROUP BY tenant_id, factory_id
            UNION ALL
            SELECT tenant_id AS TenantId, factory_id AS FactoryId, 'S2_L3_003' AS MetricCode,
                   ROUND(COUNT(1) / NULLIF(SUM(CASE WHEN IFNULL(run_crew, 0) > 0 THEN run_crew ELSE 1 END), 0), 4) AS MetricValue
            FROM mdp_std_operation_schedule
            WHERE IFNULL(work_order, '') <> ''
            GROUP BY tenant_id, factory_id
            UNION ALL
            SELECT tenant_id AS TenantId, factory_id AS FactoryId, 'S2_L3_004' AS MetricCode,
                   ROUND(AVG(
                       CASE
                           WHEN start_time IS NOT NULL AND end_time IS NOT NULL AND end_time >= start_time
                           THEN TIMESTAMPDIFF(HOUR, start_time, end_time) / 24
                           WHEN plan_date IS NOT NULL AND prod_date IS NOT NULL
                           THEN ABS(TIMESTAMPDIFF(DAY, plan_date, prod_date))
                       END
                   ), 4) AS MetricValue
            FROM mdp_std_operation_schedule
            WHERE IFNULL(COALESCE(NULLIF(work_center,''), NULLIF(line_code,'')), '') <> ''
            GROUP BY tenant_id, factory_id
            UNION ALL
            SELECT tenant_id AS TenantId, factory_id AS FactoryId, 'S2_L3_005' AS MetricCode,
                   ROUND(100 * SUM(CASE WHEN plan_date IS NOT NULL AND (prod_date IS NULL OR prod_date >= plan_date) THEN 1 ELSE 0 END) / NULLIF(COUNT(1), 0), 4) AS MetricValue
            FROM mdp_std_operation_schedule
            WHERE IFNULL(COALESCE(NULLIF(work_center,''), NULLIF(line_code,'')), '') <> ''
            GROUP BY tenant_id, factory_id
            UNION ALL
            SELECT tenant_id AS TenantId, factory_id AS FactoryId, 'S2_L3_006' AS MetricCode,
                   ROUND(COUNT(DISTINCT COALESCE(NULLIF(work_center,''), NULLIF(line_code,''), CONCAT(work_order, ':', op_no)))
                         / NULLIF(SUM(CASE WHEN IFNULL(run_crew, 0) > 0 THEN run_crew ELSE 1 END), 0), 4) AS MetricValue
            FROM mdp_std_operation_schedule
            WHERE IFNULL(COALESCE(NULLIF(work_center,''), NULLIF(line_code,'')), '') <> ''
            GROUP BY tenant_id, factory_id
            """),
            new SugarParameter("@BatchId", batchId),
            new SugarParameter("@StatDate", statDate),
            new SugarParameter("@TenantId", scope.TenantId),
            new SugarParameter("@FactoryId", scope.FactoryId));
    }

    private async Task<int> UpsertS2KpiValueAsync(S2KpiCalcRow row, DateTime statDate, DateTime now)
    {
        if (row.MetricValue == null) return 0;

        var meta = await _db.Ado.SqlQuerySingleAsync<S2KpiMetaRow>(
            """
            SELECT MetricLevel, Direction, YellowThreshold, RedThreshold
            FROM ado_smart_ops_kpi_master
            WHERE TenantId=@TenantId AND ModuleCode='S2' AND MetricCode=@MetricCode AND IsEnabled=1
            LIMIT 1
            """,
            new SugarParameter("@TenantId", row.TenantId),
            new SugarParameter("@MetricCode", row.MetricCode));
        if (meta == null) return 0;

        var table = ResolveKpiValueTable(meta.MetricLevel);
        var current = await _db.Ado.SqlQuerySingleAsync<S2KpiValueRow>(
            $"""
            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='S2'
              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<S2KpiValueRow>(
            $"""
            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='S2'
              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, "S2", statDate);
        var target = snap.TargetValue;
        var status = AidopS4KpiMerge.AchievementLevel(actual, target, meta.Direction, meta.YellowThreshold, meta.RedThreshold);
        var trend = ResolveTrendFlag(actual, prior?.MetricValue);

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

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

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

    private async Task MarkSyncLogSuccessAsync(long logId, DateTime started, int affected)
    {
        await MdpSchemaAligner.ExecuteAsync(_db, 
            """
            UPDATE mdp_sync_log
            SET 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)
        {
            Console.Error.WriteLine($"[S2MdpSyncTransform] MarkSyncLogFailed write failed (syncLogId={logId}): {ex.Message}");
        }
    }

    private async Task<long> InsertTransformRunLogAsync(string batchId, DateTime startedAt, string triggerType)
    {
        await MdpSchemaAligner.ExecuteAsync(_db, 
            """
            INSERT INTO mdp_transform_run_log
            (tenant_id, job_code, job_name, trigger_type, batch_id, status, start_time)
            VALUES (@TenantId, 'S2_MDP_SYNC_TRANSFORM', 'S2 MDP同步与KPI计算', @TriggerType, @BatchId, 'RUNNING', @StartTime)
            """,
            new SugarParameter("@TenantId", RequireScope().TenantId),
            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, S2MdpSyncTransformResult result)
    {
        var finishedAt = DateTime.Now;
        await MdpSchemaAligner.ExecuteAsync(_db, 
            """
            UPDATE mdp_transform_run_log
            SET status='SUCCESS', end_time=@EndTime, duration_ms=@DurationMs,
                stage_rows=@StageRows, standard_rows=@StandardRows, dwd_rows=@DwdRows,
                summary_json=@SummaryJson, update_time=CURRENT_TIMESTAMP
            WHERE id=@Id
            """,
            new SugarParameter("@EndTime", finishedAt),
            new SugarParameter("@DurationMs", (int)(finishedAt - startedAt).TotalMilliseconds),
            new SugarParameter("@StageRows", result.StageRows),
            new SugarParameter("@StandardRows", result.StandardRows),
            new SugarParameter("@DwdRows", result.DwdRows),
            new SugarParameter("@SummaryJson", BuildRunSummaryJson(result)),
            new SugarParameter("@Id", runLogId));
    }

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

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

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

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

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

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

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

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

    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 DefaultS2Target(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 S2ColumnRow
    {
        public string ColumnName { get; set; } = string.Empty;
    }

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

    private sealed class S2KpiCalcRow
    {
        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 S2KpiMetaRow
    {
        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 S2KpiValueRow
    {
        public long Id { get; set; }
        public decimal? MetricValue { get; set; }
        public decimal? TargetValue { get; set; }
    }
}

public sealed class S2MdpSyncTransformResult
{
    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; }
}

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

internal sealed record S2MdpEntityConfig(
    string EntityCode,
    string SourceTable,
    string TargetTable,
    string SourceRowIdExpression,
    string SourceBizKeyExpression)
{
    public static readonly IReadOnlyList<S2MdpEntityConfig> All = new List<S2MdpEntityConfig>
    {
        new("S2_WORK_ORDER_MASTER", "WorkOrdMaster", "mdp_stg_schedule", "RecID", "CONCAT(IFNULL(s.`Domain`,''), ':', IFNULL(s.`WorkOrd`,''))"),
        new("S2_WORK_ORDER_ROUTING", "WorkOrdRouting", "mdp_stg_schedule", "RecID", "CONCAT(IFNULL(s.`Domain`,''), ':', IFNULL(s.`WorkOrd`,''), ':', IFNULL(s.`OP`,''))"),
        new("S2_WORK_ORDER_DETAIL", "WorkOrdDetail", "mdp_stg_schedule", "RecID", "CONCAT(IFNULL(s.`Domain`,''), ':', IFNULL(s.`WorkOrd`,''), ':', IFNULL(s.`Op`,''), ':', IFNULL(s.`ItemNum`,''))"),
        new("S2_PERIOD_SEQUENCE_DET", "PeriodSequenceDet", "mdp_stg_schedule", "RecID", "CONCAT(IFNULL(s.`Domain`,''), ':', IFNULL(s.`WorkOrds`,''), ':', IFNULL(s.`Op`,''), ':', IFNULL(s.`Sequence`,''))"),
        new("S2_SCHEDULE_RESULT_OP", "ScheduleResultOpMaster", "mdp_stg_schedule", "RecID", "CONCAT(IFNULL(s.`Domain`,''), ':', IFNULL(s.`WorkOrd`,''), ':', IFNULL(s.`Op`,''), ':', IFNULL(s.`WorkDate`,''))")
    };
}

internal static class S2MdpDdl
{
    public static readonly IReadOnlyList<string> SqlBlocks = new[]
    {
        """
        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, v.entity_code, v.entity_name, 'TABLE', v.source_table_name, 'mdp_stg_schedule', 'FULL', 5000, 1, v.remark, CURRENT_TIMESTAMP, CURRENT_TIMESTAMP
        FROM mdp_source s
        JOIN (
            SELECT 'S2_WORK_ORDER_MASTER' AS entity_code, 'S2工单主数据' AS entity_name, 'WorkOrdMaster' AS source_table_name, '工单主数据进入 S2 贴源层' AS remark
            UNION ALL SELECT 'S2_WORK_ORDER_ROUTING', 'S2工单工艺路线', 'WorkOrdRouting', '工单工艺路线进入 S2 贴源层'
            UNION ALL SELECT 'S2_WORK_ORDER_DETAIL', 'S2工单物料明细', 'WorkOrdDetail', '工单物料需求进入 S2 贴源层'
            UNION ALL SELECT 'S2_PERIOD_SEQUENCE_DET', 'S2工序排程计划', 'PeriodSequenceDet', '工序间衔接与排程计划进入 S2 贴源层'
            UNION ALL SELECT 'S2_SCHEDULE_RESULT_OP', 'S2工序排产结果', 'ScheduleResultOpMaster', '工序排产结果进入 S2 贴源层'
        ) 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), entity_type=VALUES(entity_type),
            source_table_name=VALUES(source_table_name), target_table_name=VALUES(target_table_name),
            sync_mode=VALUES(sync_mode), batch_size=VALUES(batch_size), status=VALUES(status),
            remark=VALUES(remark), update_time=CURRENT_TIMESTAMP;
        """
    };
}