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

using Admin.NET.Plugin.AiDOP.DataPlatform.Executors;
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 SmartOpsKpiAtomicBuildService _atomicBuild;
    private readonly MdpModuleStagingPuller _stagingPuller;

    public S1MdpSyncTransformService(
        ISqlSugarClient db,
        SmartOpsKpiAtomicBuildService atomicBuild,
        MdpModuleStagingPuller stagingPuller)
    {
        _db = db;
        _atomicBuild = atomicBuild;
        _stagingPuller = stagingPuller;
    }

    public async Task<S1MdpSyncTransformResult> RunFullAsync(CancellationToken cancellationToken = default, string triggerType = "AUTO")
    {
        cancellationToken.ThrowIfCancellationRequested();
        var now = DateTime.Now;
        var batchId = $"S1_MDP_FULL_{now:yyyyMMddHHmmss}";
        var runLogId = await InsertTransformRunLogAsync(batchId, now, triggerType);
        var result = new S1MdpSyncTransformResult { BatchId = batchId, RunLogId = runLogId };

        try
        {
            await EnsureS1RuntimeObjectsAsync();
            result.StageRows = await SyncStagingAsync(batchId, now, cancellationToken);
            result.StandardRows = await TransformStandardAsync(batchId, now, cancellationToken);
            result.DwdRows = await BuildDwdAsync(batchId, now, cancellationToken);
            result.KpiRows = await BuildS1KpiValuesAsync(batchId, now, cancellationToken);
            result.AtomicRows = await _atomicBuild.BuildOrderDeliveryDomainForAllDatesAsync(batchId, cancellationToken: cancellationToken);
            await MarkTransformRunSuccessAsync(runLogId, now, result);
            return result;
        }
        catch (Exception ex)
        {
            await MarkTransformRunFailedAsync(runLogId, now, ex.Message);
            throw;
        }
    }

    private async Task EnsureS1RuntimeObjectsAsync()
    {
        await _db.Ado.ExecuteCommandAsync(
            """
            CREATE TABLE IF NOT EXISTS dwd_requirement_examine_detail (
                id BIGINT AUTO_INCREMENT PRIMARY KEY,
                tenant_id BIGINT NOT NULL,
                factory_id BIGINT NULL,
                stat_date DATE NOT NULL,
                row_id BIGINT NOT NULL,
                parent_row_id BIGINT NULL,
                examine_id BIGINT NULL,
                order_entry_id BIGINT NULL,
                bill_no VARCHAR(100) NULL,
                morder_no VARCHAR(100) NULL,
                num VARCHAR(100) NULL,
                item_number VARCHAR(100) NULL,
                item_name VARCHAR(200) NULL,
                bom_number VARCHAR(100) NULL,
                model VARCHAR(200) NULL,
                kitting_time DATE NULL,
                item_type VARCHAR(50) NULL,
                erp_cls_name VARCHAR(80) NULL,
                qty DECIMAL(18,4) NULL,
                wastage DECIMAL(18,4) NULL,
                need_count DECIMAL(18,4) NULL,
                sqty DECIMAL(18,4) NULL,
                use_qty DECIMAL(18,4) NULL,
                self_lack_qty DECIMAL(18,4) NULL,
                lack_qty DECIMAL(18,4) NULL,
                mo_qty DECIMAL(18,4) NULL,
                make_qty DECIMAL(18,4) NULL,
                purchase_qty DECIMAL(18,4) NULL,
                purchase_occupy_qty DECIMAL(18,4) NULL,
                satisfy_time DATE NULL,
                have_ic_subs VARCHAR(10) NULL,
                substitute_code VARCHAR(100) NULL,
                create_time DATETIME NULL,
                source_system VARCHAR(50) NOT NULL DEFAULT 'AIDOP',
                sync_batch_id VARCHAR(100) NOT NULL,
                calc_batch_id VARCHAR(100) NOT NULL,
                calc_time DATETIME NOT NULL,
                update_time DATETIME DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
                UNIQUE KEY uk_dwd_req_exam_detail (tenant_id, row_id, calc_batch_id),
                KEY idx_req_exam_tenant_batch (tenant_id, calc_batch_id),
                KEY idx_req_exam_bill (tenant_id, bill_no),
                KEY idx_req_exam_item (tenant_id, item_number)
            ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci COMMENT='S1需求明细核验DWD';
            """);

        await _db.Ado.ExecuteCommandAsync(
            """
            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 _db.Ado.ExecuteCommandAsync(
            """
            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 _db.Ado.ExecuteCommandAsync(
            """
            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(string batchId, DateTime now, CancellationToken cancellationToken)
    {
        _ = now;
        var codes = S1MdpEntityConfig.All.Select(x => x.EntityCode);
        // 本库源表含 tenant_id；ctx tenantId=0 仅作兜底，Writer 优先取源行值。
        return await _stagingPuller.PullEntitiesAsync(
            codes, batchId, tenantId: 0, fullRefresh: true, taskCode: "S1_MDP_INBOUND", cancellationToken);
    }

    /// <summary>
    /// 双模式入站：仅抽 stg（可指定 DB 实体码或 *_API），再跑 std/DWD/KPI。
    /// </summary>
    public async Task<S1MdpSyncTransformResult> RunInboundAsync(
        IEnumerable<string>? entityCodes = null,
        long tenantId = 0,
        bool fullRefresh = false,
        CancellationToken cancellationToken = default)
    {
        cancellationToken.ThrowIfCancellationRequested();
        var now = DateTime.Now;
        var batchId = $"S1_MDP_IN_{now:yyyyMMddHHmmss}";
        var runLogId = await InsertTransformRunLogAsync(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, tenantId, fullRefresh, "S1_MDP_INBOUND", cancellationToken);
            result.StandardRows = await TransformStandardAsync(batchId, now, cancellationToken);
            result.DwdRows = await BuildDwdAsync(batchId, now, cancellationToken);
            result.KpiRows = await BuildS1KpiValuesAsync(batchId, now, cancellationToken);
            result.AtomicRows = await _atomicBuild.BuildOrderDeliveryDomainForAllDatesAsync(batchId, cancellationToken: cancellationToken);
            await MarkTransformRunSuccessAsync(runLogId, now, result);
            return result;
        }
        catch (Exception ex)
        {
            await MarkTransformRunFailedAsync(runLogId, now, ex.Message);
            throw;
        }
    }

    /// <summary>Phase3：已停用。源→stg 改由执行器路径；保留方法供紧急回滚对照，勿再调用。</summary>
    [Obsolete("Phase3: use MdpModuleStagingPuller / executors")]
    private async Task<int> SyncOneEntityAsync(S1MdpEntityConfig entity, string batchId, DateTime now)
    {
        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(entityRow.Id, entityRow.EntityName, batchId, rowsRead);
        var started = DateTime.Now;

        try
        {
            var affected = await _db.Ado.ExecuteCommandAsync(
                $"""
                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},
                    'AIDOP',
                    @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(string batchId, DateTime now, CancellationToken cancellationToken)
    {
        var total = 0;
        foreach (var command in BuildStandardCommands(batchId, now))
        {
            cancellationToken.ThrowIfCancellationRequested();
            total += await _db.Ado.ExecuteCommandAsync(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 _db.Ado.ExecuteCommandAsync(command.Sql, command.Parameters);
        }
        return total;
    }

    private async Task<int> BuildS1KpiValuesAsync(string batchId, DateTime now, CancellationToken cancellationToken)
    {
        var statDate = now.Date;
        var rows = await CalculateS1KpiValuesAsync(batchId, statDate);
        var affected = 0;
        foreach (var row in rows)
        {
            cancellationToken.ThrowIfCancellationRequested();
            affected += await UpsertS1KpiValueAsync(row, statDate, now);
        }

        return affected;
    }

    private async Task<List<S1KpiCalcRow>> CalculateS1KpiValuesAsync(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 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 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 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, 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
            GROUP BY tenant_id
            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 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 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 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, 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
            GROUP BY tenant_id
            UNION ALL
            SELECT tenant_id AS TenantId, 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
            GROUP BY tenant_id
            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_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
            GROUP BY tenant_id
            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 source_table = 'ado_contract_review'
            ) 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 source_table = 'ado_contract_review'
            ) 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 source_table = 'ado_contract_review'
            ) 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 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 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 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 source_table = 'ado_contract_review_flow'
            ) 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 source_table = 'ado_contract_review_flow'
            ) 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 source_table = 'ado_contract_review_flow'
                  AND JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.StageNo')) = '1'
            ) 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 source_table = 'ado_contract_review_flow'
                  AND JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.StageNo')) = '1'
            ) t
            WHERE metric_code IS NOT NULL
              AND cycle_hours IS NOT NULL AND cycle_hours >= 0
            GROUP BY tenant_id, metric_code
            """,
            new SugarParameter("@BatchId", batchId),
            new SugarParameter("@StatDate", statDate));
    }

    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 target = current?.TargetValue ?? prior?.TargetValue ?? DefaultS1Target(row.MetricCode);
        var status = ResolveKpiStatus(actual, target, meta.Direction, meta.YellowThreshold, meta.RedThreshold);
        var trend = ResolveTrendFlag(actual, prior?.MetricValue);

        if (current != null)
        {
            return await _db.Ado.ExecuteCommandAsync(
                $"""
                UPDATE {table}
                SET metric_value=@MetricValue, target_value=@TargetValue, status_color=@StatusColor, trend_flag=@TrendFlag,
                    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", target),
                new SugarParameter("@StatusColor", status),
                new SugarParameter("@TrendFlag", trend),
                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 = await _db.Ado.GetLongAsync($"SELECT COALESCE(MAX(id), 0) + 1 FROM {table}");
        return await _db.Ado.ExecuteCommandAsync(
            $"""
            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)
            VALUES
            (@Id, @TenantId, @FactoryId, 'ACTIVE', @BizDate, @CalcTime, @CalcTime, 0, 1,
             'S1', @MetricCode, @MetricValue, @TargetValue, @StatusColor, @TrendFlag, @CalcTime)
            """,
            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", target),
            new SugarParameter("@StatusColor", status),
            new SugarParameter("@TrendFlag", trend));
    }

    private IEnumerable<S1MdpSqlCommand> BuildStandardCommands(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,
                         0),
                COALESCE(NULLIF(e.factory_id, 0), NULLIF(h.factory_id, 0)),
                NULLIF(e.company_id, 0),
                'AIDOP',
                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 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 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 IFNULL(COALESCE(JSON_UNQUOTE(JSON_EXTRACT(e.raw_data,'$.bill_no')), JSON_UNQUOTE(JSON_EXTRACT(h.raw_data,'$.bill_no'))), '') <> ''
            ON DUPLICATE KEY UPDATE
                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), deleted_flag=VALUES(deleted_flag),
                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_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,
                         0),
                NULLIF(d.factory_id, 0),
                NULLIF(d.company_id, 0),
                'AIDOP',
                '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 JSON_UNQUOTE(JSON_EXTRACT(m.raw_data,'$.RecID')) = JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.plan_id'))
            WHERE d.source_table='ShippingPlanDetail'
              AND IFNULL(COALESCE(JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.bill_no')), JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.OrdNbr'))), '') <> ''
            ON DUPLICATE KEY UPDATE
                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
            """, 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,
                         0),
                NULLIF(d.factory_id, 0),
                'AIDOP',
                '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 JSON_UNQUOTE(JSON_EXTRACT(m.raw_data,'$.RecID')) = JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.ASNBOLShipperRecID'))
            WHERE d.source_table='ASNBOLShipperDetail'
              AND IFNULL(COALESCE(JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.Id')), JSON_UNQUOTE(JSON_EXTRACT(m.raw_data,'$.Id'))), '') <> ''
            ON DUPLICATE KEY UPDATE
                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
            """, 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,
                         0),
                NULLIF(d.factory_id, 0),
                'AIDOP',
                '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 IFNULL(JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.bill_no')), '') <> ''
            ON DUPLICATE KEY UPDATE
                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
            """, batchId, now);
    }

    private IEnumerable<S1MdpSqlCommand> BuildDwdCommands(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)
            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,
                         0),
                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'),
                'AIDOP',
                d.sync_batch_id,
                @BatchId,
                @Now
            FROM mdp_stg_so d
            INNER JOIN mdp_stg_so r ON r.source_table='b_examine_result'
                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 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 JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.num')) = '1'
                      AND JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.is_use')) IN ('1', 'true', 'True', 'base64:type16:AQ==')
                ) 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 JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.is_use')) IN ('1', 'true', 'True', 'base64:type16:AQ==')
            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
            """, 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,
                'AIDOP',
                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'
                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'
                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, '') <> ''
                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 IFNULL(so.order_no, '') <> ''
            ON DUPLICATE KEY UPDATE
                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
            """, batchId, now);
    }

    private async Task<long> InsertSyncLogAsync(long entityId, string entityName, string batchId, int rowsRead)
    {
        await _db.Ado.ExecuteCommandAsync(
            """
            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 (0, @EntityId, 'AIDOPDEV_MYSQL', @EntityName, @BatchId, 'FULL', 'AUTO', NOW(), @RowsRead, 'RUNNING')
            """,
            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 _db.Ado.ExecuteCommandAsync(
            """
            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 _db.Ado.ExecuteCommandAsync(
                """
                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(string batchId, DateTime startedAt, string triggerType)
    {
        await _db.Ado.ExecuteCommandAsync(
            """
            INSERT INTO mdp_transform_run_log
            (tenant_id, job_code, job_name, trigger_type, batch_id, status, start_time)
            VALUES (0, @JobCode, 'S1 MDP同步与标准化转换', @TriggerType, @BatchId, 'RUNNING', @StartTime)
            """,
            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 _db.Ado.ExecuteCommandAsync(
            """
            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 _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", (int)(finishedAt - startedAt).TotalMilliseconds),
                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, string batchId, DateTime now)
    {
        return new S1MdpSqlCommand(sql, new[]
        {
            new SugarParameter("@BatchId", batchId),
            new SugarParameter("@Now", now),
            new SugarParameter("@StatDate", now.Date)
        });
    }

    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 string NormalizeTriggerType(string? triggerType)
    {
        return string.IsNullOrWhiteSpace(triggerType) ? "AUTO" : triggerType.Trim().ToUpperInvariant();
    }

    private static string BuildRunSummaryJson(S1MdpSyncTransformResult 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 DefaultS1Target(string metricCode)
    {
        return metricCode switch
        {
            "S1_L1_001" or "S1_L2_001" or "S1_L2_004" or "S1_L2_010" => 3m,
            "S1_L1_004" => 60m,
            "S1_L1_002" or "S1_L2_002" or "S1_L2_005" or "S1_L2_011" or "S1_L2_013" or "S1_L2_014" or "S1_L2_015" => 95m,
            "S1_L1_003" or "S1_L2_003" or "S1_L2_006" or "S1_L2_012" => 100m,
            "S1_L3_001" or "S1_L3_101" => 8m,
            "S1_L3_002" or "S1_L3_102" => 12m,
            "S1_L3_003" or "S1_L3_103" => 8m,
            "S1_L3_004" or "S1_L3_104" => 10m,
            "S1_L3_005" or "S1_L3_105" => 2m,
            "S1_L4_001" or "S1_L4_101" => 2m,
            "S1_L4_002" or "S1_L4_102" => 3m,
            "S1_L4_003" or "S1_L4_103" => 1m,
            "S1_L4_004" or "S1_L4_104" => 2m,
            _ => 0m
        };
    }

    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; }
}

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))")
    };
}