#!/usr/bin/env python3 """Live-capture browser: open a visible Chromium window, persist the profile, and save HTML/JSON responses as you browse so you can "download content as you browse normally". Usage: - Install Playwright and browsers: python -m pip install playwright python-dotenv; python -m playwright install - Run: python scrapers/statdx_live_capture.py --output captured --domain app.statdx.com What it does: - Launches a persistent Chromium context (profile saved to .playwright_profile by default) so cookies/session persist. - Opens a visible browser (headful). You can interact with it manually. - Listens to network responses. When it sees a response whose content-type is HTML or JSON (or document resource), it saves the response body to the `output` directory with a sanitized filename derived from the URL. Notes / limitations: - This saves HTTP responses as they arrive. It does not run playback for dynamic client-rendered content (but it will save the HTML shell and any XHR/JSON responses used by the client). - You may need to filter by domain to avoid saving lots of third-party assets. - Respect terms of service and only use for accounts you own/have permission to access. """ import os import argparse import logging import time import threading import sys from datetime import datetime, timezone import hashlib from urllib.parse import urlparse from dotenv import load_dotenv import mimetypes load_dotenv() logging.basicConfig(level=logging.INFO, format="%(levelname)s: %(message)s") def sanitize_filename(url: str) -> str: """Create a short filesystem-safe filename from a URL.""" parsed = urlparse(url) # use host + path + query, but keep it short base = parsed.netloc + parsed.path if parsed.query: base += "?" + parsed.query # remove leading/trailing slashes base = base.strip("/") if not base: base = parsed.netloc # replace problematic chars safe = "".join(c if c.isalnum() or c in ('-', '_', '.') else '_' for c in base) # add a short hash to avoid collisions and trim length h = hashlib.sha1(url.encode('utf-8')).hexdigest()[:8] filename = f"{safe[:240]}_{h}" return filename def auto_scroll(page, step_ms: int = 50, step_px: int = 300): """Scroll the page to the bottom slowly to trigger lazy loading. Note: executed in the sync Playwright context by calling page.evaluate. """ try: page.evaluate( """ async (step_px, step_ms) => { const wait = (ms) => new Promise(r => setTimeout(r, ms)); const doc = document.scrollingElement || document.documentElement; let total = 0; while (total < doc.scrollHeight) { doc.scrollBy(0, step_px); total += step_px; await wait(step_ms); } } """, step_px, step_ms, ) except Exception: logging.exception('auto_scroll failed') def save_response(resp, out_dir: str): try: url = resp.url headers = resp.headers ctype = headers.get('content-type', '') # Decide whether to save: HTML or JSON (or document) save_as_text = False save_as_binary = False if 'text/html' in ctype or resp.request.resource_type == 'document': save_as_text = True ext = '.html' elif 'application/json' in ctype or 'text/javascript' in ctype or 'application/javascript' in ctype: save_as_text = True ext = '.json' elif ctype.startswith('text/'): save_as_text = True ext = '.txt' else: # treat as binary and save (images, fonts, video, etc.) save_as_binary = True ext = '' if not (save_as_text or save_as_binary): return False filename_base = sanitize_filename(url) out_path = os.path.join(out_dir, filename_base + ext) logging.info('Saving %s -> %s (content-type: %s)', url, out_path, ctype.split(';')[0]) if save_as_text: try: body = resp.text() except Exception: # fallback to binary then decode body = resp.body().decode('utf-8', errors='replace') with open(out_path, 'w', encoding='utf-8') as fh: fh.write(body) else: body = resp.body() # try to guess extension from content-type try: ctype_main = ctype.split(';')[0] ext_guess = mimetypes.guess_extension(ctype_main) or '' except Exception: ext_guess = '' out_path = out_path + (ext_guess if ext_guess else '') with open(out_path, 'wb') as fh: fh.write(body) return True except Exception as e: logging.exception('Failed to save response: %s', e) return False def save_page_snapshot(page, out_dir: str, reason: str = ''): try: url = page.url ts = datetime.now(timezone.utc).strftime('%Y%m%dT%H%M%SZ') filename_base = sanitize_filename(url) safe_reason = ''.join(c if c.isalnum() or c in ('-', '_') else '_' for c in reason)[:40] out_path = os.path.join(out_dir, f"snapshot_{filename_base}_{ts}_{safe_reason}.html") logging.info('Saving page snapshot: %s -> %s', url, out_path) content = page.content() with open(out_path, 'w', encoding='utf-8') as fh: fh.write(content) # Screenshots are disabled (not saving visual snapshots per user request) return True except Exception: logging.exception('Failed to save page snapshot') return False def main(): parser = argparse.ArgumentParser(description='Live-capture browsing responses using Playwright') parser.add_argument('--output', '-o', default='captured', help='Directory to save captured responses') parser.add_argument('--profile', '-p', default='.playwright_profile', help='Persistent profile directory') parser.add_argument('--domain', '-d', default=os.getenv('STATDX_DOMAIN', 'app.statdx.com'), help='Only capture responses whose URL contains this domain') parser.add_argument('--headless', action='store_true', help='Run headless (no visible browser)') parser.add_argument('--snapshot-delay', type=float, default=0.5, help='Seconds to wait after an XHR before saving a snapshot') parser.add_argument('--devtools', action='store_true', help='Open browser with devtools panel') parser.add_argument('--width', type=int, default=1280, help='Viewport width') parser.add_argument('--height', type=int, default=1600, help='Viewport height') parser.add_argument('--slowmo', type=int, default=0, help='Slow down Playwright actions (ms)') parser.add_argument('--user-agent', dest='user_agent', help='Custom User-Agent header') parser.add_argument('--wait-for-selector', dest='wait_for_selector', help='CSS selector to wait for after navigation') parser.add_argument('--auto-scroll', action='store_true', help='Automatically scroll the page when snapshotting to trigger lazy loads') parser.add_argument('--channel', default='chrome', help='Playwright browser channel to use (e.g. chrome). Set to empty to use bundled Chromium') args = parser.parse_args() os.makedirs(args.output, exist_ok=True) from playwright.sync_api import sync_playwright with sync_playwright() as p: # Use persistent context so cookies/localStorage persist between runs browser_type = p.chromium logging.info('Launching Chromium (headless=%s) with profile: %s, channel=%s', args.headless, args.profile, args.channel) if args.channel: context = browser_type.launch_persistent_context(user_data_dir=args.profile, headless=args.headless, channel=args.channel) else: context = browser_type.launch_persistent_context(user_data_dir=args.profile, headless=args.headless) # Optionally you can set viewport, user agent, etc. here. # Create a new page with the requested viewport page = context.new_page() try: page.set_viewport_size({"width": args.width, "height": args.height}) except Exception: # Older playwright versions may not support set_viewport_size on persistent contexts pass def handle_response(resp): try: if args.domain not in resp.url: return # save textual responses saved = save_response(resp, args.output) # If this response is a document or an XHR/fetch that likely carried data, # also save a rendered page snapshot so client-side content is captured. rtype = resp.request.resource_type if saved and rtype in ("document", "xhr", "fetch"): # page may not have finished rendering; give it a small chance to update try: time.sleep(args.snapshot_delay) if args.auto_scroll: try: auto_scroll(page) except Exception: logging.exception('auto-scroll failed') save_page_snapshot(page, args.output, reason=rtype) except Exception: logging.exception('snapshot after response failed') except Exception: logging.exception('Error in response handler') page.on('response', handle_response) # Log console messages and page errors to help diagnose missing content def on_console(msg): try: logging.info('PAGE LOG [%s] %s', msg.type, msg.text) except Exception: logging.info('PAGE LOG: %s', msg) def on_page_error(exc): logging.error('PAGE ERROR: %s', exc) page.on('console', on_console) page.on('pageerror', on_page_error) logging.info('Open the browser window and interact with the site. Captured responses will be saved to: %s', args.output) logging.info('Press Ctrl+C here in the terminal to exit and close the browser (profile saved).') logging.info('Type "s" and press Enter (or just press Enter) in this terminal to save a manual snapshot of the current page.') # Navigate to the domain root so you can log in manually if needed start_url = f'https://{args.domain}/' page.goto(start_url) # Optionally wait for a selector that indicates the page finished rendering if args.wait_for_selector: try: logging.info('Waiting for selector: %s', args.wait_for_selector) page.wait_for_selector(args.wait_for_selector, timeout=20000) except Exception: logging.exception('Waiting for selector timed out') def stdin_watcher(): # Run in a background thread so the main thread can keep the browser running logging.debug('stdin watcher started') for line in sys.stdin: cmd = line.strip().lower() if cmd == '' or cmd == 's' or cmd == 'snapshot' or cmd == 'save': logging.info('Manual snapshot requested via stdin') try: if args.auto_scroll: try: auto_scroll(page) except Exception: logging.exception('manual auto-scroll failed') save_page_snapshot(page, args.output, reason='manual') except Exception: logging.exception('manual snapshot failed') watcher = threading.Thread(target=stdin_watcher, daemon=True) watcher.start() try: # Keep the script running while the user interacts with the headed browser while True: time.sleep(1) except KeyboardInterrupt: logging.info('Shutting down...') finally: context.close() if __name__ == '__main__': main()