diff --git a/.gitignore b/.gitignore index c570d10..ff790f1 100644 --- a/.gitignore +++ b/.gitignore @@ -66,6 +66,7 @@ secrets.json .chrome_profile .playwright_profile +playwright_profile captured/ xhr_captured/ \ No newline at end of file diff --git a/scrapers/capture_with_cdp.py b/scrapers/capture_with_cdp.py new file mode 100644 index 0000000..2a5759b --- /dev/null +++ b/scrapers/capture_with_cdp.py @@ -0,0 +1,219 @@ +#!/usr/bin/env python3 +""" +Capture all network response bodies using Chrome DevTools Protocol (CDP). +This script launches a persistent Chromium/Chrome profile (so you can log in), +attaches a CDP session to the page, enables Network events and calls +Network.getResponseBody for every responseReceived event — saving the body +(text or binary) and a JSON metadata file per response. + +Usage: + python scrapers/capture_with_cdp.py --url https://app.statdx.com/ --output-dir xhr_captured --capture-wait 10 + +Notes: +- Requires Playwright and an installed Chrome/Chromium. Use the same Python + environment you used for other scripts in this repo. +- This method retrieves bodies even for cached or service-worker responses in + many cases because it asks Chrome directly for the response payload. + +""" +import argparse +import base64 +import hashlib +import json +import mimetypes +from datetime import timezone +import time +from datetime import datetime +from pathlib import Path + +try: + from loguru import logger +except Exception: + import logging + logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s") + logger = logging.getLogger("capture_with_cdp") +from playwright.sync_api import sync_playwright + + +def sanitize_fn(s: str) -> str: + safe = [] + for ch in s: + if ch.isalnum() or ch in "._-": + safe.append(ch) + else: + safe.append("-") + name = "".join(safe) + # shorten + if len(name) > 180: + h = hashlib.sha1(s.encode("utf-8")).hexdigest()[:8] + name = name[:120] + "-" + h + return name + + +def guess_ext_from_mime(mime: str) -> str: + if not mime: + return "bin" + mt = mime.split(";")[0].strip() + ext = mimetypes.guess_extension(mt) + if ext: + return ext.lstrip('.') + # fallback mapping + if mt.startswith("image/"): + return mt.split("/")[1] + if mt == "application/json": + return "json" + if mt == "text/html": + return "html" + if mt.startswith("text/"): + return "txt" + return "bin" + + +def write_file(path: Path, data: bytes): + path.parent.mkdir(parents=True, exist_ok=True) + with open(path, "wb") as f: + f.write(data) + + +def main(): + parser = argparse.ArgumentParser() + parser.add_argument("--url", default="https://app.statdx.com/", help="URL to open") + parser.add_argument("--output-dir", default="xhr_captured", help="Directory to write captures") + parser.add_argument("--profile", default="playwright_profile", help="Persistent profile directory") + parser.add_argument("--headless", action="store_true", help="Run headless") + parser.add_argument("--capture-wait", type=int, default=8, help="Seconds to wait after reload to capture XHRs") + parser.add_argument("--keep-open", action="store_true", help="Keep browser open after capture") + args = parser.parse_args() + + out_dir = Path(args.output_dir) + out_dir.mkdir(parents=True, exist_ok=True) + + logger.info("Starting Playwright and launching Chrome/Chromium (persistent profile: {})", args.profile) + + with sync_playwright() as p: + # try to use installed Chrome first + try: + browser = p.chromium.launch_persistent_context(user_data_dir=args.profile, headless=args.headless, channel="chrome") + page = browser.pages[0] if browser.pages else browser.new_page() + except Exception as e: + logger.warning("Failed to launch Chrome channel: {}. Falling back to chromium bundle.", e) + browser = p.chromium.launch_persistent_context(user_data_dir=args.profile, headless=args.headless) + page = browser.pages[0] if browser.pages else browser.new_page() + + # attach CDP session to the page + try: + session = browser.new_cdp_session(page) + except Exception as e: + logger.error("Failed to create CDP session: {}", e) + browser.close() + return + + logger.info("Enabling Network domain on CDP") + session.send("Network.enable") + + seen = set() + + def on_response_received(params): + # params contains requestId and response metadata + try: + requestId = params.get("requestId") + response = params.get("response", {}) + url = response.get("url") + status = response.get("status") + headers = response.get("headers", {}) + mime = headers.get("content-type") or response.get("mimeType") or "" + ts = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ") + + key = f"{url}|{status}|{requestId}" + if key in seen: + return + seen.add(key) + + # attempt to get body + body = None + base64_encoded = False + try: + result = session.send("Network.getResponseBody", {"requestId": requestId}) + if isinstance(result, dict): + body = result.get("body") + base64_encoded = bool(result.get("base64Encoded")) + except Exception as e: + logger.debug("Network.getResponseBody failed for {}: {}", url, e) + + # fallback: if no body, skip writing but still store metadata + # build filenames + safe = sanitize_fn(url.replace('https://', '').replace('http://', '')) + h = hashlib.sha1((url + str(time.time())).encode("utf-8")).hexdigest()[:8] + # determine extension + ext = guess_ext_from_mime(mime) + body_filename = f"{safe}_{h}_{ts}.{ext}" + meta_filename = f"{safe}_{h}_{ts}.json" + + meta = { + "url": url, + "status": status, + "timestamp": ts, + "requestId": requestId, + "mime": mime, + "headers": headers, + } + + if body is not None: + try: + if base64_encoded: + data = base64.b64decode(body) + else: + data = body.encode("utf-8") + write_file(out_dir / body_filename, data) + meta["response_body_file"] = str(out_dir / body_filename) + # try to create a small excerpt for indexing + try: + excerpt = data.decode("utf-8", errors="ignore")[:800] + except Exception: + excerpt = "" + meta["response_excerpt"] = excerpt + logger.info("Saved body {} ({} bytes) for {}", body_filename, len(data), url) + except Exception as e: + logger.warning("Failed to save body for {}: {}", url, e) + else: + meta["response_body_file"] = None + + # write metadata + with open(out_dir / meta_filename, "w", encoding="utf-8") as mf: + json.dump(meta, mf, ensure_ascii=False, indent=2) + + except Exception as exc: + logger.exception("Error handling responseReceived: {}", exc) + + # subscribe to CDP event + session.on("Network.responseReceived", on_response_received) + + logger.info("Opening page: {}. Please log in if needed, then press ENTER to start capture.", args.url) + page.goto(args.url) + + input("After logging in and navigating to the page you want to capture, press Enter to begin capture...\n") + + logger.info("Reloading page to trigger XHRs and network activity") + page.reload() + + # wait a bit to let XHRs fire and be processed by CDP + wait = args.capture_wait + logger.info("Waiting {} seconds to collect responses...", wait) + time.sleep(wait) + + logger.info("Capture pass complete. Metadata and bodies written to {}", out_dir) + + if args.keep_open: + logger.info("Keeping browser open. Close manually when done.") + try: + while True: + time.sleep(1) + except KeyboardInterrupt: + logger.info("Received interrupt, closing.") + browser.close() + else: + browser.close() + + +if __name__ == "__main__": + main() diff --git a/scrapers/extract_sections.py b/scrapers/extract_sections.py new file mode 100644 index 0000000..401c5be --- /dev/null +++ b/scrapers/extract_sections.py @@ -0,0 +1,116 @@ +#!/usr/bin/env python3 +""" +Simple extractor for STATdx snapshot HTML files. +Scans `xhr_captured/` for .html snapshot files (or a single file) and +extracts document sections and subsection points into JSON Lines. + +Usage: + python scrapers/extract_sections.py --input-dir xhr_captured --out-file xhr_captured/extracted_topics.jsonl + +The script looks for
blocks and +collects H1 section titles and H2 subsection headers and
  • points. +""" +from bs4 import BeautifulSoup +import argparse +import json +import glob +import os +from pathlib import Path + + +def extract_from_html(html_text): + soup = BeautifulSoup(html_text, "html.parser") + result = [] + # The page seems to use
    for large groups + for sec in soup.find_all("section", class_="document-page__section"): + section = {} + # group headline (TERMINOLOGY, KEY FACTS, etc.) + headline = sec.find(lambda tag: tag.name in ("div", "header") and tag.get("class") and any("headline3" in c for c in tag.get("class"))) + if headline: + h1 = headline.find("h1") + if h1: + section_title = h1.get_text(strip=True) + else: + section_title = headline.get_text(strip=True) + section["section_title"] = section_title + section["section_id"] = headline.get("id") or None + else: + # fallback: try to find any h1 + h1 = sec.find("h1") + if h1: + section["section_title"] = h1.get_text(strip=True) + section["section_id"] = h1.get("id") + else: + # skip empty sections + continue + + subsections = [] + # common structure:

  • Subheader

    • Point
    • + for li in sec.find_all("li", class_="section-title"): + sub = {} + h2 = li.find("h2") + if h2: + sub["subheader"] = h2.get_text(strip=True) + else: + sub["subheader"] = None + points = [p.get_text(strip=True) for p in li.find_all("li", class_="text")] + # sometimes the structure nests
      directly inside the section + if not points: + # find any
    • inside this li + points = [p.get_text(strip=True) for p in li.find_all("li") if "text" in (p.get("class") or [])] + sub["points"] = points + subsections.append(sub) + + # If no
    • , collect direct
    • in section + if not subsections: + points = [p.get_text(strip=True) for p in sec.find_all("li", class_="text")] + if points: + subsections.append({"subheader": None, "points": points}) + + section["subsections"] = subsections + result.append(section) + + return result + + +def process_file(path): + with open(path, "r", encoding="utf-8") as f: + html = f.read() + sections = extract_from_html(html) + return sections + + +def main(): + parser = argparse.ArgumentParser() + parser.add_argument("--input-dir", default="xhr_captured", help="Directory with captured files") + parser.add_argument("--input-file", help="Single HTML file to process (optional)") + parser.add_argument("--out-file", default="xhr_captured/extracted_topics.jsonl", help="JSONL output file") + args = parser.parse_args() + + paths = [] + if args.input_file: + paths = [args.input_file] + else: + # prefer snapshot_after_capture files, fall back to any .html + pattern = os.path.join(args.input_dir, "snapshot_after_capture_*.html") + paths = glob.glob(pattern) + if not paths: + paths = glob.glob(os.path.join(args.input_dir, "*.html")) + + out_path = Path(args.out_file) + out_path.parent.mkdir(parents=True, exist_ok=True) + + with out_path.open("w", encoding="utf-8") as out: + for p in sorted(paths): + sections = process_file(p) + item = { + "source_file": os.path.relpath(p), + "sections": sections, + } + out.write(json.dumps(item, ensure_ascii=False) + "\n") + + print(f"Wrote {len(paths)} documents to {out_path}") + + +if __name__ == "__main__": + main() diff --git a/scrapers/statdx_play_chrome.py b/scrapers/statdx_play_chrome.py index 5370443..df7aaa3 100644 --- a/scrapers/statdx_play_chrome.py +++ b/scrapers/statdx_play_chrome.py @@ -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') diff --git a/scrapers/unpack_document_content.py b/scrapers/unpack_document_content.py new file mode 100644 index 0000000..71deb84 --- /dev/null +++ b/scrapers/unpack_document_content.py @@ -0,0 +1,129 @@ +#!/usr/bin/env python3 +""" +Extract `documentHtml` from captured document content JSON bodies in `xhr_captured/`. +Writes one .html file per document with a readable filename and prints a short report. + +Usage: + python scrapers/unpack_document_content.py --input-dir xhr_captured --out-dir xhr_captured +""" +import argparse +import glob +import json +import os +from pathlib import Path +import html + + +def slugify(s): + keep = "abcdefghijklmnopqrstuvwxyz0123456789-_" + s = s.lower() + out = [] + for ch in s: + if ch in keep: + out.append(ch) + else: + out.append("-") + res = "".join(out) + # collapse - + while "--" in res: + res = res.replace("--","-") + return res.strip("-")[:200] + + +def main(): + parser = argparse.ArgumentParser() + parser.add_argument("--input-dir", default="xhr_captured") + parser.add_argument("--out-dir", default="xhr_captured") + args = parser.parse_args() + + pattern = os.path.join(args.input_dir, "*document_content_*.json") + files = sorted(glob.glob(pattern)) + if not files: + print("No document_content JSON files found in", args.input_dir) + return + + out_dir = Path(args.out_dir) + out_dir.mkdir(parents=True, exist_ok=True) + + created = [] + for p in files: + try: + with open(p, "r", encoding="utf-8") as f: + meta = json.load(f) + except Exception as e: + print("Failed to read JSON:", p, e) + continue + doc_html = None + body_file = meta.get("response_body_file") + + # If there is an explicit body file, try to open it. Some captures wrote the + # body into a separate file; others accidentally used the same filename as + # the metadata file. Try both. + if body_file: + bf = Path(body_file) + if not bf.exists(): + bf = Path(args.input_dir) / Path(body_file).name + if bf.exists() and bf.is_file(): + try: + with open(bf, "r", encoding="utf-8") as f: + body = json.load(f) + # body is often a dict containing documentHtml + doc_html = body.get("documentHtml") + except Exception: + # not JSON or unreadable + doc_html = None + + # Fallback: sometimes the metadata includes an escaped JSON snippet in + # `response_excerpt` which contains documentHtml. Try to parse it. + if not doc_html: + candidate = meta.get("response_excerpt") + if candidate and "documentHtml" in candidate: + try: + excerpt_json = json.loads(candidate) + doc_html = excerpt_json.get("documentHtml") + except Exception: + doc_html = None + + if not doc_html: + # nothing to extract from this file + continue + + if not doc_html: + continue + + # documentHtml may contain escaped characters; ensure it's properly unescaped + # It's typically a string with HTML already; we'll write it verbatim. + # Build output filename from meta url or source filename + url = meta.get("url") or p + # try to extract last path part or uuid + name_hint = url.rstrip("/").split('/')[-1] + ts = meta.get("timestamp") or "unknown" + out_name = f"extracted_document_{name_hint}_{ts}.html" + out_path = out_dir / out_name + # If the doc_html looks like JSON-escaped, unescape HTML entities + try: + content = doc_html + # if it looks like it's escaped with backslashes, unescape using unicode-escape then html.unescape + # but avoid corrupting valid content + if "\\n" in content or "\\u" in content: + try: + content = content.encode('utf-8').decode('unicode_escape') + except Exception: + pass + content = html.unescape(content) + except Exception: + pass + + try: + with open(out_path, "w", encoding="utf-8") as f: + f.write(content) + created.append(str(out_path)) + except Exception as e: + print("Failed to write:", out_path, e) + + print(f"Extracted {len(created)} document HTML files") + for c in created[:20]: + print(" -", c) + +if __name__ == '__main__': + main()