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
import uuid
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 _set_session_expired(expired):
    """Track whether the saved session is currently known-bad, so the
    dashboard banner (app.py's _session_warning()) can reflect a failed
    "Refresh Rules" too — list_rules() writes no run_history entry, so
    inferring this from run history alone missed refresh failures
    entirely. Written only on a state change, since the success path
    runs on every rule run and every verification poll."""
    state = storage.load("session_state.json", {})
    if bool(state.get("expired")) == bool(expired):
        return
    payload = {"expired": bool(expired)}
    if expired:
        payload["at"] = datetime.now(timezone.utc).isoformat()
    storage.save("session_state.json", payload)


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):
        _set_session_expired(True)
        _notify_session_expired_once()
        raise AutomationError(
            "MassPrice session expired or missing — log in via the "
            "Session page and paste in a fresh cookie export."
        )
    _set_session_expired(False)


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 _find_latest_activity(page, rule_name):
    """Returns (activity_id: int, status_text: str) for the most recent
    Activity Log entry matching rule_name, or (None, None) if there isn't
    one. Each Activity Log row carries a stable, monotonically increasing
    `data-activitylogid` — comparing that (not the displayed timestamp
    string, which would need fragile date/timezone parsing) is how a
    "new completion appeared" check is done reliably."""
    page.goto(f"{BASE_URL}/Activity/List")
    page.wait_for_timeout(1000)
    # json.dumps() quotes and escapes the name the same way CSS does for
    # `"` and `\`; interpolating it raw meant a rule named e.g. `6" Box`
    # produced an unparseable selector and blew up the whole check.
    rows = page.locator("div.activity[data-activitylogid]").filter(
        has=page.locator(f".activity__name[title={json.dumps(rule_name)}]")
    )
    if rows.count() == 0:
        return None, None
    row = rows.first
    activity_id = int(row.get_attribute("data-activitylogid"))
    status_text = row.locator(".activity__runtime").first.inner_text().strip()
    return activity_id, status_text


def run_rule(rule_id, rule_name=None):
    """Click "Run Rule" (through its confirmation modal) for the given
    rule id. Logs the outcome to run_history.json and returns a dict:
    {"status": "sent"|"failure", "message": ..., "entry_id": ...,
    "watermark_activity_id": ...}. On "sent", the caller (scheduler.py)
    starts a poll loop (check_completion(), every VERIFY_POLL_MINUTES,
    up to VERIFY_MAX_ATTEMPTS times) using entry_id/watermark_activity_id
    to confirm the run actually completed on MassPrice's side."""
    status = "failure"
    message = ""
    watermark_id = None
    # Distinct from `watermark_id is None`, which is also the legitimate
    # "this rule has no prior Activity entries at all" result. Only a
    # successful capture makes the later comparison meaningful.
    watermark_captured = False
    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)

            # Capture a watermark before submitting, so the later
            # verification check can tell "a new completion appeared"
            # apart from "the same old entry is still sitting there".
            # Best-effort: if this fails for any reason, proceed with the
            # actual run anyway — the run itself matters more than the
            # optional follow-up verification — but skip verification
            # entirely rather than run it without a watermark, which
            # would match the *previous* run's Complete entry and report
            # a false "verified" for a run that may never have landed.
            if rule_name:
                try:
                    watermark_id, _ = _find_latest_activity(page, rule_name)
                    watermark_captured = True
                except Exception:
                    logger.exception("Failed to capture activity watermark for %s", rule_name)
                page.goto(f"{BASE_URL}/Rule/List")
                page.wait_for_timeout(500)

            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 = "sent"
            message = "Run Rule confirmed"
        except Exception as exc:
            message = str(exc)
            logger.exception("run_rule failed for rule %s", rule_id)
        finally:
            browser.close()

    entry_id = _log_history(rule_id, rule_name, status, message)
    will_verify = status == "sent" and bool(rule_name) and watermark_captured
    if will_verify:
        set_verification_pending(entry_id)
    elif status == "sent":
        logger.warning(
            "Run for rule %s sent but not verifiable (no activity watermark)", rule_id
        )

    return {
        "status": status,
        "message": message,
        "entry_id": entry_id,
        "watermark_activity_id": watermark_id,
        "will_verify": will_verify,
    }


def check_completion(rule_name, watermark_id):
    """One-shot check: has a new "Complete" entry appeared for rule_name
    past watermark_id yet? Returns ("complete", "") or ("not_yet", msg).
    Pure check — no storage side effects; the caller (scheduler.py's
    poll loop) decides what that means for the stored history entry.

    A watermark_id of None means the rule had no Activity entries at all
    when the run was submitted, so any entry that appears is new. That's
    only ever reached via a *successful* watermark capture — run_rule()
    refuses to start a poll loop when the capture itself failed."""
    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)

            activity_id, status_text = _find_latest_activity(page, rule_name)
            if activity_id is None:
                return "not_yet", "No matching activity found on MassPrice"
            if watermark_id is not None and activity_id <= watermark_id:
                return "not_yet", "No new completion has appeared yet"
            if status_text and "complete" in status_text.lower():
                return "complete", ""
            return "not_yet", f'Latest activity shows "{status_text}", not Complete yet'
        except Exception as exc:
            logger.exception("check_completion failed for %s", rule_name)
            return "not_yet", str(exc)
        finally:
            browser.close()


def _log_history(rule_id, rule_name, status, message):
    entry_id = uuid.uuid4().hex
    history = storage.load("run_history.json", [])
    history.insert(
        0,
        {
            "id": entry_id,
            "rule_id": rule_id,
            "rule_name": rule_name,
            "timestamp": datetime.now(timezone.utc).isoformat(),
            "status": status,
            "message": message,
            # "pending" only when a completion poll loop has actually
            # been started for this entry (status == "sent" and
            # rule_name was known) — set explicitly by run_rule() below,
            # not here, since _log_history() itself doesn't know whether
            # scheduling succeeded.
            "verification": None,
            "verify_message": None,
        },
    )
    storage.save("run_history.json", history[:200])
    return entry_id


def _find_history_entry(history, entry_id):
    for entry in history:
        if entry.get("id") == entry_id:
            return entry
    return None


def set_verification_pending(entry_id):
    history = storage.load("run_history.json", [])
    entry = _find_history_entry(history, entry_id)
    if entry is None:
        return
    entry["verification"] = "pending"
    storage.save("run_history.json", history)


def set_verification_message(entry_id, message):
    """Record the latest poll's note without changing the terminal
    state — used between retries, while still "pending"."""
    history = storage.load("run_history.json", [])
    entry = _find_history_entry(history, entry_id)
    if entry is None:
        # Entry aged out of the 200-cap between being logged and this
        # delayed check running — nothing to update, not an error.
        return
    entry["verify_message"] = message
    storage.save("run_history.json", history)


def update_verification(entry_id, verification, verify_message):
    """Set a terminal verification result ("complete" or "not_confirmed")."""
    history = storage.load("run_history.json", [])
    entry = _find_history_entry(history, entry_id)
    if entry is None:
        return
    entry["verification"] = verification
    entry["verify_message"] = verify_message
    entry["verified_at"] = datetime.now(timezone.utc).isoformat()
    storage.save("run_history.json", history)


# ── 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()
    same_site = _SAME_SITE_MAP.get(same_site_raw, "Lax")
    secure = bool(raw.get("secure", False))
    # Chromium rejects a SameSite=None cookie that isn't also Secure, and
    # drops it from the imported session without any error — which can
    # quietly cost the one cookie that carried the login. Everything here
    # is https-only in practice, so forcing the flag is safe.
    if same_site == "None":
        secure = True
    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": secure,
        "sameSite": same_site,
    }


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)
            _set_session_expired(False)
            storage.save("notify_state.json", {"session_expired_notified": False})
        finally:
            browser.close()