hello_js_reverse_skill/tcaptcha/protocol/proxy.py

# -*- 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 "未取到代理")