Ross
491fca2f42
- Implemented `statdx_play_chrome.py` to launch Chrome via Playwright, allowing for headful browsing and capturing XHR requests. - Created `statdx_playwright.py` for headless scraping of the STATdx application, including automatic login and HTML saving. - Developed `statdx_requests.py` for form-based login and scraping, providing a fallback for sites using traditional authentication. - Added example HTML output from Playwright scraping to `topics_playwright.html`. - Captured XHR request data in JSON format for analysis in `xhr_captured/app.statdx.com_csrf_token_8000d0ed_20251014T124122Z.json`.
287 lines
12 KiB
Python
287 lines
12 KiB
Python
#!/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()
|