#!/usr/bin/env python3 """Launch the site in an actual installed Chrome via Playwright's 'chrome' channel. This can help if the headful Playwright Chromium environment differs from your regular Chrome and the site only fully renders in a real browser environment. Usage: python scrapers/statdx_play_chrome.py --profile .chrome_profile --url https://app.statdx.com/ --screenshot out.png Notes: - Requires Playwright and that the 'chrome' browser is available on your machine. - If Playwright cannot find the channel, install Chrome or use the system browser path. """ import argparse import logging import os from time import sleep from urllib.parse import urlparse import hashlib import mimetypes from datetime import datetime, timezone from playwright.sync_api import sync_playwright from dotenv import load_dotenv # Load environment variables from .env in the repository root (if present) load_dotenv() logging.basicConfig(level=logging.INFO, format='%(levelname)s: %(message)s') def main(): parser = argparse.ArgumentParser() parser.add_argument('--profile', '-p', default='.chrome_profile') parser.add_argument('--url', '-u', default='https://app.statdx.com/') parser.add_argument('--domain', '-d', default='app.statdx.com', help='Domain to filter captures') parser.add_argument('--screenshot', '-s', default='chrome_page.png') parser.add_argument('--wait-for', help='Optional selector to wait for before screenshot') parser.add_argument('--headless', action='store_true') parser.add_argument('--username', help='STATdx username (or use STATDX_USERNAME env var)') parser.add_argument('--password', help='STATdx password (or use STATDX_PASSWORD env var)') parser.add_argument('--post-login-selector', default='#ds-app', help='Selector indicating successful login (default: #ds-app)') parser.add_argument('--wait-after-login', type=float, default=3.0, help='Seconds to wait after login before snapshot') parser.add_argument('--keep-open', action='store_true', help='Leave the browser open after actions') parser.add_argument('--capture', action='store_true', help='Capture XHR/fetch requests and responses while the browser is open') parser.add_argument('--capture-output', default='xhr_captured', help='Directory to save captured XHRs') parser.add_argument('--capture-wait', type=float, default=6.0, help='Seconds to wait after reload for XHRs to complete when capture starts') args = parser.parse_args() os.makedirs(args.profile, exist_ok=True) with sync_playwright() as p: try: logging.info('Launching Chrome channel with profile %s (headless=%s)', args.profile, args.headless) context = p.chromium.launch_persistent_context(user_data_dir=args.profile, channel='chrome', headless=args.headless) except Exception as e: logging.error('Failed to launch Chrome channel: %s', e) logging.info('Falling back to default Chromium persistent context') context = p.chromium.launch_persistent_context(user_data_dir=args.profile, headless=args.headless) page = context.new_page() page.goto(args.url) # Setup capture if requested captured = [] if args.capture: os.makedirs(args.capture_output, exist_ok=True) def sanitize_filename(url: str) -> str: parsed = urlparse(url) base = parsed.netloc + parsed.path if parsed.query: base += '?' + parsed.query base = base.strip('/') if not base: base = parsed.netloc safe = ''.join(c if c.isalnum() or c in ('-', '_', '.') else '_' for c in base) h = hashlib.sha1(url.encode('utf-8')).hexdigest()[:8] return f"{safe[:200]}_{h}" def save_binary(body: bytes, out_dir: str, url: str, ts: str, ctype: str): ext = '' try: ext = mimetypes.guess_extension(ctype.split(';')[0]) or '' except Exception: ext = '' fname = f"{sanitize_filename(url)}_{ts}{ext}" path = os.path.join(out_dir, fname) with open(path, 'wb') as fh: fh.write(body) return path def on_response(resp): try: req = resp.request rtype = req.resource_type if args.domain not in req.url: return if rtype not in ('xhr', 'fetch'): return ts = datetime.now(timezone.utc).strftime('%Y%m%dT%H%M%SZ') info = { 'url': req.url, 'method': req.method, 'resource_type': rtype, 'status': resp.status, 'request_headers': dict(req.headers), 'response_headers': dict(resp.headers), 'timestamp': ts, } try: post = req.post_data info['request_post_data'] = post except Exception: try: info['request_post_data'] = req.post_data_text() except Exception: info['request_post_data'] = None ctype = resp.headers.get('content-type', '') try: if 'application/json' in ctype or 'text/' in ctype or 'application/javascript' in ctype: txt = resp.text() info['response_text'] = txt info['response_body_file'] = None else: body = resp.body() path = save_binary(body, args.capture_output, req.url, ts, ctype) info['response_text'] = None info['response_body_file'] = path except Exception: try: body = resp.body() path = save_binary(body, args.capture_output, req.url, ts, ctype) info['response_text'] = None info['response_body_file'] = path except Exception: logging.exception('Failed to read response body for %s', req.url) fname = f"{sanitize_filename(req.url)}_{ts}.json" out_path = os.path.join(args.capture_output, fname) with open(out_path, 'w', encoding='utf-8') as fh: import json json.dump(info, fh, indent=2) logging.info('Captured XHR: %s -> %s', req.url, out_path) captured.append(out_path) except Exception: logging.exception('Error in capture on_response') page.on('response', on_response) # Automatic login if credentials provided username = args.username or os.getenv('STATDX_USERNAME') password = args.password or os.getenv('STATDX_PASSWORD') if username and password: logging.info('Attempting automatic login using provided credentials') # candidate selectors for username and password fields user_selectors = ['input[name="username"]', 'input.usernameSelector', 'input[type="email"]', 'input[name*="email" i]', 'input[name*="user" i]', 'input[type="text"]'] pass_selectors = ['input[type="password"]', 'input[name="password"]', 'input.passwordSelector'] user_sel = None for sel in user_selectors: try: if page.query_selector(sel): user_sel = sel break except Exception: continue pass_sel = None for sel in pass_selectors: try: if page.query_selector(sel): pass_sel = sel break except Exception: continue if not user_sel or not pass_sel: logging.warning('Could not auto-detect username/password fields. Skipping automatic login.') else: try: page.fill(user_sel, username) sleep(0.1) page.fill(pass_sel, password) sleep(0.1) # try submit buttons submit_selectors = ['button.primary.submitSelector', 'button[type=submit]', 'input[type=submit]', 'button:has-text("Sign in")', 'button:has-text("Sign In")', 'button:has-text("Log in")'] clicked = False for s in submit_selectors: try: btn = page.query_selector(s) if btn: btn.click() clicked = True logging.info('Clicked submit button using selector: %s', s) break except Exception: continue if not clicked: logging.info('No submit button clicked; pressing Enter in password field') page.press(pass_sel, 'Enter') # wait for either navigation or a post-login selector try: if args.post_login_selector: page.wait_for_selector(args.post_login_selector, timeout=int(args.wait_after_login * 1000)) except Exception: logging.info('Post-login selector not found within timeout') # small extra wait to let JS render sleep(args.wait_after_login) except Exception: logging.exception('Automatic login attempt failed') if args.wait_for: try: logging.info('Waiting for selector: %s', args.wait_for) page.wait_for_selector(args.wait_for, timeout=20000) except Exception: logging.exception('wait_for selector timed out') # give some time for JS to finish page.wait_for_timeout(1500) # Screenshots are disabled per user request if args.keep_open: logging.info('Leaving browser open for manual inspection. Close the browser window to finish.') try: # keep running until user closes browser context.wait_for_event('close') except Exception: pass try: context.close() except Exception: logging.exception('Error closing context') if __name__ == '__main__': main()