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（过程检验单头，归口自旧系统 dopdemo，31 列）。
/// 链路：qms_gcjyd 只读 → mdp_stg_ipqc_inspection(raw_data JSON) → mdp_std_ipqc_inspection(typed)。
///
/// 归口说明（CTO 拍板方案 A）：
///   qms_gcjyd 原仅存于旧系统 dopdemo（0 行）、aidopdev 无此表，且 dopdemo 非运行时连接。
///   本服务 EnsureTablesAsync 在 aidopdev 建同构空表 qms_gcjyd（IF NOT EXISTS），
///   作为归口源；后续由上游一次性抽数填充，届时无需改代码即随刷新流通。
///
/// 约束：
///   - 只读 qms_gcjyd，仅写 mdp_stg_*/mdp_std_*/run_log；绝不 INSERT/UPDATE/DELETE 业务源单据表。
///   - qms_gcjyd 无 tenant/org/factory 列 → 标准层按常量 tenant_id=0, factory_id=1 落库。
///   - 本批只读列表：只做头表转换，不碰明细 qms_gcjydzb（留待 DETAIL 批）。
///   - result_judgement（jgpd）为字典码，原样透出，语义解码后置。
///   - 源为 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.HeadStandardRows = await TransformHeadStandardAsync(batchId, now);
            await MarkRunSuccessAsync(runLogId, now, result);
            return result;
        }
        catch (Exception ex)
        {
            await MarkRunFailedAsync(runLogId, now, ex.Message);
            throw;
        }
    }

    /// <summary>防御式建表（幂等）：归口源空表 qms_gcjyd + 贴源层 + 标准层。</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过程检验单头标准层'
            """);
    }

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

    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),
            new SugarParameter("@StandardRows", result.HeadStandardRows),
            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; }
}