Add scripts for capturing network responses and extracting document content
- Implement capture_with_cdp.py to capture all network response bodies using Chrome DevTools Protocol. - Create extract_sections.py to extract sections from STATdx snapshot HTML files into JSON Lines. - Add unpack_document_content.py to extract `documentHtml` from captured JSON bodies and save as HTML files. - Update .gitignore to include playwright_profile.
This commit is contained in:
Ross
+172
-22
@@ -19,13 +19,14 @@ from time import sleep
|
||||
from urllib.parse import urlparse
|
||||
import hashlib
|
||||
import mimetypes
|
||||
import json
|
||||
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()
|
||||
from loguru import logger
|
||||
|
||||
|
||||
|
||||
|
||||
@@ -62,6 +63,70 @@ def main():
|
||||
page = context.new_page()
|
||||
page.goto(args.url)
|
||||
|
||||
# Automatic login if credentials provided (run before any capture)
|
||||
username = args.username or os.getenv('STATDX_USERNAME')
|
||||
password = args.password or os.getenv('STATDX_PASSWORD')
|
||||
if username and password:
|
||||
logger.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:
|
||||
logger.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
|
||||
logger.info(f'Clicked submit button using selector: {s}')
|
||||
break
|
||||
except Exception:
|
||||
continue
|
||||
if not clicked:
|
||||
logger.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:
|
||||
logger.info('Post-login selector not found within timeout')
|
||||
|
||||
# small extra wait to let JS render
|
||||
sleep(args.wait_after_login)
|
||||
except Exception:
|
||||
logger.exception('Automatic login attempt failed')
|
||||
|
||||
# Setup capture if requested
|
||||
captured = []
|
||||
if args.capture:
|
||||
@@ -134,51 +199,136 @@ def main():
|
||||
except Exception:
|
||||
info['request_post_data'] = None
|
||||
|
||||
# Response body (text or binary)
|
||||
ctype = resp.headers.get('content-type', '')
|
||||
# Response body (always save to a standalone file)
|
||||
ctype = (resp.headers.get('content-type') or '').lower()
|
||||
body_path = None
|
||||
excerpt = None
|
||||
try:
|
||||
# Save HTML/document responses as separate .html files
|
||||
# Document / HTML
|
||||
if 'text/html' in ctype or rtype == 'document':
|
||||
txt = resp.text()
|
||||
html_name = f"{sanitize_filename(req.url)}_{ts}.html"
|
||||
html_path = os.path.join(args.capture_output, html_name)
|
||||
with open(html_path, 'w', encoding='utf-8') as fh:
|
||||
ext = '.html'
|
||||
body_name = f"{sanitize_filename(req.url)}_{ts}{ext}"
|
||||
body_path = os.path.join(args.capture_output, body_name)
|
||||
with open(body_path, 'w', encoding='utf-8') as fh:
|
||||
fh.write(txt)
|
||||
info['response_text'] = None
|
||||
info['response_body_file'] = html_path
|
||||
elif 'application/json' in ctype or ctype.startswith('text/') or 'javascript' in ctype:
|
||||
# keep small textual responses inline in the metadata JSON
|
||||
excerpt = txt[:400]
|
||||
# JSON
|
||||
elif 'application/json' in ctype or req.url.endswith('.json'):
|
||||
txt = resp.text()
|
||||
info['response_text'] = txt
|
||||
info['response_body_file'] = None
|
||||
ext = '.json'
|
||||
body_name = f"{sanitize_filename(req.url)}_{ts}{ext}"
|
||||
body_path = os.path.join(args.capture_output, body_name)
|
||||
# pretty-print JSON when possible
|
||||
try:
|
||||
parsed = json.loads(txt)
|
||||
with open(body_path, 'w', encoding='utf-8') as fh:
|
||||
json.dump(parsed, fh, indent=2)
|
||||
except Exception:
|
||||
with open(body_path, 'w', encoding='utf-8') as fh:
|
||||
fh.write(txt)
|
||||
excerpt = txt[:400]
|
||||
# text/javascript or other text
|
||||
elif ctype.startswith('text/') or 'javascript' in ctype:
|
||||
txt = resp.text()
|
||||
ext = '.txt'
|
||||
body_name = f"{sanitize_filename(req.url)}_{ts}{ext}"
|
||||
body_path = os.path.join(args.capture_output, body_name)
|
||||
with open(body_path, 'w', encoding='utf-8') as fh:
|
||||
fh.write(txt)
|
||||
excerpt = txt[:400]
|
||||
else:
|
||||
# binary (images, fonts, etc.)
|
||||
body = resp.body()
|
||||
path = save_binary(body, args.capture_output, req.url, ts, ctype)
|
||||
info['response_text'] = None
|
||||
info['response_body_file'] = path
|
||||
body_path = save_binary(body, args.capture_output, req.url, ts, ctype)
|
||||
excerpt = None
|
||||
except Exception:
|
||||
try:
|
||||
# fallback to raw binary save
|
||||
body = resp.body()
|
||||
path = save_binary(body, args.capture_output, req.url, ts, ctype)
|
||||
info['response_text'] = None
|
||||
info['response_body_file'] = path
|
||||
body_path = save_binary(body, args.capture_output, req.url, ts, ctype)
|
||||
except Exception:
|
||||
logger.exception(f'Failed to read response body for {req.url}')
|
||||
|
||||
info['response_body_file'] = body_path
|
||||
info['response_excerpt'] = excerpt
|
||||
|
||||
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)
|
||||
|
||||
logger.info(f'Captured {rtype}: {req.url} -> {out_path}')
|
||||
logger.info(f'Captured {rtype}: {req.url} -> {out_path} (body: {body_path})')
|
||||
captured.append(out_path)
|
||||
# append to a simple JSONL index for quick searching
|
||||
try:
|
||||
index_entry = {
|
||||
'url': req.url,
|
||||
'resource_type': rtype,
|
||||
'timestamp': ts,
|
||||
'body_file': body_path,
|
||||
'meta_file': out_path,
|
||||
'excerpt': excerpt,
|
||||
}
|
||||
index_path = os.path.join(args.capture_output, 'capture_index.jsonl')
|
||||
with open(index_path, 'a', encoding='utf-8') as idx:
|
||||
idx.write(json.dumps(index_entry, ensure_ascii=False) + '\n')
|
||||
except Exception:
|
||||
logger.exception('Failed to write capture index entry')
|
||||
except Exception:
|
||||
logger.exception('Error in capture on_response')
|
||||
|
||||
page.on('response', on_response)
|
||||
|
||||
# Wait for user to start capture so they can log in / navigate manually if needed.
|
||||
try:
|
||||
input('Press Enter to start capture (the page will be reloaded to trigger XHR/fetch requests)...')
|
||||
except Exception:
|
||||
# non-interactive environments may raise; continue
|
||||
pass
|
||||
|
||||
logger.info('Reloading page to trigger API calls...')
|
||||
try:
|
||||
page.reload()
|
||||
except Exception:
|
||||
logger.exception('Failed to reload page; continuing capture')
|
||||
|
||||
# Try to auto-scroll to trigger lazy loading
|
||||
try:
|
||||
page.evaluate(
|
||||
"""
|
||||
async () => {
|
||||
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, 400);
|
||||
total += 400;
|
||||
await wait(150);
|
||||
}
|
||||
}
|
||||
"""
|
||||
)
|
||||
except Exception:
|
||||
logger.exception('auto-scroll failed during capture')
|
||||
|
||||
# Wait for the configured capture window so responses arrive and are saved
|
||||
import time as _time
|
||||
logger.info('Waiting %.1f seconds for responses...', args.capture_wait)
|
||||
_time.sleep(args.capture_wait)
|
||||
|
||||
# Save a final rendered page snapshot (HTML)
|
||||
try:
|
||||
html = page.content()
|
||||
ts = datetime.now(timezone.utc).strftime('%Y%m%dT%H%M%SZ')
|
||||
fname = sanitize_filename(page.url)
|
||||
snap_path = os.path.join(args.capture_output, f'snapshot_after_capture_{fname}_{ts}.html')
|
||||
with open(snap_path, 'w', encoding='utf-8') as fh:
|
||||
fh.write(html)
|
||||
logger.info('Saved final page snapshot: %s', snap_path)
|
||||
except Exception:
|
||||
logger.exception('Failed to save final snapshot after capture')
|
||||
|
||||
# Automatic login if credentials provided
|
||||
username = args.username or os.getenv('STATDX_USERNAME')
|
||||
password = args.password or os.getenv('STATDX_PASSWORD')
|
||||
|
||||
Reference in New Issue
Block a user