buff3.0/docs/confirm_poll.py

#!/usr/bin/env python3
"""
confirm_poll.py

Usage:
  # Use config and perform initial purchase to get confirmation id, then poll
  python confirm_poll.py

  # Provide an existing confirmation id to poll directly
  python confirm_poll.py --confirmation 1374710298789955445

This script mimics the browser behavior: when a buylisting returns need_confirmation,
it extracts the confirmation id and repeatedly POSTs the buylisting URL with that
confirmation id until the purchase succeeds or times out.
"""
import time
import json
import argparse
from urllib.parse import urlencode
import requests

try:
    import config
except Exception:
    print("❌ Could not import config.py. Make sure you're running this from the steam-buy directory.")
    raise


DEFAULT_HEADERS = {
    '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': '*/*',
    'Accept-Encoding': 'gzip, deflate, br',
    'Accept-Language': 'zh-CN,zh;q=0.9,en;q=0.8',
    'Cache-Control': 'no-cache',
    'Connection': 'keep-alive',
    'Content-Type': 'application/x-www-form-urlencoded; charset=UTF-8',
    'Origin': 'https://steamcommunity.com',
    'Referer': f'https://steamcommunity.com/market/listings/730/Test-Item'
}


def _post_buylisting(session, listing_id, params, headers):
    url = f"https://steamcommunity.com/market/buylisting/{listing_id}"
    data = urlencode(params)
    resp = session.post(url, data=data, headers=headers, timeout=30)
    text = resp.text or ""
    try:
        parsed = json.loads(text)
    except Exception:
        parsed = None
    return resp.status_code, text, parsed


def poll_with_confirmation(session, listing_id, purchase_params, confirmation_id, max_attempts=12, interval=3):
    """Poll buylisting using the provided confirmation_id until success or timeout."""
    headers = DEFAULT_HEADERS.copy()
    # attach cookies via Cookie header (we assume config.BROWSER_COOKIE_STRING contains full cookie string)
    if getattr(config, 'BROWSER_COOKIE_STRING', None):
        headers['Cookie'] = config.BROWSER_COOKIE_STRING

    print(f"🔁 Start polling buylisting {listing_id} with confirmation id: {confirmation_id}")
    for attempt in range(1, max_attempts + 1):
        params = purchase_params.copy()
        params['confirmation'] = str(confirmation_id)

        status, text, parsed = _post_buylisting(session, listing_id, params, headers)
        print(f"[{attempt}/{max_attempts}] HTTP {status}")
        if parsed:
            print(json.dumps(parsed, indent=2, ensure_ascii=False))
        else:
            print(text[:400])

        if parsed and parsed.get('success') == 1:
            print("🎉 Purchase succeeded.")
            return True

        # If server still requests confirmation, wait and retry
        if parsed and parsed.get('need_confirmation') or status == 406:
            print(f"⏳ Still needs confirmation, sleeping {interval}s before retry...")
            time.sleep(interval)
            continue

        # Other non-success result -> log and break (or retry depending on code)
        print("⚠️ Received non-success response and not explicitly need_confirmation. Stopping.")
        return False

    print("❌ Max polling attempts reached without success.")
    return False


def run(initial_only=False, provided_confirmation=None):
    listing_id = getattr(config, "LISTING_ID", None)
    purchase_params = getattr(config, "PURCHASE_PARAMS", {}).copy()

    if not listing_id or not purchase_params:
        print("❌ LISTING_ID or PURCHASE_PARAMS missing in config.py")
        return

    headers = DEFAULT_HEADERS.copy()
    if getattr(config, 'BROWSER_COOKIE_STRING', None):
        headers['Cookie'] = config.BROWSER_COOKIE_STRING

    session = requests.Session()

    # If user provided confirmation id, poll directly
    if provided_confirmation:
        return poll_with_confirmation(session, listing_id, purchase_params, provided_confirmation)

    # Otherwise perform initial purchase attempt to obtain confirmation id
    print("🔄 Sending initial purchase request with confirmation=0 ...")
    purchase_params['confirmation'] = '0'
    status, text, parsed = _post_buylisting(session, listing_id, purchase_params, headers)
    print(f"Initial HTTP {status}")
    if parsed:
        print(json.dumps(parsed, indent=2, ensure_ascii=False))
    else:
        print(text[:400])

    if parsed and parsed.get('success') == 1:
        print("🎉 Purchase succeeded without confirmation.")
        return True

    # Extract confirmation id
    confirmation_id = None
    if parsed:
        confirmation = parsed.get('confirmation') or {}
        confirmation_id = confirmation.get('confirmation_id') or parsed.get('confirmation_id')

    if not confirmation_id:
        print("⚠️ No confirmation_id found in initial response. If the response indicated need_confirmation,")
        print("you may need to open the account in a browser or confirm via mobileconf. Exiting.")
        return False

    print(f"🔢 Found confirmation_id: {confirmation_id}")

    # At this point we should ensure the mobile confirmation is accepted (by another process/tool).
    # This script will poll buylisting with the confirmation_id, assuming the confirmation
    # will be accepted externally (by SteamDesktopAuthenticator) or was auto-accepted.
    return poll_with_confirmation(session, listing_id, getattr(config, "PURCHASE_PARAMS", {}), confirmation_id)


if __name__ == "__main__":
    parser = argparse.ArgumentParser(description="Poll Steam buylisting using confirmation id.")
    parser.add_argument("--confirmation", "-c", help="Confirmation id to poll (optional)", default=None)
    args = parser.parse_args()

    conf = args.confirmation
    result = run(provided_confirmation=conf)
    if result:
        print("Done: purchase complete.")
    else:
        print("Done: purchase NOT completed.")