AiDOPWarehouse/server/Plugins/Admin.NET.Plugin.AiDOP/MaterialWarehouse/ProductionIssueMdpSyncService.cs

namespace Admin.NET.Plugin.AiDOP.MaterialWarehouse;

/// <summary>
/// S5 生产领料单 数据中台只读同步转换服务（DOP 内部，头级单表，独立于 S5MdpSyncTransformService 的 KPI 管线）。
///
/// 源：aidopdev NbrMaster(m)，业务类型 Type='SM'；维表 DepartmentMaster(d) LEFT JOIN（Domain+Department）。
/// 链路：NbrMaster(SM) 只读 → mdp_std_production_issue(typed)。
///     不经 stg、不建明细、不建 dwd（只读列表）；SM=领料单业务类型有 WorkOrderPickBillService 代码实证（工单下达 Type=SM）。
///
/// 约束：
///   - 只读源表，仅写 mdp_std_production_issue；绝不 INSERT/UPDATE/DELETE NbrMaster；不碰 WorkOrderPickBillService 写路径。
///   - 头过滤 m.Type='SM' AND m.IsActive=1（领料无 IsReturn 语义）。
///   - status_desc 用旧系统领料 SQL 口径：PretreatmentState 非空取其值，Status='C'→'已下架'，'A'→'备料中'，否则空。
///   - 领料人 issue_user 取 m.User1（列表 SQL 口径），原样落库。
///   - trans_type 展示：TransType='Z61'→'补料'，否则'正常'；trans_type_text：TransType='PrevProcess'→'需要前处理'，否则空（旧系统口径）。
///   - 不做 danjia/jiage/库存事务/状态流转；SM 为 0 行时转换成功完成、处理数为 0，不报错。
/// </summary>
public class ProductionIssueMdpSyncService : ITransient
{
    private const string JobCode = "S5_PRODUCTION_ISSUE_MDP_SYNC";

    private readonly ISqlSugarClient _db;

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

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

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

        try
        {
            result.HeadRows = await TransformHeadStandardAsync(batchId, now);
            await MarkRunSuccessAsync(runLogId, now, result);
            return result;
        }
        catch (Exception ex)
        {
            await MarkRunFailedAsync(runLogId, now, ex.Message);
            throw;
        }
    }

    /// <summary>防御式建表（与 UpdateScripts DDL 同构，幂等）。</summary>
    private async Task EnsureTablesAsync()
    {
        await _db.Ado.ExecuteCommandAsync(
            """
            CREATE TABLE IF NOT EXISTS mdp_std_production_issue (
                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',
                domain VARCHAR(80) NOT NULL,
                rec_id INT NOT NULL,
                nbr VARCHAR(24) NULL,
                issue_date DATETIME NULL,
                status VARCHAR(8) NULL,
                status_desc VARCHAR(20) NULL,
                work_ord VARCHAR(64) NULL,
                department VARCHAR(8) NULL,
                department_desc VARCHAR(255) NULL,
                qty_ord DECIMAL(18,5) NULL DEFAULT 0,
                prod_line VARCHAR(8) NULL,
                applicant_name VARCHAR(12) NULL,
                issue_user VARCHAR(255) NULL,
                user1 TEXT NULL,
                remark VARCHAR(200) NULL,
                create_user VARCHAR(24) NULL,
                source_create_time DATETIME NULL,
                pretreatment_state VARCHAR(50) NULL,
                trans_type VARCHAR(24) NULL,
                trans_type_text VARCHAR(50) NULL,
                eff_date DATETIME NULL,
                address VARCHAR(120) NULL,
                source_biz_key VARCHAR(200) NULL,
                sync_batch_id VARCHAR(100) NOT NULL,
                sync_time DATETIME NOT NULL,
                create_time DATETIME DEFAULT CURRENT_TIMESTAMP,
                update_time DATETIME DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
                UNIQUE KEY uk_mdp_std_prod_issue (tenant_id, domain, rec_id),
                KEY idx_mdp_std_prod_issue_nbr (tenant_id, nbr),
                KEY idx_mdp_std_prod_issue_date (tenant_id, issue_date),
                KEY idx_mdp_std_prod_issue_status (tenant_id, status),
                KEY idx_mdp_std_prod_issue_workord (tenant_id, work_ord),
                KEY idx_mdp_std_prod_issue_prodline (tenant_id, prod_line),
                KEY idx_mdp_std_prod_issue_dept (tenant_id, department)
            ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci COMMENT='S5生产领料单头标准层'
            """);
    }

    /// <summary>
    /// 标准化头：NbrMaster(SM) + DepartmentMaster -> mdp_std_production_issue。返回处理头行数。
    /// SQL 由旧系统领料头 SQL 翻译（SQL Server -&gt; MySQL）：IsNull-&gt;IFNULL、upper() 保留、
    /// rtrim(a+' '+b)-&gt;TRIM(CONCAT(...))、with(nolock) 去除、convert(varchar,EffDate,23)-&gt; DATE(EffDate)。
    /// </summary>
    private async Task<int> TransformHeadStandardAsync(string batchId, DateTime now)
    {
        var rows = await _db.Ado.GetIntAsync(
            "SELECT COUNT(1) FROM NbrMaster m WHERE m.Type='SM' AND m.IsActive=1");
        await _db.Ado.ExecuteCommandAsync(
            """
            INSERT INTO mdp_std_production_issue
            (tenant_id, factory_id, source_system, domain, rec_id, nbr, issue_date, status, status_desc,
             work_ord, department, department_desc, qty_ord, prod_line, applicant_name, issue_user,
             user1, remark, create_user, source_create_time, pretreatment_state, trans_type, trans_type_text,
             eff_date, address, source_biz_key, sync_batch_id, sync_time)
            SELECT
                IFNULL(m.tenant_id, 0), 1, 'AIDOP', m.Domain, m.RecID, m.Nbr, m.Date,
                UPPER(IFNULL(m.Status, '')),
                CASE
                    WHEN IFNULL(m.PretreatmentState,'')<>'' THEN m.PretreatmentState
                    WHEN UPPER(IFNULL(m.Status,''))='C' THEN '已下架'
                    WHEN UPPER(IFNULL(m.Status,''))='A' THEN '备料中'
                    ELSE ''
                END,
                m.WorkOrd, m.Department, TRIM(CONCAT(m.Department, ' ', IFNULL(d.Descr, ''))),
                m.QtyOrd, m.ProdLine, m.Name, CAST(m.User1 AS CHAR),
                CAST(m.User1 AS CHAR), m.Remark, m.CreateUser, m.CreateTime,
                IFNULL(m.PretreatmentState, ''),
                CASE WHEN m.TransType='Z61' THEN '补料' ELSE '正常' END,
                CASE WHEN m.TransType='PrevProcess' THEN '需要前处理' ELSE '' END,
                m.EffDate, m.Address,
                CONCAT(m.Domain, ':', m.RecID), @BatchId, @Now
            FROM NbrMaster m
            LEFT JOIN DepartmentMaster d ON d.Domain = m.Domain AND d.Department = m.Department
            WHERE m.Type='SM' AND m.IsActive=1
            ON DUPLICATE KEY UPDATE
                factory_id=VALUES(factory_id), nbr=VALUES(nbr), issue_date=VALUES(issue_date),
                status=VALUES(status), status_desc=VALUES(status_desc), work_ord=VALUES(work_ord),
                department=VALUES(department), department_desc=VALUES(department_desc), qty_ord=VALUES(qty_ord),
                prod_line=VALUES(prod_line), applicant_name=VALUES(applicant_name), issue_user=VALUES(issue_user),
                user1=VALUES(user1), remark=VALUES(remark), create_user=VALUES(create_user),
                source_create_time=VALUES(source_create_time), pretreatment_state=VALUES(pretreatment_state),
                trans_type=VALUES(trans_type), trans_type_text=VALUES(trans_type_text),
                eff_date=VALUES(eff_date), address=VALUES(address),
                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, 'S5生产领料单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, ProductionIssueMdpSyncResult 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=0, 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("@StandardRows", result.HeadRows),
            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 ProductionIssueMdpSyncResult
{
    public long RunLogId { get; set; }
    public string BatchId { get; set; } = string.Empty;
    public int HeadRows { get; set; }
}