buff3.0/docs/steam_confirm_order.py

#!/usr/bin/env python3
"""
Confirm a Steam trade/market confirmation by calling steamAPI.

Endpoints used:
1) POST /trade-confirmations/list      (optional, when confirmation_id is absent)
2) POST /trade-confirmations/operate   (required)

Example:
python docs/steam_confirm_order.py \
  --mafile "/path/to/xxxx.maFile" \
  --steam-id 7656119XXXXXXXXXX \
  --access-token "your_access_token" \
  --refresh-token "your_refresh_token" \
  --confirmation-id 1234567890123456789
"""

import argparse
import json
import os
import sys
import time
from typing import Any, Dict, Optional, Tuple
from urllib import request, error
from urllib.parse import urlencode


def post_json(url: str, payload: Dict[str, Any], timeout: int = 120) -> Tuple[int, str, Dict[str, Any]]:
    data = json.dumps(payload).encode("utf-8")
    req = request.Request(
        url=url,
        data=data,
        headers={"Content-Type": "application/json"},
        method="POST",
    )
    try:
        with request.urlopen(req, timeout=timeout) as resp:
            body = resp.read().decode("utf-8", errors="replace")
            code = resp.getcode()
    except error.HTTPError as e:
        code = e.code
        body = e.read().decode("utf-8", errors="replace")
    except Exception as e:
        return 0, str(e), {}

    try:
        parsed = json.loads(body) if body else {}
    except Exception:
        parsed = {}
    return code, body, parsed


def build_session(args: argparse.Namespace) -> Dict[str, Any]:
    return {
        "SteamId": int(args.steam_id),
        "AccessToken": args.access_token,
        "RefreshToken": args.refresh_token,
        "SessionId": args.session_id or "",
    }


def load_account_from_accounts_json(accounts_json_path: str, steam_account: str) -> Dict[str, Any]:
    if not os.path.exists(accounts_json_path):
        raise FileNotFoundError(f"accounts.json not found: {accounts_json_path}")

    with open(accounts_json_path, "r", encoding="utf-8") as f:
        root = json.load(f)

    accounts = root.get("accounts") or {}
    if steam_account not in accounts:
        raise KeyError(f"steam account not found in accounts.json: {steam_account}")

    info = accounts[steam_account]
    required_fields = ["steam_guard_path", "steamId", "accessToken", "refreshToken"]
    missing = [k for k in required_fields if not info.get(k)]
    if missing:
        raise ValueError(f"account info missing required fields: {', '.join(missing)}")

    return info


def normalize_proxy_url(raw: str) -> Optional[str]:
    value = (raw or "").strip().strip('"').strip("'")
    if not value:
        return None
    if value.startswith(("http://", "https://", "socks5://")):
        return value
    if ":" in value:
        return f"http://{value}"
    return None


def parse_proxy_api_response(body: str) -> Optional[str]:
    trimmed = (body or "").strip()
    if not trimmed:
        return None

    try:
        data = json.loads(trimmed)
        if isinstance(data, str):
            return normalize_proxy_url(data)
        if isinstance(data, dict):
            for key in ("proxyUrl", "proxy_url", "url", "proxy"):
                value = data.get(key)
                if isinstance(value, str):
                    parsed = normalize_proxy_url(value)
                    if parsed:
                        return parsed
            host = data.get("ip") or data.get("host") or data.get("server")
            port = data.get("port") or data.get("serverPort")
            if host and port:
                return normalize_proxy_url(f"{host}:{port}")
    except Exception:
        pass

    for line in trimmed.splitlines():
        candidate = line.strip().split(",")[0].strip()
        parsed = normalize_proxy_url(candidate)
        if parsed:
            return parsed
    return None


def load_proxy_config(accounts_json_path: str) -> Dict[str, Any]:
    steam_msg_dir = os.path.dirname(accounts_json_path)
    proxy_config_path = os.path.join(steam_msg_dir, "proxy_config.json")
    if not os.path.exists(proxy_config_path):
        return {}

    with open(proxy_config_path, "r", encoding="utf-8") as f:
        return json.load(f)


def resolve_auto_proxy_from_cfg(cfg: Dict[str, Any], timeout_seconds: int, force_refresh_api: bool = False) -> Tuple[str, Optional[str]]:
    mode = (cfg.get("effectiveMode") or "").strip().lower()
    if mode in ("", "none"):
        return mode or "none", None

    effective_brand_id = cfg.get("effectiveBrandId") or ""
    brand_cfg = (cfg.get("configs") or {}).get(effective_brand_id)
    if not isinstance(brand_cfg, dict):
        raise ValueError(f"[proxy] no brand config found for effectiveBrandId={effective_brand_id}")

    if mode == "account":
        host = (brand_cfg.get("serverHost") or "").strip()
        port = int(brand_cfg.get("serverPort") or 0)
        user = (brand_cfg.get("serverUsername") or "").strip()
        password = brand_cfg.get("serverPassword") or ""
        if not host or port <= 0:
            raise ValueError("[proxy] account mode requires serverHost and serverPort")
        if user:
            proxy_url = f"http://{user}:{password}@{host}:{port}"
        else:
            proxy_url = f"http://{host}:{port}"
        return mode, proxy_url

    if mode == "api":
        api_url = (brand_cfg.get("apiUrl") or "").strip()
        if not api_url:
            raise ValueError("[proxy] api mode requires apiUrl")
        if force_refresh_api:
            sep = "&" if "?" in api_url else "?"
            api_url = f"{api_url}{sep}{urlencode({'_ts': str(int(time.time() * 1000))})}"
        req = request.Request(api_url, method="GET")
        try:
            with request.urlopen(req, timeout=min(timeout_seconds, 20)) as resp:
                body = resp.read().decode("utf-8", errors="replace")
                status = resp.getcode()
        except Exception as e:
            raise RuntimeError(f"[proxy] fetch api proxy failed: {e}") from e
        if status < 200 or status >= 300:
            raise RuntimeError(f"[proxy] proxy api returned status={status}, body={body[:300]}")
        proxy_url = parse_proxy_api_response(body)
        if not proxy_url:
            raise RuntimeError(f"[proxy] cannot parse proxy from api response: {body[:300]}")
        return mode, proxy_url

    raise ValueError(f"[proxy] unsupported effectiveMode={mode}")


def proxy_precheck(proxy_url: Optional[str], timeout_seconds: int) -> bool:
    if not proxy_url:
        return True
    target = "https://steamcommunity.com/mobileconf/getlist"
    handler = request.ProxyHandler({"http": proxy_url, "https": proxy_url})
    opener = request.build_opener(handler)
    req = request.Request(target, method="GET")
    try:
        with opener.open(req, timeout=min(timeout_seconds, 15)) as resp:
            code = resp.getcode()
            print(f"[proxy] precheck success, status={code}, target=mobileconf/getlist")
            return True
    except error.HTTPError as e:
        # Reaching steam and getting an HTTP status still means proxy path is alive.
        print(f"[proxy] precheck reachable, status={e.code}, target=mobileconf/getlist")
        return True
    except Exception as e:
        print(f"[proxy] precheck failed: {e}")
        return False


def is_proxy_related_failure(code: int, body: str) -> bool:
    if code in (0, 407, 502, 503, 504):
        return True
    text = (body or "").lower()
    tokens = [
        "proxy",
        "tunnel",
        "timed out",
        "timeout",
        "connection refused",
        "connection reset",
        "failed to connect",
        "unreachable",
        "network is unreachable",
        "name or service not known",
        "an error occurred while sending the request",
    ]
    return any(token in text for token in tokens)


def resolve_confirmation_id(
    api_base: str,
    mafile_path: str,
    session: Dict[str, Any],
    timeout_seconds: int,
    proxy_url: Optional[str],
) -> Tuple[Optional[int], Optional[int]]:
    list_url = f"{api_base.rstrip('/')}/trade-confirmations/list"
    payload = {"MaFilePath": mafile_path, "Session": session}
    if proxy_url:
        payload["ProxyUrl"] = proxy_url
    code, body, parsed = post_json(list_url, payload, timeout=timeout_seconds)
    if code != 200:
        print(f"[list] failed, http={code}, body={body}")
        return None, None

    confirmations = parsed.get("confirmations") or []
    if not confirmations:
        print("[list] no pending confirmations.")
        return None, None

    first = confirmations[0]
    cid = first.get("id")
    ckey = first.get("key")
    print(f"[list] picked first pending confirmation id={cid}, key={ckey}")
    try:
        return int(cid), int(ckey) if ckey is not None else None
    except Exception:
        return None, None


def main() -> int:
    parser = argparse.ArgumentParser(description="Confirm Steam order via steamAPI /trade-confirmations/operate")
    parser.add_argument("--api-base", default="http://localhost:5000", help="steamAPI base url, default http://localhost:5000")
    parser.add_argument("--accounts-json", default="/Users/wuzilong/Library/Application Support/steam-msg/accounts.json", help="accounts.json path")
    parser.add_argument("--steam-account", default="", help="steam account key in accounts.json, e.g. gt23053")
    parser.add_argument("--mafile", default="", help="absolute path to .maFile (optional, overrides accounts.json)")
    parser.add_argument("--steam-id", type=int, default=0, help="SteamId (optional, overrides accounts.json)")
    parser.add_argument("--access-token", default="", help="AccessToken (optional, overrides accounts.json)")
    parser.add_argument("--refresh-token", default="", help="RefreshToken (optional, overrides accounts.json)")
    parser.add_argument("--session-id", default="", help="optional SessionId from /login response session.sessionId")
    parser.add_argument("--confirmation-id", type=int, required=True, help="pending confirmation id to operate")
    parser.add_argument("--confirmation-key", type=int, default=0, help="optional confirmation key; omit to let server resolve")
    parser.add_argument("--operation", default="accept", choices=["accept", "cancel"], help="accept or cancel, default accept")
    parser.add_argument("--timeout-seconds", type=int, default=120, help="HTTP timeout for steamAPI call, default 120")
    parser.add_argument("--proxy-retry-times", type=int, default=2, help="retry times when api proxy fails, default 2")
    args = parser.parse_args()

    if args.steam_account:
        try:
            account_info = load_account_from_accounts_json(args.accounts_json, args.steam_account)
        except Exception as e:
            print(f"[account] failed to load account from accounts.json: {e}")
            return 2

        if not args.mafile:
            args.mafile = account_info.get("steam_guard_path", "")
        if not args.steam_id:
            args.steam_id = int(account_info.get("steamId", 0))
        if not args.access_token:
            args.access_token = account_info.get("accessToken", "")
        if not args.refresh_token:
            args.refresh_token = account_info.get("refreshToken", "")
        if not args.session_id:
            args.session_id = account_info.get("sessionId", "")

    if not args.mafile or not args.steam_id or not args.access_token or not args.refresh_token:
        print("[args] missing required auth fields. provide --steam-account or manual --mafile/--steam-id/--access-token/--refresh-token")
        return 2

    proxy_cfg = load_proxy_config(args.accounts_json)
    if not proxy_cfg:
        print("[proxy] proxy_config.json not found, use direct mode.")
    try:
        proxy_mode, proxy_url = resolve_auto_proxy_from_cfg(proxy_cfg, args.timeout_seconds, force_refresh_api=False)
    except Exception as e:
        print(str(e))
        return 2
    print(f"[proxy] effectiveMode={proxy_mode}, selected proxy={proxy_url or '<none>'}")

    if not proxy_precheck(proxy_url, args.timeout_seconds):
        if proxy_mode == "api":
            print("[proxy] precheck failed, refresh api proxy and retry precheck.")
            try:
                _, proxy_url = resolve_auto_proxy_from_cfg(proxy_cfg, args.timeout_seconds, force_refresh_api=True)
            except Exception as e:
                print(str(e))
                return 2
            print(f"[proxy] refreshed api proxy={proxy_url or '<none>'}")
            if not proxy_precheck(proxy_url, args.timeout_seconds):
                return 2
        else:
            return 2

    session = build_session(args)
    confirmation_id = args.confirmation_id
    confirmation_key = args.confirmation_key

    if confirmation_id == 0:
        confirmation_id, resolved_key = resolve_confirmation_id(
            args.api_base,
            args.mafile,
            session,
            args.timeout_seconds,
            proxy_url,
        )
        if not confirmation_id:
            print("[operate] no valid confirmation id found.")
            return 2
        if not confirmation_key and resolved_key:
            confirmation_key = resolved_key

    operate_url = f"{args.api_base.rstrip('/')}/trade-confirmations/operate"
    payload: Dict[str, Any] = {
        "MaFilePath": args.mafile,
        "Session": session,
        "ConfirmationId": int(confirmation_id),
        "Operation": args.operation,
    }
    if proxy_url:
        payload["ProxyUrl"] = proxy_url
    if confirmation_key:
        payload["ConfirmationKey"] = int(confirmation_key)

    max_attempts = 1 + max(args.proxy_retry_times, 0) if proxy_mode == "api" else 1
    for attempt in range(1, max_attempts + 1):
        if proxy_url:
            payload["ProxyUrl"] = proxy_url
        else:
            payload.pop("ProxyUrl", None)

        code, body, parsed = post_json(operate_url, payload, timeout=args.timeout_seconds)
        print(f"[operate] attempt={attempt}/{max_attempts} http={code}")
        if parsed:
            print(json.dumps(parsed, ensure_ascii=False, indent=2))
        else:
            print(body)

        if code == 200 and parsed and parsed.get("success") is True:
            print("[operate] confirmation succeeded.")
            return 0

        if attempt >= max_attempts:
            break
        if not is_proxy_related_failure(code, body):
            print("[proxy] non-proxy failure detected, stop retry.")
            break

        print("[proxy] proxy-related failure detected, refresh api proxy then retry.")
        try:
            _, proxy_url = resolve_auto_proxy_from_cfg(proxy_cfg, args.timeout_seconds, force_refresh_api=True)
            print(f"[proxy] refreshed api proxy={proxy_url or '<none>'}")
        except Exception as e:
            print(str(e))
            break
        if not proxy_precheck(proxy_url, args.timeout_seconds):
            break

    if code != 200:
        return 1
    print("[operate] request finished but success=false.")
    return 3


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