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

using System.Text.Json;
using Admin.NET.Plugin.AiDOP.DataPlatform;
using Admin.NET.Plugin.AiDOP.DataPlatform.Executors;
using Admin.NET.Plugin.AiDOP.Infrastructure;

namespace Admin.NET.Plugin.AiDOP.Manufacturing;

/// <summary>
/// S6 报工双模式入站：执行器 → mdp_stg_s6_report → mdp_std_s6_report。
/// 源优先 T8 <c>Cj_Bg_Head_Rep</c>（实体 S6_REPORT）；API 对偶 S6_REPORT_API。
/// </summary>
public class ReportWorkMdpSyncService : ITransient
{
    private const string InboundEntityCode = "S6_REPORT";
    /// <summary>T8 报工源无 tenant_id；须经 <see cref="AidopSourceTenantMap"/> 解析。</summary>
    private const string SourceZtid = "pbxfxp";

    private readonly ISqlSugarClient _db;
    private readonly MdpSourcePullDispatcher _pullDispatcher;

    public ReportWorkMdpSyncService(ISqlSugarClient db, MdpSourcePullDispatcher pullDispatcher)
    {
        _db = db;
        _pullDispatcher = pullDispatcher;
    }

    public async Task<ReportWorkInboundResult> RunInboundAsync(
        long tenantId = 0,
        bool fullRefresh = false,
        string? entityCode = null,
        CancellationToken cancellationToken = default)
    {
        cancellationToken.ThrowIfCancellationRequested();
        await EnsureTablesAsync();

        // T8 源表无 tenant_id 列；显式 tenantId≤0 时经账套映射解析真实租户。
        var resolvedTenantId = AidopSourceTenantMap.ResolveTenantId(SourceZtid, tenantId);

        var code = string.IsNullOrWhiteSpace(entityCode) ? InboundEntityCode : entityCode.Trim();
        var now = DateTime.Now;
        var pullCtx = new MdpPullContext
        {
            TenantId = resolvedTenantId,
            FullRefresh = fullRefresh,
            TaskCode = "S6_REPORT_INBOUND",
            BatchId = $"S6_RPT_IN_{now:yyyyMMddHHmmss}"
        };
        var pull = await _pullDispatcher.PullByEntityCodeAsync(code, pullCtx, cancellationToken);
        var transformBatch = $"{pullCtx.BatchId}_STD";
        var stdRows = await TransformStandardAsync(resolvedTenantId, transformBatch, now);
        return new ReportWorkInboundResult
        {
            PullBatchId = pullCtx.BatchId,
            RowsPulled = pull.RowsWritten,
            RowsWrittenStg = pull.RowsWritten,
            TransformBatchId = transformBatch,
            StandardRows = stdRows
        };
    }

    private async Task<int> TransformStandardAsync(long tenantId, string batchId, DateTime now)
    {
        // 将 PENDING stg 投影到 std（JSON 字段兼容大小写）
        var sTenant = MdpJsonSql.TenantFromStg("s", "@tid");
        var sql = $"""
            INSERT INTO mdp_std_s6_report
              (tenant_id, factory_id, source_system, work_order_no, report_date, report_qty, ztid,
               source_row_id, source_biz_key, sync_batch_id, sync_time)
            SELECT
              {sTenant},
              1,
              IFNULL(NULLIF(s.source_system,''), 'T8'),
              IFNULL(NULLIF(JSON_UNQUOTE(JSON_EXTRACT(s.raw_data, '$.noid')), 'null'),
                     IFNULL(NULLIF(JSON_UNQUOTE(JSON_EXTRACT(s.raw_data, '$.NOID')), 'null'), s.source_biz_key)),
              COALESCE(
                STR_TO_DATE(NULLIF(NULLIF(JSON_UNQUOTE(JSON_EXTRACT(s.raw_data, '$.kgdate')), 'null'), ''), '%Y-%m-%d %H:%i:%s'),
                STR_TO_DATE(NULLIF(NULLIF(JSON_UNQUOTE(JSON_EXTRACT(s.raw_data, '$.KGDATE')), 'null'), ''), '%Y-%m-%d %H:%i:%s'),
                NULL),
              CAST(NULLIF(NULLIF(JSON_UNQUOTE(JSON_EXTRACT(s.raw_data, '$.sl')), 'null'), '') AS DECIMAL(18,6)),
              NULLIF(JSON_UNQUOTE(JSON_EXTRACT(s.raw_data, '$.ztid')), 'null'),
              IFNULL(NULLIF(s.source_row_id,''), s.source_biz_key),
              s.source_biz_key,
              @batch,
              @now
            FROM mdp_stg_s6_report s
            WHERE s.process_status = 'PENDING'
              AND s.source_biz_key IS NOT NULL
              AND s.source_biz_key <> ''
              AND {MdpJsonSql.TenantGuard(sTenant)}
            ON DUPLICATE KEY UPDATE
              work_order_no = VALUES(work_order_no),
              report_date = VALUES(report_date),
              report_qty = VALUES(report_qty),
              ztid = VALUES(ztid),
              source_row_id = VALUES(source_row_id),
              sync_batch_id = VALUES(sync_batch_id),
              sync_time = VALUES(sync_time),
              update_time = CURRENT_TIMESTAMP
            """;
        var affected = await _db.Ado.ExecuteCommandAsync(sql,
            new SugarParameter("@tid", tenantId),
            new SugarParameter("@batch", batchId),
            new SugarParameter("@now", now));

        await _db.Ado.ExecuteCommandAsync(
            tenantId > 0
                ? "UPDATE mdp_stg_s6_report SET process_status='DONE', update_time=NOW() WHERE process_status='PENDING' AND tenant_id = @tid"
                : "UPDATE mdp_stg_s6_report SET process_status='DONE', update_time=NOW() WHERE process_status='PENDING'",
            tenantId > 0 ? new SugarParameter("@tid", tenantId) : null!);
        return affected;
    }

    private async Task EnsureTablesAsync()
    {
        await _db.Ado.ExecuteCommandAsync("""
            CREATE TABLE IF NOT EXISTS mdp_stg_s6_report (
              id bigint NOT NULL AUTO_INCREMENT,
              tenant_id bigint NOT NULL DEFAULT 0,
              source_system varchar(50) DEFAULT NULL,
              source_table varchar(200) DEFAULT NULL,
              source_row_id varchar(200) DEFAULT NULL,
              source_biz_key varchar(300) DEFAULT NULL,
              raw_data json DEFAULT NULL,
              sync_batch_id varchar(100) DEFAULT NULL,
              create_time datetime NOT NULL DEFAULT CURRENT_TIMESTAMP,
              sync_time datetime DEFAULT CURRENT_TIMESTAMP,
              process_status varchar(20) NOT NULL DEFAULT 'PENDING',
              process_message varchar(500) DEFAULT NULL,
              update_time datetime DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
              PRIMARY KEY (id),
              UNIQUE KEY uk_source_key (source_system, source_table, source_biz_key)
            ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4
            """);
        await _db.Ado.ExecuteCommandAsync("""
            CREATE TABLE IF NOT EXISTS mdp_std_s6_report (
              id bigint NOT NULL AUTO_INCREMENT,
              tenant_id bigint NOT NULL DEFAULT 0,
              factory_id bigint DEFAULT 1,
              source_system varchar(50) NOT NULL DEFAULT 'T8',
              work_order_no varchar(100) NOT NULL,
              report_date datetime DEFAULT NULL,
              report_qty decimal(18,6) DEFAULT NULL,
              ztid varchar(50) DEFAULT NULL,
              source_row_id varchar(100) NOT NULL,
              source_biz_key varchar(200) NOT NULL,
              sync_batch_id varchar(100) NOT NULL,
              sync_time datetime NOT NULL,
              update_time datetime DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
              PRIMARY KEY (id),
              UNIQUE KEY uk_mdp_std_s6_report (tenant_id, source_system, source_biz_key)
            ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4
            """);
    }
}

public class ReportWorkInboundResult
{
    public string PullBatchId { get; set; } = string.Empty;
    public int RowsPulled { get; set; }
    public int RowsWrittenStg { get; set; }
    public string TransformBatchId { get; set; } = string.Empty;
    public int StandardRows { get; set; }
}