#!/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 time import threading import sys from datetime import datetime, timezone import hashlib from urllib.parse import urlparse from dotenv import load_dotenv import mimetypes from loguru import logger load_dotenv() logger.remove() logger.add(sys.stderr, level="INFO") 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: logger.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) logger.info(f'Saving {url} -> {out_path} (content-type: {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: logger.exception(f'Failed to save response: {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") logger.info(f'Saving page snapshot: {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: logger.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 logger.info(f'Launching Chromium (headless={args.headless}) with profile: {args.profile}, channel={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): logger.debug(f'RESPONSE: {resp.url} ({resp.request.resource_type})') 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: logger.exception('auto-scroll failed') save_page_snapshot(page, args.output, reason=rtype) except Exception: logger.exception('snapshot after response failed') except Exception: logger.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: logger.info(f'PAGE LOG [{msg.type}] {msg.text}') except Exception: logger.info(f'PAGE LOG: {msg}') def on_page_error(exc): logger.error(f'PAGE ERROR: {exc}') page.on('console', on_console) page.on('pageerror', on_page_error) logger.info(f'Open the browser window and interact with the site. Captured responses will be saved to: {args.output}') logger.info('Press Ctrl+C here in the terminal to exit and close the browser (profile saved).') logger.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: logger.info(f'Waiting for selector: {args.wait_for_selector}') page.wait_for_selector(args.wait_for_selector, timeout=20000) except Exception: logger.exception('Waiting for selector timed out') def stdin_watcher(): # Run in a background thread so the main thread can keep the browser running logger.debug('stdin watcher started') for line in sys.stdin: cmd = line.strip().lower() if cmd == '' or cmd == 's' or cmd == 'snapshot' or cmd == 'save': logger.info('Manual snapshot requested via stdin') try: if args.auto_scroll: try: auto_scroll(page) except Exception: logger.exception('manual auto-scroll failed') save_page_snapshot(page, args.output, reason='manual') except Exception: logger.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: logger.info('Shutting down...') finally: context.close() if __name__ == '__main__': main()