""" 防封模块 — 三级频率限制器 L1 全局限流 / L2 设备限流 / L3 动作限流（按平台+操作类型） 所有自动化操作必须经过本限流器检查，超频则等待或拒绝。 依赖 Redis；Redis 不可用时退化为内存字典（重启丢失）。 """ import asyncio import logging import random import time from typing import Dict, Optional, Tuple logger = logging.getLogger(__name__) _redis = None _redis_last_check: float = 0.0 _REDIS_CHECK_INTERVAL = 60.0 async def _get_redis(): global _redis, _redis_last_check now = time.time() if _redis is not None: if now - _redis_last_check > _REDIS_CHECK_INTERVAL: try: await _redis.ping() _redis_last_check = now except Exception: logger.warning("Redis 连接失效，重置并退化内存模式") _redis = None if _redis is not None: return _redis try: import redis.asyncio as aioredis from config import settings _redis = aioredis.from_url(settings.REDIS_URL, decode_responses=True) await _redis.ping() _redis_last_check = now return _redis except Exception as e: logger.warning(f"Redis 不可用，退化为内存限流: {e}") _redis = None return None _memory_store: Dict[str, float] = {} _memory_daily: Dict[str, int] = {} _daily_date: str = "" def _today() -> str: return time.strftime("%Y-%m-%d") PLATFORM_LIMITS: Dict[str, Dict[str, Dict]] = { "wechat": { "add_friend": {"interval": (180, 300), "daily_max": 50, "new_daily_max": 10}, "send_message": {"interval": (30, 60), "daily_max": 200, "new_daily_max": 20}, "post_moments": {"interval": (7200, 10800), "daily_max": 5, "new_daily_max": 2}, "batch_send": {"interval": (300, 600), "daily_max": 50, "new_daily_max": 5}, "like_moments": {"interval": (15, 45), "daily_max": 100, "new_daily_max": 20}, "comment_moments": {"interval": (60, 180), "daily_max": 50, "new_daily_max": 10}, "send_group_message": {"interval": (60, 120), "daily_max": 100, "new_daily_max": 15}, }, "douyin": { "send_message": {"interval": (60, 120), "daily_max": 100, "new_daily_max": 15}, "follow": {"interval": (30, 60), "daily_max": 200, "new_daily_max": 30}, "like": {"interval": (5, 15), "daily_max": 500, "new_daily_max": 100}, "comment": {"interval": (60, 180), "daily_max": 50, "new_daily_max": 10}, }, "xhs": { "send_message": {"interval": (60, 180), "daily_max": 50, "new_daily_max": 10}, "like": {"interval": (15, 30), "daily_max": 300, "new_daily_max": 50}, "collect": {"interval": (30, 60), "daily_max": 200, "new_daily_max": 30}, "comment": {"interval": (60, 180), "daily_max": 30, "new_daily_max": 5}, }, "xianyu": { "send_message": {"interval": (60, 120), "daily_max": 80, "new_daily_max": 15}, }, "soul": { "send_message": {"interval": (30, 90), "daily_max": 100, "new_daily_max": 20}, }, } L1_GLOBAL_RPM = 100 L2_DEVICE_RPM = 10 OPERATION_HOURS = (7, 23) class DailyLimitExceeded(Exception): """日操作上限已达""" pass class OutsideOperationHours(Exception): """非操作时段""" pass class RateLimitWaiting(Exception): """需要等待（信息性，不应到达调用方）""" pass class AntiDetectRateLimiter: """三级防封限流器（单例）""" def get_limit(self, platform: str, action: str) -> Optional[Dict]: plat = PLATFORM_LIMITS.get(platform) if not plat: return None for key in (action, action.replace("-", "_")): if key in plat: return plat[key] return None def check_operation_hours(self) -> bool: """检查当前时间是否在允许操作时段内（默认 7:00-23:00）""" import datetime hour = datetime.datetime.now().hour return OPERATION_HOURS[0] <= hour < OPERATION_HOURS[1] async def _check_l1_global(self) -> bool: """L1 全局限流：所有设备总请求 <= L1_GLOBAL_RPM/分钟""" now = time.time() window_key = f"rl:l1:global:{int(now // 60)}" r = await _get_redis() if r: try: count = await r.incr(window_key) if count == 1: await r.expire(window_key, 120) if count > L1_GLOBAL_RPM: logger.warning(f"[RateLimiter] L1 全局限流触发: {count}/{L1_GLOBAL_RPM} RPM") return False return True except Exception: pass win_key = f"l1:{int(now // 60)}" _memory_daily[win_key] = _memory_daily.get(win_key, 0) + 1 return _memory_daily[win_key] <= L1_GLOBAL_RPM async def _check_l2_device(self, device_id: str) -> bool: """L2 设备限流：单设备总请求 <= L2_DEVICE_RPM/分钟""" now = time.time() window_key = f"rl:l2:{device_id}:{int(now // 60)}" r = await _get_redis() if r: try: count = await r.incr(window_key) if count == 1: await r.expire(window_key, 120) if count > L2_DEVICE_RPM: logger.warning(f"[RateLimiter] L2 设备限流触发 {device_id}: {count}/{L2_DEVICE_RPM} RPM") return False return True except Exception: pass win_key = f"l2:{device_id}:{int(now // 60)}" _memory_daily[win_key] = _memory_daily.get(win_key, 0) + 1 return _memory_daily[win_key] <= L2_DEVICE_RPM async def check_and_wait( self, device_id: str, platform: str, action: str, is_new_account: bool = False, ) -> float: """ 三级检查：L1全局 → L2设备 → L3动作 + 时段检查。 返回实际等待的秒数。 若日上限已到，抛 DailyLimitExceeded。 """ if not self.check_operation_hours(): raise OutsideOperationHours( f"当前不在操作时段 ({OPERATION_HOURS[0]}:00-{OPERATION_HOURS[1]}:00)" ) if not await self._check_l1_global(): wait = random.uniform(3.0, 8.0) logger.info(f"[RateLimiter] L1 全局超限，等待 {wait:.1f}s") await asyncio.sleep(wait) if not await self._check_l2_device(device_id): wait = random.uniform(5.0, 15.0) logger.info(f"[RateLimiter] L2 设备超限 {device_id}，等待 {wait:.1f}s") await asyncio.sleep(wait) limit = self.get_limit(platform, action) if limit is None: await asyncio.sleep(random.uniform(0.5, 2.0)) return 0.0 daily_max = limit.get("new_daily_max" if is_new_account else "daily_max", 9999) daily_count = await self._get_daily_count(device_id, platform, action) if daily_count >= daily_max: raise DailyLimitExceeded( f"{platform}.{action} 已达日上限 {daily_max}（当前 {daily_count}）" ) interval_range: Tuple[float, float] = limit["interval"] min_wait = random.uniform(*interval_range) last_ts = await self._get_last_ts(device_id, platform, action) elapsed = time.time() - last_ts if last_ts else float("inf") wait = max(0.0, min_wait - elapsed) if wait > 0: jitter = random.uniform(-wait * 0.1, wait * 0.15) wait = max(0.3, wait + jitter) logger.info( f"[RateLimiter] {device_id}/{platform}.{action} 等待 {wait:.1f}s " f"(interval={interval_range}, daily={daily_count}/{daily_max})" ) await asyncio.sleep(wait) await self._record(device_id, platform, action) return wait async def get_daily_count(self, device_id: str, platform: str, action: str) -> int: return await self._get_daily_count(device_id, platform, action) def get_platform_config(self, platform: str, action: str) -> Optional[Dict]: """获取某平台某动作的限流配置（供风控看板展示）""" return self.get_limit(platform, action) # ---- 存储层 ---- async def _get_last_ts(self, device_id: str, platform: str, action: str) -> float: key = f"rl:ts:{device_id}:{platform}:{action}" r = await _get_redis() if r: try: val = await r.get(key) return float(val) if val else 0.0 except Exception: pass return _memory_store.get(key, 0.0) async def _get_daily_count(self, device_id: str, platform: str, action: str) -> int: global _daily_date today = _today() key = f"rl:daily:{today}:{device_id}:{platform}:{action}" r = await _get_redis() if r: try: val = await r.get(key) return int(val) if val else 0 except Exception: pass if _daily_date != today: _memory_daily.clear() _daily_date = today return _memory_daily.get(key, 0) async def _record(self, device_id: str, platform: str, action: str): now = time.time() today = _today() ts_key = f"rl:ts:{device_id}:{platform}:{action}" daily_key = f"rl:daily:{today}:{device_id}:{platform}:{action}" r = await _get_redis() if r: try: pipe = r.pipeline() pipe.set(ts_key, str(now), ex=86400) pipe.incr(daily_key) pipe.expire(daily_key, 86400) await pipe.execute() return except Exception: pass _memory_store[ts_key] = now global _daily_date if _daily_date != today: _memory_daily.clear() _daily_date = today _memory_daily[daily_key] = _memory_daily.get(daily_key, 0) + 1 rate_limiter = AntiDetectRateLimiter()