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