using Admin.NET.Plugin.AiDOP.DataPlatform; using Admin.NET.Plugin.AiDOP.FinishedWarehouse; using Admin.NET.Plugin.AiDOP.Manufacturing; using Admin.NET.Plugin.AiDOP.MaterialWarehouse; using Furion.Schedule; using Microsoft.Extensions.DependencyInjection; using Microsoft.Extensions.Logging; using System.Text.Json; namespace Admin.NET.Plugin.AiDOP.Job; /// /// T8 基础数据 + S5/S6/S7 KPI 统一刷新编排任务(方案④ owner)。 /// 一个 tick 内:先执行 1 次 T8 基表双模式入站(源→mdp_stg_t8_*→mdp_std_t8_*), /// 再顺序计算 S5/S6/S7 三个模块的 T8 KPI(均读同一份本轮刷新后的 std,保证强一致)。 /// JobId 保留 job_s5_t8_kpi_refresh(不迁移 Job 身份/历史);原 S6/S7 独立 Job 已合并至本 Job。 /// 每日 02:00 / 07:00 / 12:00 / 17:00 / 22:00 共 5 次。 /// /// 失败语义(关键,勿机械照抄 Bootstrap): /// ① T8 inbound 是整轮硬门——失败则记错误日志并立即结束本轮, /// 禁止继续算 S5/S6/S7(不允许退化为用上一周期 std 静默计算的 eventual consistency); /// ② inbound 成功后,S5→S6→S7 采用步级失败隔离(某模块失败记错误、继续下一模块); /// ③ OperationCanceledException 保持取消语义直接上抛,不吞。 /// 手工补偿入口 /t8-base-mdp/inbound 仍独立保留。 /// [JobDetail("job_s5_t8_kpi_refresh", Description = "T8 基础数据 + S5/S6/S7 KPI 统一刷新编排(inbound 硬门 → S5→S6→S7;5 次/天:02/07/12/17/22)", GroupName = "default", Concurrent = false)] [Cron("0 2,7,12,17,22 * * *", TriggerId = "trigger_s5_t8_kpi_refresh", Description = "每日 02:00/07:00/12:00/17:00/22:00 触发(5 字段:分 时 日 月 周,默认 CronStringFormat.Default)")] public class S5MdpRefreshJob : IJob { private readonly IServiceScopeFactory _scopeFactory; private readonly ILogger _logger; public S5MdpRefreshJob(IServiceScopeFactory scopeFactory, ILoggerFactory loggerFactory) { _scopeFactory = scopeFactory; _logger = loggerFactory.CreateLogger(nameof(S5MdpRefreshJob)); } public async Task ExecuteAsync(JobExecutingContext context, CancellationToken stoppingToken) { // ① 硬门:T8 基表入站(源→stg→std)必须先成功;失败则本轮结束,不继续下游、不降级用旧 std。 try { using var inboundScope = _scopeFactory.CreateScope(); var inbound = inboundScope.ServiceProvider.GetRequiredService(); var inboundResult = await inbound.RunInboundAsync( tenantId: 0, fullRefresh: true, entityCode: null, cancellationToken: stoppingToken); _logger.LogInformation("S5MdpRefreshJob T8 inbound 完成 {Payload}", JsonSerializer.Serialize(inboundResult)); } catch (OperationCanceledException) when (stoppingToken.IsCancellationRequested) { _logger.LogInformation("S5MdpRefreshJob T8 inbound 收到停止信号,本轮结束"); throw; } catch (Exception ex) { _logger.LogError(ex, "S5MdpRefreshJob T8 inbound 失败,本轮跳过 S5/S6/S7 KPI 刷新(不使用上一周期 std 继续计算)"); return; } // ② 下游:inbound 成功后,S5→S6→S7 顺序执行,步级失败隔离(各自独立 DI scope)。 await RunStepAsync("S5", stoppingToken, async sp => (object)await sp.GetRequiredService().RunFullAsync(stoppingToken, "AUTO")); await RunStepAsync("S6", stoppingToken, async sp => (object)await sp.GetRequiredService().RunFullAsync(stoppingToken, "AUTO")); await RunStepAsync("S7", stoppingToken, async sp => (object)await sp.GetRequiredService().RunFullAsync(stoppingToken, "AUTO")); } /// /// 下游模块步级执行:独立 DI scope + 失败隔离(仿 SmartOpsKpiMdpBootstrapJob.RunStepAsync)。 /// 仅用于 inbound 成功之后的 S5/S6/S7;不得用于包裹 inbound(inbound 失败必须中断本轮)。 /// private async Task RunStepAsync(string step, CancellationToken stoppingToken, Func> action) { using var scope = _scopeFactory.CreateScope(); try { stoppingToken.ThrowIfCancellationRequested(); var payload = await action(scope.ServiceProvider); _logger.LogInformation("S5MdpRefreshJob {Step} 完成 {Payload}", step, JsonSerializer.Serialize(payload)); } catch (OperationCanceledException) when (stoppingToken.IsCancellationRequested) { _logger.LogInformation("S5MdpRefreshJob {Step} 收到停止信号", step); throw; } catch (Exception ex) { // 失败通知由各 service 内部 MarkTransformRunFailedAsync 触发;此处记录并继续下一步 _logger.LogError(ex, "S5MdpRefreshJob {Step} 执行失败,继续后续步骤", step); } } }