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

namespace Admin.NET.Plugin.AiDOP.MaterialWarehouse;

/// <summary>
/// S5 生产入库单 数据中台只读同步转换服务（DOP 内部，扁平单表，独立于 S5MdpSyncTransformService 的 KPI 管线）。
///
/// 源：aidopdev NbrMaster(n) + NbrDetail(d)，业务类型 Type='WOI'（生产入库）；
///     扁平 LIST join：n.Domain=d.Domain AND n.Nbr=d.Nbr（一行=一条 NbrDetail 明细）。
///     维表 DepartmentMaster(p) / ItemMaster(i) / LocationMaster(lt,lf) 全 LEFT JOIN。
///     LocationMaster 无 Domain 列（DOP 重建：domain_code/location/descr），按 tenant_id+location join。
/// 链路：NbrMaster/NbrDetail(WOI) 只读 → mdp_std_production_receipt(typed 扁平)。
///     不经 stg、不建 dwd；不采用生产退料单 head-detail 双表逻辑；不碰 S2 写路径（S2 不写 WOI）。
///
/// 约束：
///   - 只读源表，仅写 mdp_std_production_receipt；绝不 INSERT/UPDATE/DELETE NbrMaster/NbrDetail。
///   - 过滤 n.Type='WOI' AND n.IsActive=1。
///   - status_desc 本批暂 = raw status（UPPER）；remark=NbrMaster.Remark（列表显示名“入库类型”待确认）。
///   - qty_rec=NbrDetail.QtyRec（列表用）；qty_to/location_from 保留落库、本批不展示。
///   - WOI 为 0 行时转换成功完成、处理数为 0，不报错。
/// </summary>
public class ProductionReceiptMdpSyncService : ITransient
{
    private const string JobCode = "S5_PRODUCTION_RECEIPT_MDP_SYNC";

    private readonly ISqlSugarClient _db;

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

    /// <summary>全量同步转换：源(WOI 扁平) → mdp_std_production_receipt。</summary>
    public async Task<ProductionReceiptMdpSyncResult> RunFullAsync(CancellationToken cancellationToken = default, string triggerType = "AUTO")
    {
        cancellationToken.ThrowIfCancellationRequested();
        await EnsureTablesAsync();

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

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

    /// <summary>防御式建表（与 UpdateScripts/1.0.214.sql 同构，幂等）。</summary>
    private async Task EnsureTablesAsync()
    {
        await _db.Ado.ExecuteCommandAsync(
            """
            CREATE TABLE IF NOT EXISTS mdp_std_production_receipt (
                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,
                master_rec_id INT NULL,
                detail_rec_id INT NOT NULL,
                nbr VARCHAR(24) NULL,
                line SMALLINT NOT NULL DEFAULT 0,
                receipt_date DATETIME NULL,
                status VARCHAR(8) NULL,
                status_desc VARCHAR(20) NULL,
                remark VARCHAR(200) NULL,
                prod_line VARCHAR(8) NULL,
                work_ord VARCHAR(64) NULL,
                erp_work_ord VARCHAR(60) NULL,
                department VARCHAR(8) NULL,
                department_desc VARCHAR(255) NULL,
                applicant_name VARCHAR(12) NULL,
                item_num VARCHAR(24) NULL,
                item_name VARCHAR(200) NULL,
                item_spec VARCHAR(200) NULL,
                um VARCHAR(8) NULL,
                location_to VARCHAR(8) NULL,
                location_to_desc VARCHAR(255) NULL,
                lot_serial VARCHAR(120) NULL,
                qty_rec DECIMAL(18,5) NULL DEFAULT 0,
                qty_to DECIMAL(18,5) NULL DEFAULT 0,
                location_from VARCHAR(8) NULL,
                location_from_desc VARCHAR(255) NULL,
                ord_nbr VARCHAR(48) 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_receipt (tenant_id, domain, detail_rec_id),
                KEY idx_mdp_std_prod_rcpt_nbr (tenant_id, nbr),
                KEY idx_mdp_std_prod_rcpt_date (tenant_id, receipt_date),
                KEY idx_mdp_std_prod_rcpt_workord (tenant_id, work_ord),
                KEY idx_mdp_std_prod_rcpt_erp (tenant_id, erp_work_ord),
                KEY idx_mdp_std_prod_rcpt_lot (tenant_id, lot_serial),
                KEY idx_mdp_std_prod_rcpt_item (tenant_id, item_num),
                KEY idx_mdp_std_prod_rcpt_locto (tenant_id, location_to)
            ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci COMMENT='S5生产入库单标准层（扁平，一行=一条明细）'
            """);
    }

    /// <summary>
    /// 标准化：NbrMaster(WOI) + NbrDetail + 维表 -> mdp_std_production_receipt（扁平，一行=一条明细）。返回处理行数。
    /// SQL 由旧系统扁平 LIST SQL 翻译（SQL Server -&gt; MySQL）：with(nolock) 去除、IsNull-&gt;IFNULL、
    /// rtrim(a+' '+b)-&gt;TRIM(CONCAT(...))、row_number() 不落库（分页在读 API 侧）。
    /// LocationMaster 无 Domain 列，按 tenant_id+location join（生产退料单 1R 教训）。
    /// </summary>
    private async Task<int> TransformStandardAsync(string batchId, DateTime now)
    {
        var rows = await _db.Ado.GetIntAsync(
            """
            SELECT COUNT(1)
            FROM NbrMaster n
            LEFT JOIN NbrDetail d ON n.Domain = d.Domain AND n.Nbr = d.Nbr
            WHERE n.Type='WOI' AND n.IsActive=1 AND d.RecID IS NOT NULL
            """);
        await _db.Ado.ExecuteCommandAsync(
            """
            INSERT INTO mdp_std_production_receipt
            (tenant_id, factory_id, source_system, domain, master_rec_id, detail_rec_id, nbr, line,
             receipt_date, status, status_desc, remark, prod_line, work_ord, erp_work_ord,
             department, department_desc, applicant_name, item_num, item_name, item_spec, um,
             location_to, location_to_desc, lot_serial, qty_rec, qty_to, location_from, location_from_desc,
             ord_nbr, source_biz_key, sync_batch_id, sync_time)
            SELECT
                IFNULL(n.tenant_id, 0), 1, 'AIDOP', n.Domain, n.RecID, d.RecID, n.Nbr, d.Line,
                n.Date, UPPER(IFNULL(n.Status, '')), UPPER(IFNULL(n.Status, '')), n.Remark,
                d.ProdLine, d.WorkOrd, d.Address,
                n.Department, TRIM(CONCAT(n.Department, ' ', IFNULL(p.Descr, ''))), n.Name,
                d.ItemNum, i.Descr, i.Descr1, i.UM,
                d.LocationTo, lt.descr, CAST(d.LotSerial AS CHAR), d.QtyRec, d.QtyTo,
                d.LocationFrom, lf.descr, d.OrdNbr,
                CONCAT(n.Domain, ':', d.RecID), @BatchId, @Now
            FROM NbrMaster n
            LEFT JOIN NbrDetail d ON n.Domain = d.Domain AND n.Nbr = d.Nbr
            LEFT JOIN DepartmentMaster p ON p.Domain = n.Domain AND p.Department = n.Department
            LEFT JOIN ItemMaster i ON i.Domain = d.Domain AND i.ItemNum = d.ItemNum
            LEFT JOIN LocationMaster lt ON lt.tenant_id = IFNULL(d.tenant_id, 0) AND lt.location = d.LocationTo
            LEFT JOIN LocationMaster lf ON lf.tenant_id = IFNULL(d.tenant_id, 0) AND lf.location = d.LocationFrom
            WHERE n.Type='WOI' AND n.IsActive=1 AND d.RecID IS NOT NULL
            ON DUPLICATE KEY UPDATE
                factory_id=VALUES(factory_id), master_rec_id=VALUES(master_rec_id), nbr=VALUES(nbr), line=VALUES(line),
                receipt_date=VALUES(receipt_date), status=VALUES(status), status_desc=VALUES(status_desc), remark=VALUES(remark),
                prod_line=VALUES(prod_line), work_ord=VALUES(work_ord), erp_work_ord=VALUES(erp_work_ord),
                department=VALUES(department), department_desc=VALUES(department_desc), applicant_name=VALUES(applicant_name),
                item_num=VALUES(item_num), item_name=VALUES(item_name), item_spec=VALUES(item_spec), um=VALUES(um),
                location_to=VALUES(location_to), location_to_desc=VALUES(location_to_desc), lot_serial=VALUES(lot_serial),
                qty_rec=VALUES(qty_rec), qty_to=VALUES(qty_to), location_from=VALUES(location_from),
                location_from_desc=VALUES(location_from_desc), ord_nbr=VALUES(ord_nbr),
                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, ProductionReceiptMdpSyncResult 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.Rows),
            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 ProductionReceiptMdpSyncResult
{
    public long RunLogId { get; set; }
    public string BatchId { get; set; } = string.Empty;
    public int Rows { get; set; }
}