buff3.0/tljb/confirm_only.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""仅确认：login + trade-confirmations/operate（读 config.json）"""
import json
import sys
import time
from pathlib import Path

import requests

ROOT = Path(__file__).resolve().parent
CFG = ROOT / "config.json"


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


def resolve_mafile(mafile: str) -> str:
    p = Path(mafile)
    if not p.is_absolute():
        p = ROOT / p
    return str(p.resolve())


def normalize_proxy(proxy: str):
    raw = (proxy or "").strip()
    if not raw:
        return None
    if raw.lower().startswith(("http://", "https://", "socks5://", "socks4://", "socks://")):
        return raw
    return f"http://{raw}"


def login(cfg, proxy):
    body = {
        "Username": cfg["username"],
        "Password": cfg["password"],
        "MaFilePath": resolve_mafile(cfg["mafile"]),
    }
    if proxy:
        body["ProxyUrl"] = proxy
    api = cfg.get("api_base") or "http://localhost:5000"
    for attempt in range(1, 6):
        print(f"[login] attempt {attempt}/5")
        r = requests.post(f"{api.rstrip('/')}/login", json=body, timeout=90)
        print(f"[login] HTTP {r.status_code} {r.text[:300]}")
        if r.status_code == 200:
            return r.json(), api
        time.sleep(3)
    raise RuntimeError("login failed")


def main():
    if len(sys.argv) < 2:
        print("用法: python confirm_only.py <confirmation_id> [proxy]", file=sys.stderr)
        return 2
    confirmation_id = sys.argv[1]
    cfg = load_cfg()
    proxy = normalize_proxy(sys.argv[2] if len(sys.argv) > 2 else cfg.get("proxy") or "")

    data, api = login(cfg, proxy)
    sess = data["session"]
    session = {
        "SteamId": sess["steamId"],
        "AccessToken": sess["accessToken"],
        "RefreshToken": sess["refreshToken"],
        "SessionId": sess["sessionId"],
    }
    mafile = resolve_mafile(cfg["mafile"])

    list_body = {"MaFilePath": mafile, "Session": session}
    if proxy:
        list_body["ProxyUrl"] = proxy
    print("[list] fetching confirmations...")
    lr = requests.post(f"{api.rstrip('/')}/trade-confirmations/list", json=list_body, timeout=60)
    print(f"[list] HTTP {lr.status_code} {lr.text[:2000]}")
    if lr.status_code == 200:
        lj = lr.json()
        if lj.get("session"):
            s = lj["session"]
            session = {
                "SteamId": s.get("steamId", session["SteamId"]),
                "AccessToken": s.get("accessToken", session["AccessToken"]),
                "RefreshToken": s.get("refreshToken", session["RefreshToken"]),
                "SessionId": s.get("sessionId", session["SessionId"]),
            }

    op_body = {
        "MaFilePath": mafile,
        "Session": session,
        "ConfirmationId": int(confirmation_id),
        "Operation": "accept",
    }
    if proxy:
        op_body["ProxyUrl"] = proxy

    for attempt in range(1, 11):
        print(f"[confirm] attempt {attempt}/10 ConfirmationId={confirmation_id}")
        r = requests.post(f"{api.rstrip('/')}/trade-confirmations/operate", json=op_body, timeout=60)
        print(f"[confirm] HTTP {r.status_code}")
        print(r.text[:2000])
        if r.status_code == 200:
            j = r.json()
            if j.get("success") in (True, 1, "1"):
                print("[DONE] confirm success")
                return 0
            if j.get("session"):
                s = j["session"]
                op_body["Session"] = {
                    "SteamId": s.get("steamId", session["SteamId"]),
                    "AccessToken": s.get("accessToken", session["AccessToken"]),
                    "RefreshToken": s.get("refreshToken", session["RefreshToken"]),
                    "SessionId": s.get("sessionId", session["SessionId"]),
                }
        time.sleep(3)
    print("[DONE] confirm failed")
    return 1


if __name__ == "__main__":
    try:
        sys.stdout.reconfigure(encoding="utf-8")
    except Exception:
        pass
    raise SystemExit(main())