using Admin.NET.Plugin.AiDOP.DataPlatform.MdpRebuild; using Admin.NET.Plugin.AiDOP.Entity.DataPlatform; using Microsoft.Extensions.Logging; namespace Admin.NET.Plugin.AiDOP.DataPlatform; /// /// 方式甲完成贴源抽取后,按任务消费模块触发既有模块重算。 /// 只负责编排,不复制任何 S1-S7 转换或 KPI 逻辑。 /// public sealed class MdpDbSyncRebuildTrigger : ITransient { private readonly ISqlSugarClient _db; private readonly ModuleRebuildService _rebuild; private readonly ILogger _logger; public MdpDbSyncRebuildTrigger( ISqlSugarClient db, ModuleRebuildService rebuild, ILogger logger) { _db = db; _rebuild = rebuild; _logger = logger; } public async Task> EnqueueForTaskAsync( MdpSyncTask task, long tenantId, long factoryId, string triggerType, CancellationToken cancellationToken, long? requestedBy = null) { if (tenantId <= 0) throw new InvalidOperationException("方式甲触发模块重算时 tenantId 必须大于 0"); var modules = ParseModules(task.ConsumerModules); if (modules.Count == 0) modules = await InferModulesAsync(task, tenantId, cancellationToken); var accepted = new List(); var rejected = new List(); foreach (var module in modules) { cancellationToken.ThrowIfCancellationRequested(); var (status, body) = await _rebuild.EnqueueAsync( module, tenantId, factoryId > 0 && factoryId != tenantId ? factoryId : 1, requestedBy, triggerType: triggerType, cancellationToken); if (status == 202) { accepted.Add(module); continue; } _logger.LogWarning( "方式甲任务 {TaskCode} 未能触发 {Module} 重算,status={Status} message={Message}", task.TaskCode, module, status, body.Message); rejected.Add($"{module}({status}:{body.Message})"); } if (rejected.Count > 0) throw new InvalidOperationException($"方式甲抽取成功,但模块重算入队失败:{string.Join(";", rejected)}"); return accepted; } internal static List ParseModules(string? raw) { if (string.IsNullOrWhiteSpace(raw)) return []; return raw .Split([',', ';', '|', ' ', '\r', '\n'], StringSplitOptions.RemoveEmptyEntries | StringSplitOptions.TrimEntries) .Select(x => x.ToUpperInvariant()) .Where(MdpRebuildScope.RebuildModules.Contains) .Distinct(StringComparer.OrdinalIgnoreCase) .OrderBy(x => x, StringComparer.OrdinalIgnoreCase) .ToList(); } private async Task> InferModulesAsync( MdpSyncTask task, long tenantId, CancellationToken cancellationToken) { var jobCode = string.IsNullOrWhiteSpace(task.JobCode) ? task.TaskCode : task.JobCode; var codes = await _db.Queryable() .Where(x => x.Status == 1 && (x.TenantId == tenantId || x.TenantId == 0) && (x.JobId == task.TaskCode || x.JobId == jobCode)) .Select(x => x.EntityCode) .ToListAsync(cancellationToken); var modules = new HashSet(StringComparer.OrdinalIgnoreCase); foreach (var code in codes) { var prefix = code.Length >= 2 ? code[..2].ToUpperInvariant() : ""; if (MdpRebuildScope.RebuildModules.Contains(prefix)) modules.Add(prefix); var std = MdpStdObjectCatalog.Find(code); if (std != null && MdpRebuildScope.RebuildModules.Contains(std.Module)) modules.Add(std.Module); } return modules.OrderBy(x => x, StringComparer.OrdinalIgnoreCase).ToList(); } }