AiDOPWarehouse/server/Plugins/Admin.NET.Plugin.AiDOP/Service/S8/S8WatchSchedulerService.cs

using Admin.NET.Plugin.AiDOP.Entity.S8;
using Admin.NET.Plugin.AiDOP.Infrastructure.S8;
using SqlSugar;
using System.Data;
using System.Globalization;
using System.Text.Json;

namespace Admin.NET.Plugin.AiDOP.Service.S8;

/// <summary>
/// 监视规则轮询调度服务（首轮存根）。
/// 后续接入 Admin.NET 定时任务机制后，由调度器周期调用 <see cref="RunOnceAsync"/>，
/// 按各规则的 PollIntervalSeconds 逐条评估并生成异常记录。
/// </summary>
public class S8WatchSchedulerService : ITransient
{
    private readonly SqlSugarRepository<AdoS8WatchRule> _ruleRep;
    private readonly SqlSugarRepository<AdoS8AlertRule> _alertRuleRep;
    private readonly SqlSugarRepository<AdoS8DataSource> _dataSourceRep;
    private readonly SqlSugarRepository<AdoS8Exception> _exceptionRep;
    private readonly S8NotificationService _notificationService;
    private readonly S8ManualReportService _manualReportService;

    private const string DefaultTriggerType = "VALUE_DEVIATION";
    private const string SqlDataSourceType = "SQL";

    // G01-05 未闭环状态集合：复用自 S8ExceptionService 当前 pendingStatuses 事实口径
    // （见 S8ExceptionService.GetPagedAsync 中 pendingStatuses 的定义，两处必须保持一致）。
    // 这不是“自定义未闭环集合”；若现有口径调整，两处需同步修改。
    private static readonly string[] UnclosedExceptionStatuses =
        { "NEW", "ASSIGNED", "IN_PROGRESS", "PENDING_VERIFICATION" };

    public S8WatchSchedulerService(
        SqlSugarRepository<AdoS8WatchRule> ruleRep,
        SqlSugarRepository<AdoS8AlertRule> alertRuleRep,
        SqlSugarRepository<AdoS8DataSource> dataSourceRep,
        SqlSugarRepository<AdoS8Exception> exceptionRep,
        S8NotificationService notificationService,
        S8ManualReportService manualReportService)
    {
        _ruleRep = ruleRep;
        _alertRuleRep = alertRuleRep;
        _dataSourceRep = dataSourceRep;
        _exceptionRep = exceptionRep;
        _notificationService = notificationService;
        _manualReportService = manualReportService;
    }

    public async Task<List<S8WatchExecutionRule>> LoadExecutionRulesAsync(long tenantId, long factoryId)
    {
        var watchRules = await _ruleRep.AsQueryable()
            .Where(x => x.TenantId == tenantId
                && x.FactoryId == factoryId
                && x.Enabled
                && x.SceneCode == S8SceneCode.S2S6Production)
            .ToListAsync();

        var deviceRules = watchRules
            .Where(x => IsDeviceWatchObjectType(x.WatchObjectType))
            .ToList();
        if (deviceRules.Count == 0) return new();

        var dataSourceIds = deviceRules
            .Select(x => x.DataSourceId)
            .Distinct()
            .ToList();
        var dataSources = await _dataSourceRep.AsQueryable()
            .Where(x => x.TenantId == tenantId
                && x.FactoryId == factoryId
                && x.Enabled
                && dataSourceIds.Contains(x.Id))
            .ToListAsync();
        var dataSourceMap = dataSources.ToDictionary(x => x.Id);
        if (dataSourceMap.Count == 0) return new();

        var alertRules = (await _alertRuleRep.AsQueryable()
            .Where(x => x.TenantId == tenantId
                && x.FactoryId == factoryId
                && x.SceneCode == S8SceneCode.S2S6Production)
            .ToListAsync())
            .Where(IsSupportedAlertRule)
            .ToList();

        // G-01 首版 AlertRule 冲突口径（C 收口）：
        // 当前场景存在多条可运行 AlertRule 时，视为“当前规则配置冲突”并跳过该规则，
        // 不按“首条”继续运行，也不扩大为“整场景停摆”。
        // 当前模型下所有 device watchRule 共享同场景 AlertRule，故冲突态下所有 device 规则均跳过，
        // 但此处按“逐规则跳过”的语义实现，避免被误读为“整场景 return empty 停摆”。
        var alertRule = alertRules.Count == 1 ? alertRules[0] : null;

        var executionRules = new List<S8WatchExecutionRule>();
        foreach (var watchRule in deviceRules.OrderBy(x => x.Id))
        {
            // 配置冲突：当前规则跳过（不停摆其他规则）。
            if (alertRule == null)
                continue;

            if (!dataSourceMap.TryGetValue(watchRule.DataSourceId, out var dataSource))
                continue;

            if (!IsSupportedSqlDataSource(dataSource))
                continue;

            executionRules.Add(new S8WatchExecutionRule
            {
                WatchRuleId = watchRule.Id,
                WatchRuleCode = watchRule.RuleCode,
                SceneCode = watchRule.SceneCode,
                TriggerType = DefaultTriggerType,
                WatchObjectType = watchRule.WatchObjectType.Trim(),
                DataSourceId = dataSource.Id,
                DataSourceCode = dataSource.DataSourceCode,
                DataSourceType = dataSource.Type,
                DataSourceConnection = dataSource.Endpoint?.Trim() ?? string.Empty,
                QueryExpression = watchRule.Expression?.Trim() ?? string.Empty,
                PollIntervalSeconds = watchRule.PollIntervalSeconds,
                AlertRuleId = alertRule.Id,
                AlertRuleCode = alertRule.RuleCode,
                TriggerCondition = alertRule.TriggerCondition!.Trim(),
                ThresholdValue = alertRule.ThresholdVal!.Trim(),
                Severity = alertRule.Severity
            });
        }

        return executionRules;
    }

    public async Task<List<S8WatchDeviceQueryResult>> QueryDeviceRowsAsync(long tenantId, long factoryId)
    {
        var executionRules = await LoadExecutionRulesAsync(tenantId, factoryId);
        var results = new List<S8WatchDeviceQueryResult>();
        foreach (var rule in executionRules)
            results.Add(await QueryDeviceRowsAsync(rule));

        return results;
    }

    public async Task<S8WatchDeviceQueryResult> QueryDeviceRowsAsync(S8WatchExecutionRule rule)
    {
        if (!string.Equals(rule.DataSourceType, SqlDataSourceType, StringComparison.OrdinalIgnoreCase))
            return S8WatchDeviceQueryResult.Fail(rule, "数据源类型不是 SQL，已跳过");

        if (string.IsNullOrWhiteSpace(rule.QueryExpression))
            return S8WatchDeviceQueryResult.Fail(rule, "查询表达式为空，已跳过");

        try
        {
            using var db = CreateSqlQueryScope(rule.DataSourceConnection);
            var table = await db.Ado.GetDataTableAsync(rule.QueryExpression);
            if (!HasRequiredColumns(table))
                return S8WatchDeviceQueryResult.Fail(rule, "查询结果缺少 required columns: related_object_code/current_value");

            var rows = table.Rows.Cast<DataRow>()
                .Select(MapDeviceRow)
                .Where(x => !string.IsNullOrWhiteSpace(x.RelatedObjectCode))
                .ToList();
            return S8WatchDeviceQueryResult.Ok(rule, rows);
        }
        catch (Exception ex)
        {
            return S8WatchDeviceQueryResult.Fail(rule, $"查询执行失败: {ex.Message}");
        }
    }

    /// <summary>
    /// G01-04：基于设备级结果行集做首版 VALUE_DEVIATION 单阈值判定，
    /// 产出命中结果对象列表，供 G01-05 去重与 G01-06 建单消费。
    /// 本方法不做去重、不做建单、不做严重度重算、不做时间线。
    /// </summary>
    public async Task<List<S8WatchHitResult>> EvaluateHitsAsync(long tenantId, long factoryId)
    {
        var executionRules = await LoadExecutionRulesAsync(tenantId, factoryId);
        var ruleMap = executionRules.ToDictionary(x => x.WatchRuleId);
        var queryResults = new List<S8WatchDeviceQueryResult>();
        foreach (var rule in executionRules)
            queryResults.Add(await QueryDeviceRowsAsync(rule));

        var hits = new List<S8WatchHitResult>();
        foreach (var queryResult in queryResults)
        {
            // G01-03 查询失败：跳过，不进入判定。
            if (!queryResult.Success) continue;

            if (!ruleMap.TryGetValue(queryResult.WatchRuleId, out var rule)) continue;

            // 判定参数缺失：跳过当前规则。
            if (string.IsNullOrWhiteSpace(rule.TriggerCondition)
                || string.IsNullOrWhiteSpace(rule.ThresholdValue))
                continue;

            // 比较符非法：跳过当前规则。
            var op = TryParseTriggerCondition(rule.TriggerCondition);
            if (op == null) continue;

            // ThresholdValue 非法：跳过当前规则。
            if (!TryParseDecimal(rule.ThresholdValue, out var threshold)) continue;

            foreach (var row in queryResult.Rows)
            {
                // CurrentValue 非法（null / 无法解析数值）：跳过当前行，不进入判定。
                if (row.CurrentValue == null) continue;

                // 未命中：不进入后续链路。
                if (!EvaluateHit(row.CurrentValue.Value, op, threshold)) continue;

                hits.Add(new S8WatchHitResult
                {
                    SourceRuleId = rule.WatchRuleId,
                    SourceRuleCode = rule.WatchRuleCode,
                    AlertRuleId = rule.AlertRuleId,
                    DataSourceId = rule.DataSourceId,
                    RelatedObjectCode = row.RelatedObjectCode,
                    CurrentValue = row.CurrentValue.Value,
                    ThresholdValue = threshold,
                    TriggerCondition = op,
                    Severity = rule.Severity,
                    OccurrenceDeptId = row.OccurrenceDeptId,
                    ResponsibleDeptId = row.ResponsibleDeptId,
                    SourcePayload = row.SourcePayload
                });
            }
        }

        return hits;
    }

    /// <summary>
    /// G01-05：未闭环异常去重最小实现。
    /// 消费 G01-04 产出的 <see cref="S8WatchHitResult"/>，按 (SourceRuleId + RelatedObjectCode)
    /// 在未闭环状态集合内判重，只回答“是否允许建单”。
    /// 首版明确不做：原单刷新 / 时间线追加 / payload 更新 / 次数累计 / 严重度重算 / 状态修复。
    /// </summary>
    public async Task<List<S8WatchDedupResult>> EvaluateDedupAsync(long tenantId, long factoryId)
    {
        var hits = await EvaluateHitsAsync(tenantId, factoryId);
        var results = new List<S8WatchDedupResult>(hits.Count);
        foreach (var hit in hits)
        {
            // 防御性分支：正常情况下 SourceRuleId 与 RelatedObjectCode 已由上游
            // （G01-02 规则装配 + G01-03 查询结果列校验）保证；此处仅作兜底，
            // 不是首版正常路径。
            if (hit.SourceRuleId <= 0 || string.IsNullOrWhiteSpace(hit.RelatedObjectCode))
            {
                results.Add(new S8WatchDedupResult
                {
                    Hit = hit,
                    CanCreate = false,
                    MatchedExceptionId = null,
                    Reason = "missing_dedup_key"
                });
                continue;
            }

            long matchedId;
            try
            {
                matchedId = await FindPendingExceptionIdAsync(
                    tenantId, factoryId, hit.SourceRuleId, hit.RelatedObjectCode);
            }
            catch
            {
                // 去重查询失败：首版偏保守，宁可阻止也不重复建单，不扩展为补偿。
                results.Add(new S8WatchDedupResult
                {
                    Hit = hit,
                    CanCreate = false,
                    MatchedExceptionId = null,
                    Reason = "query_failed"
                });
                continue;
            }

            if (matchedId > 0)
            {
                // 命中已有未闭环异常 → 阻止建单。
                results.Add(new S8WatchDedupResult
                {
                    Hit = hit,
                    CanCreate = false,
                    MatchedExceptionId = matchedId,
                    Reason = "duplicate_pending"
                });
            }
            else
            {
                // 未命中 → 允许建单，交 G01-06 消费。
                results.Add(new S8WatchDedupResult
                {
                    Hit = hit,
                    CanCreate = true,
                    MatchedExceptionId = null,
                    Reason = "no_pending"
                });
            }
        }

        return results;
    }

    // 取任意一条匹配的未闭环异常 Id 作为“是否存在重复单”的拦截依据。
    // 首版只需要“存在性”，不关心“最早 / 最新”；不在 G01-05 处理排序语义。
    private async Task<long> FindPendingExceptionIdAsync(
        long tenantId, long factoryId, long sourceRuleId, string relatedObjectCode)
    {
        return await _exceptionRep.AsQueryable()
            .Where(x => x.TenantId == tenantId
                && x.FactoryId == factoryId
                && !x.IsDeleted
                && x.SourceRuleId == sourceRuleId
                && x.RelatedObjectCode == relatedObjectCode
                && UnclosedExceptionStatuses.Contains(x.Status))
            .Select(x => x.Id)
            .FirstAsync();
    }

    /// <summary>
    /// G01-06：自动建单入口。消费 G01-05 去重结果，对 CanCreate==true 的命中
    /// 复用 S8ManualReportService.CreateFromWatchAsync（同一主链的自动建单分支）落成标准 AdoS8Exception。
    /// CanCreate==false 直接跳过；创建失败返回最小失败结果，不补偿、不重试、不对账。
    /// </summary>
    public async Task<List<S8WatchCreationResult>> CreateExceptionsAsync(long tenantId, long factoryId)
    {
        var dedupResults = await EvaluateDedupAsync(tenantId, factoryId);
        var results = new List<S8WatchCreationResult>(dedupResults.Count);
        foreach (var dedup in dedupResults)
        {
            if (!dedup.CanCreate)
            {
                results.Add(new S8WatchCreationResult
                {
                    DedupResult = dedup,
                    Created = false,
                    Skipped = true,
                    CreatedExceptionId = null,
                    Reason = dedup.Reason,
                    ErrorMessage = null
                });
                continue;
            }

            try
            {
                var entity = await _manualReportService.CreateFromWatchAsync(dedup.Hit);
                results.Add(new S8WatchCreationResult
                {
                    DedupResult = dedup,
                    Created = true,
                    Skipped = false,
                    CreatedExceptionId = entity.Id,
                    Reason = "auto_created",
                    ErrorMessage = null
                });
            }
            catch (Exception ex)
            {
                results.Add(new S8WatchCreationResult
                {
                    DedupResult = dedup,
                    Created = false,
                    Skipped = false,
                    CreatedExceptionId = null,
                    Reason = "create_failed",
                    ErrorMessage = ex.Message
                });
            }
        }

        return results;
    }

    /// <summary>
    /// 单次轮询入口。当前仅完成规则读取与组装，返回可执行规则数量，不做实际数据采集。
    /// </summary>
    public async Task<int> RunOnceAsync()
    {
        var executionRules = await LoadExecutionRulesAsync(1, 1);
        return executionRules.Count;
    }

    // G01-04 首版最小比较符集合：>, >=, <, <=。
    // 允许首尾空格；非此集合的一律视为“比较符非法”，由调用方跳过。
    private static string? TryParseTriggerCondition(string raw)
    {
        var normalized = raw.Trim();
        return normalized switch
        {
            ">" => ">",
            ">=" => ">=",
            "<" => "<",
            "<=" => "<=",
            _ => null
        };
    }

    private static bool TryParseDecimal(string raw, out decimal value) =>
        decimal.TryParse(raw.Trim(), NumberStyles.Any, CultureInfo.InvariantCulture, out value);

    private static bool EvaluateHit(decimal current, string op, decimal threshold) => op switch
    {
        ">" => current > threshold,
        ">=" => current >= threshold,
        "<" => current < threshold,
        "<=" => current <= threshold,
        _ => false
    };

    private static bool IsSupportedAlertRule(AdoS8AlertRule alertRule) =>
        !string.IsNullOrWhiteSpace(alertRule.TriggerCondition)
        && !string.IsNullOrWhiteSpace(alertRule.ThresholdVal);

    private static bool IsSupportedSqlDataSource(AdoS8DataSource dataSource) =>
        dataSource.Enabled
        && string.Equals(dataSource.Type?.Trim(), SqlDataSourceType, StringComparison.OrdinalIgnoreCase)
        && !string.IsNullOrWhiteSpace(dataSource.Endpoint);

    private static bool IsDeviceWatchObjectType(string? watchObjectType)
    {
        if (string.IsNullOrWhiteSpace(watchObjectType)) return false;

        var normalized = watchObjectType.Trim().ToUpperInvariant();
        return normalized is "DEVICE" or "EQUIPMENT" || watchObjectType.Trim() == "设备";
    }

    private SqlSugarScope CreateSqlQueryScope(string connectionString)
    {
        var dbType = _ruleRep.Context.CurrentConnectionConfig.DbType;
        return new SqlSugarScope(new ConnectionConfig
        {
            ConfigId = $"s8-watch-sql-{Guid.NewGuid():N}",
            DbType = dbType,
            ConnectionString = connectionString,
            InitKeyType = InitKeyType.Attribute,
            IsAutoCloseConnection = true
        });
    }

    private static bool HasRequiredColumns(DataTable table) =>
        TryGetColumnName(table.Columns, "related_object_code") != null
        && TryGetColumnName(table.Columns, "current_value") != null;

    private static S8WatchDeviceRow MapDeviceRow(DataRow row)
    {
        var columns = row.Table.Columns;
        var relatedObjectCodeColumn = TryGetColumnName(columns, "related_object_code");
        var currentValueColumn = TryGetColumnName(columns, "current_value");
        var occurrenceDeptIdColumn = TryGetColumnName(columns, "occurrence_dept_id");
        var responsibleDeptIdColumn = TryGetColumnName(columns, "responsible_dept_id");

        return new S8WatchDeviceRow
        {
            RelatedObjectCode = ReadString(row, relatedObjectCodeColumn),
            CurrentValue = ReadDecimal(row, currentValueColumn),
            OccurrenceDeptId = ReadLong(row, occurrenceDeptIdColumn),
            ResponsibleDeptId = ReadLong(row, responsibleDeptIdColumn),
            SourcePayload = BuildSourcePayload(row)
        };
    }

    private static string? TryGetColumnName(DataColumnCollection columns, string expectedName)
    {
        var normalizedExpected = NormalizeColumnName(expectedName);
        foreach (DataColumn column in columns)
        {
            if (NormalizeColumnName(column.ColumnName) == normalizedExpected)
                return column.ColumnName;
        }

        return null;
    }

    private static string NormalizeColumnName(string columnName) =>
        columnName.Replace("_", string.Empty, StringComparison.Ordinal).Trim().ToUpperInvariant();

    private static string ReadString(DataRow row, string? columnName)
    {
        if (string.IsNullOrWhiteSpace(columnName)) return string.Empty;
        var value = row[columnName];
        return value == DBNull.Value ? string.Empty : Convert.ToString(value)?.Trim() ?? string.Empty;
    }

    private static long? ReadLong(DataRow row, string? columnName)
    {
        if (string.IsNullOrWhiteSpace(columnName)) return null;
        var value = row[columnName];
        if (value == DBNull.Value) return null;
        return long.TryParse(Convert.ToString(value, CultureInfo.InvariantCulture), out var result) ? result : null;
    }

    private static decimal? ReadDecimal(DataRow row, string? columnName)
    {
        if (string.IsNullOrWhiteSpace(columnName)) return null;
        var value = row[columnName];
        if (value == DBNull.Value) return null;
        return decimal.TryParse(Convert.ToString(value, CultureInfo.InvariantCulture), NumberStyles.Any, CultureInfo.InvariantCulture, out var result)
            ? result
            : null;
    }

    private static string BuildSourcePayload(DataRow row)
    {
        var payload = new Dictionary<string, object?>(StringComparer.OrdinalIgnoreCase);
        foreach (DataColumn column in row.Table.Columns)
        {
            var value = row[column];
            payload[column.ColumnName] = value == DBNull.Value ? null : value;
        }

        return JsonSerializer.Serialize(payload);
    }
}

public sealed class S8WatchExecutionRule
{
    public long WatchRuleId { get; set; }
    public string WatchRuleCode { get; set; } = string.Empty;
    public string SceneCode { get; set; } = string.Empty;
    public string TriggerType { get; set; } = string.Empty;
    public string WatchObjectType { get; set; } = string.Empty;
    public long DataSourceId { get; set; }
    public string DataSourceCode { get; set; } = string.Empty;
    public string DataSourceType { get; set; } = string.Empty;
    public string DataSourceConnection { get; set; } = string.Empty;
    public string QueryExpression { get; set; } = string.Empty;
    public int PollIntervalSeconds { get; set; }
    public long AlertRuleId { get; set; }
    public string AlertRuleCode { get; set; } = string.Empty;
    public string TriggerCondition { get; set; } = string.Empty;
    public string ThresholdValue { get; set; } = string.Empty;
    public string Severity { get; set; } = string.Empty;
}

public sealed class S8WatchDeviceQueryResult
{
    public long WatchRuleId { get; set; }
    public string WatchRuleCode { get; set; } = string.Empty;
    public bool Success { get; set; }
    public string? FailureReason { get; set; }
    public List<S8WatchDeviceRow> Rows { get; set; } = new();

    public static S8WatchDeviceQueryResult Ok(S8WatchExecutionRule rule, List<S8WatchDeviceRow> rows) =>
        new()
        {
            WatchRuleId = rule.WatchRuleId,
            WatchRuleCode = rule.WatchRuleCode,
            Success = true,
            Rows = rows
        };

    public static S8WatchDeviceQueryResult Fail(S8WatchExecutionRule rule, string reason) =>
        new()
        {
            WatchRuleId = rule.WatchRuleId,
            WatchRuleCode = rule.WatchRuleCode,
            Success = false,
            FailureReason = reason
        };
}

public sealed class S8WatchDeviceRow
{
    public string RelatedObjectCode { get; set; } = string.Empty;
    public decimal? CurrentValue { get; set; }
    public long? OccurrenceDeptId { get; set; }
    public long? ResponsibleDeptId { get; set; }
    public string SourcePayload { get; set; } = string.Empty;
}

/// <summary>
/// G01-05 去重结果对象。仅服务 G01-06 建单前拦截，由 CanCreate 单决策位决定是否建单。
/// 只服务首版唯一场景 S2S6_PRODUCTION + 唯一 trigger_type VALUE_DEVIATION + 设备对象。
/// 不预留多 trigger_type / 平台化去重扩展结构。
/// Reason 值域：no_pending / duplicate_pending / missing_dedup_key / query_failed。
/// </summary>
public sealed class S8WatchDedupResult
{
    public S8WatchHitResult Hit { get; set; } = new();
    public bool CanCreate { get; set; }
    public long? MatchedExceptionId { get; set; }
    public string Reason { get; set; } = string.Empty;
}

/// <summary>
/// G01-06 建单结果对象。仅服务 G-01 首版主线验收，由 Created / Skipped 两位决定结局。
/// 只服务首版唯一场景 S2S6_PRODUCTION + 唯一 trigger_type VALUE_DEVIATION + 设备对象。
/// 不预留多 trigger_type / 平台化工单扩展结构。
/// Reason 值域：auto_created / create_failed / 透传自 DedupResult.Reason。
/// </summary>
public sealed class S8WatchCreationResult
{
    public S8WatchDedupResult DedupResult { get; set; } = new();
    public bool Created { get; set; }
    public bool Skipped { get; set; }
    public long? CreatedExceptionId { get; set; }
    public string Reason { get; set; } = string.Empty;
    public string? ErrorMessage { get; set; }
}

/// <summary>
/// G01-04 命中结果对象。承载 G01-05 去重与 G01-06 建单所需最小追溯字段，
/// 仅服务首版唯一场景 S2S6_PRODUCTION + 唯一 trigger_type VALUE_DEVIATION + 设备对象。
/// 不预留多 trigger_type / 多场景 / 平台化扩展结构。
/// </summary>
public sealed class S8WatchHitResult
{
    public long SourceRuleId { get; set; }
    public string SourceRuleCode { get; set; } = string.Empty;
    public long AlertRuleId { get; set; }
    public long DataSourceId { get; set; }
    public string RelatedObjectCode { get; set; } = string.Empty;
    public decimal CurrentValue { get; set; }
    public decimal ThresholdValue { get; set; }
    public string TriggerCondition { get; set; } = string.Empty;
    public string Severity { get; set; } = string.Empty;
    public long? OccurrenceDeptId { get; set; }
    public long? ResponsibleDeptId { get; set; }
    public string SourcePayload { get; set; } = string.Empty;
}