buff3.0/test_steam_proxy.py

#!/usr/bin/env python3
"""
Steam Render API 代理测试脚本 v2
优化：代理冷却机制 + 智能调度

使用方法: python test_steam_proxy.py
"""

import requests
import time
import random
import threading
import queue
import collections
from datetime import datetime
from urllib.parse import urlparse

# 配置
# PROXY_API_URL = "https://api.haiwaiip.net/api/pull?token=3d706c31d8d0267d757706b3e4d81454&num=100&format=text&protocol=http&country=&state=&city=&ip_type=datacenter&sep=1&area="
# STEAM_RENDER_URL = "https://steamcommunity.com/market/listings/730/AK-47%20%7C%20Crossfade%20%28Field-Tested%29/render/?query=&start=10&count=10&country=SG&language=schinese&currency=1"

# 配置 - 新的代理 API
PROXY_API_URL = "https://api.cliproxy.io/white/api?region=US&num=100&time=10&format=n&type=txt"
STEAM_RENDER_URL = "https://steamcommunity.com/market/listings/730/AK-47%20%7C%20Crossfade%20%28Field-Tested%29/render/?query=&start=10&count=10&country=SG&language=schinese&currency=1"

# 固定端口（根据 API 返回格式调整）
# 如果 API 返回的是 IP:PORT 格式，使用默认端口
PROXY_PORT = 20000  # 默认端口，可根据实际情况修改

# 请求头（模拟浏览器）
HEADERS = {
    "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/146.0.0.0 Safari/537.36",
    "Accept": "text/javascript, text/html, application/xml, text/xml, */*",
    "Accept-Language": "zh-CN,zh;q=0.9",
    "Accept-Encoding": "gzip, deflate, br, zstd",
    "Connection": "keep-alive",
    "Referer": "https://steamcommunity.com/market/listings/730/AK-47%20%7C%20Crossfade%20%28Field-Tested%29",
    "X-Requested-With": "XMLHttpRequest",
    "X-Prototype-Version": "1.7",
    "Sec-Fetch-Site": "same-origin",
    "Sec-Fetch-Mode": "cors",
    "Sec-Fetch-Dest": "empty",
    "sec-ch-ua": "\"Chromium\";v=\"146\", \"Not-A.Brand\";v=\"24\", \"Google Chrome\";v=\"146\"",
    "sec-ch-ua-mobile": "?0",
    "sec-ch-ua-platform": "\"Windows\"",
    "Cookie": "timezoneOffset=28800,0; steamCountry=SG%7C3b4993028d9b610ab149cd29f164f361",
}

# 冷却配置
PROXY_COOLDOWN_SECONDS = 120  # 代理被 429 后冷却 2 分钟
MAX_CONSECUTIVE_429 = 3      # 连续 429 超过 3 次标记为"坏代理"

# 全局变量
proxy_pool = queue.Queue()
proxy_cooldown = {}  # proxy -> 冷却结束时间
proxy_429_count = collections.defaultdict(int)  # proxy -> 429 次数
bad_proxies = set()  # 坏代理集合
stats = {
    "total_requests": 0,
    "status_200": 0,
    "status_429": 0,
    "status_other": 0,
    "errors": 0,
    "proxies_used": 0,
}
stats_lock = threading.Lock()
running = True
proxy_lock = threading.Lock()

def get_proxies():
    """从API获取代理列表"""
    try:
        response = requests.get(PROXY_API_URL, timeout=10)
        if response.status_code == 200:
            text = response.text.strip()
            # 解析返回的代理列表（可能每行一个 IP:PORT）
            proxies = []
            for line in text.split('\n'):
                line = line.strip()
                if line and ':' in line:
                    proxies.append(line)
                elif line and not line.startswith('not added'):
                    # 只有 IP 的情况，添加默认端口
                    proxies.append(f"{line}:{PROXY_PORT}")

            if proxies:
                print(f"[{datetime.now().strftime('%H:%M:%S')}] 获取到 {len(proxies)} 个代理")
                return proxies
            else:
                print(f"[{datetime.now().strftime('%H:%M:%S')}] API返回为空或格式未知: {text[:80]}")
                return []
        else:
            print(f"[{datetime.now().strftime('%H:%M:%S')}] API失败: HTTP {response.status_code}")
            return []
    except Exception as e:
        print(f"[{datetime.now().strftime('%H:%M:%S')}] 获取代理异常: {e}")
        return []

def is_proxy_on_cooldown(proxy):
    """检查代理是否在冷却中"""
    if proxy in proxy_cooldown:
        if time.time() < proxy_cooldown[proxy]:
            return True
        else:
            # 冷却结束，移除
            del proxy_cooldown[proxy]
    return False

def mark_proxy_cooldown(proxy):
    """标记代理进入冷却"""
    proxy_cooldown[proxy] = time.time() + PROXY_COOLDOWN_SECONDS
    print(f"[{datetime.now().strftime('%H:%M:%S')}] 代理 {proxy} 进入冷却 ({PROXY_COOLDOWN_SECONDS}秒)")

def mark_bad_proxy(proxy):
    """标记为坏代理"""
    with proxy_lock:
        bad_proxies.add(proxy)
    proxy_429_count[proxy] += 1
    print(f"[{datetime.now().strftime('%H:%M:%S')}] 代理 {proxy} 标记为坏代理 (429次数: {proxy_429_count[proxy]})")

def is_bad_proxy(proxy):
    """检查是否为坏代理"""
    with proxy_lock:
        return proxy in bad_proxies

def ensure_proxy_pool(min_count=20):
    """确保代理池有足够的可用代理"""
    while proxy_pool.qsize() < min_count and running:
        proxies = get_proxies()
        if proxies:
            for proxy in proxies:
                if running:
                    # 过滤掉坏代理
                    if not is_bad_proxy(proxy):
                        proxy_pool.put(proxy)
        else:
            print(f"[{datetime.now().strftime('%H:%M:%S')}] 获取代理为空，等待5秒后重试...")
            time.sleep(5)

def get_valid_proxy():
    """获取一个有效的代理（不在冷却中且不是坏代理）"""
    max_attempts = 10
    for _ in range(max_attempts):
        if proxy_pool.empty():
            ensure_proxy_pool(10)
        try:
            proxy = proxy_pool.get(block=False)
            if is_proxy_on_cooldown(proxy):
                # 仍在冷却，放回去并继续等待
                proxy_pool.put(proxy)
                time.sleep(0.5)
                continue
            if is_bad_proxy(proxy):
                continue
            return proxy
        except queue.Empty:
            continue
    return None

def print_stats():
    """打印统计信息"""
    with stats_lock:
        total = stats["total_requests"]
        print(f"\n{'='*60}")
        print(f"  统计信息")
        print(f"{'='*60}")
        print(f"  总请求数:     {total}")
        print(f"  状态码 200:   {stats['status_200']:>6}  ({stats['status_200']/max(total,1)*100:>5.1f}%)")
        print(f"  状态码 429:   {stats['status_429']:>6}  ({stats['status_429']/max(total,1)*100:>5.1f}%)")
        print(f"  其他状态:     {stats['status_other']:>6}")
        print(f"  请求错误:     {stats['errors']:>6}")
        print(f"  使用代理数:   {stats['proxies_used']:>6}")
        print(f"{'='*60}")
        print(f"  代理池剩余:   {proxy_pool.qsize():>6}")
        print(f"  冷却中代理:   {len(proxy_cooldown):>6}")
        print(f"  坏代理数:     {len(bad_proxies):>6}")
        print(f"{'='*60}\n")

def make_request(thread_id, request_interval):
    """单次请求"""
    proxy = get_valid_proxy()
    if not proxy:
        print(f"[T{thread_id}] [{datetime.now().strftime('%H:%M:%S')}] 无可用代理，等待...")
        time.sleep(5)
        return

    proxy_dict = {
        "http": f"http://{proxy}",
        "https": f"http://{proxy}",
    }

    try:
        start_time = time.time()
        response = requests.get(
            STEAM_RENDER_URL,
            headers=HEADERS,
            proxies=proxy_dict,
            timeout=15,
            allow_redirects=False
        )
        elapsed = (time.time() - start_time) * 1000
        status = response.status_code

        with stats_lock:
            stats["total_requests"] += 1

            if status == 200:
                stats["status_200"] += 1
                stats["proxies_used"] += 1
                # 成功，将代理归还
                proxy_pool.put(proxy)
                print(f"[T{thread_id}] [{datetime.now().strftime('%H:%M:%S')}] 200 OK  ({elapsed:.0f}ms) - {proxy}")

            elif status == 429:
                stats["status_429"] += 1
                proxy_429_count[proxy] += 1
                print(f"[T{thread_id}] [{datetime.now().strftime('%H:%M:%S')}] 429 TOO_MANY ({elapsed:.0f}ms) - {proxy} (429次数:{proxy_429_count[proxy]}) ***")

                # 标记冷却
                mark_proxy_cooldown(proxy)

                # 连续 429 超过阈值，标记为坏代理
                if proxy_429_count[proxy] >= MAX_CONSECUTIVE_429:
                    mark_bad_proxy(proxy)

            else:
                stats["status_other"] += 1
                proxy_pool.put(proxy)
                print(f"[T{thread_id}] [{datetime.now().strftime('%H:%M:%S')}] {status} ({elapsed:.0f}ms) - {proxy}")

    except requests.exceptions.Timeout:
        with stats_lock:
            stats["errors"] += 1
        print(f"[T{thread_id}] [{datetime.now().strftime('%H:%M:%S')}] TIMEOUT - {proxy}")
        if proxy:
            proxy_pool.put(proxy)

    except requests.exceptions.ProxyError as e:
        with stats_lock:
            stats["errors"] += 1
        err_msg = str(e)
        print(f"[T{thread_id}] [{datetime.now().strftime('%H:%M:%S')}] PROXY_ERROR: {err_msg[:80]} - {proxy}")
        if proxy:
            proxy_pool.put(proxy)

    except requests.exceptions.ConnectionError as e:
        with stats_lock:
            stats["errors"] += 1
        err_msg = str(e)
        print(f"[T{thread_id}] [{datetime.now().strftime('%H:%M:%S')}] CONN_ERROR: {err_msg[:80]} - {proxy}")
        if proxy:
            proxy_pool.put(proxy)

    except Exception as e:
        with stats_lock:
            stats["errors"] += 1
        print(f"[T{thread_id}] [{datetime.now().strftime('%H:%M:%S')}] ERROR: {str(e)[:80]} - {proxy}")
        if proxy:
            proxy_pool.put(proxy)

def worker_thread(thread_id):
    """工作线程"""
    while running:
        make_request(thread_id, request_interval=0)
        # 无间隔，高并发模式

def stats_thread():
    """统计线程"""
    while running:
        time.sleep(60)
        print_stats()

def cooldown_cleanup_thread():
    """定期清理过期的冷却记录"""
    while running:
        time.sleep(30)
        current_time = time.time()
        expired = [p for p, end_time in proxy_cooldown.items() if current_time >= end_time]
        for p in expired:
            del proxy_cooldown[p]

def main():
    global running
    num_threads = 5

    print("=" * 60)
    print("  Steam Render API 代理测试 v3")
    print("  模式：高并发 + 快速请求")
    print("=" * 60)
    print(f"  目标URL: {STEAM_RENDER_URL[:70]}...")
    print(f"  线程数: {num_threads}")
    print(f"  请求间隔: 0秒 (全速)")
    print(f"  代理冷却: {PROXY_COOLDOWN_SECONDS}秒")
    print(f"  坏代理阈值: {MAX_CONSECUTIVE_429}次429")
    print("=" * 60)
    print()

    # 预热
    print("预热：获取初始代理池...")
    ensure_proxy_pool(50)
    print(f"代理池已准备，当前数量: {proxy_pool.qsize()}\n")

    # 启动守护线程
    threading.Thread(target=stats_thread, daemon=True).start()
    threading.Thread(target=cooldown_cleanup_thread, daemon=True).start()

    # 启动工作线程
    num_threads = 5
    threads = []

    for i in range(num_threads):
        t = threading.Thread(target=worker_thread, args=(i+1,), daemon=True)
        t.start()
        threads.append(t)

    print(f"已启动 {num_threads} 个工作线程\n")

    try:
        while True:
            time.sleep(10)
            if proxy_pool.qsize() < 15:
                print(f"[{datetime.now().strftime('%H:%M:%S')}] 代理池不足，补充...")
                ensure_proxy_pool(30)
    except KeyboardInterrupt:
        print("\n正在停止...")
        running = False

    for t in threads:
        t.join(timeout=2)

    print_stats()
    print("测试结束")

if __name__ == "__main__":
    main()