yml2213
60ae14d54a
- 题目图片自动下载: 所有题型原图统一落盘至 scratch/captcha_debug，文字点选自动生成带编号红圈标注图便于肉眼核验 - 字形严格一致性校验: 在 SolveResult 增加 text_match_ok 检查，对图字不符或混淆干扰题直接拒答换题，避免错答引发 errorCode=12 - 防秒换与拟人延时: 遇到非预期题目换题等待 2.8~4.2s (对齐 HAR 真实抓包)，求解失败等待 2.5~3.8s，作答前模拟交互 5.5~7.5s - 代理容错增强: 在 OneMsAdapter 捕获代理断流与鉴权异常，支持动态轮换新代理重试
102 lines
3.4 KiB
Python
102 lines
3.4 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""
|
|
代理 IP 提取与格式化 (用于绕过腾讯天御的 IP 级滚动风控)
|
|
|
|
腾讯验证码按出口 IP 做滚动窗口风控, 触发时会返回 `errorCode=12` (f_limit)。
|
|
本模块从常见的「代理 IP 提取 API」取一个可用代理, 供 `TencentCaptchaClient(proxy=...)`
|
|
与 `OneMsAdapter(proxy=...)` 使用。
|
|
|
|
已适配的返回格式:
|
|
* JSON: {"status":100,"data":[{"ip":"1.2.3.4","port":"8080"}]} (讯客 xkdaili 等)
|
|
* JSON: {"data":{"ip":"1.2.3.4","port":8080}} / [{"ip":..,"port":..}]
|
|
* 纯文本: 一行一个 "1.2.3.4:8080" 或 "1.2.3.4,8080"
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import re
|
|
from typing import List, Optional
|
|
|
|
import httpx
|
|
|
|
_IP_PORT_RE = re.compile(r"(\d{1,3}(?:\.\d{1,3}){3})\s*[:,]\s*(\d{1,5})")
|
|
|
|
|
|
def format_proxy(ip: str, port, scheme: str = "http") -> str:
|
|
return f"{scheme}://{ip}:{port}"
|
|
|
|
|
|
def parse_proxies(payload: str, scheme: str = "http") -> List[str]:
|
|
"""从 API 返回体里解析出全部代理 (兼容 JSON 与纯文本两种格式)"""
|
|
out: List[str] = []
|
|
text = (payload or "").strip()
|
|
if not text:
|
|
return out
|
|
|
|
try:
|
|
import json
|
|
|
|
data = json.loads(text)
|
|
|
|
def walk(node):
|
|
if isinstance(node, dict):
|
|
ip = node.get("ip") or node.get("host") or node.get("server")
|
|
port = node.get("port")
|
|
if ip and port:
|
|
out.append(format_proxy(str(ip), port, scheme))
|
|
for key in ("data", "result", "proxy", "list", "proxies"):
|
|
if key in node:
|
|
walk(node[key])
|
|
elif isinstance(node, list):
|
|
for item in node:
|
|
walk(item)
|
|
|
|
walk(data)
|
|
if out:
|
|
return out
|
|
except Exception:
|
|
pass
|
|
|
|
for ip, port in _IP_PORT_RE.findall(text):
|
|
out.append(format_proxy(ip, port, scheme))
|
|
return out
|
|
|
|
|
|
def fetch_proxy(api_url: str, scheme: str = "http", timeout: float = 15.0) -> Optional[str]:
|
|
"""调用代理提取 API 并返回第一个可用代理字符串, 例如 http://1.2.3.4:8080"""
|
|
import logging
|
|
_log = logging.getLogger("proxy")
|
|
resp = httpx.get(api_url, timeout=timeout, trust_env=False, follow_redirects=True)
|
|
proxies = parse_proxies(resp.text, scheme=scheme)
|
|
if not proxies:
|
|
try:
|
|
import json
|
|
err = json.loads(resp.text)
|
|
if "info" in err or "message" in err:
|
|
_log.error("代理 API 返回失败: %s", err.get("info") or err.get("message"))
|
|
except Exception:
|
|
pass
|
|
return proxies[0] if proxies else None
|
|
|
|
|
|
def check_proxy(proxy: str, url: str = "https://turing.captcha.qcloud.com/",
|
|
timeout: float = 20.0) -> bool:
|
|
"""探测代理是否可用 (能连上目标域名即可, 不判断业务状态码)"""
|
|
try:
|
|
httpx.get(url, proxy=proxy, timeout=timeout, trust_env=False,
|
|
follow_redirects=True, headers={"User-Agent": "Mozilla/5.0"})
|
|
return True
|
|
except Exception:
|
|
return False
|
|
|
|
|
|
if __name__ == "__main__":
|
|
import argparse
|
|
|
|
ap = argparse.ArgumentParser(description="从代理 API 取一个可用代理")
|
|
ap.add_argument("api_url")
|
|
ap.add_argument("--scheme", default="http")
|
|
args = ap.parse_args()
|
|
proxy = fetch_proxy(args.api_url, scheme=args.scheme)
|
|
print(proxy or "未取到代理")
|