buff3.0/tljb/confirm_only.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""浠呯‘璁わ細login + trade-confirmations/operate"""
import json
import sys
import time

import requests

API = "http://localhost:5000"
USERNAME = "tsi19338"
PASSWORD = "ayx71486"
MAFILE = r"c:\Users\Administrator\Desktop\tsi19338.maFile"
CONFIRMATION_ID = "1184627388269539651"
PROXY = "http://127.0.0.1:7897"


def login():
    body = {
        "Username": USERNAME,
        "Password": PASSWORD,
        "MaFilePath": MAFILE,
    }
    if PROXY:
        body["ProxyUrl"] = PROXY
    for attempt in range(1, 6):
        print(f"[login] attempt {attempt}/5")
        r = requests.post(f"{API}/login", json=body, timeout=90)
        print(f"[login] HTTP {r.status_code} {r.text[:300]}")
        if r.status_code == 200:
            return r.json()
        time.sleep(3)
    raise RuntimeError("login failed")


def main():
    data = login()
    sess = data["session"]
    session = {
        "SteamId": sess["steamId"],
        "AccessToken": sess["accessToken"],
        "RefreshToken": sess["refreshToken"],
        "SessionId": sess["sessionId"],
    }
    # list first
    list_body = {"MaFilePath": MAFILE, "Session": session}
    if PROXY:
        list_body["ProxyUrl"] = PROXY
    print("[list] fetching confirmations...")
    lr = requests.post(f"{API}/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}/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
            # refresh session for next try
            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())