160 lines
5.2 KiB
Python
160 lines
5.2 KiB
Python
"""携趣代理(xiequ)白名单适配器。
|
|
|
|
从 core.douyu.whitelist.WhitelistManager 迁移而来。
|
|
"""
|
|
|
|
import json
|
|
import re
|
|
import time
|
|
from urllib.parse import urlencode
|
|
|
|
import requests
|
|
from loguru import logger
|
|
|
|
from .base import BaseWhitelistAdapter
|
|
|
|
|
|
class XiequAdapter(BaseWhitelistAdapter):
|
|
"""携趣代理白名单适配器。
|
|
|
|
API格式：
|
|
基础URL: http://op.xiequ.cn/IpWhiteList.aspx
|
|
认证参数: uid + ukey
|
|
操作区分: act=add / del / getjson
|
|
"""
|
|
|
|
BASE_URL = "http://op.xiequ.cn/IpWhiteList.aspx"
|
|
|
|
def __init__(self, credentials: dict):
|
|
self.uid = credentials.get("uid", "")
|
|
self.ukey = credentials.get("ukey", "")
|
|
|
|
@property
|
|
def platform_name(self) -> str:
|
|
return "xiequ"
|
|
|
|
@property
|
|
def platform_label(self) -> str:
|
|
return "携趣"
|
|
|
|
def _build_url(self, **params) -> str:
|
|
"""构建请求URL"""
|
|
base_params = {"uid": self.uid, "ukey": self.ukey}
|
|
base_params.update(params)
|
|
return f"{self.BASE_URL}?{urlencode(base_params)}"
|
|
|
|
def get_whitelist(self) -> list[dict]:
|
|
"""
|
|
获取白名单列表。
|
|
|
|
协固原始格式: [{"IP": "x.x.x.x", "MEMO": "备注"}, ...]
|
|
统一格式: [{"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, dict):
|
|
data = data.get("data", [])
|
|
if not isinstance(data, list):
|
|
return []
|
|
|
|
# 映射到统一格式
|
|
return [
|
|
{"ip": r.get("IP", r.get("ip", "")), "memo": r.get("MEMO", r.get("memo", ""))}
|
|
for r in data
|
|
if r.get("IP") or r.get("ip")
|
|
]
|
|
|
|
except Exception as e:
|
|
logger.error(f"获取白名单失败: {e}")
|
|
return []
|
|
|
|
def add_ip(self, ip: str, retry: bool = True) -> tuple[bool, str]:
|
|
"""添加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}")
|
|
|
|
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) -> tuple[bool, str]:
|
|
"""删除白名单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, text
|
|
|
|
logger.warning(f"白名单删除结果: {text}")
|
|
return False, text
|
|
|
|
except Exception as e:
|
|
logger.error(f"删除白名单失败: {e}")
|
|
return False, str(e)
|
|
|
|
def test_connection(self) -> tuple[bool, str]:
|
|
"""测试白名单API连接。"""
|
|
try:
|
|
records = self.get_whitelist()
|
|
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
|