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

using Admin.NET.Core.Service;
using Admin.NET.Plugin.AiDOP.Infrastructure;
using Microsoft.Extensions.Logging;
using System.Text.Json;

namespace Admin.NET.Plugin.AiDOP.MaterialWarehouse;

/// <summary>
/// S5 物料仓储 — KPI 计算与刷新转换服务。边界：8 迁 1 留。
/// 已中台化（读本地标准层 mdp_std_t8_*，由 T8BaseInboundMdpSyncService 从 T8 贴源→标准）：
///   S5_L1_001 物料上线周期 / S5_L1_002 物料上线满足率（依赖 Cj_Bg_Head_Rep + kc_dd_list_cllist，已贴源）/ S5_L1_003 物料仓储人效。
/// 仍 legacy 直连 T8（QueryT8Async，ConfigId=t8_v5）：
///   S5_L1_004 品类物料库存周转（依赖 T8 TVF Rep_总账_存货_V3，报表聚合结果无逐行主键，暂不中台化）。
/// 结果统一落 dwd_t8_* 与 ado_s9_kpi_value_l1_day。计算口径沿用方老师 v5.4 KPI J 列。
/// </summary>
public class S5MdpSyncTransformService : ITransient
{
    private readonly ISqlSugarClient _db;
    private readonly SysNoticeService _sysNoticeService;
    private readonly ILogger<S5MdpSyncTransformService> _logger;
    private const string JobCode = "S5_MDP_SYNC_TRANSFORM";
    private const string JobName = "S5 物料仓储 MDP 同步与转换";
    private const string T8ConfigId = "t8_v5";
    // 加固：legacy 直连 T8 查询的命令超时上限（秒）。防止 T8 TVF(Rep_总账_存货_V3) 慢或挂起时
    // 查询无限期阻塞、进而长时间持有刷新锁（见 AidopT8KpiManualRefreshService）。
    // 仅加超时护栏，不改 S5_L1_004 的 SQL / 参数 / 计算口径。（S5_L1_002 已中台化，不再走 QueryT8Async。）
    private const int T8CommandTimeoutSeconds = 60;
    private const string ModuleCode = "S5";
    // FAILURE-NOTIFICATION-1：超级管理员 superAdmin.NET（AccountType=999）
    private const long NoticeReceiverUserId = 1300000000101L;
    private const string NoticeReceiverUserName = "超级管理员";

    private readonly SmartOps.KpiCalcDispatcher _kpiCalcDispatcher;
    private readonly SmartOps.KpiDimensionRunService _dimensionRun;

    public S5MdpSyncTransformService(
        ISqlSugarClient db,
        SysNoticeService sysNoticeService,
        ILogger<S5MdpSyncTransformService> logger,
        SmartOps.KpiCalcDispatcher kpiCalcDispatcher,
        SmartOps.KpiDimensionRunService dimensionRun)
    {
        _db = db;
        _sysNoticeService = sysNoticeService;
        _logger = logger;
        _kpiCalcDispatcher = kpiCalcDispatcher;
        _dimensionRun = dimensionRun;
    }

    public async Task<S5MdpSyncTransformResult> RunFullAsync(
        CancellationToken cancellationToken = default,
        string triggerType = "AUTO",
        S5MdpRefreshOption? option = null)
    {
        cancellationToken.ThrowIfCancellationRequested();
        option ??= S5MdpRefreshOption.Default();
        NormalizeOption(option);

        var now = DateTime.Now;
        var batchId = $"S5_MDP_FULL_{now:yyyyMMddHHmmss}";
        var normalizedTrigger = NormalizeTriggerType(triggerType);
        var runLogId = await InsertTransformRunLogAsync(batchId, now, normalizedTrigger, option);

        var result = new S5MdpSyncTransformResult
        {
            BatchId = batchId,
            RunLogId = runLogId,
            TriggerType = normalizedTrigger,
            SourceZtid = option.SourceZtid,
            TargetTenantId = option.TargetTenantId,
            TargetFactoryId = option.TargetFactoryId,
            BizDate = option.BizDate,
            BizMonth = option.BizMonth,
            DailyPeriodStart = option.DailyPeriodStart,
            DailyPeriodEnd = option.DailyPeriodEnd,
            MonthlyPeriodStart = option.MonthlyPeriodStart,
            MonthlyPeriodEnd = option.MonthlyPeriodEnd
        };

        try
        {
            // 路径 A：直发 T8 SQL，不做 stg / std 中间层
            result.StageRows = 0;
            result.StandardRows = 0;

            var sub16 = await BuildS5L1001MaterialOnlineCycleAsync(batchId, now, option, normalizedTrigger, cancellationToken);
            result.MergeSub("S5_L1_001", sub16);

            var sub17 = await BuildS5L1002MaterialOnlineFulfillmentAsync(batchId, now, option, normalizedTrigger, cancellationToken);
            result.MergeSub("S5_L1_002", sub17);

            var sub18 = await BuildS5L1003MaterialWarehouseEfficiencyAsync(batchId, now, option, normalizedTrigger, cancellationToken);
            result.MergeSub("S5_L1_003", sub18);

            var sub19 = await BuildS5L1004MaterialInventoryTurnoverAsync(batchId, now, option, normalizedTrigger, cancellationToken);
            result.MergeSub("S5_L1_004", sub19);

            await MarkTransformRunSuccessAsync(runLogId, now, result);
            return result;
        }
        catch (Exception ex)
        {
            await MarkTransformRunFailedAsync(runLogId, now, ex.Message, batchId);
            throw;
        }
    }

    // ─────────────────────────────────────────────────────────────────────────
    // KPI 实现（方老师 v5.4 KPI J 列 SQL 原逻辑直发 T8）
    // ─────────────────────────────────────────────────────────────────────────

    /// <summary>S5_L1_001 物料上线周期 = 配送到产线日期(lbs=生产领料) - 收货日期(lbs=采购入库)。
    /// 数据准备(写 dwd 明细)始终执行；最终 KPI 聚合由计算配置分发器接管(LEGACY_CODE/CONFIG_SQL)。</summary>
    private async Task<KpiBuildSubResult> BuildS5L1001MaterialOnlineCycleAsync(
        string batchId, DateTime now, S5MdpRefreshOption option, string triggerType, CancellationToken ct)
    {
        var sub = new KpiBuildSubResult();

        // 双模式：读本地标准层 mdp_std_t8_*（源 identity Id→src_id），语义等价于原直发 T8 SQL。
        const string sqlOnline = @"
select b.code as code, min(a.shtime) as shtime
 from mdp_std_t8_kc_tz_head a
 inner join mdp_std_t8_kc_tz_list b on a.src_id=b.idid
 where a.ztid=@ztid and a.lbs='生产领料' and a.hzyn=0 and a.zfyn=0 and a.shyn=1
 group by b.code";
        const string sqlReceipt = @"
select b.code as code, min(a.shtime) as shtime
 from mdp_std_t8_kc_tz_head a
 inner join mdp_std_t8_kc_tz_list b on a.src_id=b.idid
 where a.ztid=@ztid and a.lbs='采购入库' and a.hzyn=0 and a.zfyn=0 and a.shyn=1
 group by b.code";

        var p = new[] { new SugarParameter("@ztid", option.SourceZtid) };
        var onlineRows = await _db.Ado.SqlQueryAsync<S5OnlineCycleRow>(sqlOnline, p);
        var receiptRows = await _db.Ado.SqlQueryAsync<S5OnlineCycleRow>(sqlReceipt, p);
        sub.T8Rows = onlineRows.Count + receiptRows.Count;

        var onlineByCode = onlineRows.Where(r => !string.IsNullOrEmpty(r.code))
            .ToDictionary(r => r.code!, r => r.shtime, StringComparer.OrdinalIgnoreCase);
        var receiptByCode = receiptRows.Where(r => !string.IsNullOrEmpty(r.code))
            .ToDictionary(r => r.code!, r => r.shtime, StringComparer.OrdinalIgnoreCase);

        var allCodes = new HashSet<string>(onlineByCode.Keys, StringComparer.OrdinalIgnoreCase);
        allCodes.UnionWith(receiptByCode.Keys);

        var dwdAffected = 0;
        var cycleDaysList = new List<int>();
        foreach (var code in allCodes)
        {
            ct.ThrowIfCancellationRequested();
            var online = onlineByCode.GetValueOrDefault(code);
            var receipt = receiptByCode.GetValueOrDefault(code);
            int? cycleDays = null;
            if (online.HasValue && receipt.HasValue)
            {
                cycleDays = (int)(online.Value.Date - receipt.Value.Date).TotalDays;
                cycleDaysList.Add(cycleDays.Value);
            }

            dwdAffected += await _db.Ado.ExecuteCommandAsync(@"
INSERT INTO dwd_t8_material_online_cycle
  (tenant_id, factory_id, biz_date, source_ztid, item_code, online_date, receipt_date, cycle_days, batch_id, create_time)
VALUES
  (@tenantId, @factoryId, @bizDate, @ztid, @itemCode, @online, @receipt, @cycleDays, @batchId, @now)
ON DUPLICATE KEY UPDATE
  online_date=VALUES(online_date), receipt_date=VALUES(receipt_date),
  cycle_days=VALUES(cycle_days), batch_id=VALUES(batch_id), update_time=@now",
                new SugarParameter("@tenantId", option.TargetTenantId),
                new SugarParameter("@factoryId", option.TargetFactoryId),
                new SugarParameter("@bizDate", option.BizDate),
                new SugarParameter("@ztid", option.SourceZtid),
                new SugarParameter("@itemCode", code),
                new SugarParameter("@online", online),
                new SugarParameter("@receipt", receipt),
                new SugarParameter("@cycleDays", cycleDays),
                new SugarParameter("@batchId", batchId),
                new SugarParameter("@now", now));
        }
        sub.DwdRows = dwdAffected;

        // 数据准备(dwd 明细)已完成。最终 KPI 聚合交计算配置分发器：
        //   无配置/LEGACY_CODE → 用下面 legacy 均值；CONFIG_SQL → 执行已发布只读 SQL；
        //   CONFIG_SQL 失败不 fallback、不写值、保留上一成功值（ShouldUpsert=false）。
        decimal? legacyValue = cycleDaysList.Count > 0 ? (decimal)cycleDaysList.Average() : null;
        var legacyDenom = cycleDaysList.Count > 0 ? "OK" : "NO_NUMERATOR";
        var dispatch = await _kpiCalcDispatcher.DispatchAsync(
            "S5_L1_001", ModuleCode, option.TargetTenantId, option.TargetFactoryId,
            option.BizDate, option.DailyPeriodStart, option.DailyPeriodEnd, option.SourceZtid,
            batchId, triggerType, legacyValue, legacyDenom, ct);
        sub.KpiRows = dispatch.ShouldUpsert
            ? await UpsertKpiValueAsync("S5_L1_001", option.BizDate, dispatch.MetricValue, now, option)
            : 0;
        sub.DenominatorStatus = dispatch.DenominatorStatus;

        // 调度：SUMMARY 成功/NO_DATA 后触发对应 DIMENSION 跑批（共享 BatchId；SUMMARY FAILED 不触发）。
        if (dispatch.ShouldUpsert)
        {
            try
            {
                await _dimensionRun.RunDimensionAsync(
                    "S5_L1_001", ModuleCode, option.TargetTenantId, option.BizDate, batchId, triggerType, ct);
            }
            catch (Exception ex)
            {
                _logger.LogWarning(ex, "S5_L1_001 维度跑批异常（不影响汇总链路）");
            }
        }
        return sub;
    }

    /// <summary>S5_L1_002 物料上线满足率 = 开工日期前完成上线行数 / 工单物料总行数。</summary>
    private async Task<KpiBuildSubResult> BuildS5L1002MaterialOnlineFulfillmentAsync(
        string batchId, DateTime now, S5MdpRefreshOption option, string triggerType, CancellationToken ct)
    {
        var sub = new KpiBuildSubResult();

        // 中台化：读本地标准层 mdp_std_t8_*（kc_tz_head.Id→src_id、Cj_Bg_Head_Rep/kc_dd_list_cllist 已贴源），
        // 口径与原直发 T8 SQL 逐项等价（分子/分母/JOIN/去重/shtime<=kgdate/count 均不变），不再直连 t8_v5。
        const string sqlNumer = @"
select lynoid as lynoid, count(*) as codenum
 from (
   select h.lynoid as lynoid, l.code as code
    from mdp_std_t8_kc_tz_head h
    inner join mdp_std_t8_kc_tz_list l on h.src_id=l.idid
    left join (
      select noid as noid, min(kgdate) as kgdate
       from mdp_std_t8_cj_bg_head_rep
       where ztid=@ztid group by noid
    ) c on h.lynoid=c.noid
    where h.ztid=@ztid and h.lbs='生产领料' and h.hzyn=0 and h.zfyn=0 and h.shyn=1 and h.shtime<=c.kgdate
    group by h.lynoid, l.code
 ) n
 group by lynoid";
        const string sqlDenom = @"
select h.noid as noid, count(l.src_id) as listnum
 from mdp_std_t8_kc_dd_head h
 left join mdp_std_t8_kc_dd_list_cllist l on h.src_id=l.idid
 where h.ztid=@ztid and h.lbs='生产任务' and h.zf=0 and h.shyn=1
 group by h.noid";

        var p = new[] { new SugarParameter("@ztid", option.SourceZtid) };
        var numerRows = await _db.Ado.SqlQueryAsync<S5FulfillmentNumerRow>(sqlNumer, p);
        var denomRows = await _db.Ado.SqlQueryAsync<S5FulfillmentDenomRow>(sqlDenom, p);
        sub.T8Rows = numerRows.Count + denomRows.Count;

        var numerByOrder = numerRows.Where(r => !string.IsNullOrEmpty(r.lynoid))
            .ToDictionary(r => r.lynoid!, r => r.codenum, StringComparer.OrdinalIgnoreCase);

        var dwdAffected = 0;
        var rateList = new List<decimal>();
        foreach (var d in denomRows)
        {
            ct.ThrowIfCancellationRequested();
            if (string.IsNullOrEmpty(d.noid)) continue;
            var beforeKg = numerByOrder.GetValueOrDefault(d.noid, 0);
            decimal? rate = d.listnum > 0
                ? Math.Round((decimal)beforeKg / d.listnum, 4)
                : null; // 分母为 0 时不伪装真实 0
            if (rate.HasValue) rateList.Add(rate.Value);

            dwdAffected += await _db.Ado.ExecuteCommandAsync(@"
INSERT INTO dwd_t8_material_online_fulfillment
  (tenant_id, factory_id, biz_date, source_ztid, work_order_no,
   before_kgdate_rows, total_rows, fulfillment_rate, batch_id, create_time)
VALUES
  (@tenantId, @factoryId, @bizDate, @ztid, @workOrderNo, @beforeKg, @total, @rate, @batchId, @now)
ON DUPLICATE KEY UPDATE
  before_kgdate_rows=VALUES(before_kgdate_rows),
  total_rows=VALUES(total_rows),
  fulfillment_rate=VALUES(fulfillment_rate),
  batch_id=VALUES(batch_id), update_time=@now",
                new SugarParameter("@tenantId", option.TargetTenantId),
                new SugarParameter("@factoryId", option.TargetFactoryId),
                new SugarParameter("@bizDate", option.BizDate),
                new SugarParameter("@ztid", option.SourceZtid),
                new SugarParameter("@workOrderNo", d.noid),
                new SugarParameter("@beforeKg", beforeKg),
                new SugarParameter("@total", d.listnum),
                new SugarParameter("@rate", rate),
                new SugarParameter("@batchId", batchId),
                new SugarParameter("@now", now));
        }
        sub.DwdRows = dwdAffected;

        // 数据准备(dwd 逐单明细)已完成。最终 KPI 聚合交计算配置分发器：
        //   无配置/LEGACY_CODE → 用下面 legacy 均值-of-比率×100；CONFIG_SQL → 执行已发布只读 SQL；
        //   CONFIG_SQL 失败不 fallback、不写值、保留上一成功值（ShouldUpsert=false）。
        decimal? legacyValue = rateList.Count > 0
            ? Math.Round(rateList.Average() * 100m, 4) // 百分号
            : null;
        var legacyDenom = rateList.Count > 0 ? "OK" : "NO_VALID_ORDER";
        var dispatch = await _kpiCalcDispatcher.DispatchAsync(
            "S5_L1_002", ModuleCode, option.TargetTenantId, option.TargetFactoryId,
            option.BizDate, option.DailyPeriodStart, option.DailyPeriodEnd, option.SourceZtid,
            batchId, triggerType, legacyValue, legacyDenom, ct);
        sub.KpiRows = dispatch.ShouldUpsert
            ? await UpsertKpiValueAsync("S5_L1_002", option.BizDate, dispatch.MetricValue, now, option)
            : 0;
        sub.DenominatorStatus = dispatch.DenominatorStatus;

        // 调度：SUMMARY 成功/NO_DATA 后触发对应 DIMENSION 跑批（共享 BatchId；SUMMARY FAILED 不触发）。
        if (dispatch.ShouldUpsert)
        {
            try
            {
                await _dimensionRun.RunDimensionAsync(
                    "S5_L1_002", ModuleCode, option.TargetTenantId, option.BizDate, batchId, triggerType, ct);
            }
            catch (Exception ex)
            {
                _logger.LogWarning(ex, "S5_L1_002 维度跑批异常（不影响汇总链路）");
            }
        }
        return sub;
    }

    /// <summary>S5_L1_003 物料仓储人效 = SUM(slzx where lbs=生产领料) / count(gw=仓管)。</summary>
    private async Task<KpiBuildSubResult> BuildS5L1003MaterialWarehouseEfficiencyAsync(
        string batchId, DateTime now, S5MdpRefreshOption option, string triggerType, CancellationToken ct)
    {
        var sub = new KpiBuildSubResult();

        const string sqlNumer = @"
select sum(b.slzx) as slzx
 from mdp_std_t8_kc_tz_head a
 inner join mdp_std_t8_kc_tz_list b on a.src_id=b.idid
 where a.ztid=@ztid and a.lbs='生产领料' and a.hzyn=0 and a.zfyn=0 and a.shyn=1
  and a.shtime between @startDate and @endDate";
        const string sqlDenom = @"
select count(*) as penum
 from mdp_std_t8_sys_pelist
 where ztid=@ztid and zzzt='在职' and gw='仓管'";

        var pNumer = new[]
        {
            new SugarParameter("@ztid", option.SourceZtid),
            new SugarParameter("@startDate", option.MonthlyPeriodStart),
            new SugarParameter("@endDate", option.MonthlyPeriodEnd)
        };
        var pDenom = new[] { new SugarParameter("@ztid", option.SourceZtid) };

        var numerRows = await _db.Ado.SqlQueryAsync<S5SumQtyRow>(sqlNumer, pNumer);
        var denomRows = await _db.Ado.SqlQueryAsync<S5CountRow>(sqlDenom, pDenom);
        sub.T8Rows = numerRows.Count + denomRows.Count;

        decimal? onlineQty = numerRows.FirstOrDefault()?.slzx;
        int? headcount = denomRows.FirstOrDefault()?.penum;

        // 分母 = 0 或 NULL：efficiency 写 NULL，并标记 denominator_status；不伪装真实 0
        decimal? efficiency = null;
        string denomStatus;
        if (!headcount.HasValue || headcount.Value <= 0)
        {
            denomStatus = "NO_HEADCOUNT";
        }
        else if (!onlineQty.HasValue)
        {
            denomStatus = "NO_NUMERATOR";
        }
        else
        {
            efficiency = Math.Round(onlineQty.Value / headcount.Value, 4);
            denomStatus = "OK";
        }
        sub.DenominatorStatus = denomStatus;

        // 月度 KPI 用 biz_month 唯一键，整月 1 行
        var dwdAffected = await _db.Ado.ExecuteCommandAsync(@"
INSERT INTO dwd_t8_material_warehouse_efficiency
  (tenant_id, factory_id, biz_month, source_ztid, period_start, period_end,
   online_qty, warehouse_headcount, efficiency, denominator_status, batch_id, create_time)
VALUES
  (@tenantId, @factoryId, @bizMonth, @ztid, @periodStart, @periodEnd,
   @onlineQty, @headcount, @efficiency, @denomStatus, @batchId, @now)
ON DUPLICATE KEY UPDATE
  period_start=VALUES(period_start), period_end=VALUES(period_end),
  online_qty=VALUES(online_qty), warehouse_headcount=VALUES(warehouse_headcount),
  efficiency=VALUES(efficiency), denominator_status=VALUES(denominator_status),
  batch_id=VALUES(batch_id), update_time=@now",
            new SugarParameter("@tenantId", option.TargetTenantId),
            new SugarParameter("@factoryId", option.TargetFactoryId),
            new SugarParameter("@bizMonth", option.BizMonth),
            new SugarParameter("@ztid", option.SourceZtid),
            new SugarParameter("@periodStart", option.MonthlyPeriodStart),
            new SugarParameter("@periodEnd", option.MonthlyPeriodEnd),
            new SugarParameter("@onlineQty", onlineQty),
            new SugarParameter("@headcount", headcount),
            new SugarParameter("@efficiency", efficiency),
            new SugarParameter("@denomStatus", denomStatus),
            new SugarParameter("@batchId", batchId),
            new SugarParameter("@now", now));
        sub.DwdRows = dwdAffected;

        // 月度 KPI 最终聚合交分发器；bizDate=月末、period=当月窗口；
        //   legacyValue=efficiency、legacyDenom 保留 NO_HEADCOUNT/NO_NUMERATOR（CONFIG_SQL 下塌缩为 NO_DATA）。
        var dispatch = await _kpiCalcDispatcher.DispatchAsync(
            "S5_L1_003", ModuleCode, option.TargetTenantId, option.TargetFactoryId,
            option.MonthlyPeriodEnd, option.MonthlyPeriodStart, option.MonthlyPeriodEnd, option.SourceZtid,
            batchId, triggerType, efficiency, denomStatus, ct);
        sub.KpiRows = dispatch.ShouldUpsert
            ? await UpsertKpiValueAsync("S5_L1_003", option.MonthlyPeriodEnd, dispatch.MetricValue, now, option)
            : 0;
        sub.DenominatorStatus = dispatch.DenominatorStatus;

        // 调度：SUMMARY 成功/NO_DATA 后触发人效月度 DIMENSION 跑批（月度：@biz_date=月末派生 biz_month，与 SUMMARY 同月）。
        if (dispatch.ShouldUpsert)
        {
            try
            {
                await _dimensionRun.RunDimensionAsync(
                    "S5_L1_003", ModuleCode, option.TargetTenantId, option.MonthlyPeriodEnd, batchId, triggerType, ct);
            }
            catch (Exception ex)
            {
                _logger.LogWarning(ex, "S5_L1_003 维度跑批异常（不影响汇总链路）");
            }
        }
        return sub;
    }

    /// <summary>S5_L1_004 品类物料库存周转 = D1/D2 × 30；D1=je3 月均库存金额，D2=je2 出库成本。</summary>
    private async Task<KpiBuildSubResult> BuildS5L1004MaterialInventoryTurnoverAsync(
        string batchId, DateTime now, S5MdpRefreshOption option, string triggerType, CancellationToken ct)
    {
        var sub = new KpiBuildSubResult();

        // TVF：Rep_总账_存货_V3(账套, '普通', '正常', 起期 YYYYMM, 止期 YYYYMM)
        const string sqlTvf = @"
select ckcode as ckcode, ckname as ckname,
       code as code, cname as cname,
       pcode as pcode, pname as pname,
       je3 as je3, je2 as je2
 from dbo.Rep_总账_存货_V3(@ztid, N'普通', N'正常', @startYm, @endYm)";

        var p = new[]
        {
            new SugarParameter("@ztid", option.SourceZtid),
            new SugarParameter("@startYm", option.TvfPeriodStartYyyymm),
            new SugarParameter("@endYm", option.TvfPeriodEndYyyymm)
        };
        var tvfRows = await QueryT8Async<S5InventoryTurnoverRow>(sqlTvf, p);
        sub.T8Rows = tvfRows.Count;

        var dwdAffected = 0;
        var turnoverDaysList = new List<decimal>();
        foreach (var r in tvfRows)
        {
            ct.ThrowIfCancellationRequested();
            // 周转天数：D2=0 或 NULL 时 NULL，不伪装 0
            decimal? turnoverDays = (r.je2.HasValue && r.je2.Value > 0m && r.je3.HasValue)
                ? Math.Round(r.je3.Value / r.je2.Value * 30m, 4)
                : null;
            if (turnoverDays.HasValue) turnoverDaysList.Add(turnoverDays.Value);

            dwdAffected += await _db.Ado.ExecuteCommandAsync(@"
INSERT INTO dwd_t8_material_inventory_turnover
  (tenant_id, factory_id, biz_month, source_ztid, period_start_yyyymm, period_end_yyyymm,
   warehouse_code, warehouse_name, item_code, item_name, category_code, category_name,
   avg_inventory_value, monthly_outbound_cost, turnover_days, batch_id, create_time)
VALUES
  (@tenantId, @factoryId, @bizMonth, @ztid, @startYm, @endYm,
   @ckcode, @ckname, @itemCode, @itemName, @pcode, @pname,
   @je3, @je2, @turnoverDays, @batchId, @now)
ON DUPLICATE KEY UPDATE
  warehouse_name=VALUES(warehouse_name), item_name=VALUES(item_name),
  category_code=VALUES(category_code), category_name=VALUES(category_name),
  avg_inventory_value=VALUES(avg_inventory_value),
  monthly_outbound_cost=VALUES(monthly_outbound_cost),
  turnover_days=VALUES(turnover_days),
  period_start_yyyymm=VALUES(period_start_yyyymm),
  period_end_yyyymm=VALUES(period_end_yyyymm),
  batch_id=VALUES(batch_id), update_time=@now",
                new SugarParameter("@tenantId", option.TargetTenantId),
                new SugarParameter("@factoryId", option.TargetFactoryId),
                new SugarParameter("@bizMonth", option.BizMonth),
                new SugarParameter("@ztid", option.SourceZtid),
                new SugarParameter("@startYm", option.TvfPeriodStartYyyymm),
                new SugarParameter("@endYm", option.TvfPeriodEndYyyymm),
                new SugarParameter("@ckcode", r.ckcode ?? ""),
                new SugarParameter("@ckname", r.ckname),
                new SugarParameter("@itemCode", r.code ?? ""),
                new SugarParameter("@itemName", r.cname),
                new SugarParameter("@pcode", r.pcode),
                new SugarParameter("@pname", r.pname),
                new SugarParameter("@je3", r.je3),
                new SugarParameter("@je2", r.je2),
                new SugarParameter("@turnoverDays", turnoverDays),
                new SugarParameter("@batchId", batchId),
                new SugarParameter("@now", now));
        }
        sub.DwdRows = dwdAffected;

        // KPI 值：所有品类周转天数算术平均；无任一可计算品类时 NULL
        decimal? metricValue = turnoverDaysList.Count > 0
            ? Math.Round(turnoverDaysList.Average(), 4)
            : null;
        // S5_L1_004 保留 LEGACY_TVF：分发器 LEGACY_TVF 分支直接回传上面的 TVF 均值，
        //   不经 KpiSqlReadOnlyExecutor；仅统一 run-log 记录引擎状态（配置登记为 LEGACY_TVF）。
        //   TVF(Rep_总账_存货_V3)/参数/口径/CommandTimeout 全不变，不建 CONFIG_SQL 版本。
        var legacyDenom = turnoverDaysList.Count > 0 ? "OK" : "NO_VALID_OUTBOUND_COST";
        var dispatch = await _kpiCalcDispatcher.DispatchAsync(
            "S5_L1_004", ModuleCode, option.TargetTenantId, option.TargetFactoryId,
            option.MonthlyPeriodEnd, option.MonthlyPeriodStart, option.MonthlyPeriodEnd, option.SourceZtid,
            batchId, triggerType, metricValue, legacyDenom, ct);
        sub.KpiRows = dispatch.ShouldUpsert
            ? await UpsertKpiValueAsync("S5_L1_004", option.MonthlyPeriodEnd, dispatch.MetricValue, now, option)
            : 0;
        sub.DenominatorStatus = dispatch.DenominatorStatus;

        // 调度：SUMMARY 成功/NO_DATA 后触发对应 DIMENSION 跑批（月度：DIMENSION_SQL 用 @biz_date 派生 biz_month，与 SUMMARY 同月）。
        if (dispatch.ShouldUpsert)
        {
            try
            {
                await _dimensionRun.RunDimensionAsync(
                    "S5_L1_004", ModuleCode, option.TargetTenantId, option.MonthlyPeriodEnd, batchId, triggerType, ct);
            }
            catch (Exception ex)
            {
                _logger.LogWarning(ex, "S5_L1_004 维度跑批异常（不影响汇总链路）");
            }
        }
        return sub;
    }

    // ─────────────────────────────────────────────────────────────────────────
    // 跨库 / 写入 / 日志 封装
    // ─────────────────────────────────────────────────────────────────────────

    // legacy 直连 T8：仅 S5_L1_004（Rep_总账_存货_V3 TVF，报表聚合结果无逐行主键）仍用；
    // 已中台化的 S5_L1_001/002/003 均读 mdp_std_t8_*，不再走此方法。
    private async Task<List<T>> QueryT8Async<T>(string sql, SugarParameter[] parameters)
    {
        var t8 = _db.AsTenant().GetConnectionScope(T8ConfigId);
        t8.Ado.CommandTimeOut = T8CommandTimeoutSeconds;
        return await t8.Ado.SqlQueryAsync<T>(sql, parameters);
    }

    private async Task<int> UpsertKpiValueAsync(string metricCode, DateTime bizDate, decimal? metricValue, DateTime now, S5MdpRefreshOption option)
    {
        // 沿用 S3 UpsertS3KpiValueAsync 范式：先查现存行 → UPDATE；不存在 → SELECT MAX(id)+1 显式生成 id 后 INSERT。
        // ado_s9_kpi_value_l1_day.id 为手工分配主键（无 AUTO_INCREMENT），必须显式 set；
        // metric_value 允许 NULL（分母缺失不得伪装真实 0）。
        // FIX-2：截断时分秒（月度 KPI 入参可能为 YYYY-MM-DD 23:59:59），保证 SELECT WHERE biz_date=@BizDate 与 DB date 列匹配，避免重复 INSERT。
        // FIX-1：tenant_id/factory_id 取自 option，默认仍为 1300000000001/1，不破坏 Demo。
        bizDate = bizDate.Date;
        var existingId = await _db.Ado.GetLongAsync(
            "SELECT IFNULL((SELECT id FROM ado_s9_kpi_value_l1_day WHERE tenant_id=@TenantId AND factory_id=@FactoryId " +
            "AND module_code=@ModuleCode AND metric_code=@MetricCode AND biz_date=@BizDate AND is_deleted=0 " +
            "ORDER BY id LIMIT 1), 0)",
            new List<SugarParameter>
            {
                new("@TenantId", option.TargetTenantId),
                new("@FactoryId", option.TargetFactoryId),
                new("@ModuleCode", ModuleCode),
                new("@MetricCode", metricCode),
                new("@BizDate", bizDate)
            });

        if (existingId > 0)
        {
            return await _db.Ado.ExecuteCommandAsync(
                "UPDATE ado_s9_kpi_value_l1_day SET metric_value=@MetricValue, calc_time=@Now, " +
                "update_time=@Now, is_deleted=0, is_active=1 WHERE id=@Id",
                new SugarParameter("@MetricValue", metricValue),
                new SugarParameter("@Now", now),
                new SugarParameter("@Id", existingId));
        }

        var nextId = await _db.Ado.GetLongAsync(
            "SELECT COALESCE(MAX(id), 0) + 1 FROM ado_s9_kpi_value_l1_day");
        return await _db.Ado.ExecuteCommandAsync(@"
INSERT INTO ado_s9_kpi_value_l1_day
  (id, tenant_id, org_id, company_id, factory_id, status, biz_date,
   create_time, update_time, is_deleted, is_active,
   module_code, metric_code, metric_value, calc_time)
VALUES
  (@Id, @TenantId, NULL, NULL, @FactoryId, NULL, @BizDate,
   @Now, @Now, 0, 1,
   @ModuleCode, @MetricCode, @MetricValue, @Now)",
            new SugarParameter("@Id", nextId),
            new SugarParameter("@TenantId", option.TargetTenantId),
            new SugarParameter("@FactoryId", option.TargetFactoryId),
            new SugarParameter("@BizDate", bizDate),
            new SugarParameter("@Now", now),
            new SugarParameter("@ModuleCode", ModuleCode),
            new SugarParameter("@MetricCode", metricCode),
            new SugarParameter("@MetricValue", metricValue));
    }

    private async Task<long> InsertTransformRunLogAsync(string batchId, DateTime startedAt, string triggerType, S5MdpRefreshOption option)
    {
        await _db.Ado.ExecuteCommandAsync(@"
INSERT INTO mdp_transform_run_log
  (tenant_id, job_code, job_name, trigger_type, batch_id, status, start_time, stage_rows, standard_rows, dwd_rows, create_time, update_time)
VALUES
  (@TenantId, @JobCode, @JobName, @TriggerType, @BatchId, 'RUNNING', @StartTime, 0, 0, 0, @StartTime, @StartTime)",
            new SugarParameter("@TenantId", option.TargetTenantId),
            new SugarParameter("@JobCode", JobCode),
            new SugarParameter("@JobName", JobName),
            new SugarParameter("@TriggerType", 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 MarkTransformRunSuccessAsync(long runLogId, DateTime startedAt, S5MdpSyncTransformResult 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=@DwdRows,
    summary_json=@SummaryJson, update_time=CURRENT_TIMESTAMP
WHERE id=@Id",
            new SugarParameter("@EndTime", finishedAt),
            new SugarParameter("@DurationMs", (int)(finishedAt - startedAt).TotalMilliseconds),
            new SugarParameter("@StageRows", result.StageRows),
            new SugarParameter("@StandardRows", result.StandardRows),
            new SugarParameter("@DwdRows", result.DwdRows),
            new SugarParameter("@SummaryJson", BuildRunSummaryJson(result)),
            new SugarParameter("@Id", runLogId));
    }

    private async Task MarkTransformRunFailedAsync(long runLogId, DateTime startedAt, string message, string batchId)
    {
        bool runLogUpdated = false;
        try
        {
            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", Truncate(message, 2000)),
                new SugarParameter("@Id", runLogId));
            runLogUpdated = true;
        }
        catch (Exception ex)
        {
            // 写库本身失败兜底：远端 MySQL 瞬断导致 MarkFailed 自身也连不上
            Console.Error.WriteLine($"[S5MdpSyncTransform] MarkTransformRunFailed write failed (runLogId={runLogId}): {ex.Message}");
        }

        // FAILURE-NOTIFICATION-1：写库 FAILED 成功后发通知给超级管理员；通知失败不影响主流程
        if (!runLogUpdated) return;
        try
        {
            await _sysNoticeService.AddNotice(new AddNoticeInput
            {
                Title = "S5 物料仓储 T8 KPI 跑批失败",
                Content = $"模块：S5 物料仓储\n批次ID：{batchId}\n失败时间：{DateTime.Now:yyyy-MM-dd HH:mm:ss}\n错误信息：{Truncate(message, 1000)}\n\n请查看 mdp_transform_run_log 获取完整错误与重试记录。",
                Type = NoticeTypeEnum.NOTICE,
                PublicTime = DateTime.Now,
                Status = NoticeStatusEnum.PUBLIC,
                PublicUserId = NoticeReceiverUserId,
                PublicUserName = NoticeReceiverUserName
            });
        }
        catch (Exception notifyEx)
        {
            _logger.LogError(notifyEx, "[S5MdpSyncTransform] SysNotice 发送失败 (runLogId={RunLogId}, batchId={BatchId})", runLogId, batchId);
        }
    }

    private static string BuildRunSummaryJson(S5MdpSyncTransformResult r)
    {
        var summary = new
        {
            batchId = r.BatchId,
            sourceZtid = r.SourceZtid,
            bizDate = r.BizDate.ToString("yyyy-MM-dd"),
            bizMonth = r.BizMonth,
            triggerType = r.TriggerType,
            dwdRows = r.DwdRows,
            kpiRows = r.KpiRows,
            perKpiDwdRows = r.PerKpiDwdRows,
            perKpiKpiRows = r.PerKpiKpiRows,
            denominatorStatus = r.KpiDenominatorStatus,
            tvfPeriod = $"{r.MonthlyPeriodStart:yyyy-MM-dd}~{r.MonthlyPeriodEnd:yyyy-MM-dd}"
        };
        return JsonSerializer.Serialize(summary);
    }

    private static string NormalizeTriggerType(string s) =>
        string.IsNullOrWhiteSpace(s) ? "AUTO" : s.Trim().ToUpperInvariant();

    private static void NormalizeOption(S5MdpRefreshOption option)
    {
        var d = S5MdpRefreshOption.Default();
        if (option.TargetFactoryId <= 0) option.TargetFactoryId = d.TargetFactoryId;
        if (string.IsNullOrWhiteSpace(option.SourceZtid)) option.SourceZtid = d.SourceZtid;
        // 目标租户由 T8 账套(ztid)映射决定，禁止固定默认/兜底
        option.TargetTenantId = AidopSourceTenantMap.ResolveTenantId(option.SourceZtid, option.TargetTenantId);
        if (option.BizDate == default) option.BizDate = d.BizDate;
        if (string.IsNullOrWhiteSpace(option.BizMonth)) option.BizMonth = d.BizMonth;
        if (option.DailyPeriodStart == default) option.DailyPeriodStart = d.DailyPeriodStart;
        if (option.DailyPeriodEnd == default) option.DailyPeriodEnd = d.DailyPeriodEnd;
        if (option.MonthlyPeriodStart == default) option.MonthlyPeriodStart = d.MonthlyPeriodStart;
        if (option.MonthlyPeriodEnd == default) option.MonthlyPeriodEnd = d.MonthlyPeriodEnd;
        if (string.IsNullOrWhiteSpace(option.TvfPeriodStartYyyymm)) option.TvfPeriodStartYyyymm = d.TvfPeriodStartYyyymm;
        if (string.IsNullOrWhiteSpace(option.TvfPeriodEndYyyymm)) option.TvfPeriodEndYyyymm = d.TvfPeriodEndYyyymm;
    }

    private static string Truncate(string s, int max) =>
        string.IsNullOrEmpty(s) ? "" : (s.Length <= max ? s : s.Substring(0, max));
}

// ─────────────────────────────────────────────────────────────────────────────
// Refresh 入参与结果 DTO
// ─────────────────────────────────────────────────────────────────────────────

public sealed class S5MdpRefreshOption
{
    /// <summary>T8 账套（kc_tz_head.ztid）；实测当前唯一账套为 pbxfxp。</summary>
    public string SourceZtid { get; set; } = "pbxfxp";
    /// <summary>KPI/DWD 落库目标租户；≤0 时由 <see cref="AidopSourceTenantMap"/> 按 SourceZtid 解析。</summary>
    public long TargetTenantId { get; set; }
    /// <summary>KPI/DWD 落库目标工厂；默认 1。</summary>
    public long TargetFactoryId { get; set; } = 1L;
    /// <summary>日 T+1 KPI 的业务日期（默认昨天）。</summary>
    public DateTime BizDate { get; set; }
    /// <summary>月 M+1 KPI 的业务月 YYYY-MM（默认上月）。</summary>
    public string BizMonth { get; set; } = "";
    /// <summary>日 T+1 KPI 区间起（含），默认昨天 00:00。</summary>
    public DateTime DailyPeriodStart { get; set; }
    /// <summary>日 T+1 KPI 区间止（含），默认昨天 23:59:59。</summary>
    public DateTime DailyPeriodEnd { get; set; }
    /// <summary>月 M+1 KPI 区间起（含），默认上月 1 日。</summary>
    public DateTime MonthlyPeriodStart { get; set; }
    /// <summary>月 M+1 KPI 区间止（含），默认上月末日。</summary>
    public DateTime MonthlyPeriodEnd { get; set; }
    /// <summary>TVF Rep_总账_存货_V3 入参起期 YYYYMM。</summary>
    public string TvfPeriodStartYyyymm { get; set; } = "";
    /// <summary>TVF Rep_总账_存货_V3 入参止期 YYYYMM。</summary>
    public string TvfPeriodEndYyyymm { get; set; } = "";

    public static S5MdpRefreshOption Default()
    {
        var today = DateTime.Today;
        var yesterday = today.AddDays(-1);
        var lastMonth = today.AddMonths(-1);
        var monthStart = new DateTime(lastMonth.Year, lastMonth.Month, 1);
        var monthEnd = monthStart.AddMonths(1).AddDays(-1);
        return new S5MdpRefreshOption
        {
            SourceZtid = "pbxfxp",
            TargetTenantId = 0,
            TargetFactoryId = 1L,
            BizDate = yesterday,
            BizMonth = lastMonth.ToString("yyyy-MM"),
            DailyPeriodStart = yesterday,
            DailyPeriodEnd = yesterday.AddDays(1).AddSeconds(-1),
            MonthlyPeriodStart = monthStart,
            MonthlyPeriodEnd = monthEnd.AddDays(1).AddSeconds(-1),
            TvfPeriodStartYyyymm = monthStart.ToString("yyyyMM"),
            TvfPeriodEndYyyymm = monthEnd.ToString("yyyyMM")
        };
    }
}

public sealed class S5MdpSyncTransformResult
{
    public string BatchId { get; set; } = "";
    public long RunLogId { get; set; }
    public string TriggerType { get; set; } = "AUTO";
    public string SourceZtid { get; set; } = "";
    public long TargetTenantId { get; set; }
    public long TargetFactoryId { get; set; }
    public DateTime BizDate { get; set; }
    public string BizMonth { get; set; } = "";
    public DateTime DailyPeriodStart { get; set; }
    public DateTime DailyPeriodEnd { get; set; }
    public DateTime MonthlyPeriodStart { get; set; }
    public DateTime MonthlyPeriodEnd { get; set; }

    public int StageRows { get; set; }
    public int StandardRows { get; set; }
    public int DwdRows { get; set; }
    public int KpiRows { get; set; }

    public Dictionary<string, int> PerKpiDwdRows { get; } = new();
    public Dictionary<string, int> PerKpiKpiRows { get; } = new();
    public List<string> KpiDenominatorStatus { get; } = new();

    public void MergeSub(string kpiCode, KpiBuildSubResult sub)
    {
        PerKpiDwdRows[kpiCode] = sub.DwdRows;
        PerKpiKpiRows[kpiCode] = sub.KpiRows;
        DwdRows += sub.DwdRows;
        KpiRows += sub.KpiRows;
        KpiDenominatorStatus.Add($"{kpiCode}:{sub.DenominatorStatus}");
    }
}

public sealed class KpiBuildSubResult
{
    public int T8Rows { get; set; }
    public int DwdRows { get; set; }
    public int KpiRows { get; set; }
    public string DenominatorStatus { get; set; } = "OK";
}

// ─────────────────────────────────────────────────────────────────────────────
// T8 result set 投影类型（与方老师 SQL SELECT 列名严格一致；SqlSugar 映射）
// ─────────────────────────────────────────────────────────────────────────────

internal sealed class S5OnlineCycleRow
{
    public string? code { get; set; }
    public DateTime? shtime { get; set; }
}

internal sealed class S5FulfillmentNumerRow
{
    public string? lynoid { get; set; }
    public int codenum { get; set; }
}

internal sealed class S5FulfillmentDenomRow
{
    public string? noid { get; set; }
    public int listnum { get; set; }
}

internal sealed class S5SumQtyRow
{
    public decimal? slzx { get; set; }
}

internal sealed class S5CountRow
{
    public int penum { get; set; }
}

internal sealed class S5InventoryTurnoverRow
{
    public string? ckcode { get; set; }
    public string? ckname { get; set; }
    public string? code { get; set; }
    public string? cname { get; set; }
    public string? pcode { get; set; }
    public string? pname { get; set; }
    public decimal? je3 { get; set; }
    public decimal? je2 { get; set; }
}