60 lines
1.9 KiB
Python
60 lines
1.9 KiB
Python
"""代理 API 响应解析。"""
|
||
|
||
import json
|
||
import re
|
||
from typing import Optional
|
||
|
||
|
||
def parse_proxy_response(text: str) -> tuple[list[str], Optional[str]]:
|
||
"""
|
||
解析代理 API 响应，支持 JSON 格式与旧版纯文本格式。
|
||
|
||
Returns:
|
||
(proxy_urls, whitelist_ip)
|
||
- proxy_urls: 解析到的所有代理地址列表（http://ip:port）
|
||
- whitelist_ip: 需要添加到白名单的 IP，无错误时为 None
|
||
"""
|
||
text = (text or "").strip()
|
||
if not text:
|
||
return [], None
|
||
|
||
try:
|
||
data = json.loads(text)
|
||
if isinstance(data, dict):
|
||
code = data.get("code")
|
||
msg = data.get("msg", "") or ""
|
||
if code != 0 and ("白名单" in msg or "添加白名单" in msg):
|
||
ip_match = re.search(r"(\d+\.\d+\.\d+\.\d+)", msg)
|
||
if ip_match:
|
||
return [], ip_match.group(1)
|
||
|
||
data_field = data.get("data")
|
||
proxies: list[str] = []
|
||
if isinstance(data_field, list):
|
||
for item in data_field:
|
||
if not isinstance(item, dict):
|
||
continue
|
||
ip = item.get("IP") or item.get("ip")
|
||
port = item.get("Port") or item.get("port")
|
||
if ip and port:
|
||
proxies.append(f"http://{ip}:{port}")
|
||
if proxies:
|
||
return proxies, None
|
||
|
||
if code == 0 and not proxies:
|
||
return [], None
|
||
except (json.JSONDecodeError, ValueError):
|
||
pass
|
||
|
||
if "添加白名单" in text or "白名单" in text:
|
||
ip_match = re.search(r"(\d+\.\d+\.\d+\.\d+)", text)
|
||
if ip_match:
|
||
return [], ip_match.group(1)
|
||
|
||
matches = re.findall(r"(\d+\.\d+\.\d+\.\d+):(\d+)", text)
|
||
proxies = [f"http://{ip}:{port}" for ip, port in matches]
|
||
if proxies:
|
||
return proxies, None
|
||
|
||
return [], None
|