AiDOPWarehouse/server/Plugins/Admin.NET.Plugin.AiDOP/FinishedWarehouse/FqcMdpSyncService.cs

using System.Linq;
using Admin.NET.Plugin.AiDOP.DataPlatform;
using Admin.NET.Plugin.AiDOP.DataPlatform.Executors;
using Admin.NET.Plugin.AiDOP.Infrastructure;

namespace Admin.NET.Plugin.AiDOP.FinishedWarehouse;

/// <summary>
/// S7 FQC 成品质量 数据中台只读同步转换服务（DOP 内部，独立于 KPI 管线）。
///
/// B-①/C 迁通用管线（CTO 拍板：可接受批次快照延迟、全迁 std）：
///   源(qms_fqcbj 任务 / qms_qcpp_inspbill 结果头 / qms_qcpp_inspbillst 明细) 经执行器灌 mdp_stg_fqc_pull（三表共表，source_table 区分）
///   → transform 读 pull stg（MdpJsonSql 跨源类型兼容）→ mdp_std_fqc_task / mdp_std_fqc_result / mdp_std_fqc_result_detail。
///   result 富化 SAP 工单号：本库 WorkOrdMaster 相关子查询 MIN(WorkOrd) WHERE Batch=sczld（维表本库，不入 stg）。
///   result_detail 回填 std_result_id：glid → mdp_std_fqc_result.source_row_id。样本 j1..j50 逐列落 std（utf8mb3 控行宽）。
///
/// 双源：本库实体 S7_FQC_TASK/RESULT/RESULT_DETAIL（AIDOPDEV_MYSQL, status=1）；
///   dopdemorq 第二源 *_SQLSERVER（DOPDEMORQ_SQLSERVER, status=0 就位不启用）。qms_* 无 UpdateTime → FULL。
///
/// 约束：只读源/贴源，仅写 mdp_stg_fqc_pull / mdp_std_fqc_*；绝不 INSERT/UPDATE/DELETE 业务源表；不接 ApprovalFlow、不做录入/审核/处置。
///   tenant_id 从 raw_data 保留（源含该列 → std 租户不回归；SQL Server 源无此列 → 回落入站上下文）。
///   源为 0 行时转换成功完成、处理数 0，不报错。切源 FULL Replace 三表各一次单事务、仅按 tenant、Pull 成功后才 destructive。
/// </summary>
public class FqcMdpSyncService : ITransient
{
    private const string JobCode = "S7_FQC_MDP_SYNC";
    private const string InboundTaskEntityCode = "S7_FQC_TASK";
    private const string InboundResultEntityCode = "S7_FQC_RESULT";
    private const string InboundDetailEntityCode = "S7_FQC_RESULT_DETAIL";

    private const string SqlServerSourceCode = "DOPDEMORQ_SQLSERVER";
    private const string SqlServerTaskEntityCode = "S7_FQC_TASK_SQLSERVER";
    private const string SqlServerResultEntityCode = "S7_FQC_RESULT_SQLSERVER";
    private const string SqlServerDetailEntityCode = "S7_FQC_RESULT_DETAIL_SQLSERVER";
    /// <summary>165 SQL Server 源无 tenant_id 列；切源时须经账套映射解析。</summary>
    private const string SourceZtid = "pbxfxp";

    private const string StdTaskTable = "mdp_std_fqc_task";
    private const string StdResultTable = "mdp_std_fqc_result";
    private const string StdResultDetailTable = "mdp_std_fqc_result_detail";

    private readonly ISqlSugarClient _db;
    private readonly MdpSourcePullDispatcher _pullDispatcher;

    public FqcMdpSyncService(ISqlSugarClient db, MdpSourcePullDispatcher pullDispatcher)
    {
        _db = db;
        _pullDispatcher = pullDispatcher;
    }

    /// <summary>全量：本地 DB 执行器灌 pull stg（任务/结果/明细） → 标准层三表（读 stg）。</summary>
    public async Task<FqcMdpSyncResult> RunFullAsync(CancellationToken cancellationToken = default, string triggerType = "AUTO")
    {
        cancellationToken.ThrowIfCancellationRequested();
        await EnsureTablesAsync();
        await EnsurePullStgTableAsync();

        // 1E-0：并发收口——同库同时只允许一个 FQC 全量刷新（多实例/多次触发串行化，抢不到则跳过，交下一轮）
        var locked = await _db.Ado.GetIntAsync("SELECT GET_LOCK(@k, 15)", new List<SugarParameter> { new("@k", FqcRefreshLockKey) });
        if (locked != 1)
            return new FqcMdpSyncResult { BatchId = "SKIPPED_LOCK_BUSY", Skipped = true };

        var now = DateTime.Now;
        var batchId = $"S7_FQC_FULL_{now:yyyyMMddHHmmss}";
        var runLogId = await InsertRunLogAsync(batchId, now, triggerType);
        var result = new FqcMdpSyncResult { BatchId = batchId, RunLogId = runLogId };

        try
        {
            // 本库 qms_* 含 tenant_id；ctx tenantId=0 仅兜底，Writer 优先取源行值。
            var pullCtx = new MdpPullContext
            {
                TenantId = 0,
                FullRefresh = true,
                TaskCode = "S7_FQC_INBOUND",
                BatchId = $"{batchId}_PULL"
            };
            var pull = await PopulateStgAsync(pullCtx, cancellationToken);
            result.StageRows = pull.RowsWritten;
            result.TaskRows = await TransformTaskStandardAsync(batchId, now);
            result.ResultRows = await TransformResultStandardAsync(batchId, now);
            result.DetailRows = await TransformResultDetailStandardAsync(batchId, now);
            // 1F：全量刷新后清理本库(AIDOP)贴源 orphan——源单据已删的 std 行（transform 只 upsert 不删，
            // pull stg 跨全量批不清致旧会话行残留 → std 累积孤儿 → 结果列表虚增行）。仅清 AIDOP 源，
            // SQLSERVER 双源(status=0 就位不启用)行按 source_system 排除，绝不误删。
            await PurgeStdOrphansAsync();
            await MarkRunSuccessAsync(runLogId, now, result);
            return result;
        }
        catch (Exception ex)
        {
            await MarkRunFailedAsync(runLogId, now, ex.Message);
            throw;
        }
        finally
        {
            await _db.Ado.ExecuteCommandAsync("SELECT RELEASE_LOCK(@k)", new List<SugarParameter> { new("@k", FqcRefreshLockKey) });
        }
    }

    /// <summary>FQC 全量刷新并发锁键（同库同时只跑一个）。</summary>
    private const string FqcRefreshLockKey = "aidop:s7:fqc:refresh";

    /// <summary>本库 MySQL 贴源 source_system（与 MdpDbPullExecutor 落 stg 的取值一致，决定 uk_source_key 归属）。</summary>
    private const string NativeSourceSystem = "AIDOPDEV_MYSQL";

    /// <summary>targeted sync 的单据作用域（租户 + 检验单 id + 报检单号）。</summary>
    private sealed class FqcBillScope
    {
        public long TenantId { get; init; }
        public long BillId { get; init; }
        /// <summary>qms_qcpp_inspbill.lydjbh = qms_fqcbj.FBILLNO；为空则跳过任务表同步。</summary>
        public string? SourceBillNo { get; init; }
    }

    /// <summary>把作用域参数补进 transform 的参数表（未启用作用域时不加，保持全量 SQL 原样）。</summary>
    private static void AddScopePars(List<SugarParameter> pars, FqcBillScope? scope)
    {
        if (scope == null) return;
        pars.Add(new SugarParameter("@ScopeTenant", scope.TenantId));
        pars.Add(new SugarParameter("@ScopeBillId", scope.BillId.ToString()));
        pars.Add(new SugarParameter("@ScopeBillNo", scope.SourceBillNo));
    }

    /// <summary>
    /// 单据级 targeted 同步（业务写入后立即可见，替代把 12s 全量 <see cref="RunFullAsync"/> 挂到高频提交路径）。
    ///
    /// 只处理 billId 这一张检验单：源 → stg（贴源 upsert）→ std 三表（复用全量同一套 transform，字段映射零分叉）。
    ///   qms_qcpp_inspbill  (id=billId)                → mdp_std_fqc_result
    ///   qms_qcpp_inspbillst(glid=billId)              → mdp_std_fqc_result_detail
    ///   qms_fqcbj          (FBILLNO=inspbill.lydjbh)  → mdp_std_fqc_task（submit-result/退回会改 FINSPECTSTATUS 等进度列）
    ///
    /// 边界：只读业务源、只写 mdp_stg_fqc_pull / mdp_std_fqc_*；全程 tenant_id 强约束（源查询与 transform 双重）；
    ///   纯 upsert，不 DELETE、不跑 orphan purge、不抢全量 GET_LOCK、不产生 S7_FQC_FULL 批次。
    /// </summary>
    public async Task<FqcMdpSyncResult> SyncBillAsync(
        long billId, long tenantId, string triggerType = "BILL", CancellationToken cancellationToken = default)
    {
        cancellationToken.ThrowIfCancellationRequested();
        if (billId <= 0 || tenantId <= 0) return new FqcMdpSyncResult { BatchId = "SKIPPED_INVALID_SCOPE", Skipped = true };

        var now = DateTime.Now;
        var batchId = $"S7_FQC_BILL_{now:yyyyMMddHHmmssfff}_{billId}";
        var runLogId = await InsertRunLogAsync(batchId, now, triggerType);
        var result = new FqcMdpSyncResult { BatchId = batchId, RunLogId = runLogId };

        try
        {
            // 来源单号：决定是否需要同步任务表；同时用租户约束再证一次单据归属
            var sourceBillNo = await _db.Ado.SqlQuerySingleAsync<string>(
                "SELECT lydjbh FROM qms_qcpp_inspbill WHERE id=@id AND tenant_id=@tid LIMIT 1",
                new SugarParameter("@id", billId), new SugarParameter("@tid", tenantId));
            var scope = new FqcBillScope { TenantId = tenantId, BillId = billId, SourceBillNo = sourceBillNo };

            // ① 贴源：把该单据的源行刷进 stg（transform 一律读 stg，故必须先刷，否则同步的还是旧快照）
            result.StageRows = await UpsertStgFromSourceAsync(
                "qms_qcpp_inspbill", "s.id=@id AND s.tenant_id=@tid",
                new List<SugarParameter> { new("@id", billId), new("@tid", tenantId) }, batchId, now);
            result.StageRows += await UpsertStgFromSourceAsync(
                "qms_qcpp_inspbillst", "s.glid=@id AND s.tenant_id=@tid",
                new List<SugarParameter> { new("@id", billId), new("@tid", tenantId) }, batchId, now);
            if (!string.IsNullOrWhiteSpace(sourceBillNo))
                result.StageRows += await UpsertStgFromSourceAsync(
                    "qms_fqcbj", "s.FBILLNO=@bjbh AND s.tenant_id=@tid",
                    new List<SugarParameter> { new("@bjbh", sourceBillNo), new("@tid", tenantId) }, batchId, now);

            // ② 标准层：复用全量 transform，仅加单据作用域。result 必须先于 detail（detail 回填 std_result_id）
            result.ResultRows = await TransformResultStandardAsync(batchId, now, null, scope);
            result.DetailRows = await TransformResultDetailStandardAsync(batchId, now, null, scope);
            if (!string.IsNullOrWhiteSpace(sourceBillNo))
                result.TaskRows = await TransformTaskStandardAsync(batchId, now, null, scope);

            await MarkRunSuccessAsync(runLogId, now, result);
            return result;
        }
        catch (Exception ex)
        {
            await MarkRunFailedAsync(runLogId, now, ex.Message);
            throw;
        }
    }

    /// <summary>
    /// 单表贴源 upsert：源行 → mdp_stg_fqc_pull，raw_data 按源表<b>全列</b>动态生成 JSON_OBJECT，
    /// 与 MdpDbPullExecutor 的信封形状一致（source_row_id=source_biz_key=主键，JSON 键=源列名），
    /// 因此绝不会因为 transform 未来新增字段而漂移。命中既有 uk_source_key 则原地更新，不产生重复 stg 行。
    /// </summary>
    private async Task<int> UpsertStgFromSourceAsync(
        string sourceTable, string whereSql, List<SugarParameter> pars, string batchId, DateTime now)
    {
        var cols = await _db.Ado.SqlQueryAsync<string>(
            "SELECT COLUMN_NAME FROM information_schema.COLUMNS WHERE TABLE_SCHEMA=DATABASE() AND TABLE_NAME=@t ORDER BY ORDINAL_POSITION",
            new List<SugarParameter> { new("@t", sourceTable) });
        // 列名来自 information_schema 且强制标识符白名单，非用户输入，无注入面
        cols = cols.Where(c => !string.IsNullOrWhiteSpace(c) && System.Text.RegularExpressions.Regex.IsMatch(c, "^[A-Za-z0-9_]+$")).ToList();
        if (cols.Count == 0) return 0;

        var jsonObj = "JSON_OBJECT(" + string.Join(", ", cols.Select(c => $"'{c}', s.`{c}`")) + ")";
        var allPars = new List<SugarParameter>(pars)
        {
            new("@BatchId", batchId),
            new("@Now", now),
            new("@SrcSys", NativeSourceSystem),
            new("@SrcTab", sourceTable),
        };

        return await _db.Ado.ExecuteCommandAsync(
            $"""
            INSERT INTO mdp_stg_fqc_pull
            (tenant_id, source_system, source_table, source_row_id, source_biz_key, raw_data, sync_batch_id, sync_time, process_status)
            SELECT s.tenant_id, @SrcSys, @SrcTab, CAST(s.id AS CHAR), CAST(s.id AS CHAR), {jsonObj}, @BatchId, @Now, 'PENDING'
            FROM `{sourceTable}` s
            WHERE {whereSql}
            ON DUPLICATE KEY UPDATE
                tenant_id=VALUES(tenant_id), source_row_id=VALUES(source_row_id), raw_data=VALUES(raw_data),
                sync_batch_id=VALUES(sync_batch_id), sync_time=VALUES(sync_time), update_time=CURRENT_TIMESTAMP
            """,
            allPars);
    }

    /// <summary>
    /// 全量刷新后清理本库(AIDOP)贴源 orphan：源单据（qms_qcpp_inspbill / qms_fqcbj / qms_qcpp_inspbillst）
    /// 已删除但 std 层仍残留的行。仅清 source_system='AIDOP'（含历史 NULL/空）的行，SQLSERVER 双源行不受影响；
    /// detail 无 source_system 列，按父 result（glid=result.source_row_id）是否存在判定，与源系统无关且安全。
    /// </summary>
    private async Task PurgeStdOrphansAsync()
    {
        // 仅限本库 MySQL 源(source_system=AIDOPDEV_MYSQL, mdp_source.source_code)；DEMO 合成源、
        // DOPDEMORQ_SQLSERVER 双源均按 source_system 排除，绝不误删。
        const string nativeScope = "s.source_system = 'AIDOPDEV_MYSQL'";
        // 结果头：源 qms_qcpp_inspbill（source_row_id=inspbill.id）
        await _db.Ado.ExecuteCommandAsync(
            $"""
            DELETE s FROM {StdResultTable} s
            LEFT JOIN qms_qcpp_inspbill q ON q.id = CAST(s.source_row_id AS UNSIGNED) AND q.tenant_id = s.tenant_id
            WHERE {nativeScope} AND q.id IS NULL
            """);
        // 任务：源 qms_fqcbj（source_row_id=fqcbj.id）
        await _db.Ado.ExecuteCommandAsync(
            $"""
            DELETE s FROM {StdTaskTable} s
            LEFT JOIN qms_fqcbj q ON q.id = CAST(s.source_row_id AS UNSIGNED) AND q.tenant_id = s.tenant_id
            WHERE {nativeScope} AND q.id IS NULL
            """);
        // 明细：本库源 qms_qcpp_inspbillst（source_row_id=inspbillst.id, 数值型）；非本库(DEMO 的 source_row_id
        // 形如 DEMO-FQC-SRC-*)由数值正则排除，绝不误删。父结果被上面清理后本库孤儿明细亦随源缺失被清。
        await _db.Ado.ExecuteCommandAsync(
            $"""
            DELETE d FROM {StdResultDetailTable} d
            LEFT JOIN qms_qcpp_inspbillst q ON q.id = CAST(d.source_row_id AS UNSIGNED) AND q.tenant_id = d.tenant_id
            WHERE d.source_row_id REGEXP '^[0-9]+$' AND q.id IS NULL
            """);
    }

    /// <summary>双模式入站：执行器抽三表 → pull stg，再跑标准层三表（读 stg）。</summary>
    public async Task<FqcInboundResult> RunInboundAsync(
        long tenantId = 0,
        bool fullRefresh = false,
        CancellationToken cancellationToken = default)
    {
        cancellationToken.ThrowIfCancellationRequested();
        await EnsureTablesAsync();
        await EnsurePullStgTableAsync();

        var now = DateTime.Now;
        // 入参 tenantId：本库源含 tenant_id 时可传 0（Writer 优先取源行）。
        var pullCtx = new MdpPullContext
        {
            TenantId = tenantId,
            FullRefresh = fullRefresh,
            TaskCode = "S7_FQC_INBOUND",
            BatchId = $"S7_FQC_IN_{now:yyyyMMddHHmmss}"
        };
        var pull = await PopulateStgAsync(pullCtx, cancellationToken);

        var batchId = $"S7_FQC_STD_{now:yyyyMMddHHmmss}";
        var runLogId = await InsertRunLogAsync(batchId, now, "INBOUND");
        var result = new FqcMdpSyncResult { BatchId = batchId, RunLogId = runLogId };
        try
        {
            result.TaskRows = await TransformTaskStandardAsync(batchId, now);
            result.ResultRows = await TransformResultStandardAsync(batchId, now);
            result.DetailRows = await TransformResultDetailStandardAsync(batchId, now);
            await MarkRunSuccessAsync(runLogId, now, result);
        }
        catch (Exception ex)
        {
            await MarkRunFailedAsync(runLogId, now, ex.Message);
            throw;
        }

        return new FqcInboundResult
        {
            PullBatchId = pullCtx.BatchId,
            RowsPulled = pull.RowsPulled,
            RowsWrittenStg = pull.RowsWritten,
            NewCursor = pull.NewCursor,
            TransformBatchId = batchId,
            TaskRows = result.TaskRows,
            ResultRows = result.ResultRows,
            DetailRows = result.DetailRows
        };
    }

    /// <summary>
    /// 双源切换 FULL Replace（三表）：从指定 DB 源（默认 DOPDEMORQ_SQLSERVER）PullAll 全量灌 pull stg，
    /// 成功后 task/result/result_detail 各一次单事务 FULL 重建（按 tenant 精确隔离）。
    /// 顺序：result 先（写 std 结果，供明细回填 std_result_id），再 result_detail；task 独立。
    /// SQLSERVER 实体默认 status=0（就位不启用）；dopdemorq 三表当前空，未实际执行 destructive 切换。
    /// </summary>
    public async Task<FqcInboundResult> RunSourceSwitchFullAsync(
        string sourceCode = SqlServerSourceCode,
        string taskEntityCode = SqlServerTaskEntityCode,
        string resultEntityCode = SqlServerResultEntityCode,
        string detailEntityCode = SqlServerDetailEntityCode,
        long tenantId = 0,
        CancellationToken cancellationToken = default)
    {
        cancellationToken.ThrowIfCancellationRequested();
        await EnsureTablesAsync();
        await EnsurePullStgTableAsync();

        // 165 SQL Server 源无 tenant_id；显式 tenantId≤0 时经账套映射解析。
        tenantId = AidopSourceTenantMap.ResolveTenantId(SourceZtid, tenantId);

        var now = DateTime.Now;
        var pullCtx = new MdpPullContext
        {
            TenantId = tenantId,
            FullRefresh = true,
            TaskCode = "S7_FQC_INBOUND",
            BatchId = $"S7_FQC_SW_{now:yyyyMMddHHmmss}"
        };
        var task = await _pullDispatcher.PullAllByEntityCodeAsync(taskEntityCode, pullCtx, cancellationToken);
        var res = await _pullDispatcher.PullAllByEntityCodeAsync(resultEntityCode, pullCtx, cancellationToken);
        var det = await _pullDispatcher.PullAllByEntityCodeAsync(detailEntityCode, pullCtx, cancellationToken);

        var batchId = $"S7_FQC_SWSTD_{now:yyyyMMddHHmmss}";
        var runLogId = await InsertRunLogAsync(batchId, now, "SOURCE_SWITCH");
        var result = new FqcMdpSyncResult { BatchId = batchId, RunLogId = runLogId };
        try
        {
            result.TaskRows = await MdpStdFullReplace.ReplaceAsync(
                _db, StdTaskTable, tenantId, extraWhere: null,
                insertScopedAsync: () => TransformTaskStandardAsync(batchId, now, sourceCode), cancellationToken);
            result.ResultRows = await MdpStdFullReplace.ReplaceAsync(
                _db, StdResultTable, tenantId, extraWhere: null,
                insertScopedAsync: () => TransformResultStandardAsync(batchId, now, sourceCode), cancellationToken);
            result.DetailRows = await MdpStdFullReplace.ReplaceAsync(
                _db, StdResultDetailTable, tenantId, extraWhere: null,
                insertScopedAsync: () => TransformResultDetailStandardAsync(batchId, now, sourceCode), cancellationToken);
            await MarkRunSuccessAsync(runLogId, now, result);
        }
        catch (Exception ex)
        {
            await MarkRunFailedAsync(runLogId, now, ex.Message);
            throw;
        }

        return new FqcInboundResult
        {
            PullBatchId = pullCtx.BatchId,
            RowsPulled = task.RowsPulled + res.RowsPulled + det.RowsPulled,
            RowsWrittenStg = task.RowsWritten + res.RowsWritten + det.RowsWritten,
            NewCursor = det.NewCursor ?? res.NewCursor ?? task.NewCursor,
            TransformBatchId = batchId,
            TaskRows = result.TaskRows,
            ResultRows = result.ResultRows,
            DetailRows = result.DetailRows
        };
    }

    private async Task<(int RowsPulled, int RowsWritten, string? NewCursor)> PopulateStgAsync(
        MdpPullContext pullCtx, CancellationToken cancellationToken)
    {
        // 1E-0：改分页抽尽，消除单页 5000 行截断（源表 > batch_size 时单页只灌前 5000 → std 缺尾）
        var task = await _pullDispatcher.PullAllByEntityCodeAsync(InboundTaskEntityCode, pullCtx, cancellationToken);
        var res = await _pullDispatcher.PullAllByEntityCodeAsync(InboundResultEntityCode, pullCtx, cancellationToken);
        var det = await _pullDispatcher.PullAllByEntityCodeAsync(InboundDetailEntityCode, pullCtx, cancellationToken);
        return (
            task.RowsPulled + res.RowsPulled + det.RowsPulled,
            task.RowsWritten + res.RowsWritten + det.RowsWritten,
            det.NewCursor ?? res.NewCursor ?? task.NewCursor);
    }

    private async Task EnsurePullStgTableAsync()
    {
        await _db.Ado.ExecuteCommandAsync(
            """
            CREATE TABLE IF NOT EXISTS mdp_stg_fqc_pull (
                id BIGINT PRIMARY KEY AUTO_INCREMENT,
                tenant_id BIGINT NOT NULL,
                source_system VARCHAR(50) NULL,
                source_table VARCHAR(200),
                source_row_id VARCHAR(200),
                source_biz_key VARCHAR(300) NULL,
                raw_data JSON,
                sync_batch_id VARCHAR(100),
                sync_time DATETIME NULL DEFAULT CURRENT_TIMESTAMP,
                process_status VARCHAR(20) NOT NULL DEFAULT 'PENDING',
                process_message VARCHAR(500) NULL,
                create_time DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
                update_time DATETIME NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
                UNIQUE KEY uk_source_key (source_system, source_table, source_biz_key),
                KEY idx_batch (sync_batch_id),
                KEY idx_src (source_table, source_row_id),
                KEY idx_tenant (tenant_id)
            ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='S7 FQC 执行器贴源层（任务/结果/明细共表，source_table 区分）'
            """);
    }

    /// <summary>
    /// 防御式建表（幂等）：归口源空表(qms_fqcbj/qms_qcpp_inspbillst，方案 A，供执行器贴源) + 任务 std + 结果 std + 明细 std（j1..j50 utf8mb3 控行宽）。
    /// qms_qcpp_inspbill(结果头) 由既有 FQC 批次建，沿用原 Service 假设已存在。
    /// </summary>
    private async Task EnsureTablesAsync()
    {
        // 归口源空表（方案 A，DDL 例外、幂等、零 DML）：同构自旧系统 dopdemo，待上游抽数；供执行器 pull 贴源。
        await _db.Ado.ExecuteCommandAsync(
            """
            CREATE TABLE IF NOT EXISTS qms_fqcbj (
                id BIGINT NOT NULL PRIMARY KEY COMMENT '主表id',
                tenant_id BIGINT NOT NULL COMMENT 'tenant_id WP0-D4',
                FBILLNO VARCHAR(80) NULL COMMENT '单据编号',
                FBILLSTATUS VARCHAR(50) NULL COMMENT '单据状态',
                FCREATORID BIGINT NULL, FMODIFIERID BIGINT NULL, FAUDITORID BIGINT NULL,
                FAUDITDATE DATETIME NULL, FMODIFYTIME DATETIME NULL, FCREATETIME DATETIME NULL,
                FORGID BIGINT NULL COMMENT '申请组织', FQUALITYORG BIGINT NULL,
                FAPPLYUSER VARCHAR(50) NULL COMMENT '申请人', FAPPLYTIME DATETIME NULL COMMENT '申请时间',
                FBILLTYPE VARCHAR(50) NULL COMMENT '单据类型', FBIZTYPE BIGINT NULL COMMENT '业务类型',
                FINSPECORGID BIGINT NULL, FCOMMENT TEXT NULL COMMENT '备注', FAUTHORIZEOBJID BIGINT NULL, FINTERFACEID VARCHAR(50) NULL,
                cpmc VARCHAR(255) NULL COMMENT '产品名称', cpxh VARCHAR(255) NULL COMMENT '产品型号',
                scph VARCHAR(255) NULL COMMENT '生产批号', sl DECIMAL(10,0) NULL COMMENT '数量',
                jfbh TEXT NULL COMMENT '击发编号', wlbm VARCHAR(255) NULL COMMENT '物料编码',
                jyfzr VARCHAR(255) NULL COMMENT '检验负责人', yxj VARCHAR(255) NULL COMMENT '检验优先级',
                jykssj DATETIME NULL COMMENT '检验开始时间', jywcsj DATETIME NULL COMMENT '检验完成时间',
                mjpc VARCHAR(255) NULL, tsyq VARCHAR(255) NULL,
                FINSPECTSTATUS VARCHAR(255) NULL COMMENT '检验进度', sczld VARCHAR(255) NULL COMMENT '生产指令单',
                sczldsl DECIMAL(10,0) NULL COMMENT '指令单数量', lydjbh VARCHAR(100) NULL COMMENT '来源单据编号',
                ISscrap INT NULL, CirculationCard TEXT NULL
            ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci COMMENT='FQC检验任务(报检) 归口自旧系统dopdemo，空表待上游抽数'
            """);
        var jGkDdl = string.Join(", ", Enumerable.Range(1, 50).Select(i => $"j{i} VARCHAR(255) NULL"));
        await _db.Ado.ExecuteCommandAsync(
            $"""
            CREATE TABLE IF NOT EXISTS qms_qcpp_inspbillst (
                id BIGINT NOT NULL PRIMARY KEY COMMENT '子表id',
                tenant_id BIGINT NOT NULL COMMENT 'tenant_id WP0-D4',
                glid BIGINT NULL COMMENT '关联id(=qms_qcpp_inspbill.id)',
                jyxm VARCHAR(255) CHARACTER SET utf8mb3 COLLATE utf8mb3_general_ci NULL COMMENT '检验项目',
                sx VARCHAR(255) CHARACTER SET utf8mb3 COLLATE utf8mb3_general_ci NULL COMMENT '上限',
                xx VARCHAR(255) CHARACTER SET utf8mb3 COLLATE utf8mb3_general_ci NULL COMMENT '下限',
                bz VARCHAR(255) NULL COMMENT '备注', czf VARCHAR(255) NULL,
                {jGkDdl},
                pd VARCHAR(255) CHARACTER SET utf8mb3 COLLATE utf8mb3_general_ci NULL COMMENT '判定',
                ybl VARCHAR(255) CHARACTER SET utf8mb3 COLLATE utf8mb3_general_ci NULL COMMENT '样本量',
                txfl BIGINT NULL COMMENT '指标分类', bztk TEXT NULL, jsyq TEXT NULL COMMENT '技术要求',
                jyyqsbgj TEXT NULL COMMENT '检验仪器设备工具', jyff TEXT NULL, jytk TEXT NULL,
                jsbz TEXT NULL COMMENT '接收标准', cyfa TEXT NULL COMMENT '抽样方案', jl TEXT NULL,
                jlqjbh VARCHAR(255) NULL COMMENT '计量器具编号', jlbh VARCHAR(255) NULL COMMENT '计量器具编号',
                jf1 VARCHAR(255) CHARACTER SET utf8mb3 COLLATE utf8mb3_general_ci NULL COMMENT '击发1',
                jf2 VARCHAR(255) CHARACTER SET utf8mb3 COLLATE utf8mb3_general_ci NULL COMMENT '击发2',
                jf3 VARCHAR(255) CHARACTER SET utf8mb3 COLLATE utf8mb3_general_ci NULL COMMENT '击发3',
                jf4 VARCHAR(255) CHARACTER SET utf8mb3 COLLATE utf8mb3_general_ci NULL COMMENT '击发4',
                jf5 VARCHAR(255) CHARACTER SET utf8mb3 COLLATE utf8mb3_general_ci NULL COMMENT '击发5',
                NonNumericalTypeOK BIGINT NULL COMMENT '非数值OK', NonNumericalTypeNG BIGINT NULL COMMENT '非数值NG',
                ThresholdSwitch VARCHAR(255) CHARACTER SET utf8mb3 COLLATE utf8mb3_general_ci NULL COMMENT '自动判定按钮',
                ThresholdSwitchInt BIGINT NULL COMMENT '自动判定按钮',
                SizeSpecifications VARCHAR(255) CHARACTER SET utf8mb3 COLLATE utf8mb3_general_ci NULL COMMENT '规范标准',
                ybl1 VARCHAR(255) CHARACTER SET utf8mb3 COLLATE utf8mb3_general_ci NULL,
                xh INT NULL COMMENT '序号',
                KEY idx_qms_qcpp_inspbillst_glid (glid)
            ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci COMMENT='FQC检验单明细(品检项目) 归口自旧系统dopdemo，空表待上游抽数'
            """);
        await _db.Ado.ExecuteCommandAsync(
            """
            CREATE TABLE IF NOT EXISTS mdp_std_fqc_task (
                id BIGINT AUTO_INCREMENT PRIMARY KEY,
                tenant_id BIGINT NOT NULL DEFAULT 0,
                source_system VARCHAR(50) NOT NULL DEFAULT 'AIDOP',
                bill_no VARCHAR(80) NULL,
                production_order_no VARCHAR(255) NULL,
                apply_org_id BIGINT NULL,
                apply_time DATETIME NULL,
                material_code VARCHAR(255) NULL,
                product_name VARCHAR(255) NULL,
                product_model VARCHAR(255) NULL,
                production_batch_no VARCHAR(255) NULL,
                order_qty DECIMAL(18,6) NULL,
                qty DECIMAL(18,6) NULL,
                applicant VARCHAR(50) NULL,
                remark TEXT NULL,
                priority VARCHAR(255) NULL,
                inspect_start_time DATETIME NULL,
                inspect_finish_time DATETIME NULL,
                inspector VARCHAR(255) NULL,
                inspect_progress VARCHAR(255) NULL,
                source_row_id VARCHAR(100) NOT NULL,
                source_biz_key VARCHAR(200) NULL,
                sync_batch_id VARCHAR(100) NOT NULL,
                sync_time DATETIME NOT NULL,
                update_time DATETIME DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
                UNIQUE KEY uk_mdp_std_fqc_task (tenant_id, source_row_id),
                KEY idx_mdp_std_fqc_task_bill (tenant_id, bill_no),
                KEY idx_mdp_std_fqc_task_apply (tenant_id, apply_time)
            ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci COMMENT='S7 FQC 检验任务标准层'
            """);
        await _db.Ado.ExecuteCommandAsync(
            """
            CREATE TABLE IF NOT EXISTS mdp_std_fqc_result (
                id BIGINT AUTO_INCREMENT PRIMARY KEY,
                tenant_id BIGINT NOT NULL DEFAULT 0,
                source_system VARCHAR(50) NOT NULL DEFAULT 'AIDOP',
                bill_no VARCHAR(80) NULL,
                inspect_time DATETIME NULL,
                production_batch_no VARCHAR(255) NULL,
                material_code VARCHAR(255) NULL,
                material_name VARCHAR(255) NULL,
                spec VARCHAR(255) NULL,
                inspector VARCHAR(255) NULL,
                inspect_qty DECIMAL(18,6) NULL,
                source_bill_no VARCHAR(100) NULL,
                production_order_no VARCHAR(255) NULL,
                order_qty DECIMAL(18,6) NULL,
                sap_work_order_no VARCHAR(64) NULL,
                judgment VARCHAR(32) NULL,
                qualified_qty DECIMAL(18,6) NULL,
                unqualified_qty DECIMAL(18,6) NULL,
                fire_no TEXT NULL,
                inspect_spec_no VARCHAR(255) NULL,
                inspect_spec_version VARCHAR(255) NULL,
                source_row_id VARCHAR(100) NOT NULL,
                source_biz_key VARCHAR(200) NULL,
                sync_batch_id VARCHAR(100) NOT NULL,
                sync_time DATETIME NOT NULL,
                update_time DATETIME DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
                UNIQUE KEY uk_mdp_std_fqc_result (tenant_id, source_row_id),
                KEY idx_mdp_std_fqc_result_bill (tenant_id, bill_no),
                KEY idx_mdp_std_fqc_result_inspect (tenant_id, inspect_time)
            ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci COMMENT='S7 FQC 检验单(结果)标准层'
            """);
        var jDdl = string.Join(",\n                ", Enumerable.Range(1, 50).Select(i =>
            $"j{i} VARCHAR(255) CHARACTER SET utf8mb3 COLLATE utf8mb3_general_ci NULL"));
        await _db.Ado.ExecuteCommandAsync(
            $"""
            CREATE TABLE IF NOT EXISTS mdp_std_fqc_result_detail (
                id BIGINT AUTO_INCREMENT PRIMARY KEY,
                tenant_id BIGINT NOT NULL DEFAULT 0,
                std_result_id BIGINT NULL,
                glid VARCHAR(100) NULL,
                seq_no INT NULL,
                inspection_item TEXT NULL,
                accept_standard TEXT NULL,
                tech_requirement TEXT NULL,
                sampling_plan TEXT NULL,
                sample_qty VARCHAR(255) CHARACTER SET utf8mb3 COLLATE utf8mb3_general_ci NULL,
                upper_limit VARCHAR(255) CHARACTER SET utf8mb3 COLLATE utf8mb3_general_ci NULL,
                lower_limit VARCHAR(255) CHARACTER SET utf8mb3 COLLATE utf8mb3_general_ci NULL,
                size_specification VARCHAR(255) CHARACTER SET utf8mb3 COLLATE utf8mb3_general_ci NULL,
                judgment VARCHAR(255) CHARACTER SET utf8mb3 COLLATE utf8mb3_general_ci NULL,
                non_numeric_ok BIGINT NULL,
                non_numeric_ng BIGINT NULL,
                remark TEXT NULL,
                {jDdl},
                source_row_id VARCHAR(100) NOT NULL,
                source_biz_key VARCHAR(200) NULL,
                sync_batch_id VARCHAR(100) NOT NULL,
                sync_time DATETIME NOT NULL,
                update_time DATETIME DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
                UNIQUE KEY uk_mdp_std_fqc_result_dtl (tenant_id, source_row_id),
                KEY idx_mdp_std_fqc_result_dtl_head (tenant_id, glid),
                KEY idx_mdp_std_fqc_result_dtl_stdres (std_result_id)
            ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci COMMENT='S7 FQC 检验单明细标准层(j1-j50样本)'
            """);
    }

    /// <summary>标准化任务：pull stg(qms_fqcbj) → mdp_std_fqc_task。返回处理行数。</summary>
    private async Task<int> TransformTaskStandardAsync(string batchId, DateTime now, string? sourceSystem = null, FqcBillScope? scope = null)
    {
        var srcClause = sourceSystem == null ? "" : " AND t.source_system=@Src";
        var tTenant = MdpJsonSql.TenantFromStgJsonCol("t", MdpJsonSql.Int("t", "tenant_id"));
        var where = $"t.source_table='qms_fqcbj'{srcClause} AND {MdpJsonSql.TenantGuard(tTenant)}";
        // 单据级作用域（targeted sync）：任务经报检单号关联（qms_fqcbj.FBILLNO = inspbill.lydjbh），并强制租户相等
        if (scope != null)
            where += $" AND ({tTenant})=@ScopeTenant AND {MdpJsonSql.Str("t", "FBILLNO")}=@ScopeBillNo";

        var countPars = new List<SugarParameter>();
        if (sourceSystem != null) countPars.Add(new SugarParameter("@Src", sourceSystem));
        AddScopePars(countPars, scope);
        var rows = await _db.Ado.GetIntAsync(
            $"SELECT COUNT(1) FROM mdp_stg_fqc_pull t WHERE {where}", countPars);

        var insertSql =
            $"""
            INSERT INTO mdp_std_fqc_task
            (tenant_id, source_system, bill_no, production_order_no, apply_org_id, apply_time, material_code,
             product_name, product_model, production_batch_no, order_qty, qty, applicant, remark, priority,
             inspect_start_time, inspect_finish_time, inspector, inspect_progress,
             source_row_id, source_biz_key, sync_batch_id, sync_time)
            SELECT
                {tTenant}, IFNULL(NULLIF(t.source_system,''), 'AIDOP'),
                {MdpJsonSql.Str("t", "FBILLNO")}, {MdpJsonSql.Str("t", "sczld")}, {MdpJsonSql.Int("t", "FORGID")}, {MdpJsonSql.DateTimeSec("t", "FAPPLYTIME")}, {MdpJsonSql.Str("t", "wlbm")},
                {MdpJsonSql.Str("t", "cpmc")}, {MdpJsonSql.Str("t", "cpxh")}, {MdpJsonSql.Str("t", "scph")}, {MdpJsonSql.Dec("t", "sczldsl", 18, 6)}, {MdpJsonSql.Dec("t", "sl", 18, 6)}, {MdpJsonSql.Str("t", "FAPPLYUSER")}, {MdpJsonSql.Str("t", "FCOMMENT")}, {MdpJsonSql.Str("t", "yxj")},
                {MdpJsonSql.DateTimeSec("t", "jykssj")}, {MdpJsonSql.DateTimeSec("t", "jywcsj")}, {MdpJsonSql.Str("t", "jyfzr")}, {MdpJsonSql.Str("t", "FINSPECTSTATUS")},
                IFNULL(t.source_row_id, {MdpJsonSql.Str("t", "id")}), IFNULL(NULLIF(t.source_biz_key,''), {MdpJsonSql.Str("t", "FBILLNO")}),
                @BatchId, @Now
            FROM mdp_stg_fqc_pull t
            WHERE {where}
            ON DUPLICATE KEY UPDATE
                bill_no=VALUES(bill_no), production_order_no=VALUES(production_order_no), apply_org_id=VALUES(apply_org_id),
                apply_time=VALUES(apply_time), material_code=VALUES(material_code), product_name=VALUES(product_name),
                product_model=VALUES(product_model), production_batch_no=VALUES(production_batch_no), order_qty=VALUES(order_qty),
                qty=VALUES(qty), applicant=VALUES(applicant), remark=VALUES(remark), priority=VALUES(priority),
                inspect_start_time=VALUES(inspect_start_time), inspect_finish_time=VALUES(inspect_finish_time),
                inspector=VALUES(inspector), inspect_progress=VALUES(inspect_progress),
                sync_batch_id=VALUES(sync_batch_id), sync_time=VALUES(sync_time), update_time=CURRENT_TIMESTAMP
            """;
        var insPars = new List<SugarParameter> { new("@BatchId", batchId), new("@Now", now) };
        if (sourceSystem != null) insPars.Add(new SugarParameter("@Src", sourceSystem));
        AddScopePars(insPars, scope);
        await _db.Ado.ExecuteCommandAsync(insertSql, insPars);
        return rows;
    }

    /// <summary>标准化结果：pull stg(qms_qcpp_inspbill) + WorkOrdMaster(本库富化 SAP 工单号) → mdp_std_fqc_result。</summary>
    private async Task<int> TransformResultStandardAsync(string batchId, DateTime now, string? sourceSystem = null, FqcBillScope? scope = null)
    {
        var srcClause = sourceSystem == null ? "" : " AND a.source_system=@Src";
        var aTenant = MdpJsonSql.TenantFromStgJsonCol("a", MdpJsonSql.Int("a", "tenant_id"));
        var where = $"a.source_table='qms_qcpp_inspbill'{srcClause} AND {MdpJsonSql.TenantGuard(aTenant)}";
        // 单据级作用域（targeted sync）：按 stg 的 source_row_id（=inspbill.id）收敛，并强制租户相等，杜绝跨租户命中
        if (scope != null)
            where += $" AND ({aTenant})=@ScopeTenant AND IFNULL(a.source_row_id, {MdpJsonSql.Str("a", "id")})=@ScopeBillId";
        var sapSub = $"(SELECT MIN(wo.WorkOrd) FROM WorkOrdMaster wo WHERE wo.Batch = {MdpJsonSql.Str("a", "sczld")})";
        // 判定码表归一（FIX-A2）：源 pd 存两套编码——业务写入 0/1（FqcInspBillFlowService 校验口径 0=合格/1=不合格），
        // 历史归口数据直存中文。std 面向展示统一为中文；非 0/1 的历史值 ELSE 原样透传，绝不破坏既有数据。
        // 仅映射整单 pd；明细行 pd（qms_qcpp_inspbillst.pd）源本身即中文，不映射。源表 pd 保持 0/1 不动。
        var judgmentExpr = $"CASE {MdpJsonSql.Str("a", "pd")} WHEN '0' THEN '合格' WHEN '1' THEN '不合格' ELSE {MdpJsonSql.Str("a", "pd")} END";

        var countPars = new List<SugarParameter>();
        if (sourceSystem != null) countPars.Add(new SugarParameter("@Src", sourceSystem));
        AddScopePars(countPars, scope);
        var rows = await _db.Ado.GetIntAsync(
            $"SELECT COUNT(1) FROM mdp_stg_fqc_pull a WHERE {where}", countPars);

        var insertSql =
            $"""
            INSERT INTO mdp_std_fqc_result
            (tenant_id, source_system, bill_no, inspect_time, production_batch_no, material_code, material_name, spec,
             inspector, inspect_qty, source_bill_no, production_order_no, order_qty, sap_work_order_no, judgment,
             qualified_qty, unqualified_qty, fire_no, inspect_spec_no, inspect_spec_version,
             source_row_id, source_biz_key, sync_batch_id, sync_time)
            SELECT
                {aTenant}, IFNULL(NULLIF(a.source_system,''), 'AIDOP'),
                {MdpJsonSql.Str("a", "FBILLNO")}, {MdpJsonSql.DateTimeSec("a", "FINSPEENDDATE")}, {MdpJsonSql.Str("a", "scph")}, {MdpJsonSql.Str("a", "FMATERIALCFG")}, {MdpJsonSql.Str("a", "wlmc")}, {MdpJsonSql.Str("a", "ggxh")},
                {MdpJsonSql.Str("a", "jyr")}, {MdpJsonSql.Dec("a", "jysj", 18, 6)}, {MdpJsonSql.Str("a", "lydjbh")}, {MdpJsonSql.Str("a", "sczld")}, {MdpJsonSql.Dec("a", "sczldsl", 18, 6)}, {sapSub}, {judgmentExpr},
                {MdpJsonSql.Dec("a", "hgsl", 18, 6)}, {MdpJsonSql.Dec("a", "bhgsl", 18, 6)}, {MdpJsonSql.Str("a", "jfbh")}, {MdpJsonSql.Str("a", "jgbh")}, {MdpJsonSql.Str("a", "jgbb")},
                IFNULL(a.source_row_id, {MdpJsonSql.Str("a", "id")}), IFNULL(NULLIF(a.source_biz_key,''), {MdpJsonSql.Str("a", "FBILLNO")}),
                @BatchId, @Now
            FROM mdp_stg_fqc_pull a
            WHERE {where}
            ON DUPLICATE KEY UPDATE
                bill_no=VALUES(bill_no), inspect_time=VALUES(inspect_time), production_batch_no=VALUES(production_batch_no),
                material_code=VALUES(material_code), material_name=VALUES(material_name), spec=VALUES(spec),
                inspector=VALUES(inspector), inspect_qty=VALUES(inspect_qty), source_bill_no=VALUES(source_bill_no),
                production_order_no=VALUES(production_order_no), order_qty=VALUES(order_qty), sap_work_order_no=VALUES(sap_work_order_no),
                judgment=VALUES(judgment), qualified_qty=VALUES(qualified_qty), unqualified_qty=VALUES(unqualified_qty),
                fire_no=VALUES(fire_no), inspect_spec_no=VALUES(inspect_spec_no), inspect_spec_version=VALUES(inspect_spec_version),
                sync_batch_id=VALUES(sync_batch_id), sync_time=VALUES(sync_time), update_time=CURRENT_TIMESTAMP
            """;
        var insPars = new List<SugarParameter> { new("@BatchId", batchId), new("@Now", now) };
        if (sourceSystem != null) insPars.Add(new SugarParameter("@Src", sourceSystem));
        AddScopePars(insPars, scope);
        await _db.Ado.ExecuteCommandAsync(insertSql, insPars);
        return rows;
    }

    /// <summary>标准化明细：pull stg(qms_qcpp_inspbillst) → mdp_std_fqc_result_detail（回填 std_result_id，j1..j50 逐列）。</summary>
    private async Task<int> TransformResultDetailStandardAsync(string batchId, DateTime now, string? sourceSystem = null, FqcBillScope? scope = null)
    {
        var srcClause = sourceSystem == null ? "" : " AND d.source_system=@Src";
        var dTenant = MdpJsonSql.TenantFromStgJsonCol("d", MdpJsonSql.Int("d", "tenant_id"));
        // 单据级作用域（targeted sync）：明细按 glid（=inspbill.id）收敛，并强制租户相等
        var scopeClause = scope == null
            ? ""
            : $"\n              AND ({dTenant})=@ScopeTenant AND {MdpJsonSql.Str("d", "glid")}=@ScopeBillId";
        var fromWhere =
            $"""
            FROM mdp_stg_fqc_pull d
            LEFT JOIN mdp_std_fqc_result r
              ON r.tenant_id = {dTenant} AND r.source_row_id = {MdpJsonSql.Str("d", "glid")}
            WHERE d.source_table='qms_qcpp_inspbillst'{srcClause}
              AND {MdpJsonSql.TenantGuard(dTenant)}{scopeClause}
            """;

        var countPars = new List<SugarParameter>();
        if (sourceSystem != null) countPars.Add(new SugarParameter("@Src", sourceSystem));
        AddScopePars(countPars, scope);
        var rows = await _db.Ado.GetIntAsync($"SELECT COUNT(1) {fromWhere}", countPars);

        var jCols = string.Join(", ", Enumerable.Range(1, 50).Select(i => $"j{i}"));
        var jSelect = string.Join(", ", Enumerable.Range(1, 50).Select(i => MdpJsonSql.Str("d", $"j{i}")));
        var jUpdate = string.Join(", ", Enumerable.Range(1, 50).Select(i => $"j{i}=VALUES(j{i})"));

        var insertSql =
            $"""
            INSERT INTO mdp_std_fqc_result_detail
            (tenant_id, std_result_id, glid, seq_no, inspection_item, accept_standard, tech_requirement, sampling_plan,
             sample_qty, upper_limit, lower_limit, size_specification, judgment, non_numeric_ok, non_numeric_ng, remark,
             {jCols},
             source_row_id, source_biz_key, sync_batch_id, sync_time)
            SELECT
                {dTenant}, r.id, {MdpJsonSql.Str("d", "glid")}, {MdpJsonSql.Int("d", "xh")},
                {MdpJsonSql.Str("d", "jyxm")}, {MdpJsonSql.Str("d", "jsbz")}, {MdpJsonSql.Str("d", "jsyq")}, {MdpJsonSql.Str("d", "cyfa")},
                {MdpJsonSql.Str("d", "ybl")}, {MdpJsonSql.Str("d", "sx")}, {MdpJsonSql.Str("d", "xx")}, {MdpJsonSql.Str("d", "SizeSpecifications")}, {MdpJsonSql.Str("d", "pd")}, {MdpJsonSql.Int("d", "NonNumericalTypeOK")}, {MdpJsonSql.Int("d", "NonNumericalTypeNG")}, {MdpJsonSql.Str("d", "bz")},
                {jSelect},
                IFNULL(d.source_row_id, {MdpJsonSql.Str("d", "id")}), IFNULL(NULLIF(d.source_biz_key,''), {MdpJsonSql.Str("d", "id")}),
                @BatchId, @Now
            {fromWhere}
            ON DUPLICATE KEY UPDATE
                std_result_id=VALUES(std_result_id), glid=VALUES(glid), seq_no=VALUES(seq_no),
                inspection_item=VALUES(inspection_item), accept_standard=VALUES(accept_standard), tech_requirement=VALUES(tech_requirement),
                sampling_plan=VALUES(sampling_plan), sample_qty=VALUES(sample_qty), upper_limit=VALUES(upper_limit),
                lower_limit=VALUES(lower_limit), size_specification=VALUES(size_specification), judgment=VALUES(judgment),
                non_numeric_ok=VALUES(non_numeric_ok), non_numeric_ng=VALUES(non_numeric_ng), remark=VALUES(remark),
                {jUpdate},
                sync_batch_id=VALUES(sync_batch_id), sync_time=VALUES(sync_time), update_time=CURRENT_TIMESTAMP
            """;
        var insPars = new List<SugarParameter> { new("@BatchId", batchId), new("@Now", now) };
        if (sourceSystem != null) insPars.Add(new SugarParameter("@Src", sourceSystem));
        AddScopePars(insPars, scope);
        await _db.Ado.ExecuteCommandAsync(insertSql, insPars);
        return rows;
    }

    private async Task<long> InsertRunLogAsync(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, 'S7 FQC 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 MarkRunSuccessAsync(long runLogId, DateTime startedAt, FqcMdpSyncResult 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=0, 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.TaskRows + result.ResultRows + result.DetailRows),
            new SugarParameter("@Id", runLogId));
    }

    private async Task MarkRunFailedAsync(long runLogId, DateTime startedAt, string message)
    {
        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", message.Length > 2000 ? message[..2000] : message),
            new SugarParameter("@Id", runLogId));
    }

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

/// <summary>S7 FQC MDP 同步转换结果。</summary>
public sealed class FqcMdpSyncResult
{
    public long RunLogId { get; set; }
    public string BatchId { get; set; } = string.Empty;
    public int StageRows { get; set; }
    public int TaskRows { get; set; }
    public int ResultRows { get; set; }
    public int DetailRows { get; set; }
    /// <summary>1E-0：并发锁未抢到、本次跳过（交由持锁者或下一轮刷新）。</summary>
    public bool Skipped { get; set; }
}

/// <summary>S7 FQC 双模式入站结果。</summary>
public sealed class FqcInboundResult
{
    public string PullBatchId { get; set; } = string.Empty;
    public int RowsPulled { get; set; }
    public int RowsWrittenStg { get; set; }
    public string? NewCursor { get; set; }
    public string TransformBatchId { get; set; } = string.Empty;
    public int TaskRows { get; set; }
    public int ResultRows { get; set; }
    public int DetailRows { get; set; }
}