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 the tool logs
in once and slides the session forward on every successful run rather
than hitting the login form every time. A fresh login (via
MASSPRICE_USERNAME/MASSPRICE_PASSWORD) only happens when the saved state
is missing or has expired.

NOTE — the real logged-out login form was not inspected during
development (the account was already signed in during discovery, and
forcing a logout to inspect it would have disrupted the user's real
MassPrice session). The locators in _login() are written to be resilient
(role/label/placeholder based, not brittle CSS) but may need a one-time
adjustment against the live form on first deploy. See CLAUDE.md.
"""
import logging
import os
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")

MASSPRICE_USERNAME = os.environ.get("MASSPRICE_USERNAME")
MASSPRICE_PASSWORD = os.environ.get("MASSPRICE_PASSWORD")

# 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):
    if "login" in page.url.lower() or "signin" in page.url.lower():
        return True
    return page.locator('input[type="password"]').count() > 0


def _login(page):
    if not MASSPRICE_USERNAME or not MASSPRICE_PASSWORD:
        raise AutomationError(
            "MASSPRICE_USERNAME/MASSPRICE_PASSWORD not set — cannot log in"
        )
    page.goto(f"{BASE_URL}/Dashboard")
    if not _is_logged_out(page):
        return

    email_field = (
        page.get_by_label("Email", exact=False)
        .or_(page.get_by_placeholder("Email", exact=False))
        .or_(page.locator('input[type="email"]'))
        .or_(page.locator('input[name="username" i]'))
    )
    password_field = page.locator('input[type="password"]')

    email_field.first.fill(MASSPRICE_USERNAME)
    password_field.first.fill(MASSPRICE_PASSWORD)

    submit = (
        page.get_by_role("button", name="Log In", exact=False)
        .or_(page.get_by_role("button", name="Sign In", exact=False))
        .or_(page.locator('button[type="submit"]'))
    )
    submit.first.click()
    page.wait_for_load_state("networkidle")

    if _is_logged_out(page):
        raise AutomationError(
            "Login failed — still on a login page after submit "
            "(bad credentials, or a CAPTCHA/2FA challenge that automated "
            "login cannot get past — see CLAUDE.md)"
        )


def _ensure_authenticated(page):
    page.goto(f"{BASE_URL}/Rule/List")
    if _is_logged_out(page):
        _login(page)
        page.goto(f"{BASE_URL}/Rule/List")
        if _is_logged_out(page):
            raise AutomationError("Still logged out after a login attempt")


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