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

namespace Admin.NET.Plugin.AiDOP.Order;

/// <summary>
/// S1 首批 MDP 同步和标准化转换服务。
/// </summary>
public class S1MdpSyncTransformService : ITransient
{
    private const string JobCode = "S1_MDP_SYNC_TRANSFORM";
    private readonly ISqlSugarClient _db;

    public S1MdpSyncTransformService(ISqlSugarClient db)
    {
        _db = db;
    }

    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
        {
            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(now, cancellationToken);
            await MarkTransformRunSuccessAsync(runLogId, now, result);
            return result;
        }
        catch (Exception ex)
        {
            await MarkTransformRunFailedAsync(runLogId, now, ex.Message);
            throw;
        }
    }

    private async Task<int> SyncStagingAsync(string batchId, DateTime now, CancellationToken cancellationToken)
    {
        var total = 0;
        foreach (var entity in S1MdpEntityConfig.All)
        {
            cancellationToken.ThrowIfCancellationRequested();
            total += await SyncOneEntityAsync(entity, batchId, now);
        }
        return total;
    }

    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(DateTime now, CancellationToken cancellationToken)
    {
        var statDate = now.Date;
        var rows = await CalculateS1KpiValuesAsync(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(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, 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)
            """,
            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(e.tenant_id, h.tenant_id, 0),
                COALESCE(e.factory_id, h.factory_id),
                e.company_id,
                '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
                IFNULL(d.tenant_id, 0),
                d.factory_id,
                d.company_id,
                '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
                IFNULL(d.tenant_id, 0),
                d.factory_id,
                '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);
    }

    private IEnumerable<S1MdpSqlCommand> BuildDwdCommands(string batchId, DateTime 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, 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, IFNULL(order_line, ''), IFNULL(item_code, '')
            ) p ON so.tenant_id=p.tenant_id AND so.order_no=p.order_no AND IFNULL(so.order_line, '')=p.order_line AND IFNULL(so.item_code, '')=p.item_code
            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_010" => 3m,
            "S1_L1_002" or "S1_L2_011" => 95m,
            "S1_L1_003" or "S1_L2_012" => 100m,
            _ => 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; }
}

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