using dodohold.core; using Google.Protobuf.WellKnownTypes; using molilian.core; using System; using System.Collections.Concurrent; using System.Linq; public partial class TkEndpointManager { // 存储每个账号的端点状态(Key: accountId) private static readonly ConcurrentDictionary accountStates = new(); // 记录每个账号最后一次挂起操作的时间 private static readonly ConcurrentDictionary lastSuspendTimes = new(); // 每个账号的挂起锁对象 private static readonly ConcurrentDictionary accountLocks = new(); // 挂起冷却时间(3秒) private static readonly TimeSpan SuspendCooldown = TimeSpan.FromSeconds(3); // 替换原来的 ConcurrentDictionary private static readonly ConcurrentDictionary accountSemaphores = new(); public static void Refresh() { accountStates.Clear(); lastSuspendTimes.Clear(); accountLocks.Clear(); accountSemaphores.Clear(); } // 从数据库获取所有可用的API端点 private static async Task> GetAllAvailableApisAsync(int accountId) { var endpoints = await TkEndpointCore.GetEndpointsByAccountAsync(accountId); return endpoints? .Where(e => e.status) .ToDictionary(e => e.ep_id, e => e.endpoint) ?? new Dictionary(); } // 从数据库获取端点的挂起时间配置 private static async Task> GetEndpointHoldMinutesAsync(int accountId) { var endpoints = await TkEndpointCore.GetEndpointsByAccountAsync(accountId); 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) { if (!accountStates.TryGetValue(accountId, out var state)) return false; return state.IsEndpointSuspended(endpoint); } // 新增:直接接收端点对象的版本 public static bool IsEndpointSuspended(TkEndpointConfigDTO endpoint) { if (endpoint == null) return true; if (!accountStates.TryGetValue(endpoint.tk_pool_id, out var state)) return false; return state.IsEndpointSuspended(endpoint.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 => !state.IsEndpointSuspended(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 (state.IsEndpointSuspended(api)) { var suspendUntil = state.GetSuspendTime(api); var remainingTime = suspendUntil - DateTime.Now; status[api] = $"挂起 (剩余时间: {remainingTime:mm\\:ss})"; } else { status[api] = "正常"; } } return status; } public static async Task<(int, string)> GetConvertApiAsync(int accountId, string parseEndpoint) { var endpoints = await TkEndpointCore.GetEndpointsByAccountAsync(accountId); if (endpoints == null || !endpoints.Any(e => e.status)) return (0, string.Empty); // 1. 过滤可用端点(状态正常) var availableApis = endpoints.Where(e => e.status); // 2. 如果指定了 parseEndpoint,按 ID 进一步过滤 if (!string.IsNullOrEmpty(parseEndpoint)) { var endpointIds = parseEndpoint.Split(',') .Select(idStr => int.TryParse(idStr.Trim(), out var id) ? id : (int?)null) .Where(id => id.HasValue) .Select(id => id.Value) .ToHashSet(); // 添加日志,看看过滤前后的端点 var beforeCount = availableApis.Count(); availableApis = availableApis.Where(e => endpointIds.Contains(e.ep_id)); var afterCount = availableApis.Count(); if (!availableApis.Any()) return (0, string.Empty); } var state = accountStates.GetOrAdd(accountId, _ => new AccountEndpointState()); var now = DateTime.Now; // 3. 先找出所有未被挂起、未达限制的端点 var availableEndpoints = availableApis .Where(e => !state.IsEndpointSuspended(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(); // 4. 如果存在可用端点,直接返回 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); } // 5. 检查是否仅因时间间隔限制 var intervalLimitedEndpoints = availableApis .Where(e => !state.IsEndpointSuspended(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); } // 6. 检查所有端点的状态 var allEndpointsSuspended = availableApis.All(e => state.IsEndpointSuspended(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"); } // 7. 其他情况(理论上不应该到达这里) 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 semaphore = accountSemaphores.GetOrAdd(accountId, _ => new SemaphoreSlim(1, 1)); await semaphore.WaitAsync(); try { // 在锁外部获取所需数据(避免在锁内await) var holdMinutes = await GetEndpointHoldMinutesAsync(accountId); // 如果是被动暂停,检查冷却时间 if (lastSuspendTimes.TryGetValue(accountId, out var lastSuspendTime) && DateTime.UtcNow - lastSuspendTime < SuspendCooldown) { return; } // 确定挂起时长 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)) }; } // 执行挂起操作 var state = accountStates.GetOrAdd(accountId, _ => new AccountEndpointState()); state.SuspendEndpoint(endpoint, suspendDuration); lastSuspendTimes[accountId] = DateTime.UtcNow; _ = new LoggerLibrary("转链接口风控", $"{accountId}").Info($"{accountId}_{endpoint}", reason.ToString()).SaveAsync(); _ = NotifyCore.NotifyAsync($"【转链接口风控】{accountId}_{endpoint} ({reason})"); } finally { semaphore.Release(); } } /// /// 主动释放指定账号的挂起状态 /// /// 账号ID /// 要释放的端点名称,如果为null则释放所有端点的挂起状态 public static void ReleaseSuspend(int accountId, string endpoint = null) { // 获取或创建账号特定的锁对象 var accountLock = accountLocks.GetOrAdd(accountId, _ => new object()); lock (accountLock) { if (accountStates.TryGetValue(accountId, out var state)) { if (endpoint == null) { // 释放所有端点的挂起状态 state.suspendedEndpoints.Clear(); } else { // 释放指定端点的挂起状态 state.suspendedEndpoints.TryRemove(endpoint, out _); } // 记录日志 var action = endpoint == null ? "释放所有挂起" : $"释放挂起({endpoint})"; _ = new LoggerLibrary("转链接口风控", $"{accountId}").Info($"{accountId}", action).SaveAsync(); _ = NotifyCore.NotifyAsync($"【转链接口风控】{accountId} {action}"); } } } private static TimeSpan CalculateHourlySuspendDuration() { var now = DateTime.UtcNow; var nextHour = now.AddHours(1).Date.AddHours(now.Hour + 1); // 下一个整点(如 14:30 → 15:00) return nextHour - now; } private static TimeSpan CalculateDailySuspendDuration() { var now = DateTime.UtcNow; var tomorrow = now.Date.AddDays(1); // 次日零点 return tomorrow - now; } public enum SuspendReason { /// /// 主动暂停(手动操作) /// Active, /// /// 风控导致的被动暂停 /// Passive, /// /// 每小时总量限制触发的暂停 /// HourlyLimit, /// /// 每日总量限制触发的暂停 /// DailyLimit } private class AccountEndpointState { internal readonly ConcurrentDictionary suspendedEndpoints = new(); // 为每个账号和 parseEndpoint 组合维护独立的索引 private readonly ConcurrentDictionary roundRobinIndices = new(); public bool IsEndpointSuspended(string endpoint) { if (!suspendedEndpoints.TryGetValue(endpoint, out var suspendUntil)) return false; if (DateTime.Now >= suspendUntil) { suspendedEndpoints.TryRemove(endpoint, out _); return false; } return true; } public DateTime? GetSuspendTime(string endpoint) { if (suspendedEndpoints.TryGetValue(endpoint, out var suspendUntil)) { return suspendUntil; } return null; } public void SuspendEndpoint(string endpoint, TimeSpan duration) { suspendedEndpoints[endpoint] = DateTime.Now.Add(duration); } public int GetOrAddAccountIndex(string key) => roundRobinIndices.GetOrAdd(key, -1); public void UpdateAccountIndex(string key, int newIndex) => roundRobinIndices[key] = newIndex; } }