"""代理IP白名单管理模块""" import json import re import threading import time from typing import Optional from urllib.parse import urlencode import requests from loguru import logger # 全局锁：防止多个并发登录任务同时同步白名单，触发代理服务商的30秒限流 _whitelist_sync_lock = threading.Lock() class WhitelistManager: """协固代理IP白名单管理器""" MEMO_PREFIX = "douyu_auto" BASE_URL = "http://op.xiequ.cn/IpWhiteList.aspx" def __init__(self, uid: str, ukey: str): self.uid = uid self.ukey = ukey self._memo = self.MEMO_PREFIX @property def memo(self) -> str: """当前机器的固定备注""" return self._memo def _build_url(self, **params) -> str: """构建请求URL""" base_params = { "uid": self.uid, "ukey": self.ukey, } base_params.update(params) query = urlencode(base_params) return f"{self.BASE_URL}?{query}" def get_whitelist_json(self) -> list[dict]: """ 获取白名单列表（JSON格式） Returns: [{"IP": "x.x.x.x", "MEMO": "备注"}, ...] """ try: url = self._build_url(act="getjson") response = requests.get(url, timeout=10) response.raise_for_status() text = response.text.strip() if not text or text == "[]": return [] data = json.loads(text) if isinstance(data, list): return data # 处理 {"data": [...]} 格式 if isinstance(data, dict): items = data.get("data", []) if isinstance(items, list): return items return [] except Exception as e: logger.error(f"获取白名单失败: {e}") return [] def add_ip(self, ip: str, retry: bool = True) -> tuple[bool, str]: """ 添加IP到白名单 Returns: (是否成功, API原始响应) """ try: url = self._build_url(act="add", ip=ip, meno=self._memo) response = requests.get(url, timeout=10) response.raise_for_status() text = response.text.strip() logger.debug(f"添加白名单响应: {text}") if "ok" in text.lower() or "success" in text.lower() or "添加成功" in text: logger.info(f"白名单添加成功: {ip} (备注: {self._memo})") return True, text if "已存在" in text or "exist" in text.lower() or "IpRep" in text: logger.info(f"白名单已存在: {ip}") return True, text # 频率限制，等待后重试一次 if retry and ("频率过快" in text or "稍后" in text): wait = 5 match = re.search(r'(\d+)\s*秒', text) if match: wait = int(match.group(1)) logger.info(f"白名单添加被限流，等待 {wait} 秒后重试...") time.sleep(wait) return self.add_ip(ip, retry=False) # UKEY 错误 if "Err:Key" in text: logger.error("白名单UKEY错误，请检查配置") return False, "UKEY错误，请检查白名单配置" # 超出白名单数量限制 if "Err:Max" in text or "超过" in text or "上限" in text: logger.error(f"白名单数量超限: {text}") return False, f"白名单数量超限: {text}" logger.warning(f"白名单添加结果: {text}") return False, text except Exception as e: logger.error(f"添加白名单失败: {e}") return False, str(e) def delete_ip(self, ip: str) -> bool: """ 删除指定IP Args: ip: 要删除的IP地址 """ try: url = self._build_url(act="del", ip=ip) response = requests.get(url, timeout=10) response.raise_for_status() text = response.text.strip() logger.debug(f"删除白名单响应: {text}") if "ok" in text.lower() or "success" in text.lower() or "删除成功" in text: logger.info(f"白名单删除成功: {ip}") return True logger.warning(f"白名单删除结果: {text}") return False except Exception as e: logger.error(f"删除白名单失败: {e}") return False def get_memo_ip(self) -> Optional[str]: """ 获取当前备注对应的IP Returns: IP地址，如果不存在返回None """ records = self.get_whitelist_json() for record in records: if record.get("MEMO") == self._memo: return record.get("IP") return None def get_memo_records(self) -> list[dict]: """ 获取当前备注的所有记录 Returns: 匹配备注的记录列表 """ records = self.get_whitelist_json() return [r for r in records if r.get("MEMO") == self._memo] def sync_ip(self, current_ip: str, keep_recent: int = 3) -> tuple[bool, str]: """ 同步白名单IP（保留多个近期出口IP，应对移动网络IP漂移） 策略： - 当前IP已在白名单（同备注），无需操作 - 当前IP不在白名单，添加（不删旧的，保留多个出口IP） - 超过 keep_recent 个同备注IP时，删除最老的（按列表顺序） - 当前IP已存在但备注不同，不处理（避免误删他人配置） 线程安全：使用全局锁串行化同步操作，避免并发调用触发代理服务商限流。 Args: current_ip: 当前出口IP keep_recent: 保留的同备注IP数量上限（默认3个，应对IP漂移） Returns: (是否成功, 消息) """ with _whitelist_sync_lock: try: records = self.get_whitelist_json() memo_records = [r for r in records if r.get("MEMO") == self._memo] memo_ips = [r.get("IP") for r in memo_records] # 当前IP已在白名单（同备注），无需操作 if current_ip in memo_ips: msg = f"白名单IP已是最新的: {current_ip}" logger.info(msg) return True, msg # 当前IP已存在但备注不同，不处理（避免误删他人配置） if any(r.get('IP') == current_ip for r in records): logger.info(f"白名单IP {current_ip} 已存在（备注不同），无需重复添加") return True, f"白名单IP已存在: {current_ip}" # 超过上限，删除最老的（列表前面的） if len(memo_ips) >= keep_recent: # 删除最早添加的（列表顺序） to_delete = memo_ips[:len(memo_ips) - keep_recent + 1] for old_ip in to_delete: logger.info(f"白名单同备注IP超限，删除旧的: {old_ip}") self.delete_ip(old_ip) time.sleep(1) # 添加新IP logger.info(f"白名单添加新出口IP: {current_ip} (当前 {len(memo_ips)} 个同备注)") ok, resp = self.add_ip(current_ip) if ok: msg = f"白名单IP已添加: {current_ip}" logger.info(msg) return True, msg return False, f"白名单添加失败，API响应: {resp}" except Exception as e: msg = f"白名单同步失败: {e}" logger.error(msg) return False, msg def test_connection(self) -> tuple[bool, str]: """ 测试白名单API连接 Returns: (是否成功, 消息) """ try: records = self.get_whitelist_json() count = len(records) my_records = [r for r in records if r.get("MEMO", "").startswith(self.MEMO_PREFIX)] msg = f"连接成功，白名单共 {count} 条记录，其中本机相关 {len(my_records)} 条" logger.info(msg) return True, msg except Exception as e: msg = f"连接失败: {e}" logger.error(msg) return False, msg def get_local_exit_ip() -> Optional[str]: """ 获取本机当前公网出口IP（不走代理）。 用于在获取代理前主动同步白名单，避免出口IP漂移导致代理拒绝连接。 Returns: 出口IP地址，获取失败返回 None """ targets = [ "https://myip.ipip.net", "https://4.ipw.cn", "https://api.ipify.org", ] for url in targets: try: response = requests.get(url, timeout=8, headers={"User-Agent": "Mozilla/5.0"}) response.raise_for_status() text = response.text.strip() # 从文本中提取IP match = re.search(r'(\d{1,3}(?:\.\d{1,3}){3})', text) if match: return match.group(1) except Exception: continue return None def get_exit_ip_via_proxy(proxy: str) -> Optional[str]: """ 通过代理获取出口IP Args: proxy: 代理URL，格式 http://ip:port Returns: 出口IP地址 """ targets = [ "https://qifu-api.baidubce.com/ip/local/geo/v1/district", "https://myip.ipip.net", "https://4.ipw.cn", ] proxies = {"http": proxy, "https": proxy} for url in targets: try: response = requests.get( url, proxies=proxies, timeout=6, headers={"User-Agent": "Mozilla/5.0"}, ) response.raise_for_status() # 尝试解析IP content_type = response.headers.get("content-type", "").lower() if "json" in content_type: data = response.json() ip = data.get("ip") or data.get("origin") if ip: return str(ip).split(",")[0].strip() # 从文本中提取IP text = response.text.strip() match = re.search(r"(\d{1,3}(?:\.\d{1,3}){3})", text) if match: return match.group(1) except Exception: continue return None