白名单 ok

2026-06-22 11:29:22 +08:00
parent 66b03a7d89
commit 4c924375aa
5 changed files with 544 additions and 2 deletions
@@ -0,0 +1,268 @@
+"""代理IP白名单管理模块"""
+
+import json
+import re
+from typing import Optional
+from urllib.parse import urlencode
+
+import requests
+from loguru import logger
+
+
+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) -> bool:
+        """
+        添加IP到白名单
+
+        Args:
+            ip: 要添加的IP地址
+        """
+        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}")
+
+            # 成功通常返回 "ok" 或类似信息
+            if "ok" in text.lower() or "success" in text.lower() or "添加成功" in text:
+                logger.info(f"白名单添加成功: {ip} (备注: {self._memo})")
+                return True
+
+            # 检查是否已存在
+            if "已存在" in text or "exist" in text.lower():
+                logger.info(f"白名单已存在: {ip}")
+                return True
+
+            logger.warning(f"白名单添加结果: {text}")
+            return False
+
+        except Exception as e:
+            logger.error(f"添加白名单失败: {e}")
+            return False
+
+    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) -> tuple[bool, str]:
+        """
+        同步白名单IP
+
+        检查当前备注是否有记录：
+        - 如果IP相同，无需操作
+        - 如果IP不同，删除旧的并添加新的
+        - 如果IP已存在但备注不同（如手动添加无备注），删除后重新添加
+        - 如果无记录，添加新的
+
+        Args:
+            current_ip: 当前出口IP
+
+        Returns:
+            (是否成功, 消息)
+        """
+        try:
+            existing_ip = self.get_memo_ip()
+
+            # IP相同，无需更新
+            if existing_ip == current_ip:
+                msg = f"白名单IP已是最新的: {current_ip}"
+                logger.info(msg)
+                return True, msg
+
+            # 有旧记录，先删除
+            if existing_ip:
+                logger.info(f"白名单IP变化: {existing_ip} -> {current_ip}")
+                self.delete_ip(existing_ip)
+
+            # IP已存在但备注不同（或备注为空），先删除后重新添加
+            records = self.get_whitelist_json()
+            if any(r.get('IP') == current_ip for r in records):
+                logger.info(f"白名单IP {current_ip} 已存在但备注不同，删除后重新添加")
+                self.delete_ip(current_ip)
+
+            # 添加新IP
+            if self.add_ip(current_ip):
+                if existing_ip:
+                    msg = f"白名单IP已更新: {existing_ip} -> {current_ip}"
+                else:
+                    msg = f"白名单IP已添加: {current_ip}"
+                logger.info(msg)
+                return True, msg
+
+            return False, "白名单添加失败"
+
+        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_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