using dodohold.core; using Google.Protobuf.WellKnownTypes; using molilian.core; using System; using System.Collections.Concurrent; using System.Linq; public partial class TkEndpointManager { // 只保留轮询索引 private static readonly ConcurrentDictionary accountStates = new(); // 记录每个账号最后一次挂起操作的时间 private static readonly ConcurrentDictionary lastSuspendTimes = new(); public static void Refresh() { accountStates.Clear(); lastSuspendTimes.Clear(); } // 从数据库获取所有可用的API端点 private static async Task> GetAllAvailableApisAsync(int accountId, bool? isTaobaoUrl = null, string parseEndpoint = null) { var endpoints = await TkEndpointCore.GetEndpointsByAccountAsync(accountId, isTaobaoUrl, parseEndpoint); return endpoints? .Where(e => e.status) .ToDictionary(e => e.ep_id, e => e.endpoint) ?? new Dictionary(); } // 从数据库获取端点的挂起时间配置 private static async Task> GetEndpointHoldMinutesAsync(int accountId, bool? isTaobaoUrl = null, string parseEndpoint = null) { var endpoints = await TkEndpointCore.GetEndpointsByAccountAsync(accountId, isTaobaoUrl, parseEndpoint); var dict = new ConcurrentDictionary(); if (endpoints != null) { foreach (var endpoint in endpoints) { dict.TryAdd(endpoint.endpoint, endpoint.suspend_duration); } } return dict; } public static bool IsEndpointSuspended(int accountId, string endpoint) { return RedisHelper.Exists($"tk_suspend:{accountId}:{endpoint}"); } /// /// 检查指定账号是否至少拥有一个可用的endpoint /// /// 账号ID /// true如果至少有一个可用endpoint,否则false public static async Task HasAvailableEndpointAsync(int accountId) { var availableApis = await GetAllAvailableApisAsync(accountId); if (availableApis.Count == 0) return false; if (!accountStates.TryGetValue(accountId, out var state)) { return true; // 如果账号没有状态记录,所有节点都是正常的 } return availableApis.Any(api => !IsEndpointSuspended(accountId, api.Value)); } /// /// 获取多个账号的节点状态信息 /// public static async Task>> GetStatus(IEnumerable accountIds) { var result = new Dictionary>(); foreach (var accountId in accountIds.Distinct()) { var availableApis = await GetAllAvailableApisAsync(accountId); result[accountId] = await GetStatus(accountId, availableApis); } return result; } /// /// 获取指定账号的节点状态信息 /// public static async Task> GetStatus(int accountId, Dictionary availableApis) { var status = new Dictionary(); if (availableApis.Count == 0) { return status; } if (!accountStates.TryGetValue(accountId, out var state)) { // 如果账号没有状态记录,所有节点都是正常的 foreach ((var api_id, var api) in availableApis) { status[api] = "正常"; } return status; } foreach ((var api_id, var api) in availableApis) { if (IsEndpointSuspended(accountId, api)) { // Redis没有挂起到期时间,展示"挂起"即可 status[api] = $"挂起"; } else { status[api] = "正常"; } } return status; } public static async Task<(int, string)> GetConvertApiAsync(int accountId, bool isTaobaoUrl, string parseEndpoint) { var endpoints = await TkEndpointCore.GetEndpointsByAccountAsync(accountId, isTaobaoUrl, parseEndpoint); if (endpoints == null || !endpoints.Any(e => e.status)) return (0, string.Empty); // 1. 过滤可用端点(状态正常) var availableApis = endpoints.Where(e => e.status); var state = accountStates.GetOrAdd(accountId, _ => new AccountEndpointState()); var now = DateTime.Now; // 2. 先找出所有未被挂起、未达限制的端点 var availableEndpoints = availableApis .Where(e => !IsEndpointSuspended(accountId, e.endpoint)) .Where(e => e.hourly_calls_limit <= 0 || e.current_hourly_calls < e.hourly_calls_limit) .Where(e => e.daily_calls_limit <= 0 || e.current_daily_calls < e.daily_calls_limit) .ToArray(); // 3. 如果存在可用端点,直接返回 if (availableEndpoints.Length > 0) { // 生成唯一的轮询key,确保每种可用端点组合都有独立索引 string roundRobinKey = $"{accountId}:{string.Join(",", availableEndpoints.Select(e => e.ep_id).OrderBy(x => x))}"; var currentIndex = state.GetOrAddAccountIndex(roundRobinKey); var newIndex = (currentIndex + 1) % availableEndpoints.Length; state.UpdateAccountIndex(roundRobinKey, newIndex); var selectedEndpoint = availableEndpoints[newIndex]; // 异步更新调用统计(不阻塞当前请求) _ = UpdateEndpointStatsAsync(accountId, selectedEndpoint, now); return (selectedEndpoint.ep_id, selectedEndpoint.endpoint); } // 4. 检查是否仅因时间间隔限制 var intervalLimitedEndpoints = availableApis .Where(e => !IsEndpointSuspended(accountId, e.endpoint)) .Where(e => e.hourly_calls_limit <= 0 || e.current_hourly_calls < e.hourly_calls_limit) .Where(e => e.daily_calls_limit <= 0 || e.current_daily_calls < e.daily_calls_limit) .Where(e => e.interval_seconds > 0 && e.last_call_time != null && (now - e.last_call_time).TotalSeconds < e.interval_seconds) .ToArray(); if (intervalLimitedEndpoints.Length > 0) { // 如果是时间间隔限制,返回空字符串 return (0, string.Empty); } // 5. 检查所有端点的状态 var allEndpointsSuspended = availableApis.All(e => IsEndpointSuspended(accountId, e.endpoint)); var allEndpointsLimited = availableApis.All(e => (e.hourly_calls_limit > 0 && e.current_hourly_calls >= e.hourly_calls_limit) || (e.daily_calls_limit > 0 && e.current_daily_calls >= e.daily_calls_limit)); // 如果所有端点都被挂起或达到限制,返回 "ALL" if (allEndpointsSuspended || allEndpointsLimited) { return (0, "ALL"); } // 6. 其他情况(理论上不应该到达这里) return (0, string.Empty); } // 辅助方法:异步更新调用统计(原逻辑,仅拆分以保持清晰) private static async Task UpdateEndpointStatsAsync(int accountId, TkEndpointConfigDTO selectedEndpoint, DateTime now) { var currentHour = now.ToString("yyyyMMddHH"); var currentDay = now.ToString("yyyyMMdd"); // 更新调用计数 selectedEndpoint.current_hourly_calls++; selectedEndpoint.current_daily_calls++; selectedEndpoint.last_call_time = now; // 保存到 Redis await RiskControlCore.SetTkEndpointCallsAsync(accountId, selectedEndpoint.id, currentHour, selectedEndpoint.current_hourly_calls); await RiskControlCore.SetTkEndpointCallsAsync(accountId, selectedEndpoint.id, currentDay, selectedEndpoint.current_daily_calls); // 检查是否需要挂起(原逻辑) if (selectedEndpoint.hourly_calls_limit > 0 && selectedEndpoint.current_hourly_calls >= selectedEndpoint.hourly_calls_limit) { _ = SuspendAsync(accountId, selectedEndpoint.endpoint, SuspendReason.HourlyLimit, selectedEndpoint.suspend_duration); } if (selectedEndpoint.daily_calls_limit > 0 && selectedEndpoint.current_daily_calls >= selectedEndpoint.daily_calls_limit) { _ = SuspendAsync(accountId, selectedEndpoint.endpoint, SuspendReason.DailyLimit, selectedEndpoint.suspend_duration); } } public static async Task SuspendAsync(int accountId, string endpoint, SuspendReason reason = SuspendReason.Passive, int? customHoldMinutes = null) { var holdMinutes = await GetEndpointHoldMinutesAsync(accountId); TimeSpan suspendDuration; if (customHoldMinutes.HasValue) { suspendDuration = TimeSpan.FromMinutes(customHoldMinutes.Value); } else { suspendDuration = reason switch { SuspendReason.HourlyLimit => CalculateHourlySuspendDuration(), SuspendReason.DailyLimit => CalculateDailySuspendDuration(), _ => TimeSpan.FromMinutes(holdMinutes.GetValueOrDefault(endpoint, 240)) }; } RedisHelper.Set($"tk_suspend:{accountId}:{endpoint}", 1, (int)suspendDuration.TotalSeconds); _ = new LoggerLibrary("转链接口风控", accountId.ToString()).Info($"{accountId}_{endpoint}", reason.ToString()).SaveAsync(); _ = NotifyCore.NotifyAsync($"【转链接口风控】{accountId}_{endpoint} ({reason})"); } /// /// 主动释放指定账号的挂起状态 /// /// 账号ID /// 要释放的端点名称,如果为null则释放所有端点的挂起状态 public static void ReleaseSuspend(int accountId, string endpoint = null) { if (endpoint == null) { var pattern = $"tk_suspend:{accountId}:*"; var keys = RedisHelper.Keys(pattern); foreach (var key in keys) { RedisHelper.Del(key); } } else { RedisHelper.Del($"tk_suspend:{accountId}:{endpoint}"); } var action = endpoint == null ? "释放所有挂起" : $"释放挂起({endpoint})"; _ = new LoggerLibrary("转链接口风控", accountId.ToString()).Info($"{accountId}", action).SaveAsync(); _ = NotifyCore.NotifyAsync($"【转链接口风控】{accountId} {action}"); } private static TimeSpan CalculateHourlySuspendDuration() { var now = DateTime.Now; var nextHour = now.AddHours(1).Date.AddHours(now.Hour + 1); // 下一个整点(如 14:30 → 15:00) return nextHour - now; } private static TimeSpan CalculateDailySuspendDuration() { var now = DateTime.Now; var tomorrow = now.Date.AddDays(1); // 次日零点 return tomorrow - now; } public enum SuspendReason { /// /// 主动暂停(手动操作) /// Active, /// /// 风控导致的被动暂停 /// Passive, /// /// 每小时总量限制触发的暂停 /// HourlyLimit, /// /// 每日总量限制触发的暂停 /// DailyLimit } private class AccountEndpointState { // 只保留轮询索引 private readonly ConcurrentDictionary roundRobinIndices = new(); public int GetOrAddAccountIndex(string key) => roundRobinIndices.GetOrAdd(key, -1); public void UpdateAccountIndex(string key, int newIndex) => roundRobinIndices[key] = newIndex; } }