buff3.0/docs/test_steam_market_proxy.py

#!/usr/bin/env python3
"""
测试 Steam Market render 接口在直连与代理下的可达性。

默认测试 URL:
https://steamcommunity.com/market/listings/730/P90%20%7C%20Sand%20Spray%20%28Minimal%20Wear%29/render/?query=&start=0&count=10&country=US&language=schinese&currency=1

默认代理:
http://192.154.231.161:10609
"""

from __future__ import annotations

import argparse
import json
import ssl
import sys
import time
from typing import Any, Dict, Optional
from urllib.error import HTTPError, URLError
from urllib.request import HTTPSHandler, ProxyHandler, Request, build_opener


DEFAULT_URL = (
    "https://steamcommunity.com/market/listings/730/"
    "P90%20%7C%20Sand%20Spray%20%28Minimal%20Wear%29/"
    "render/?query=&start=0&count=10&country=US&language=schinese&currency=1"
)
DEFAULT_PROXY = "http://192.154.231.161:10795"


def fetch(url: str, timeout: int, proxy: Optional[str], verify_ssl: bool = True) -> Dict[str, Any]:
    handlers = []
    if proxy:
        handlers.append(ProxyHandler({"http": proxy, "https": proxy}))

    if not verify_ssl:
        # Only for temporary connectivity testing.
        handlers.append(HTTPSHandler(context=ssl._create_unverified_context()))

    opener = build_opener(*handlers)
    req = Request(
        url=url,
        headers={
            "User-Agent": "Mozilla/5.0 (compatible; steam-proxy-check/1.0)",
            "Accept": "application/json,text/plain,*/*",
            "Connection": "close",
        },
        method="GET",
    )

    start = time.perf_counter()
    try:
        with opener.open(req, timeout=timeout) as resp:
            status = getattr(resp, "status", None) or resp.getcode()
            body_bytes = resp.read()
            elapsed = (time.perf_counter() - start) * 1000
            text = body_bytes.decode("utf-8", errors="replace")
            content_type = resp.headers.get("Content-Type", "")
            return {
                "ok": True,
                "status": status,
                "elapsed_ms": round(elapsed, 2),
                "content_type": content_type,
                "body_len": len(body_bytes),
                "body_preview": text[:400],
            }
    except HTTPError as e:
        elapsed = (time.perf_counter() - start) * 1000
        err_body = ""
        try:
            err_body = e.read().decode("utf-8", errors="replace")[:400]
        except Exception:
            err_body = ""
        return {
            "ok": False,
            "status": e.code,
            "elapsed_ms": round(elapsed, 2),
            "error": f"HTTPError: {e}",
            "body_preview": err_body,
        }
    except URLError as e:
        elapsed = (time.perf_counter() - start) * 1000
        return {
            "ok": False,
            "status": None,
            "elapsed_ms": round(elapsed, 2),
            "error": f"URLError: {e}",
        }
    except Exception as e:
        elapsed = (time.perf_counter() - start) * 1000
        return {
            "ok": False,
            "status": None,
            "elapsed_ms": round(elapsed, 2),
            "error": f"{type(e).__name__}: {e}",
        }


def main() -> int:
    parser = argparse.ArgumentParser(description="Test steam market render URL with/without proxy")
    parser.add_argument("--url", default=DEFAULT_URL, help="Target URL")
    parser.add_argument("--proxy", default=DEFAULT_PROXY, help="HTTP/HTTPS proxy URL")
    parser.add_argument("--timeout", type=int, default=20, help="Timeout seconds")
    parser.add_argument(
        "--insecure",
        action="store_true",
        help="Disable SSL certificate verification (testing only)",
    )
    parser.add_argument("--skip-direct", action="store_true", help="Skip direct test")
    args = parser.parse_args()

    print("=== Steam Market Proxy Test ===")
    print(f"URL   : {args.url}")
    print(f"Proxy : {args.proxy}")
    print(f"Timeout: {args.timeout}s")
    print(f"SSL verify: {not args.insecure}")
    print()

    if not args.skip_direct:
        print("[1/2] Direct request (no proxy)")
        direct = fetch(args.url, args.timeout, proxy=None, verify_ssl=not args.insecure)
        print(json.dumps(direct, ensure_ascii=False, indent=2))
        print()
    else:
        direct = None

    print("[2/2] Proxy request")
    via_proxy = fetch(args.url, args.timeout, proxy=args.proxy, verify_ssl=not args.insecure)
    print(json.dumps(via_proxy, ensure_ascii=False, indent=2))
    print()

    if direct is not None:
        print("=== Summary ===")
        print(f"Direct ok: {direct.get('ok')} | status: {direct.get('status')}")
        print(f"Proxy  ok: {via_proxy.get('ok')} | status: {via_proxy.get('status')}")
    else:
        print("=== Summary ===")
        print(f"Proxy  ok: {via_proxy.get('ok')} | status: {via_proxy.get('status')}")

    return 0


if __name__ == "__main__":
    sys.exit(main())