AiDOPWarehouse/server/Plugins/Admin.NET.Plugin.AiDOP/Manufacturing/IpqcInspectionMdpSyncService.cs

namespace Admin.NET.Plugin.AiDOP.Manufacturing;

/// <summary>
/// S6 过程检验单（IPQC）数据中台只读同步转换服务（DOP 内部，独立于 S6MdpSyncTransformService 的 KPI 管线）。
///
/// 源：aidopdev.qms_gcjyd（过程检验单头，31 列）+ aidopdev.qms_gcjydzb（明细，100 列），归口自旧系统 dopdemo。
/// 链路：qms_gcjyd/qms_gcjydzb 只读 → mdp_stg_ipqc_inspection(_detail)(raw_data JSON) → mdp_std_ipqc_inspection(_detail)(typed)。
///
/// 归口说明（CTO 拍板方案 A）：
///   qms_gcjyd/qms_gcjydzb 原仅存于旧系统 dopdemo（0 行）、aidopdev 无此表，且 dopdemo 非运行时连接。
///   本服务 EnsureTablesAsync 在 aidopdev 建同构空表（IF NOT EXISTS）作为归口源；
///   后续由上游一次性抽数填充，届时无需改代码即随刷新流通。
///
/// 约束：
///   - 只读 qms_gcjyd/qms_gcjydzb，仅写 mdp_stg_*/mdp_std_*/run_log；绝不 INSERT/UPDATE/DELETE 业务源单据表。
///   - 源表无 tenant/org/factory 列 → 标准层按常量 tenant_id=0, factory_id=1 落库。
///   - 明细头关联：qms_gcjydzb.glid -> qms_gcjyd.id；标准层经 std_head_id 回填 mdp_std_ipqc_inspection.id。
///   - 明细只转换已确认子集列（检验项目/依据/工序/方法/技术标准/上下限/测量仪器/检验人/检验时间/样本量/检验数量/不合格数量/判定/备注等）；
///     j1~j70 动态检验值列语义不明、后置不转换、不落 typed 字段。
///   - result_judgement（jgpd/pd）为字典码，原样透出，语义解码后置。
///   - 源为 0 行时转换成功完成、处理数为 0，不报错。
/// </summary>
public class IpqcInspectionMdpSyncService : ITransient
{
    private const string JobCode = "S6_IPQC_INSPECTION_MDP_SYNC";

    private readonly ISqlSugarClient _db;

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

    /// <summary>
    /// 全量同步转换：贴源(头/明细) → 标准化(头/明细)。
    /// </summary>
    public async Task<IpqcInspectionMdpSyncResult> RunFullAsync(CancellationToken cancellationToken = default, string triggerType = "AUTO")
    {
        cancellationToken.ThrowIfCancellationRequested();
        await EnsureTablesAsync();

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

        try
        {
            result.HeadStageRows = await SyncHeadStagingAsync(batchId, now);
            result.DetailStageRows = await SyncDetailStagingAsync(batchId, now);
            result.HeadStandardRows = await TransformHeadStandardAsync(batchId, now);
            result.DetailStandardRows = await TransformDetailStandardAsync(batchId, now);
            await MarkRunSuccessAsync(runLogId, now, result);
            return result;
        }
        catch (Exception ex)
        {
            await MarkRunFailedAsync(runLogId, now, ex.Message);
            throw;
        }
    }

    /// <summary>防御式建表（幂等）：归口源空表 qms_gcjyd/qms_gcjydzb + 贴源层(头/明细) + 标准层(头/明细)。</summary>
    private async Task EnsureTablesAsync()
    {
        // 归口源表（方案 A）：同构自旧系统 dopdemo.qms_gcjyd 头表，空表，待上游抽数填充。
        // id 不设 AUTO_INCREMENT：抽数将携带旧系统主键。
        await _db.Ado.ExecuteCommandAsync(
            """
            CREATE TABLE IF NOT EXISTS qms_gcjyd (
                id BIGINT NOT NULL PRIMARY KEY COMMENT '主表id',
                djbh VARCHAR(255) NULL COMMENT '单据编号',
                cplx VARCHAR(255) NULL COMMENT '产品型号',
                scph VARCHAR(255) NULL COMMENT '生产批号',
                jyrq VARCHAR(255) NULL COMMENT '检验日期',
                lydjbh VARCHAR(255) NULL COMMENT '来源单据编号',
                jgpd BIGINT NULL COMMENT '结果判定',
                jyhgsl DECIMAL(10,0) NULL COMMENT '检验合格数量',
                jybhgsl DECIMAL(10,0) NULL COMMENT '检验不合格数量',
                fj TEXT NULL COMMENT '附件',
                bz TEXT NULL COMMENT '备注',
                jyr VARCHAR(255) NULL COMMENT '检验人',
                gxbm VARCHAR(255) NULL COMMENT '工序编码',
                gxmc VARCHAR(255) NULL COMMENT '工序名称',
                sczyry VARCHAR(255) NULL COMMENT '生产作业人员',
                ybl DECIMAL(10,0) NULL COMMENT '样本量',
                bdbh VARCHAR(255) NULL COMMENT '表单编号',
                bbh VARCHAR(255) NULL COMMENT '版本号',
                sxrq DATETIME NULL COMMENT '生效日期',
                wlbm VARCHAR(255) NULL COMMENT '物料编码',
                wlmc VARCHAR(255) NULL COMMENT '物料名称',
                lysl DECIMAL(10,0) NULL COMMENT '留样数量',
                phsl DECIMAL(10,0) NULL COMMENT '破坏数量',
                jygfid VARCHAR(255) NULL COMMENT '检验规范',
                jygfbbm VARCHAR(255) NULL COMMENT '没有用',
                jgbb VARCHAR(255) NULL COMMENT '检规版本',
                jgbh VARCHAR(255) NULL COMMENT '检规编号',
                title VARCHAR(255) NULL,
                jyr1 VARCHAR(255) NULL,
                ChangeStatus INT NULL,
                status VARCHAR(255) NULL COMMENT '检验状态'
            ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci COMMENT='过程检验单(头) 归口自旧系统dopdemo，空表待上游抽数'
            """);
        await _db.Ado.ExecuteCommandAsync(
            """
            CREATE TABLE IF NOT EXISTS mdp_stg_ipqc_inspection (
                id BIGINT AUTO_INCREMENT PRIMARY KEY,
                tenant_id BIGINT NOT NULL DEFAULT 0,
                factory_id BIGINT NULL DEFAULT 1,
                source_system VARCHAR(50) NOT NULL DEFAULT 'AIDOP',
                source_table VARCHAR(100) NOT 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,
                process_status VARCHAR(20) NOT NULL DEFAULT 'PENDING',
                raw_data JSON NOT NULL,
                update_time DATETIME DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
                UNIQUE KEY uk_mdp_stg_ipqc_insp (tenant_id, source_table, source_row_id)
            ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci COMMENT='S6过程检验单头贴源层'
            """);
        await _db.Ado.ExecuteCommandAsync(
            """
            CREATE TABLE IF NOT EXISTS mdp_std_ipqc_inspection (
                id BIGINT AUTO_INCREMENT PRIMARY KEY,
                tenant_id BIGINT NOT NULL DEFAULT 0,
                factory_id BIGINT NULL DEFAULT 1,
                source_system VARCHAR(50) NOT NULL DEFAULT 'AIDOP',
                bill_no VARCHAR(64) NOT NULL,
                product_model VARCHAR(255) NULL,
                production_batch_no VARCHAR(255) NULL,
                production_work_order VARCHAR(255) NULL,
                result_judgement VARCHAR(32) NULL,
                attachment TEXT NULL,
                remark TEXT NULL,
                inspector VARCHAR(255) NULL,
                process_code VARCHAR(255) NULL,
                process_name VARCHAR(255) NULL,
                production_person VARCHAR(255) NULL,
                sample_qty DECIMAL(18,6) NULL,
                form_no VARCHAR(255) NULL,
                version_no VARCHAR(255) NULL,
                effective_date DATETIME NULL,
                material_code VARCHAR(255) NULL,
                material_name VARCHAR(255) NULL,
                inspec_standard_version VARCHAR(255) NULL,
                inspec_standard_code VARCHAR(255) NULL,
                inspection_status 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_ipqc_insp (tenant_id, source_row_id),
                KEY idx_mdp_std_ipqc_insp_bill (tenant_id, bill_no),
                KEY idx_mdp_std_ipqc_insp_effdate (tenant_id, effective_date)
            ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci COMMENT='S6过程检验单头标准层'
            """);
        // 归口源明细表（方案 A）：同构自旧系统 dopdemo.qms_gcjydzb（100 列），空表，待上游抽数填充。
        // 关联键 glid -> qms_gcjyd.id。id 不设 AUTO_INCREMENT：抽数将携带旧系统主键。
        await _db.Ado.ExecuteCommandAsync(
            """
            CREATE TABLE IF NOT EXISTS qms_gcjydzb (
                id BIGINT NOT NULL PRIMARY KEY COMMENT '子表id',
                glid BIGINT NULL COMMENT '关联id',
                dwsj TEXT NULL, jyxm TEXT NULL, jybz TEXT NULL, jyyj TEXT NULL,
                j1 TEXT NULL, j2 TEXT NULL, j3 TEXT NULL, j4 TEXT NULL, j5 TEXT NULL,
                ljbh TEXT NULL, sczyry TEXT NULL, fj TEXT NULL, bz TEXT NULL,
                j6 TEXT NULL, j7 TEXT NULL, j8 TEXT NULL, j9 TEXT NULL, j10 TEXT NULL,
                j11 TEXT NULL, j12 TEXT NULL, j13 TEXT NULL, j14 TEXT NULL, j15 TEXT NULL,
                j16 TEXT NULL, j17 TEXT NULL, j18 TEXT NULL, j19 TEXT NULL, j20 TEXT NULL,
                j21 TEXT NULL, j22 TEXT NULL, j23 TEXT NULL, j24 TEXT NULL, j25 TEXT NULL,
                j26 TEXT NULL, j27 TEXT NULL, j28 TEXT NULL, j29 TEXT NULL, j30 TEXT NULL,
                j31 TEXT NULL, j32 TEXT NULL, j33 TEXT NULL, j34 TEXT NULL, j35 TEXT NULL,
                j36 TEXT NULL, j37 TEXT NULL, j38 TEXT NULL, j39 TEXT NULL, j40 TEXT NULL,
                pd BIGINT NULL COMMENT '结果判定',
                jysl DECIMAL(10,0) NULL COMMENT '检验数量',
                bhgsl DECIMAL(10,0) NULL COMMENT '不合格数量',
                j41 TEXT NULL, j42 TEXT NULL, j43 TEXT NULL, j44 TEXT NULL, j45 TEXT NULL,
                j46 TEXT NULL, j47 TEXT NULL, j48 TEXT NULL, j49 TEXT NULL, j50 TEXT NULL,
                j51 TEXT NULL, j52 TEXT NULL, j53 TEXT NULL, j54 TEXT NULL, j55 TEXT NULL,
                j56 TEXT NULL, j57 TEXT NULL, j58 TEXT NULL, j59 TEXT NULL, j60 TEXT NULL,
                j61 TEXT NULL, j62 TEXT NULL, j63 TEXT NULL, j64 TEXT NULL, j65 TEXT NULL,
                j66 TEXT NULL, j67 TEXT NULL, j68 TEXT NULL, j69 TEXT NULL, j70 TEXT NULL,
                nature TEXT NULL, jsbz TEXT NULL, jypc TEXT NULL, ggbh TEXT NULL,
                ProcessCode TEXT NULL, ProcessName TEXT NULL, jyff TEXT NULL,
                ybl INT NULL COMMENT '样本量',
                NonNumericalTypeOK INT NULL COMMENT '非数值型OK',
                NonNumericalTypeNG INT NULL COMMENT '非数值型NG',
                Inspector TEXT NULL,
                InspectionTime DATETIME NULL COMMENT '检验时间',
                jgid BIGINT NULL,
                AbnormalNumber TEXT NULL, MeasurementInstrument TEXT NULL, sx TEXT NULL, xx TEXT NULL,
                KEY idx_qms_gcjydzb_glid (glid)
            ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci COMMENT='过程检验单(明细) 归口自旧系统dopdemo，空表待上游抽数'
            """);
        await _db.Ado.ExecuteCommandAsync(
            """
            CREATE TABLE IF NOT EXISTS mdp_stg_ipqc_inspection_detail (
                id BIGINT AUTO_INCREMENT PRIMARY KEY,
                tenant_id BIGINT NOT NULL DEFAULT 0,
                factory_id BIGINT NULL DEFAULT 1,
                source_system VARCHAR(50) NOT NULL DEFAULT 'AIDOP',
                source_table VARCHAR(100) NOT 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,
                process_status VARCHAR(20) NOT NULL DEFAULT 'PENDING',
                raw_data JSON NOT NULL,
                update_time DATETIME DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
                UNIQUE KEY uk_mdp_stg_ipqc_insp_dtl (tenant_id, source_table, source_row_id)
            ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci COMMENT='S6过程检验单明细贴源层'
            """);
        await _db.Ado.ExecuteCommandAsync(
            """
            CREATE TABLE IF NOT EXISTS mdp_std_ipqc_inspection_detail (
                id BIGINT AUTO_INCREMENT PRIMARY KEY,
                tenant_id BIGINT NOT NULL DEFAULT 0,
                std_head_id BIGINT NULL,
                source_head_id VARCHAR(100) NULL,
                inspection_item TEXT NULL,
                inspection_basis TEXT NULL,
                technical_standard TEXT NULL,
                inspection_method TEXT NULL,
                upper_limit VARCHAR(255) NULL,
                lower_limit VARCHAR(255) NULL,
                process_code VARCHAR(255) NULL,
                process_name VARCHAR(255) NULL,
                measurement_instrument VARCHAR(255) NULL,
                inspector VARCHAR(255) NULL,
                inspection_time DATETIME NULL,
                sample_qty INT NULL,
                inspection_qty DECIMAL(18,6) NULL,
                unqualified_qty DECIMAL(18,6) NULL,
                result_judgement VARCHAR(32) NULL,
                non_numeric_ok INT NULL,
                non_numeric_ng INT NULL,
                remark TEXT NULL,
                attachment TEXT 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_ipqc_insp_dtl (tenant_id, source_row_id),
                KEY idx_mdp_std_ipqc_insp_dtl_head (std_head_id)
            ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci COMMENT='S6过程检验单明细标准层(确定子集列;j1-j70后置)'
            """);
    }

    /// <summary>贴源头：qms_gcjyd -> mdp_stg_ipqc_inspection（raw_data JSON 快照）。返回源头行数。</summary>
    private async Task<int> SyncHeadStagingAsync(string batchId, DateTime now)
    {
        var rows = await _db.Ado.GetIntAsync("SELECT COUNT(1) FROM qms_gcjyd");
        await _db.Ado.ExecuteCommandAsync(
            """
            INSERT INTO mdp_stg_ipqc_inspection
            (tenant_id, factory_id, source_system, source_table, source_row_id, source_biz_key, sync_batch_id, sync_time, process_status, raw_data)
            SELECT
                0, 1, 'AIDOP', 'qms_gcjyd',
                CAST(m.id AS CHAR), CAST(m.djbh AS CHAR), @BatchId, @Now, 'PENDING',
                JSON_OBJECT(
                    'id', m.id, 'djbh', m.djbh, 'cplx', m.cplx, 'scph', m.scph, 'lydjbh', m.lydjbh,
                    'jgpd', m.jgpd, 'fj', CAST(m.fj AS CHAR), 'bz', CAST(m.bz AS CHAR), 'jyr', m.jyr,
                    'gxbm', m.gxbm, 'gxmc', m.gxmc, 'sczyry', m.sczyry, 'ybl', m.ybl,
                    'bdbh', m.bdbh, 'bbh', m.bbh, 'sxrq', m.sxrq, 'wlbm', m.wlbm, 'wlmc', m.wlmc,
                    'jgbb', m.jgbb, 'jgbh', m.jgbh, 'status', m.status
                )
            FROM qms_gcjyd m
            ON DUPLICATE KEY UPDATE
                source_biz_key=VALUES(source_biz_key), 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("@BatchId", batchId),
            new SugarParameter("@Now", now));
        return rows;
    }

    /// <summary>标准化头：qms_gcjyd -> mdp_std_ipqc_inspection（typed）。返回处理头行数。</summary>
    private async Task<int> TransformHeadStandardAsync(string batchId, DateTime now)
    {
        var rows = await _db.Ado.GetIntAsync("SELECT COUNT(1) FROM qms_gcjyd");
        await _db.Ado.ExecuteCommandAsync(
            """
            INSERT INTO mdp_std_ipqc_inspection
            (tenant_id, factory_id, source_system, bill_no, product_model, production_batch_no, production_work_order,
             result_judgement, attachment, remark, inspector, process_code, process_name, production_person,
             sample_qty, form_no, version_no, effective_date, material_code, material_name,
             inspec_standard_version, inspec_standard_code, inspection_status,
             source_row_id, source_biz_key, sync_batch_id, sync_time)
            SELECT
                0, 1, 'AIDOP', IFNULL(m.djbh, CAST(m.id AS CHAR)), m.cplx, m.scph, m.lydjbh,
                CAST(m.jgpd AS CHAR), CAST(m.fj AS CHAR), CAST(m.bz AS CHAR), m.jyr, m.gxbm, m.gxmc, m.sczyry,
                m.ybl, m.bdbh, m.bbh, m.sxrq, m.wlbm, m.wlmc,
                m.jgbb, m.jgbh, m.status,
                CAST(m.id AS CHAR), CAST(m.djbh AS CHAR), @BatchId, @Now
            FROM qms_gcjyd m
            ON DUPLICATE KEY UPDATE
                bill_no=VALUES(bill_no), product_model=VALUES(product_model), production_batch_no=VALUES(production_batch_no),
                production_work_order=VALUES(production_work_order), result_judgement=VALUES(result_judgement),
                attachment=VALUES(attachment), remark=VALUES(remark), inspector=VALUES(inspector),
                process_code=VALUES(process_code), process_name=VALUES(process_name), production_person=VALUES(production_person),
                sample_qty=VALUES(sample_qty), form_no=VALUES(form_no), version_no=VALUES(version_no),
                effective_date=VALUES(effective_date), material_code=VALUES(material_code), material_name=VALUES(material_name),
                inspec_standard_version=VALUES(inspec_standard_version), inspec_standard_code=VALUES(inspec_standard_code),
                inspection_status=VALUES(inspection_status),
                sync_batch_id=VALUES(sync_batch_id), sync_time=VALUES(sync_time), update_time=CURRENT_TIMESTAMP
            """,
            new SugarParameter("@BatchId", batchId),
            new SugarParameter("@Now", now));
        return rows;
    }

    /// <summary>贴源明细：qms_gcjydzb -> mdp_stg_ipqc_inspection_detail（raw_data JSON 快照，确定子集列）。返回源明细行数。</summary>
    private async Task<int> SyncDetailStagingAsync(string batchId, DateTime now)
    {
        var rows = await _db.Ado.GetIntAsync("SELECT COUNT(1) FROM qms_gcjydzb");
        await _db.Ado.ExecuteCommandAsync(
            """
            INSERT INTO mdp_stg_ipqc_inspection_detail
            (tenant_id, factory_id, source_system, source_table, source_row_id, source_biz_key, sync_batch_id, sync_time, process_status, raw_data)
            SELECT
                0, 1, 'AIDOP', 'qms_gcjydzb',
                CAST(d.id AS CHAR), CAST(d.glid AS CHAR), @BatchId, @Now, 'PENDING',
                JSON_OBJECT(
                    'id', d.id, 'glid', d.glid, 'jyxm', CAST(d.jyxm AS CHAR), 'jyyj', CAST(d.jyyj AS CHAR),
                    'jsbz', CAST(d.jsbz AS CHAR), 'jyff', CAST(d.jyff AS CHAR), 'sx', CAST(d.sx AS CHAR), 'xx', CAST(d.xx AS CHAR),
                    'ProcessCode', CAST(d.ProcessCode AS CHAR), 'ProcessName', CAST(d.ProcessName AS CHAR),
                    'MeasurementInstrument', CAST(d.MeasurementInstrument AS CHAR), 'Inspector', CAST(d.Inspector AS CHAR),
                    'InspectionTime', d.InspectionTime, 'ybl', d.ybl, 'jysl', d.jysl, 'bhgsl', d.bhgsl,
                    'pd', d.pd, 'NonNumericalTypeOK', d.NonNumericalTypeOK, 'NonNumericalTypeNG', d.NonNumericalTypeNG,
                    'bz', CAST(d.bz AS CHAR), 'fj', CAST(d.fj AS CHAR)
                )
            FROM qms_gcjydzb d
            ON DUPLICATE KEY UPDATE
                source_biz_key=VALUES(source_biz_key), 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("@BatchId", batchId),
            new SugarParameter("@Now", now));
        return rows;
    }

    /// <summary>标准化明细：qms_gcjydzb -> mdp_std_ipqc_inspection_detail（确定子集列；经 glid 回填 std_head_id）。返回处理明细行数。</summary>
    private async Task<int> TransformDetailStandardAsync(string batchId, DateTime now)
    {
        var rows = await _db.Ado.GetIntAsync("SELECT COUNT(1) FROM qms_gcjydzb");
        await _db.Ado.ExecuteCommandAsync(
            """
            INSERT INTO mdp_std_ipqc_inspection_detail
            (tenant_id, std_head_id, source_head_id, inspection_item, inspection_basis, technical_standard, inspection_method,
             upper_limit, lower_limit, process_code, process_name, measurement_instrument, inspector, inspection_time,
             sample_qty, inspection_qty, unqualified_qty, result_judgement, non_numeric_ok, non_numeric_ng, remark, attachment,
             source_row_id, source_biz_key, sync_batch_id, sync_time)
            SELECT
                0, h.id, CAST(d.glid AS CHAR), CAST(d.jyxm AS CHAR), CAST(d.jyyj AS CHAR), CAST(d.jsbz AS CHAR), CAST(d.jyff AS CHAR),
                CAST(d.sx AS CHAR), CAST(d.xx AS CHAR), CAST(d.ProcessCode AS CHAR), CAST(d.ProcessName AS CHAR),
                CAST(d.MeasurementInstrument AS CHAR), CAST(d.Inspector AS CHAR), d.InspectionTime,
                d.ybl, d.jysl, d.bhgsl, CAST(d.pd AS CHAR), d.NonNumericalTypeOK, d.NonNumericalTypeNG,
                CAST(d.bz AS CHAR), CAST(d.fj AS CHAR),
                CAST(d.id AS CHAR), CAST(d.glid AS CHAR), @BatchId, @Now
            FROM qms_gcjydzb d
            LEFT JOIN mdp_std_ipqc_inspection h ON h.tenant_id = 0 AND h.source_row_id = CAST(d.glid AS CHAR)
            ON DUPLICATE KEY UPDATE
                std_head_id=VALUES(std_head_id), source_head_id=VALUES(source_head_id), inspection_item=VALUES(inspection_item),
                inspection_basis=VALUES(inspection_basis), technical_standard=VALUES(technical_standard), inspection_method=VALUES(inspection_method),
                upper_limit=VALUES(upper_limit), lower_limit=VALUES(lower_limit), process_code=VALUES(process_code), process_name=VALUES(process_name),
                measurement_instrument=VALUES(measurement_instrument), inspector=VALUES(inspector), inspection_time=VALUES(inspection_time),
                sample_qty=VALUES(sample_qty), inspection_qty=VALUES(inspection_qty), unqualified_qty=VALUES(unqualified_qty),
                result_judgement=VALUES(result_judgement), non_numeric_ok=VALUES(non_numeric_ok), non_numeric_ng=VALUES(non_numeric_ng),
                remark=VALUES(remark), attachment=VALUES(attachment),
                sync_batch_id=VALUES(sync_batch_id), sync_time=VALUES(sync_time), update_time=CURRENT_TIMESTAMP
            """,
            new SugarParameter("@BatchId", batchId),
            new SugarParameter("@Now", now));
        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, 'S6过程检验单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, IpqcInspectionMdpSyncResult 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.HeadStageRows + result.DetailStageRows),
            new SugarParameter("@StandardRows", result.HeadStandardRows + result.DetailStandardRows),
            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>过程检验单 MDP 同步转换结果。</summary>
public sealed class IpqcInspectionMdpSyncResult
{
    public long RunLogId { get; set; }
    public string BatchId { get; set; } = string.Empty;
    public int HeadStageRows { get; set; }
    public int HeadStandardRows { get; set; }
    public int DetailStageRows { get; set; }
    public int DetailStandardRows { get; set; }
}