using Admin.NET.Core.Service;
using Admin.NET.Plugin.AiDOP.Infrastructure;
using Admin.NET.Plugin.AiDOP.SmartOps;
using Microsoft.Extensions.Logging;
using System.Text.Json;
namespace Admin.NET.Plugin.AiDOP.MaterialWarehouse;
///
/// S5 物料仓储 — KPI 计算与刷新。L1 读中立标准层。
/// 结果落 dwd_* 与 ado_s9_kpi_value_l1_day。
///
public class S5MdpSyncTransformService : ITransient
{
private readonly ISqlSugarClient _db;
private readonly TransformRunLogFinalizer _runLogFinalizer;
private readonly SysNoticeService _sysNoticeService;
private readonly ILogger _logger;
private const string JobCode = "S5_MDP_SYNC_TRANSFORM";
private const string JobName = "S5 物料仓储 MDP 同步与转换";
private const string ModuleCode = "S5";
private const string L2ValueTable = "ado_s9_kpi_value_l2_day";
private const string L3ValueTable = "ado_s9_kpi_value_l3_day";
// 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;
private readonly IKpiTargetResolver _kpiTargetResolver;
private readonly InventoryMdpSyncService _inventoryMdpSync;
private readonly PurchaseReceiptMdpSyncService _purchaseReceiptMdpSync;
private readonly DataPlatform.T8BaseInboundMdpSyncService _t8Inbound;
private readonly SmartOps.S9CompositeKpiWriter _s9Composite;
public S5MdpSyncTransformService(
ISqlSugarClient db,
SysNoticeService sysNoticeService,
ILogger logger,
SmartOps.KpiCalcDispatcher kpiCalcDispatcher,
SmartOps.KpiDimensionRunService dimensionRun,
IKpiTargetResolver kpiTargetResolver,
InventoryMdpSyncService inventoryMdpSync,
PurchaseReceiptMdpSyncService purchaseReceiptMdpSync,
DataPlatform.T8BaseInboundMdpSyncService t8Inbound,
TransformRunLogFinalizer runLogFinalizer,
SmartOps.S9CompositeKpiWriter s9Composite)
{
_db = db;
_runLogFinalizer = runLogFinalizer;
_sysNoticeService = sysNoticeService;
_logger = logger;
_kpiCalcDispatcher = kpiCalcDispatcher;
_dimensionRun = dimensionRun;
_kpiTargetResolver = kpiTargetResolver;
_inventoryMdpSync = inventoryMdpSync;
_purchaseReceiptMdpSync = purchaseReceiptMdpSync;
_t8Inbound = t8Inbound;
_s9Composite = s9Composite;
}
public async Task 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
{
var receiptSync = await _purchaseReceiptMdpSync.RunInboundAsync(
option.TargetTenantId, true, cancellationToken);
result.StageRows = receiptSync.RowsWrittenStg;
result.StandardRows = receiptSync.StdRows
+ await _inventoryMdpSync.TransformTransStdFromStgAsync(
option.TargetTenantId, cancellationToken);
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);
var currentBizDate = option.BizDate;
var currentPeriodStart = option.DailyPeriodStart;
var currentPeriodEnd = option.DailyPeriodEnd;
const int backfillDays = 14;
for (var dayOffset = backfillDays - 1; dayOffset >= 0; dayOffset--)
{
option.BizDate = currentBizDate.AddDays(-dayOffset);
option.DailyPeriodStart = option.BizDate.Date;
option.DailyPeriodEnd = option.BizDate.Date.AddDays(1).AddSeconds(-1);
result.MergeSub("S5_L2_001", await BuildS5L2001ReceiptCycleAsync(
batchId, now, option, normalizedTrigger, cancellationToken));
result.MergeSub("S5_L2_002", await BuildS5L2002ReceiptFulfillmentAsync(
batchId, now, option, normalizedTrigger, cancellationToken));
result.MergeSub("S5_L2_003", await BuildS5L2003IqcCycleAsync(
batchId, now, option, normalizedTrigger, cancellationToken));
result.MergeSub("S5_L2_004", await BuildS5L2004IqcFulfillmentAsync(
batchId, now, option, normalizedTrigger, cancellationToken));
foreach (var metricCode in new[]
{
"S5_L2_005", "S5_L2_006", "S5_L2_007", "S5_L2_008", "S5_L2_009",
"S5_L2_010", "S5_L2_011", "S5_L2_012", "S5_L2_013", "S5_L2_014",
"S5_L2_015", "S5_L3_001", "S5_L3_002", "S5_L3_003", "S5_L3_004",
"S5_L3_005"
})
{
var sub = await BuildS5WarehouseStageKpiAsync(
metricCode, batchId, now, option, normalizedTrigger, cancellationToken);
result.MergeSub(metricCode, sub);
}
}
option.BizDate = currentBizDate;
option.DailyPeriodStart = currentPeriodStart;
option.DailyPeriodEnd = currentPeriodEnd;
result.KpiRows += await _s9Composite.WriteRecentAsync(
option.TargetTenantId, option.TargetFactoryId, option.BizDate, cancellationToken);
await MarkTransformRunSuccessAsync(runLogId, now, result);
return result;
}
catch (Exception ex)
{
// 宿主关停不是转换失败:交给 finally 收口为 ABORTED,不污染 FAILED 语义。
if (!_runLogFinalizer.IsHostStopping)
await MarkTransformRunFailedAsync(runLogId, now, ex.Message, batchId);
throw;
}
finally
{
await _runLogFinalizer.FinalizeIfHostStoppingAsync(runLogId, now);
}
}
// ─────────────────────────────────────────────────────────────────────────
// KPI 实现(方老师 v5.4 KPI J 列 SQL 原逻辑直发 T8)
// ─────────────────────────────────────────────────────────────────────────
/// S5_L1_001 物料上线周期 = 领料到线时间减采购收货时间。数据准备始终执行,最终聚合由计算配置分发器接管。
private async Task BuildS5L1001MaterialOnlineCycleAsync(
string batchId, DateTime now, S5MdpRefreshOption option, string triggerType, CancellationToken ct)
{
var sub = new KpiBuildSubResult();
const string sqlOnline = @"
select item_num as item_code, min(approved_time) as approved_time
from mdp_std_inv_trans
where tenant_id=@tenantId and biz_doc_type='PROD_ISSUE'
and summary_flag=0 and void_flag=0 and approved_flag=1 AND (@sourceSystem='' OR source_system=@sourceSystem) AND (source_system<>'T8' OR domain=@sourceDomain)
group by item_num";
const string sqlReceipt = @"
select item_num as item_code, min(approved_time) as approved_time
from mdp_std_inv_trans
where tenant_id=@tenantId and biz_doc_type='PUR_RECEIPT'
and summary_flag=0 and void_flag=0 and approved_flag=1 AND (@sourceSystem='' OR source_system=@sourceSystem) AND (source_system<>'T8' OR domain=@sourceDomain)
group by item_num";
var p = new[]
{
new SugarParameter("@tenantId", option.TargetTenantId),
new SugarParameter("@sourceDomain", option.SourceZtid),
new SugarParameter("@sourceSystem", "")
};
var onlineRows = await _db.Ado.SqlQueryAsync(sqlOnline, p);
var receiptRows = await _db.Ado.SqlQueryAsync(sqlReceipt, p);
sub.T8Rows = onlineRows.Count + receiptRows.Count;
var onlineByCode = onlineRows.Where(r => !string.IsNullOrEmpty(r.item_code))
.ToDictionary(r => r.item_code!, r => r.approved_time, StringComparer.OrdinalIgnoreCase);
var receiptByCode = receiptRows.Where(r => !string.IsNullOrEmpty(r.item_code))
.ToDictionary(r => r.item_code!, r => r.approved_time, StringComparer.OrdinalIgnoreCase);
var allCodes = new HashSet(onlineByCode.Keys, StringComparer.OrdinalIgnoreCase);
allCodes.UnionWith(receiptByCode.Keys);
var dwdAffected = 0;
var cycleDaysList = new List();
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_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, @sourceDomain, @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("@sourceDomain", 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;
}
/// S5_L1_002 物料上线满足率 = 开工日期前完成上线行数 / 工单物料总行数。
private async Task BuildS5L1002MaterialOnlineFulfillmentAsync(
string batchId, DateTime now, S5MdpRefreshOption option, string triggerType, CancellationToken ct)
{
var sub = new KpiBuildSubResult();
const string sqlNumer = @"
select ref_task_no as task_no, count(*) as codenum
from (
select ref_task_no, item_num
from mdp_std_inv_trans h
left join (
select work_order_no as task_no, min(start_work_date) as start_work_date
from mdp_std_s6_report
where tenant_id=@tenantId
group by work_order_no
) c on h.ref_task_no=c.task_no
where h.tenant_id=@tenantId and h.biz_doc_type='PROD_ISSUE'
and h.summary_flag=0 and h.void_flag=0 and h.approved_flag=1
AND (@sourceSystem='' OR h.source_system=@sourceSystem) AND (h.source_system<>'T8' OR h.domain=@sourceDomain)
and h.approved_time<=c.start_work_date
group by h.ref_task_no, h.item_num
) n
group by task_no";
// Q14:以工单头为驱动 LEFT JOIN BOM。无 BOM 行的工单仍出现,行数为 0(计入分母、不计入分子)。
const string sqlDenom = @"
select h.work_order as order_no, count(b.source_row_id) as listnum
from mdp_std_work_order_schedule h
left join mdp_std_work_order_bom b
on b.tenant_id=h.tenant_id and b.source_system=h.source_system and b.order_no=h.work_order
where h.tenant_id=@tenantId and h.doc_type='PROD_TASK'
and h.void_flag=0 and h.approved_flag=1
group by h.work_order";
var p = new[]
{
new SugarParameter("@tenantId", option.TargetTenantId),
new SugarParameter("@sourceDomain", option.SourceZtid),
new SugarParameter("@sourceSystem", "")
};
var numerRows = await _db.Ado.SqlQueryAsync(sqlNumer, p);
var denomRows = await _db.Ado.SqlQueryAsync(sqlDenom, p);
sub.T8Rows = numerRows.Count + denomRows.Count;
var numerByOrder = numerRows.Where(r => !string.IsNullOrEmpty(r.task_no))
.ToDictionary(r => r.task_no!, r => r.codenum, StringComparer.OrdinalIgnoreCase);
var dwdAffected = 0;
var rateList = new List();
foreach (var d in denomRows)
{
ct.ThrowIfCancellationRequested();
if (string.IsNullOrEmpty(d.order_no)) continue;
var beforeKg = numerByOrder.GetValueOrDefault(d.order_no, 0);
// 无 BOM 行:满足行数为 0,该工单以 0 计入各单比率的平均(分母含它,分子不含)。
var rate = d.listnum > 0
? Math.Round((decimal)beforeKg / d.listnum, 4)
: 0m;
rateList.Add(rate);
dwdAffected += await _db.Ado.ExecuteCommandAsync(@"
INSERT INTO dwd_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, @sourceDomain, @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("@sourceDomain", option.SourceZtid),
new SugarParameter("@workOrderNo", d.order_no),
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;
}
/// S5_L1_003 物料仓储人效 = 领料数量 / 在职仓储岗位人数。
private async Task BuildS5L1003MaterialWarehouseEfficiencyAsync(
string batchId, DateTime now, S5MdpRefreshOption option, string triggerType, CancellationToken ct)
{
var sub = new KpiBuildSubResult();
const string sqlNumer = @"
select sum(qty_change) as qty_change
from mdp_std_inv_trans
where tenant_id=@tenantId and biz_doc_type='PROD_ISSUE'
and summary_flag=0 and void_flag=0 and approved_flag=1 AND (@sourceSystem='' OR source_system=@sourceSystem) AND (source_system<>'T8' OR domain=@sourceDomain)
and approved_time between @startDate and @endDate";
const string sqlDenom = @"
select count(*) as penum
from mdp_std_employee
where tenant_id=@tenantId and employment_status='ACTIVE' and position_code='WAREHOUSE' AND (@sourceSystem='' OR source_system=@sourceSystem) AND (source_system<>'T8' OR domain=@sourceDomain)";
var pNumer = new[]
{
new SugarParameter("@tenantId", option.TargetTenantId),
new SugarParameter("@startDate", option.MonthlyPeriodStart),
new SugarParameter("@endDate", option.MonthlyPeriodEnd),
new SugarParameter("@sourceDomain", option.SourceZtid),
new SugarParameter("@sourceSystem", "")
};
var pDenom = new[]
{
new SugarParameter("@tenantId", option.TargetTenantId),
new SugarParameter("@sourceDomain", option.SourceZtid),
new SugarParameter("@sourceSystem", "")
};
var numerRows = await _db.Ado.SqlQueryAsync(sqlNumer, pNumer);
var denomRows = await _db.Ado.SqlQueryAsync(sqlDenom, pDenom);
sub.T8Rows = numerRows.Count + denomRows.Count;
decimal? onlineQty = numerRows.FirstOrDefault()?.qty_change;
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_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, @sourceDomain, @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("@sourceDomain", 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;
}
/// S5_L1_004 品类物料库存周转 = D1/D2 × 30;D1=je3 月均库存金额,D2=je2 出库成本。
private async Task BuildS5L1004MaterialInventoryTurnoverAsync(
string batchId, DateTime now, S5MdpRefreshOption option, string triggerType, CancellationToken ct)
{
var sub = new KpiBuildSubResult();
await _t8Inbound.TryMaterializeInventoryBalanceAsync(
option.TargetTenantId, option.TargetFactoryId, option.SourceZtid,
option.TvfPeriodEndYyyymm, batchId, now);
const string sqlMonthly = @"
select warehouse_code as ckcode, warehouse_name as ckname,
item_code as code, item_code as cname,
category_code as pcode, category_name as pname,
avg_balance_amount as je3, issue_cost_amount as je2
from mdp_std_inventory_balance_monthly
where tenant_id=@tenantId and period_ym=@periodYm";
var p = new[]
{
new SugarParameter("@tenantId", option.TargetTenantId),
new SugarParameter("@periodYm", option.TvfPeriodEndYyyymm)
};
var tvfRows = await _db.Ado.SqlQueryAsync(sqlMonthly, p);
sub.T8Rows = tvfRows.Count;
var dwdAffected = 0;
var turnoverDaysList = new List();
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_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, @sourceDomain, @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("@sourceDomain", 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;
if (!metricValue.HasValue)
{
var warehouseFallback = await _db.Ado.SqlQuerySingleAsync(
WarehouseTurnoverSql("MAT_RECEIPT", "MAT_RECEIPT"),
new SugarParameter("@TenantId", option.TargetTenantId),
new SugarParameter("@FactoryId", option.TargetFactoryId),
new SugarParameter("@MetricCode", "S5_L1_004"),
new SugarParameter("@sourceDomain", option.SourceZtid),
new SugarParameter("@sourceSystem", ""));
metricValue = warehouseFallback?.MetricValue;
}
// 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;
}
// ─────────────────────────────────────────────────────────────────────────
// 写入 / 日志 封装
// ─────────────────────────────────────────────────────────────────────────
/// S5_L2_001 物料收货周期 = 实际收货日期 - 约定履约日期。
private async Task BuildS5L2001ReceiptCycleAsync(
string batchId, DateTime now, S5MdpRefreshOption option, string triggerType, CancellationToken ct)
{
const string sql = """
SELECT ROUND(AVG(TIMESTAMPDIFF(HOUR, perform_date, rct_date) / 24), 4) AS MetricValue,
COUNT(1) AS RowCount
FROM (
SELECT STR_TO_DATE(NULLIF(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.PerformDate')), 'null'), '%Y-%m-%d %H:%i:%s.%f') AS perform_date,
STR_TO_DATE(NULLIF(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.RctDate')), 'null'), '%Y-%m-%d %H:%i:%s.%f') AS rct_date
FROM mdp_stg_purchase_receipt
WHERE tenant_id=@TenantId
AND source_table='PurOrdRctMaster'
) t
WHERE perform_date IS NOT NULL AND rct_date IS NOT NULL AND rct_date >= perform_date
""";
return await DispatchAverageAsync("S5_L2_001", sql, "NO_RECEIPT_CYCLE", batchId, now, option, triggerType, ct, L2ValueTable);
}
/// S5_L2_002 物料收货满足率 = 约定日期内完成收货的收货单占比。
private async Task BuildS5L2002ReceiptFulfillmentAsync(
string batchId, DateTime now, S5MdpRefreshOption option, string triggerType, CancellationToken ct)
{
const string sql = """
SELECT ROUND(100 * SUM(CASE WHEN rct_date <= perform_date THEN 1 ELSE 0 END) / NULLIF(COUNT(1), 0), 4) AS MetricValue,
COUNT(1) AS RowCount
FROM (
SELECT STR_TO_DATE(NULLIF(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.PerformDate')), 'null'), '%Y-%m-%d %H:%i:%s.%f') AS perform_date,
STR_TO_DATE(NULLIF(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.RctDate')), 'null'), '%Y-%m-%d %H:%i:%s.%f') AS rct_date
FROM mdp_stg_purchase_receipt
WHERE tenant_id=@TenantId
AND source_table='PurOrdRctMaster'
) t
WHERE perform_date IS NOT NULL AND rct_date IS NOT NULL
""";
return await DispatchAverageAsync("S5_L2_002", sql, "NO_RECEIPT_FULFILLMENT", batchId, now, option, triggerType, ct, L2ValueTable);
}
/// S5_L2_003 物料检验周期 = 检验完成时间 - 检验开始时间。
private async Task BuildS5L2003IqcCycleAsync(
string batchId, DateTime now, S5MdpRefreshOption option, string triggerType, CancellationToken ct)
{
const string sql = """
SELECT ROUND(AVG(TIMESTAMPDIFF(HOUR, FINSPESTARTDATE, FINSPEENDDATE) / 24), 4) AS MetricValue,
COUNT(1) AS RowCount
FROM qms_qcp_inspbill
WHERE tenant_id=@TenantId
AND FINSPESTARTDATE IS NOT NULL
AND FINSPEENDDATE IS NOT NULL
AND FINSPEENDDATE >= FINSPESTARTDATE
""";
return await DispatchAverageAsync("S5_L2_003", sql, "NO_IQC_CYCLE", batchId, now, option, triggerType, ct, L2ValueTable);
}
/// S5_L2_004 物料检验满足率 = 报检后 72 小时内完成检验的单据占比。
private async Task BuildS5L2004IqcFulfillmentAsync(
string batchId, DateTime now, S5MdpRefreshOption option, string triggerType, CancellationToken ct)
{
const string sql = """
SELECT ROUND(100 * SUM(CASE WHEN FINSPEENDDATE IS NOT NULL
AND TIMESTAMPDIFF(HOUR, FCREATETIME, FINSPEENDDATE) <= 72 THEN 1 ELSE 0 END)
/ NULLIF(COUNT(1), 0), 4) AS MetricValue,
COUNT(1) AS RowCount
FROM qms_qcp_inspbill
WHERE tenant_id=@TenantId
AND FCREATETIME IS NOT NULL
""";
return await DispatchAverageAsync("S5_L2_004", sql, "NO_IQC_FULFILLMENT", batchId, now, option, triggerType, ct, L2ValueTable);
}
private Task BuildS5WarehouseStageKpiAsync(
string metricCode, string batchId, DateTime now, S5MdpRefreshOption option,
string triggerType, CancellationToken ct)
{
var (sql, emptyDenom, table) = metricCode switch
{
"S5_L2_005" => (WarehouseCycleSql("MAT_IQC_RELEASE", "MAT_PUTAWAY"), "NO_MATERIAL_PUTAWAY_CYCLE", L2ValueTable),
"S5_L2_006" => (WarehouseEfficiencySql("MAT_PUTAWAY"), "NO_MATERIAL_PUTAWAY_OPERATOR", L2ValueTable),
"S5_L2_007" => (WarehouseTurnoverSql("MAT_PUTAWAY", "MAT_RECEIPT"), "NO_MATERIAL_RECEIPT_COST", L2ValueTable),
"S5_L2_008" => (WarehouseCycleSql("MAT_PUTAWAY", "MAT_PICK"), "NO_MATERIAL_PICK_CYCLE", L2ValueTable),
"S5_L2_009" => (WarehouseSatisfactionSql("MAT_PICK"), "NO_MATERIAL_PICK_REQUIRED_DATE", L2ValueTable),
"S5_L2_010" => (WarehouseEfficiencySql("MAT_PICK"), "NO_MATERIAL_PICK_OPERATOR", L2ValueTable),
"S5_L2_011" => (WarehouseTurnoverSql("MAT_PICK", "MAT_ISSUE"), "NO_MATERIAL_ISSUE_COST", L2ValueTable),
"S5_L2_012" => (WarehouseCycleSql("MAT_ISSUE", "MAT_LINE"), "NO_MATERIAL_LINE_CYCLE", L2ValueTable),
"S5_L2_013" => (WarehouseSatisfactionSql("MAT_LINE"), "NO_MATERIAL_LINE_REQUIRED_DATE", L2ValueTable),
"S5_L2_014" => (WarehouseEfficiencySql("MAT_LINE"), "NO_MATERIAL_LINE_OPERATOR", L2ValueTable),
"S5_L2_015" => (WarehouseTurnoverSql("MAT_LINE", "MAT_ISSUE"), "NO_MATERIAL_LINE_COST", L2ValueTable),
"S5_L3_001" => (WarehouseEfficiencySql("MAT_RECEIPT"), "NO_MATERIAL_RECEIPT_OPERATOR", L3ValueTable),
"S5_L3_002" => (WarehouseTurnoverSql("MAT_RECEIPT", "MAT_RECEIPT"), "NO_MATERIAL_RECEIPT_COST", L3ValueTable),
"S5_L3_003" => (WarehouseEfficiencySql("MAT_IQC_RELEASE"), "NO_MATERIAL_IQC_OPERATOR", L3ValueTable),
"S5_L3_004" => (WarehouseTurnoverSql("MAT_IQC_RELEASE", "MAT_RECEIPT"), "NO_MATERIAL_IQC_COST", L3ValueTable),
"S5_L3_005" => (WarehouseSatisfactionSql("MAT_PUTAWAY"), "NO_MATERIAL_PUTAWAY_REQUIRED_DATE", L3ValueTable),
_ => throw new ArgumentOutOfRangeException(nameof(metricCode), metricCode, "不支持的 S5 仓储阶段指标")
};
return DispatchAverageAsync(
metricCode, sql, emptyDenom, batchId, now, option, triggerType, ct, table);
}
private static string WarehouseCycleSql(string fromStage, string toStage) =>
$"""
SELECT ROUND(AVG(TIMESTAMPDIFF(MINUTE,a.trans_time,b.trans_time)/1440),4) AS MetricValue,
COUNT(*) AS RowCount
FROM mdp_std_inv_trans a
INNER JOIN mdp_std_inv_trans b
ON b.tenant_id=a.tenant_id AND b.source_system=a.source_system
AND b.item_num=a.item_num AND b.lot_serial=a.lot_serial
AND b.trans_type='{toStage}'
WHERE a.tenant_id=@TenantId AND a.trans_type='{fromStage}'
AND (@sourceSystem='' OR a.source_system=@sourceSystem) AND (a.source_system<>'T8' OR a.domain=@sourceDomain)
AND a.trans_time IS NOT NULL AND b.trans_time>=a.trans_time
""";
private static string WarehouseSatisfactionSql(string stage) =>
$"""
SELECT ROUND(100 * SUM(CASE WHEN trans_time<=eff_date THEN 1 ELSE 0 END)
/ NULLIF(COUNT(*),0),4) AS MetricValue,
COUNT(*) AS RowCount
FROM mdp_std_inv_trans
WHERE tenant_id=@TenantId AND trans_type='{stage}'
AND (@sourceSystem='' OR source_system=@sourceSystem) AND (source_system<>'T8' OR domain=@sourceDomain)
AND trans_time IS NOT NULL AND eff_date IS NOT NULL
""";
private static string WarehouseEfficiencySql(string stage) =>
$"""
SELECT ROUND(COUNT(DISTINCT NULLIF(lot_serial,''))
/ NULLIF(COUNT(DISTINCT NULLIF(TRIM(create_user),'')),0),4) AS MetricValue,
COUNT(*) AS RowCount
FROM mdp_std_inv_trans
WHERE tenant_id=@TenantId AND trans_type='{stage}'
AND (@sourceSystem='' OR source_system=@sourceSystem) AND (source_system<>'T8' OR domain=@sourceDomain)
""";
private static string WarehouseTurnoverSql(string inventoryStage, string flowStage) =>
$"""
SELECT ROUND(
30 * SUM(CASE WHEN trans_type='{inventoryStage}'
THEN IFNULL(end_balance,0)
* IFNULL(CAST(NULLIF(dimension1,'') AS DECIMAL(18,4)),1)
ELSE 0 END)
/ NULLIF(SUM(CASE WHEN trans_type='{flowStage}'
THEN ABS(IFNULL(qty_change,0))
* IFNULL(CAST(NULLIF(dimension1,'') AS DECIMAL(18,4)),1)
ELSE 0 END),0),
4) AS MetricValue,
SUM(CASE WHEN trans_type='{inventoryStage}' THEN 1 ELSE 0 END) AS RowCount
FROM mdp_std_inv_trans
WHERE tenant_id=@TenantId AND trans_type IN ('{inventoryStage}','{flowStage}')
AND (@sourceSystem='' OR source_system=@sourceSystem) AND (source_system<>'T8' OR domain=@sourceDomain)
""";
private async Task DispatchAverageAsync(
string metricCode, string sql, string emptyDenom,
string batchId, DateTime now, S5MdpRefreshOption option, string triggerType, CancellationToken ct,
string valueTable)
{
var row = await _db.Ado.SqlQuerySingleAsync(sql,
new SugarParameter("@TenantId", option.TargetTenantId),
new SugarParameter("@sourceDomain", option.SourceZtid),
new SugarParameter("@sourceSystem", ""));
decimal? value = row?.RowCount > 0 ? row.MetricValue : null;
var denom = value.HasValue ? "OK" : emptyDenom;
var dispatch = await _kpiCalcDispatcher.DispatchAsync(
metricCode, ModuleCode, option.TargetTenantId, option.TargetFactoryId,
option.BizDate, option.BizDate, option.BizDate, option.SourceZtid,
batchId, triggerType, value, denom, ct);
var sub = new KpiBuildSubResult
{
T8Rows = row?.RowCount ?? 0,
KpiRows = dispatch.ShouldUpsert
? await UpsertKpiValueAsync(metricCode, option.BizDate, dispatch.MetricValue, now, option, valueTable)
: 0,
DenominatorStatus = dispatch.DenominatorStatus
};
try
{
if (dispatch.ShouldUpsert)
{
await _dimensionRun.RunDimensionAsync(
metricCode, ModuleCode, option.TargetTenantId, option.BizDate, batchId, triggerType, ct);
}
}
catch (Exception ex)
{
_logger.LogWarning(ex, "{MetricCode} 维度跑批异常(不影响汇总链路)", metricCode);
}
return sub;
}
private async Task UpsertKpiValueAsync(string metricCode, DateTime bizDate, decimal? metricValue, DateTime now, S5MdpRefreshOption option, string valueTable = "ado_s9_kpi_value_l1_day")
{
// 沿用 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 snap = await _kpiTargetResolver.ResolveAsync(option.TargetTenantId, option.TargetFactoryId, metricCode, ModuleCode, bizDate);
var meta = await _db.Ado.SqlQuerySingleAsync(
"SELECT Direction, YellowThreshold, RedThreshold FROM ado_smart_ops_kpi_master WHERE TenantId=@TenantId AND MetricCode=@MetricCode AND IsEnabled=1 LIMIT 1",
new SugarParameter("@TenantId", option.TargetTenantId),
new SugarParameter("@MetricCode", metricCode));
var status = AidopS4KpiMerge.AchievementLevel(
metricValue,
snap.TargetValue,
(string?)meta?.Direction ?? "higher_is_better",
(decimal?)meta?.YellowThreshold,
(decimal?)meta?.RedThreshold);
var existingId = await _db.Ado.GetLongAsync(
$"SELECT IFNULL((SELECT id FROM {valueTable} 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
{
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 {valueTable} SET metric_value=@MetricValue, target_value=@TargetValue, status_color=@StatusColor, " +
"target_config_id=@TargetConfigId, target_source=@TargetSource, target_resolved_at=@TargetResolvedAt, " +
"calc_time=@Now, update_time=@Now, is_deleted=0, is_active=1 WHERE id=@Id",
new SugarParameter("@MetricValue", metricValue),
new SugarParameter("@TargetValue", KpiTargetSnapshotSql.ValueOrDbNull(snap)),
new SugarParameter("@StatusColor", status),
new SugarParameter("@TargetConfigId", KpiTargetSnapshotSql.ConfigIdOrDbNull(snap)),
new SugarParameter("@TargetSource", KpiTargetSnapshotSql.SourceOrDbNull(snap)),
new SugarParameter("@TargetResolvedAt", snap.ResolvedAt),
new SugarParameter("@Now", now),
new SugarParameter("@Id", existingId));
}
var nextId = Yitter.IdGenerator.YitIdHelper.NextId();
return await _db.Ado.ExecuteCommandAsync($@"
INSERT INTO {valueTable}
(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, target_value, status_color, calc_time,
target_config_id, target_source, target_resolved_at)
VALUES
(@Id, @TenantId, NULL, NULL, @FactoryId, NULL, @BizDate,
@Now, @Now, 0, 1,
@ModuleCode, @MetricCode, @MetricValue, @TargetValue, @StatusColor, @Now,
@TargetConfigId, @TargetSource, @TargetResolvedAt)",
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),
new SugarParameter("@TargetValue", KpiTargetSnapshotSql.ValueOrDbNull(snap)),
new SugarParameter("@StatusColor", status),
new SugarParameter("@TargetConfigId", KpiTargetSnapshotSql.ConfigIdOrDbNull(snap)),
new SugarParameter("@TargetSource", KpiTargetSnapshotSql.SourceOrDbNull(snap)),
new SugarParameter("@TargetResolvedAt", snap.ResolvedAt));
}
private async Task 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 { 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;
// 目标租户由源账套映射决定,禁止固定默认/兜底
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
{
/// 源账套编码,实测当前唯一值为 pbxfxp。
public string SourceZtid { get; set; } = "pbxfxp";
/// KPI/DWD 落库目标租户;≤0 时由 按 SourceZtid 解析。
public long TargetTenantId { get; set; }
/// KPI/DWD 落库目标工厂;默认 1。
public long TargetFactoryId { get; set; } = 1L;
/// 日 T+1 KPI 的业务日期(默认昨天)。
public DateTime BizDate { get; set; }
/// 月 M+1 KPI 的业务月 YYYY-MM(默认上月)。
public string BizMonth { get; set; } = "";
/// 日 T+1 KPI 区间起(含),默认昨天 00:00。
public DateTime DailyPeriodStart { get; set; }
/// 日 T+1 KPI 区间止(含),默认昨天 23:59:59。
public DateTime DailyPeriodEnd { get; set; }
/// 月 M+1 KPI 区间起(含),默认上月 1 日。
public DateTime MonthlyPeriodStart { get; set; }
/// 月 M+1 KPI 区间止(含),默认上月末日。
public DateTime MonthlyPeriodEnd { get; set; }
/// TVF Rep_总账_存货_V3 入参起期 YYYYMM。
public string TvfPeriodStartYyyymm { get; set; } = "";
/// TVF Rep_总账_存货_V3 入参止期 YYYYMM。
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 PerKpiDwdRows { get; } = new();
public Dictionary PerKpiKpiRows { get; } = new();
public List 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? item_code { get; set; }
public DateTime? approved_time { get; set; }
}
internal sealed class S5FulfillmentNumerRow
{
public string? task_no { get; set; }
public int codenum { get; set; }
}
internal sealed class S5FulfillmentDenomRow
{
public string? order_no { get; set; }
public int listnum { get; set; }
}
internal sealed class S5SumQtyRow
{
public decimal? qty_change { get; set; }
}
internal sealed class S5CountRow
{
public int penum { get; set; }
}
internal sealed class S5StageKpiRow
{
public decimal? MetricValue { get; set; }
public int RowCount { 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; }
}