dotnet/molilian.core/Core/taoke/TkPoolCore.cs

using dodohold.core;
using Org.BouncyCastle.Bcpg.OpenPgp;
using YunhuiKit;


namespace molilian.core
{
    public partial class TkPoolCore
    {
        public enum TkAction
        {
            all,
            parse,
            coupon,
            promotionQuery,
            activity
        }

        private static string _end_point;
        static TkPoolCore()
        {
            _end_point = Environment.GetEnvironmentVariable("EndPoint");
        }

        private static SemaphoreSlim _semaphore = new SemaphoreSlim(1, 1);

        private static IEnumerable<TkPoolDTO> _cached;
        private static IEnumerable<TkPoolDTO> _all_cached;

        public static IEnumerable<TkPoolDTO> GetCachedAccounts()
        {
            return _all_cached ?? _cached ?? Enumerable.Empty<TkPoolDTO>();
        }

        public static async Task<TkPoolDTO> GetOneAsync(int id)
        {
            var list = await ListAsync();
            if (!list.Any()) return null;
            return list.Where(e => e.id == id).FirstOrDefault();
        }
        public static async Task<TkPoolDTO> ALLGetOneAsync(int id)
        {
            var list = await AllListAsync();
            if (!list.Any()) return null;
            return list.Where(e => e.id == id).FirstOrDefault();
        }

        public static async Task<TkPoolDTO> GetOneAsync2(TkAction action)
        {
            var list = await ListAsync().ConfigureAwait(false);
            if (!list.Any()) return null;

            // 线程安全的随机排序
            var random = new Random(Guid.NewGuid().GetHashCode());
            var shuffled = list.OrderBy(_ => random.Next()).ToList();

            foreach (var item in shuffled)
            {
                if (await FilterNodesAsync(item, action).ConfigureAwait(false))
                    return item;
            }
            return null;
        }
        public static async Task<TkPoolDTO> GetOneAsync(TkAction action, bool isTaobaoUrl, string riskStrategy, int launchScene, string parse_type = "", string strategy_id = "")
        {
            var list = await ListAsync().ConfigureAwait(false);
            if (!list.Any())
                return null;

            // 获取所有账号（包括离线账号）
            var allAccounts = await AllListAsync().ConfigureAwait(false);
            if (allAccounts == null)
                return null;

            if (!string.IsNullOrEmpty(strategy_id))
            {
                list = list.Where(item => strategy_id.Equals(item.strategy_id)).ToList();
            }
            else
            {
                list = list.Where(item => string.IsNullOrEmpty(item.strategy_id)).ToList();
            }
            if (!list.Any()) return null;


            if ("dp".Equals(parse_type))
            {
                list = list.Where(item => "dp".Equals(item.parse_type)).ToList();
                if (!list.Any()) return null;
            }
            else if ("ym".Equals(parse_type))
            {
                list = list.Where(item => "ym".Equals(item.parse_type)).ToList();
                if (!list.Any()) return null;
            }
            else
            {
                list = list.Where(item => string.IsNullOrEmpty(item.parse_type)).ToList();
            }
            if (!list.Any()) return null;


            if (!string.IsNullOrEmpty(riskStrategy) && !"os".Equals(riskStrategy) && launchScene != -1)
            {
                list = list.Where(item => riskStrategy.Equals(item.riskStrategy) && item.launchScene == launchScene).ToList();
            }
            else
            {
                list = list.Where(item => string.IsNullOrEmpty(item.riskStrategy)).ToList();
            }
            if (!list.Any()) return null;

            list = FilterByAction(list, action).ToList();
            if (!list.Any()) return null;

            list = list.Where(item => PassesFastAccountChecks(item, action)).ToList();
            if (!list.Any()) return null;

            var ownerByRelatedId = BuildOwnerByRelatedAccountId(allAccounts);
            var onlineAccountIds = list.Select(a => a.id).ToHashSet();
            var filteredList = list
                .Where(item =>
                    !ownerByRelatedId.TryGetValue(item.id, out var ownerAccount) ||
                    onlineAccountIds.Contains(ownerAccount.id) ||
                    !ownerAccount.enable_fake_click)
                .ToList();

            if (filteredList.Count == 0) return null;

            // 1. 筛选出有可用端点的账号池，并计算权重
            var weightedAccounts = new List<(TkPoolDTO account, int weight)>();
            var random = new Random(Guid.NewGuid().GetHashCode()); // 避免重复种子问题

            foreach (var item in filteredList)
            {
                // 获取账号关联的所有端点配置
                var endpoints = await TkEndpointCore.GetEndpointsByAccountReadonlyAsync(item.id, isTaobaoUrl, item.parseEndpoint).ConfigureAwait(false);
                if (endpoints == null || endpoints.Count == 0)
                    continue;

                // 过滤可用端点（状态正常且未达限制）
                var availableEndpoints = endpoints
                    .Where(e => e.status)
                    .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 => !TkEndpointManager.IsEndpointSuspended(item.id, e.endpoint)) // 直接检查端点对象
                    .ToList();

                if (availableEndpoints.Count == 0) continue;

                // 计算账号权重（基于剩余调用量）
                int weight = availableEndpoints.Sum(e =>
                    e.hourly_calls_limit <= 0 ? 3600 : e.hourly_calls_limit - e.current_hourly_calls);

                weightedAccounts.Add((item, weight));
            }

            if (weightedAccounts.Count == 0) return null;

            // 2. 保留原有加权随机选择，选中账号超出 QPS 时继续尝试其他账号
            int totalWeight = weightedAccounts.Sum(x => x.weight);
            int randomNumber = random.Next(0, totalWeight);

            foreach (var (account, weight) in weightedAccounts)
            {
                if (randomNumber < weight)
                {
                    if (await FilterNodesAsync(account, action).ConfigureAwait(false) && TryAcquireQps(account))
                        return account;
                    break;
                }
                randomNumber -= weight;
            }

            // 3. 选中账号不可用或超过 QPS 时，回退尝试其他候选账号
            var fallbackCandidates = weightedAccounts
                .OrderBy(_ => random.Next())
                .Select(x => x.account);

            foreach (var account in fallbackCandidates)
            {
                if (await FilterNodesAsync(account, action).ConfigureAwait(false) && TryAcquireQps(account))
                    return account;
            }
            return null;
        }

        private static bool TryAcquireQps(TkPoolDTO account)
        {
            if (account == null) return false;

            return SpecialBusinessRateLimiter.TryAcquire(
                $"tk_pool_qps:{account.id}",
                account.qps_limit,
                TimeSpan.FromSeconds(1));
        }

        private static async Task<bool> FilterNodesAsync(TkPoolDTO item, TkAction action)
        {
            if (!PassesFastAccountChecks(item, action)) return false;

            if (item.daily_calls_limit > 0)
            {
                int daily_num = RiskControlCore.GetCalls(TkChannelEnum.tb, item.id, DateTime.Now.ToString("yyyyMMdd"));
                if (daily_num >= item.daily_calls_limit) return false;
            }
            if (item.hourly_calls_limit > 0)
            {
                int hourly_num = RiskControlCore.GetCalls(TkChannelEnum.tb, item.id, DateTime.Now.ToString("yyyyMMddHH"));
                if (hourly_num >= item.hourly_calls_limit) return false;
            }
            var endpointCheck = await TkEndpointManager.HasAvailableEndpointAsync(item.id).ConfigureAwait(false);

            if (!endpointCheck) return false;


            switch (action)
            {
                case TkAction.parse:
                    if (!item.enable_parse) return false;
                    break;
                case TkAction.promotionQuery:
                    if (!item.enable_promotionQuery) return false;
                    break;
                case TkAction.coupon:
                case TkAction.activity:
                    if (!item.enable_coupon) return false;
                    break;
            }

            if (item.daily_income_limit == 0) return true;
            decimal income_amt = RiskControlCore.GetIncomeAmt(TkChannelEnum.tb, $"{item.id}");
            return income_amt < item.daily_income_limit;
        }

        private static IEnumerable<TkPoolDTO> FilterByAction(IEnumerable<TkPoolDTO> accounts, TkAction action)
        {
            return action switch
            {
                TkAction.parse => accounts.Where(item => item.enable_parse),
                TkAction.promotionQuery => accounts.Where(item => item.enable_promotionQuery),
                TkAction.coupon => accounts.Where(item => item.enable_coupon),
                TkAction.activity => accounts.Where(item => item.enable_coupon),
                _ => accounts
            };
        }

        private static bool PassesFastAccountChecks(TkPoolDTO item, TkAction action)
        {
            if (!string.IsNullOrEmpty(item.suspended_endpoint) && item.suspended_endpoint.Contains($"{_end_point}|")) return false;

            if (!string.IsNullOrEmpty(_end_point) && !string.IsNullOrEmpty(item.end_point) && item.end_point != _end_point)
                return false;

            if (!new WorkSchedule(item.time_range).IsWorkHour()) return false;

            return action switch
            {
                TkAction.parse => item.enable_parse,
                TkAction.promotionQuery => item.enable_promotionQuery,
                TkAction.coupon => item.enable_coupon,
                TkAction.activity => item.enable_coupon,
                _ => true
            };
        }

        private static Dictionary<int, TkPoolDTO> BuildOwnerByRelatedAccountId(IEnumerable<TkPoolDTO> allAccounts)
        {
            var result = new Dictionary<int, TkPoolDTO>();
            foreach (var account in allAccounts)
            {
                if (string.IsNullOrWhiteSpace(account.related_account_ids)) continue;

                foreach (var idStr in account.related_account_ids.Split(',', StringSplitOptions.RemoveEmptyEntries | StringSplitOptions.TrimEntries))
                {
                    if (!int.TryParse(idStr, out var relatedId)) continue;
                    result.TryAdd(relatedId, account);
                }
            }

            return result;
        }


        public static async Task<IEnumerable<TkPoolDTO>> AllListAsync(bool force = false)
        {
            if (!force && _all_cached != default) return _all_cached;

            try
            {
                await _semaphore.WaitAsync();

                string cache_key = $"cache:all_tk_pool";
                var list = await RedisKit.GetAsync<IEnumerable<TkPoolDTO>>(cache_key);

                if (force || list == default)
                {
                    list = new DBContext.Table("tk_pool").Select<TkPoolDTO>();
                    if (list == null) return default;

                    // 等待 Redis 写入完成
                    await RedisKit.SetAsync(cache_key, list, 30 * 86400);
                }

                _all_cached = list;
                return list;
            }
            catch (Exception)
            {
                // 发生异常时返回上一次的缓存，如果没有则返回默认值
                return _all_cached ?? default;
            }
            finally
            {
                _semaphore.Release();
            }
        }

        public static async Task<IEnumerable<TkPoolDTO>> ListAsync(bool force = false)
        {
            // 内存缓存检查
            if (!force && _cached != null) return _cached;
            string cache_key = "cache:tk_pool";

            if (!force)
            {
                var cachedList = await RedisHelper.GetAsync<IEnumerable<TkPoolDTO>>(cache_key);
                if (cachedList != null)
                {
                    _cached = cachedList;
                    return cachedList;
                }
            }
            // 获取新数据
            await _semaphore.WaitAsync();
            try
            {
                // 双重检查，防止并发情况下重复加载
                if (!force)
                {
                    var cachedList = await RedisHelper.GetAsync<IEnumerable<TkPoolDTO>>(cache_key);
                    if (cachedList != null)
                    {
                        _cached = cachedList;
                        return cachedList;
                    }
                }

                // 从数据库加载数据
                var list = new DBContext.Table("tk_pool")
                    .Where("status=@status", new { status = 1 })
                    .Select<TkPoolDTO>();

                if (list == null) return default;

                // 更新调用次数
                foreach (var item in list)
                {
                    _ = RiskControlCore.SetCallsAsync(TkChannelEnum.tb, item.id, DateTime.Now.ToString("yyyyMMddHH"), item.current_hourly_calls);
                    _ = RiskControlCore.SetCallsAsync(TkChannelEnum.tb, item.id, DateTime.Now.ToString("yyyyMMdd"), item.current_daily_calls);
                }

                // 更新缓存
                await RedisHelper.SetAsync(cache_key, list, 30 * 86400);
                _cached = list;
                return list;
            }
            finally
            {
                _semaphore.Release();
            }
        }

        public static void Refresh()
        {
            _ = ListAsync(true);
            _ = AllListAsync(true);
            TkEndpointCore.Refresh();
            TkEndpointManager.Refresh();
        }
        public static void UpdateDrawBalance(string name, decimal amout)
        {
            new DBContext.Table("tk_pool")
                .Add("draw_balance", amout)
                .Where("name=@name", new { name })
                .Update();
            _ = ListAsync(true);
        }

        public static void Suspend(string endpoint, int accountId, string name, string content)
        {
            string cache_key = $"cache:tk_pool:{accountId}:suspend";
            long count = RedisHelper.IncrBy(cache_key);
            RedisHelper.Expire(cache_key, 10);
            if (count > 1) return;

            var update = new DBContext.Table("tk_pool").Add("suspended_endpoint:", $"CONCAT(suspended_endpoint, '{endpoint},')");

            if (accountId > 0)
            {
                update.Where("id=@accountId", new { accountId }).Update();
            }
            else
            {
                update.Where("name=@name", new { name }).Update();
            }
            _ = ListAsync(true);
            NotifyCore.Notify(new NifyMessage
            {
                message = $"【淘客{accountId}:{name}】{endpoint} 暂停",
                priority = NifyMessagePriority.high,
                tags = ["red_circle"]
            });

            _ = EndPointCore.NotifyReload(true);
        }

        public static void Disabled(int accountId, string name, string content, bool is_hide)
        {
#if DEBUG
#else
#endif
            string cache_key = $"cache:tk_pool:{accountId}:disabled";
            long count = RedisHelper.IncrBy(cache_key);
            RedisHelper.Expire(cache_key, 10);
            if (count > 1) return;

            var update = new DBContext.Table("tk_pool").Add("status", 0);
            if (accountId > 0)
            {
                update.Where("id=@accountId", new { accountId }).Update();
            }
            else
            {
                update.Where("name=@name", new { name }).Update();
            }

            _ = ListAsync(true);
            _ = AllListAsync(true);

            NotifyCore.Notify(new NifyMessage
            {
                message = $"【淘客{accountId}:{name}】cookie 掉线\n\n{content}",
                priority = NifyMessagePriority.high,
                tags = ["red_circle"]
            });

            if (!is_hide)
            {
                _ = NotifyCore.QYWeixinPushNotifyAsync("掉线", $"【淘客{accountId}:{name}】cookie 掉线");
            }
            _ = EndPointCore.NotifyReload(true);
        }

        public static async Task<int> UpdateCookies(string cookies, string user_agent)
        {
            if (string.IsNullOrEmpty(cookies)) return 0;
            cookies += ";";
            string dnk = cookies.GetContentPart("dnk=", ";");
            string company = dnk;
            int accountId = 0;
            string tb_token = cookies.GetContentPart("_tb_token_=", ";");
            if (string.IsNullOrEmpty(dnk) || string.IsNullOrEmpty(tb_token)) return 0;
            var exist = new DBContext.Table("tk_pool").Fields("id, name, company, refpid, status").Get<dynamic>("name=@dnk", new { dnk });
            if (exist != null)
            {
                int status = exist.status;
                string refpid = exist.refpid;
                company = exist.company;
                accountId = exist.id;
                if (!string.IsNullOrEmpty(refpid)) status = 1;
                new DBContext.Table("tk_pool")
                   .Add("name", dnk)
                   .Add("tb_token", tb_token)
                   .Add("cookies", cookies)
                   .Add("user_agent", user_agent)
                   .Add("status", status)
                   .Add("suspended_endpoint", string.Empty)
                   .Add("last_time", DateTime.Now)
                   .Add("login_time", DateTime.Now)
                   .Where("id=@id", new { exist.id })
                   .Update();
                RiskControlCore.SetRiskCookie($"tb:{exist.id}", "");
                if (status == 1) _ = ListAsync(true);
            }
            else
            {
                accountId = new DBContext.Table("tk_pool")
                    .Add("name", dnk)
                    .Add("description", "由cookies上报创建此记录")
                    .Add("tb_token", tb_token)
                    .Add("refpid", string.Empty)
                    .Add("cookies", cookies)
                    .Add("user_agent", user_agent)
                    .Add("create_time", DateTime.Now)
                    .Add("last_time", DateTime.Now)
                    .Add("login_time", DateTime.Now)
                    .Add("status", 0)
                    .Create();
                RiskControlCore.SetRiskCookie($"tb:{accountId}", "");
            }

            NotifyCore.Notify(new NifyMessage
            {
                message = $"【淘客{accountId}:{company}】cookie 上线",
                tags = ["green_circle"]
            });
            _ = EndPointCore.NotifyReload(true);
            //NotifyCore.QYWeixinPushNotifyAsync("上线", $"【淘宝联盟:{dnk}】cookie 上报更新");
            return accountId;
        }

        internal static void AccountExhausted()
        {
            string cache_key = $"cache:tk_pool:account:exhausted";
            long count = RedisHelper.IncrBy(cache_key);
            if (count > 1) return;
            RedisHelper.Expire(cache_key, 3600);
            NotifyCore.Notify(new NifyMessage
            {
                message = $"【淘客】没有匹配账号",
                priority = NifyMessagePriority.high,
                tags = ["red_circle"]
            });
            _ = NotifyCore.QYWeixinPushNotifyAsync("没账号", $"【淘客】没有匹配账号");
        }
    }

}