Massprice/automation.py

"""Playwright automation against massprice.tcgplayer.com.

Drives the real page rather than replaying a raw HTTP request: "Run Rule"
is bound to a JS/AJAX click handler with no plain href, and the request
it fires is almost certainly anti-forgery-token protected.

Session handling: a Playwright `storage_state` (cookies + localStorage)
is persisted to AUTH_STATE_PATH and reused across runs, so scheduled runs
never touch the login form at all. That's deliberate, not just an
optimization — MassPrice's real login (on store.tcgplayer.com, TCGplayer's
shared OAuth identity provider) is behind hCaptcha, confirmed by live
testing: filling the real username/password and submitting silently
never navigates away, because the invisible captcha check never resolves
for a headless/automated browser. **There is no way to fully automate
logging in.** A human has to solve that captcha at least once per
session lifetime, via the "Chrome remote debugging" capture flow below.

Capture flow: `start_debug_session()` launches a headless-but-CDP-exposed
Chromium instance (`--remote-debugging-port`), which the user connects to
from their own Chrome via chrome://inspect to log in interactively —
solving the hCaptcha themselves, since it's a real click. Once logged in,
`capture_debug_session()` saves that browser's cookies to
AUTH_STATE_PATH. All of this must happen on a single dedicated thread —
Playwright's sync API is not safe to call from arbitrary threads, and
Flask/gunicorn's gthread workers hand each request to a different one —
so `_debug_worker()` owns every Playwright object here and talks to
Flask request threads only through `_debug_cmd_queue`.
"""
import logging
import os
import queue
import subprocess
import threading
from datetime import datetime, timezone

from playwright.sync_api import sync_playwright

import storage

logger = logging.getLogger("massprice.automation")

BASE_URL = "https://massprice.tcgplayer.com"
AUTH_STATE_PATH = os.path.join(storage.DATA_DIR, "auth_state.json")

# Chromium's remote-debugging port binds to 127.0.0.1 only inside the
# container, no matter what --remote-debugging-address is set to —
# deliberate Chromium security hardening (confirmed by live testing, not
# a flag mistake). A loopback-only listener is unreachable through
# Docker's port mapping, so `socat` re-exposes it on all interfaces at a
# second, different internal port (they can't share one — Chromium's
# 127.0.0.1 bind and a 0.0.0.0 bind on the same port conflict). Neither
# of these is the DEBUG_PORT env var app.py reads for display — that one
# is the user-facing HOST port for chrome://inspect instructions, mapped
# in docker-compose.yml to _PROXY_PORT, not to Chromium's own port.
_INTERNAL_DEBUG_PORT = 9222
_PROXY_PORT = 9333
DEBUG_SESSION_TIMEOUT = 600  # auto-close an un-captured debug session after 10 minutes

# Standard, necessary mitigations for headless Chromium inside Docker.
LAUNCH_ARGS = ["--no-sandbox", "--disable-dev-shm-usage"]


class AutomationError(Exception):
    pass


def _new_context(browser):
    if os.path.exists(AUTH_STATE_PATH):
        return browser.new_context(storage_state=AUTH_STATE_PATH)
    return browser.new_context()


def _save_state(context):
    os.makedirs(storage.DATA_DIR, exist_ok=True)
    context.storage_state(path=AUTH_STATE_PATH)


def _is_logged_out(page):
    """An unauthenticated visit to any /Rule/* page silently redirects to
    the public marketing homepage (`/`) — there's no `/login` URL and no
    password field to detect, so "did we get bounced off of /Rule" is the
    only reliable signal (verified by live inspection)."""
    return "/Rule" not in page.url


def _ensure_authenticated(page):
    page.goto(f"{BASE_URL}/Rule/List")
    if _is_logged_out(page):
        raise AutomationError(
            "MassPrice session expired or missing — log in via the "
            "Session page (chrome://inspect) and click Capture Session."
        )


def session_status():
    if not os.path.exists(AUTH_STATE_PATH):
        return {"state": "missing", "saved_at": None}
    mtime = os.path.getmtime(AUTH_STATE_PATH)
    saved_at = datetime.fromtimestamp(mtime, tz=timezone.utc).isoformat()
    return {"state": "saved", "saved_at": saved_at}


def list_rules():
    """Scrape /Rule/List for current rule id/name pairs.

    Returns a list of {"id": ..., "name": ...} dicts and caches it to
    rules_cache.json.
    """
    with sync_playwright() as p:
        browser = p.chromium.launch(headless=True, args=LAUNCH_ARGS)
        try:
            context = _new_context(browser)
            page = context.new_page()
            _ensure_authenticated(page)
            _save_state(context)

            rows = page.locator("li.rule.widget-rule")
            rules = []
            for i in range(rows.count()):
                row = rows.nth(i)
                run_link = row.locator("a.run-rule").first
                rule_id = run_link.get_attribute("data-rule-id")
                if not rule_id:
                    continue
                name_text = None
                try:
                    # The rule name renders as the first line of text in
                    # the row, ahead of the "My Selling Prices" /
                    # "Schedule Active…" detail lines.
                    full_text = row.inner_text().strip()
                    name_text = full_text.splitlines()[0].strip()
                except Exception:
                    pass
                rules.append({"id": rule_id, "name": name_text or f"Rule {rule_id}"})

            storage.save(
                "rules_cache.json",
                {"updated_at": datetime.now(timezone.utc).isoformat(), "rules": rules},
            )
            return rules
        finally:
            browser.close()


def run_rule(rule_id, rule_name=None):
    """Click "Run Rule" for the given rule id. Logs the outcome to
    run_history.json and returns (status, message)."""
    status = "failure"
    message = ""
    with sync_playwright() as p:
        browser = p.chromium.launch(headless=True, args=LAUNCH_ARGS)
        try:
            context = _new_context(browser)
            page = context.new_page()
            _ensure_authenticated(page)
            _save_state(context)

            run_selector = f'a.run-rule[data-rule-id="{rule_id}"]'
            row = page.locator("li.rule.widget-rule").filter(
                has=page.locator(run_selector)
            )
            if row.count() == 0:
                raise AutomationError(
                    f"Rule id {rule_id} not found on /Rule/List "
                    "(renamed or deleted on the MassPrice side?)"
                )

            row.first.locator(".dropdown-ellipsis").click()
            run_link = row.first.locator(run_selector)
            run_link.wait_for(state="visible", timeout=5000)
            run_link.click()
            page.wait_for_timeout(1500)  # let the click's AJAX call fire

            status = "success"
            message = "Run Rule clicked"
        except Exception as exc:
            message = str(exc)
            logger.exception("run_rule failed for rule %s", rule_id)
        finally:
            browser.close()

    _log_history(rule_id, rule_name, status, message)
    return status, message


def _log_history(rule_id, rule_name, status, message):
    history = storage.load("run_history.json", [])
    history.insert(
        0,
        {
            "rule_id": rule_id,
            "rule_name": rule_name,
            "timestamp": datetime.now(timezone.utc).isoformat(),
            "status": status,
            "message": message,
        },
    )
    storage.save("run_history.json", history[:200])


# ── Interactive login-session capture (Chrome remote debugging) ─────────
#
# Everything below runs exclusively on `_debug_worker`'s thread. Flask
# routes call start_debug_session()/capture_debug_session()/
# stop_debug_session(), which just enqueue a command and block for the
# result — never touch the Playwright objects directly.

_debug_cmd_queue: "queue.Queue" = queue.Queue()
_debug_thread = None
_debug_active = False  # read from other threads; plain bool assignment is atomic in CPython


def _debug_worker():
    global _debug_active
    playwright = browser = context = page = timer = socat_proc = None

    def do_stop():
        nonlocal playwright, browser, context, page, timer, socat_proc
        global _debug_active
        if timer:
            timer.cancel()
            timer = None
        if socat_proc:
            try:
                socat_proc.terminate()
            except Exception:
                pass
            socat_proc = None
        if browser:
            try:
                browser.close()
            except Exception:
                pass
        if playwright:
            try:
                playwright.stop()
            except Exception:
                pass
        playwright = browser = context = page = None
        _debug_active = False

    while True:
        cmd, result_q = _debug_cmd_queue.get()
        try:
            if cmd == "start":
                if browser is None:
                    playwright = sync_playwright().start()
                    browser = playwright.chromium.launch(
                        headless=True,
                        args=LAUNCH_ARGS
                        + [
                            f"--remote-debugging-port={_INTERNAL_DEBUG_PORT}",
                            "--remote-allow-origins=*",
                        ],
                    )
                    # Re-expose Chromium's loopback-only debug port on
                    # all interfaces so Docker's port mapping (which
                    # forwards to _PROXY_PORT, not Chromium's own port —
                    # see the comment above) can actually reach it.
                    socat_proc = subprocess.Popen(
                        [
                            "socat",
                            f"TCP-LISTEN:{_PROXY_PORT},fork,reuseaddr",
                            f"TCP:127.0.0.1:{_INTERNAL_DEBUG_PORT}",
                        ]
                    )
                    context = _new_context(browser)
                    page = context.new_page()
                    page.goto(f"{BASE_URL}/")
                    _debug_active = True
                    timer = threading.Timer(
                        DEBUG_SESSION_TIMEOUT, lambda: _debug_cmd_queue.put(("stop", None))
                    )
                    timer.daemon = True
                    timer.start()
                if result_q:
                    result_q.put(("ok", None))
            elif cmd == "capture":
                if context is None:
                    if result_q:
                        result_q.put(
                            ("error", "No active login session — click Start Login Session first")
                        )
                else:
                    page.goto(f"{BASE_URL}/Rule/List")
                    page.wait_for_timeout(1000)
                    if _is_logged_out(page):
                        if result_q:
                            result_q.put(
                                (
                                    "error",
                                    "Still not logged in — finish logging in via "
                                    "chrome://inspect, then try Capture Session again",
                                )
                            )
                    else:
                        _save_state(context)
                        do_stop()
                        if result_q:
                            result_q.put(("ok", None))
            elif cmd == "stop":
                do_stop()
                if result_q:
                    result_q.put(("ok", None))
        except Exception as exc:
            logger.exception("debug session command %r failed", cmd)
            if result_q:
                result_q.put(("error", str(exc)))


def _ensure_debug_thread():
    global _debug_thread
    if _debug_thread is None or not _debug_thread.is_alive():
        _debug_thread = threading.Thread(target=_debug_worker, daemon=True)
        _debug_thread.start()


def _debug_call(cmd, timeout=30):
    result_q: "queue.Queue" = queue.Queue()
    _debug_cmd_queue.put((cmd, result_q))
    status, payload = result_q.get(timeout=timeout)
    if status == "error":
        raise AutomationError(payload)


def start_debug_session():
    _ensure_debug_thread()
    _debug_call("start")


def capture_debug_session():
    _debug_call("capture")


def stop_debug_session():
    _ensure_debug_thread()
    _debug_call("stop")


def debug_session_active():
    return _debug_active