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