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 a hard requirement, not just an
optimization — MassPrice's real login (on store.tcgplayer.com, TCGplayer's
shared OAuth identity provider) is behind hCaptcha. Two automated
approaches were tried and confirmed broken during development:
  1. Filling the real username/password and submitting — the invisible
     captcha check never resolves for a headless/automated browser, so
     the form silently never navigates.
  2. Driving a real human's clicks through a headless Chromium exposed
     via Chrome remote debugging (chrome://inspect) — still fails, because
     that Chromium's fingerprint (a `HeadlessChrome` user-agent,
     `navigator.webdriver = true`) is itself what hCaptcha flags,
     independent of who's actually clicking.
Deliberately not pursued further: trying to disguise that browser to get
past hCaptcha's bot detection is captcha-evasion, not a session-handling
technique, regardless of how legitimate the underlying goal is.

What actually works: the user logs into MassPrice in their own ordinary,
non-automated Chrome (hCaptcha passes the way it always does for a real
person), exports that session's cookies with a browser extension, and
pastes them into the Session page. `import_cookies()` below converts that
paste into Playwright's storage_state format and verifies it actually
authenticates before accepting it.
"""
import json
import logging
import os
from datetime import datetime, timezone

from playwright.sync_api import sync_playwright

import notify
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")

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

# Chrome's internal sameSite cookie values (what export extensions like
# Cookie-Editor report) mapped to what Playwright's storage_state expects.
_SAME_SITE_MAP = {
    "no_restriction": "None",
    "lax": "Lax",
    "strict": "Strict",
    "unspecified": "Lax",
}


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 _notify_session_expired_once():
    """Send the email alert only on the first failure after the session
    goes bad, not on every scheduled run that fails while it's down —
    the flag resets when a fresh session is imported successfully."""
    state = storage.load("notify_state.json", {})
    if state.get("session_expired_notified"):
        return
    sent = notify.send_alert(
        "MassPrice Scheduler: login session expired",
        "The saved MassPrice session has expired or is missing. "
        "Scheduled rule runs will keep failing until you log back in "
        "via the Session page in the MassPrice Scheduler webui and "
        "import a fresh cookie export.",
    )
    # Only mark as notified if the email actually sent — an unset/broken
    # mail config shouldn't silently swallow the next attempt forever.
    if sent:
        storage.save(
            "notify_state.json",
            {"session_expired_notified": True, "notified_at": datetime.now(timezone.utc).isoformat()},
        )


def _ensure_authenticated(page):
    page.goto(f"{BASE_URL}/Rule/List")
    if _is_logged_out(page):
        _notify_session_expired_once()
        raise AutomationError(
            "MassPrice session expired or missing — log in via the "
            "Session page and paste in a fresh cookie export."
        )


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()

            # Clicking the dropdown item only opens a confirmation modal
            # ("You are about to run the rule ... Are you sure?") — it
            # does not trigger the run itself. Confirmed by live testing:
            # without this step, zero network requests ever fire and the
            # rule silently never runs, despite the initial click
            # "succeeding" with no error. #ConfirmationModal / its
            # #Confirmation_Run button are used generically across the
            # site (also for Delete Rule etc.), so scope the confirm
            # click narrowly to this specific modal instance.
            modal = page.locator("#ConfirmationModal")
            modal.wait_for(state="visible", timeout=5000)
            modal.locator("#Confirmation_Run").click()
            modal.wait_for(state="hidden", timeout=10000)
            page.wait_for_timeout(1000)  # let the confirmed run's request land

            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])


# ── Login-session import (paste cookies exported from a real browser) ───


def _convert_cookie(raw):
    """Convert one cookie from a Cookie-Editor-style export into
    Playwright's storage_state cookie shape."""
    expires = raw.get("expirationDate")
    same_site_raw = (raw.get("sameSite") or "unspecified").lower()
    return {
        "name": raw["name"],
        "value": raw["value"],
        "domain": raw["domain"],
        "path": raw.get("path", "/"),
        "expires": expires if expires is not None else -1,
        "httpOnly": bool(raw.get("httpOnly", False)),
        "secure": bool(raw.get("secure", False)),
        "sameSite": _SAME_SITE_MAP.get(same_site_raw, "Lax"),
    }


def import_cookies(raw_text):
    """Parse a pasted cookie export, convert it, verify it actually
    authenticates against MassPrice, and only then save it as the new
    session. Raises AutomationError with a user-facing message on any
    failure; never overwrites a working session with a bad paste."""
    raw_text = (raw_text or "").strip()
    if not raw_text:
        raise AutomationError("Paste your exported cookies first — the box was empty")

    try:
        parsed = json.loads(raw_text)
    except json.JSONDecodeError:
        raise AutomationError(
            "Couldn't parse that as JSON — make sure you copied the full "
            "export from the cookie extension"
        )

    if not isinstance(parsed, list) or not parsed:
        raise AutomationError(
            "Expected a JSON array of cookies (Cookie-Editor's export format) — "
            "got something else"
        )

    relevant = [c for c in parsed if "tcgplayer.com" in (c.get("domain") or "")]
    if not relevant:
        raise AutomationError(
            "None of the pasted cookies are for tcgplayer.com — make sure you "
            "exported while on the massprice.tcgplayer.com tab, after logging in"
        )

    try:
        cookies = [_convert_cookie(c) for c in relevant]
    except KeyError as exc:
        raise AutomationError(f"A pasted cookie was missing its {exc} field")

    candidate_state = {"cookies": cookies, "origins": []}

    with sync_playwright() as p:
        browser = p.chromium.launch(headless=True, args=LAUNCH_ARGS)
        try:
            context = browser.new_context(storage_state=candidate_state)
            page = context.new_page()
            page.goto(f"{BASE_URL}/Rule/List")
            if _is_logged_out(page):
                raise AutomationError(
                    "Imported cookies didn't authenticate — still logged out. "
                    "Make sure you exported them from massprice.tcgplayer.com "
                    "itself, after finishing login, not from the login page."
                )
            _save_state(context)
            storage.save("notify_state.json", {"session_expired_notified": False})
        finally:
            browser.close()