增加了代理平台和日志

2026-06-24 14:22:29 +08:00
parent 0d03b2c242
commit 83345dfdc5
21 changed files with 1134 additions and 444 deletions
@@ -0,0 +1,162 @@
+"""星空代理(xkdaili)白名单适配器。
+
+API格式：
+  基础URL: http://api2.xkdaili.com/tools/XApi.ashx
+  白名单认证参数: apikey + sign(白名单专用) + flag
+  获取代理认证参数: apikey + sign(提取专用) — 由用户在代理API地址中填写完整URL
+
+  白名单接口的 sign 和获取代理的 sign 不同！
+
+响应格式:
+  status=100: 操作成功
+  status=508: 白名单已存在（视为成功）
+  status=202: 秘钥验证失败
+  status=808: 其它错误
+
+注意：没有查询白名单列表的接口
+"""
+
+import json
+
+import requests
+from loguru import logger
+
+from .base import BaseWhitelistAdapter
+
+
+class XkdailiAdapter(BaseWhitelistAdapter):
+    """星空代理(xkdaili)白名单适配器。"""
+
+    BASE_URL = "http://api2.xkdaili.com/tools/XApi.ashx"
+
+    def __init__(self, credentials: dict):
+        self.apikey = credentials.get("apikey", "")
+        self.wl_sign = credentials.get("wl_sign", "")
+        self.flag = credentials.get("flag", "")
+
+    @property
+    def platform_name(self) -> str:
+        return "xkdaili"
+
+    @property
+    def platform_label(self) -> str:
+        return "星空"
+
+    def _build_url(self, **params) -> str:
+        """构建白名单请求URL"""
+        base_params = {
+            "apikey": self.apikey,
+            "sign": self.wl_sign,
+            "flag": self.flag,
+        }
+        base_params.update(params)
+        from urllib.parse import urlencode
+        return f"{self.BASE_URL}?{urlencode(base_params)}"
+
+    def _parse_response(self, text: str) -> tuple[bool, str]:
+        """解析星空API响应。
+
+        status=100: 成功
+        status=508: 白名单已存在（视为成功）
+        status=202: 秘钥验证失败
+        status=808: 其它错误
+        """
+        try:
+            data = json.loads(text)
+            status = data.get("status", -1)
+            info = data.get("info", "")
+
+            # 100=成功
+            if status == 100:
+                return True, info
+
+            # 508=已存在，视为成功
+            if status == 508:
+                return True, info
+
+            # 202=秘钥验证失败
+            if status == 202:
+                return False, f"认证失败: {info}"
+
+            # 其他错误
+            return False, f"失败(status={status}): {info}"
+
+        except (json.JSONDecodeError, ValueError):
+            # 非JSON响应，按文本判断
+            if "成功" in text or "ok" in text.lower():
+                return True, text
+            if "已存在" in text:
+                return True, text
+            return False, text
+
+    def get_whitelist(self) -> list[dict]:
+        """
+        星空不支持查询白名单列表，返回空列表。
+
+        sync_ip 会检测到空列表后直接走添加逻辑。
+        """
+        return []
+
+    def add_ip(self, ip: str) -> tuple[bool, str]:
+        """添加IP到白名单。多个IP用英文逗号隔开。"""
+        try:
+            url = self._build_url(type="addwhiteip", ip=ip)
+            response = requests.get(url, timeout=10)
+            response.raise_for_status()
+
+            text = response.text.strip()
+            logger.debug(f"星空添加白名单响应: {text}")
+
+            ok, msg = self._parse_response(text)
+            if ok:
+                logger.info(f"星空白名单添加成功: {ip} - {msg}")
+            else:
+                logger.warning(f"星空白名单添加失败: {msg}")
+            return ok, msg
+
+        except Exception as e:
+            logger.error(f"星空添加白名单失败: {e}")
+            return False, str(e)
+
+    def delete_ip(self, ip: str) -> tuple[bool, str]:
+        """删除白名单IP。多个IP用英文逗号隔开。"""
+        try:
+            url = self._build_url(type="delwhiteip", ip=ip)
+            response = requests.get(url, timeout=10)
+            response.raise_for_status()
+
+            text = response.text.strip()
+            logger.debug(f"星空删除白名单响应: {text}")
+
+            ok, msg = self._parse_response(text)
+            if ok:
+                logger.info(f"星空白名单删除成功: {ip} - {msg}")
+            else:
+                logger.warning(f"星空白名单删除失败: {msg}")
+            return ok, msg
+
+        except Exception as e:
+            logger.error(f"星空删除白名单失败: {e}")
+            return False, str(e)
+
+    def test_connection(self) -> tuple[bool, str]:
+        """测试白名单API连接。"""
+        try:
+            # 尝试添加无效IP来验证认证
+            url = self._build_url(type="addwhiteip", ip="0.0.0.0")
+            response = requests.get(url, timeout=10)
+            response.raise_for_status()
+
+            text = response.text.strip()
+            ok, msg = self._parse_response(text)
+
+            if ok:
+                return True, "连接成功，API认证正常"
+            # 认证失败
+            if "认证失败" in msg:
+                return False, msg
+            # 其他错误（如无效IP），说明认证通过了
+            return True, f"连接成功，API可访问: {msg}"
+
+        except Exception as e:
+            return False, f"连接失败: {e}"