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();
}
}