AiDOPWarehouse/server/Plugins/Admin.NET.Plugin.AiDOP/DataPlatform/MdpDbSyncRebuildTrigger.cs

using Admin.NET.Plugin.AiDOP.DataPlatform.MdpRebuild;
using Admin.NET.Plugin.AiDOP.Entity.DataPlatform;
using Microsoft.Extensions.Logging;

namespace Admin.NET.Plugin.AiDOP.DataPlatform;

/// <summary>
/// 方式甲完成贴源抽取后，按任务消费模块触发既有模块重算。
/// 只负责编排，不复制任何 S1-S7 转换或 KPI 逻辑。
/// </summary>
public sealed class MdpDbSyncRebuildTrigger : ITransient
{
    private readonly ISqlSugarClient _db;
    private readonly ModuleRebuildService _rebuild;
    private readonly ILogger<MdpDbSyncRebuildTrigger> _logger;

    public MdpDbSyncRebuildTrigger(
        ISqlSugarClient db,
        ModuleRebuildService rebuild,
        ILogger<MdpDbSyncRebuildTrigger> logger)
    {
        _db = db;
        _rebuild = rebuild;
        _logger = logger;
    }

    public async Task<IReadOnlyList<string>> 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<string>();
        var rejected = new List<string>();
        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<string> 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<List<string>> InferModulesAsync(
        MdpSyncTask task,
        long tenantId,
        CancellationToken cancellationToken)
    {
        var jobCode = string.IsNullOrWhiteSpace(task.JobCode) ? task.TaskCode : task.JobCode;
        var codes = await _db.Queryable<MdpEntity>()
            .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<string>(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();
    }
}