buff3.0/docs/test/debug_buy_502.py

"""
Steam Market Buy API 调试脚本
复现 502 错误场景

日志信息:
- URL: https://steamcommunity.com/market/buylisting/634586007392692709
- listing_id: 634586007392692709
- goods_id: 35820
- goods_name: 新星 | 毒蜥 (略有磨损)
- steam_account: cea10388
- target_price_usd: 5.5
"""

import requests
import urllib.parse
import json
from datetime import datetime

# 请求配置
LISTING_ID = "689755100902207377"
SESSION_ID = "C859D89325CCF1A3963504BBA006038D"
STEAM_LOGIN_SECURE = (
    "76561199354580595%7C%7CeyAidHlwIjogIkpXVCIsICJhbGciOiAiRWREU0EiIH0."
    "eyAiaXNzIjogInI6MDAwRl8yODA5QzM3N182NERDMyIsICJzdWIiOiAiNzY1NjExOTkzNTQ1ODA1OTUiLCAiYXVkIjogWyAiY2xpZW50IiwgIndlYiIgXSwgImV4cCI6IDE3NzY2MTA2NDAsICJuYmYiOiAxNzY3ODgzNjc1LCAiaWF0IjogMTc3NjUyMzY3NSwgImp0aSI6ICIwMDEyXzI4MDlDMUQ0X0I0MEM0IiwgIm9hdCI6IDE3NzY1MjM2NzUsICJydF9leHAiOiAxNzc5MTMwMjMyLCAicGVyIjogMCwgImlwX3N1YmplY3QiOiAiMTAzLjEwNi4xODguMTU4IiwgImlwX2NvbmZpcm1lciI6ICIxMDMuMTA2LjE4OC4xNTgiIH0."
    "UcD-w8xcSCRFQKS6CDLzAiRocQzW0UX54O8HwAXv_hm8ULiUvgdQYdZe2eZeqw_Hel66Zf9T5uNgJSbQNzogDw;"
)
ITEM_NAME = "Negev%20%7C%20Wall%20Bang%20%28Minimal%20Wear%29"

# 价格信息（从日志中获取）
SUBTOTAL = "21"
FEE = "3"
TOTAL = "24"


def build_cookie() -> str:
    """构建完整的 Cookie 字符串"""
    return (
        f"timezoneOffset=28800,0; "
        f"sessionid={SESSION_ID}; "
        f"steamLoginSecure={STEAM_LOGIN_SECURE}"
    )


def build_request_body() -> dict:
    """构建请求体"""
    return {
        "sessionid": SESSION_ID,
        "currency": "1",
        "subtotal": SUBTOTAL,
        "fee": FEE,
        "total": TOTAL,
        "quantity": "1",
        "save_my_address": "0",
        "tradefee_tax": "0",
        "confirmation": "0",
        "billing_state": "",
    }


def build_headers() -> dict:
    """构建请求头"""
    return {
        "origin": "https://steamcommunity.com",
        "cookie": build_cookie(),
        "user-agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36",
        "accept": "*/*",
        "content-type": "application/x-www-form-urlencoded; charset=UTF-8",
        "referer": f"https://steamcommunity.com/market/listings/730/{ITEM_NAME}",
    }


def send_buy_request(
    listing_id: str,
    use_proxy: bool = False,
    proxy_url: str = None,
    session: requests.Session = None,
) -> tuple[requests.Response, float]:
    """
    发送购买请求

    Returns:
        (response, elapsed_ms)
    """
    url = f"https://steamcommunity.com/market/buylisting/{listing_id}"
    data = build_request_body()
    headers = build_headers()

    if session is None:
        session = requests.Session()

    proxies = None
    if use_proxy and proxy_url:
        proxies = {"http": proxy_url, "https": proxy_url}

    start_time = datetime.now()
    try:
        response = session.post(
            url,
            data=data,
            headers=headers,
            proxies=proxies,
            timeout=30,
            allow_redirects=False,
        )
    except requests.exceptions.RequestException as e:
        elapsed = (datetime.now() - start_time).total_seconds() * 1000
        print(f"[ERROR] 请求异常: {e}")
        raise

    elapsed_ms = (datetime.now() - start_time).total_seconds() * 1000
    return response, elapsed_ms


def print_response_info(response: requests.Response, elapsed_ms: float):
    """打印响应信息"""
    print("\n" + "=" * 60)
    print(f"状态码: {response.status_code}")
    print(f"耗时: {elapsed_ms:.0f}ms")
    print(f"响应头: {json.dumps(dict(response.headers), indent=2, ensure_ascii=False)}")
    print(f"\n响应体:")
    try:
        body_json = response.json()
        print(json.dumps(body_json, indent=2, ensure_ascii=False))
    except json.JSONDecodeError:
        print(response.text)
    print("=" * 60)


def main():
    print("Steam Market Buy API 调试脚本")
    print(f"Listing ID: {LISTING_ID}")
    print(f"Session ID: {SESSION_ID}")
    print(f"价格: subtotal={SUBTOTAL}, fee={FEE}, total={TOTAL}")
    print()

    session = requests.Session()

    # 测试 1: 直接请求（不使用代理）
    print("[测试 1] 直接请求（不使用代理）...")
    try:
        response, elapsed_ms = send_buy_request(LISTING_ID, use_proxy=False, session=session)
        print_response_info(response, elapsed_ms)

        if response.status_code == 200:
            print("✅ 购买成功!")
        elif response.status_code == 502:
            print("❌ 502 错误 - 他人已购买该物品")
            try:
                error_body = response.json()
                print(f"错误信息: {error_body.get('message', 'Unknown')}")
            except:
                print(f"错误信息: {response.text}")
        elif response.status_code == 429:
            print("⚠️ 429 错误 - 请求过于频繁")
        else:
            print(f"❌ 其他错误: {response.status_code}")

    except Exception as e:
        print(f"[ERROR] {e}")


if __name__ == "__main__":
    main()