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;

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

    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();

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

    /// <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;
        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();

        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)
    {
        var task = await _pullDispatcher.PullByEntityCodeAsync(InboundTaskEntityCode, pullCtx, cancellationToken);
        var res = await _pullDispatcher.PullByEntityCodeAsync(InboundResultEntityCode, pullCtx, cancellationToken);
        var det = await _pullDispatcher.PullByEntityCodeAsync(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 DEFAULT 1300000000001 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 DEFAULT 1300000000001 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)
    {
        var srcClause = sourceSystem == null ? "" : " AND t.source_system=@Src";
        var tTenant = $"IFNULL({MdpJsonSql.Int("t", "tenant_id")}, IFNULL(t.tenant_id, 0))";
        var where = $"t.source_table='qms_fqcbj'{srcClause}";

        var countPars = new List<SugarParameter>();
        if (sourceSystem != null) countPars.Add(new SugarParameter("@Src", sourceSystem));
        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));
        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)
    {
        var srcClause = sourceSystem == null ? "" : " AND a.source_system=@Src";
        var aTenant = $"IFNULL({MdpJsonSql.Int("a", "tenant_id")}, IFNULL(a.tenant_id, 0))";
        var where = $"a.source_table='qms_qcpp_inspbill'{srcClause}";
        var sapSub = $"(SELECT MIN(wo.WorkOrd) FROM WorkOrdMaster wo WHERE wo.Batch = {MdpJsonSql.Str("a", "sczld")})";

        var countPars = new List<SugarParameter>();
        if (sourceSystem != null) countPars.Add(new SugarParameter("@Src", sourceSystem));
        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}, {MdpJsonSql.Str("a", "pd")},
                {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));
        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)
    {
        var srcClause = sourceSystem == null ? "" : " AND d.source_system=@Src";
        var dTenant = $"IFNULL({MdpJsonSql.Int("d", "tenant_id")}, IFNULL(d.tenant_id, 0))";
        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}
            """;

        var countPars = new List<SugarParameter>();
        if (sourceSystem != null) countPars.Add(new SugarParameter("@Src", sourceSystem));
        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));
        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>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; }
}