Ross
395 lines
17 KiB
Python
395 lines
17 KiB
Python
#!/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")
|
|
# Provide a lightweight wrapper that supports loguru-style "{}" formatting
|
|
class _LoggerWrapper:
|
|
def __init__(self, name: str):
|
|
self._lg = logging.getLogger(name)
|
|
|
|
def _fmt(self, msg: str, *args):
|
|
try:
|
|
if args:
|
|
return msg.format(*args)
|
|
except Exception:
|
|
pass
|
|
return msg
|
|
|
|
def info(self, msg: str, *args, **kwargs):
|
|
self._lg.info(self._fmt(msg, *args), **kwargs)
|
|
|
|
def warning(self, msg: str, *args, **kwargs):
|
|
self._lg.warning(self._fmt(msg, *args), **kwargs)
|
|
|
|
def debug(self, msg: str, *args, **kwargs):
|
|
self._lg.debug(self._fmt(msg, *args), **kwargs)
|
|
|
|
def exception(self, msg: str = "", *args, **kwargs):
|
|
try:
|
|
self._lg.exception(self._fmt(msg, *args), **kwargs)
|
|
except Exception:
|
|
# fallback: log the exception without formatted message
|
|
self._lg.exception(msg)
|
|
|
|
logger = _LoggerWrapper("capture_with_cdp") # type: ignore
|
|
from playwright.sync_api import sync_playwright
|
|
from dotenv import load_dotenv
|
|
import os
|
|
|
|
# load .env
|
|
load_dotenv()
|
|
|
|
|
|
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")
|
|
parser.add_argument("--continuous", action="store_true", help="Continue capturing requests as you browse until interrupted")
|
|
parser.add_argument("--no-prompt", action="store_true", help="Do not prompt before starting capture (useful for automated runs)")
|
|
parser.add_argument("--username", help="STATdx username (or use STATDX_USERNAME env var)")
|
|
parser.add_argument("--password", help="STATdx password (or use STATDX_PASSWORD env var)")
|
|
parser.add_argument("--post-login-selector", default="#ds-app", help="Selector to wait for after login")
|
|
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:
|
|
context = p.chromium.launch_persistent_context(user_data_dir=args.profile, headless=args.headless, channel="chrome")
|
|
page = context.pages[0] if context.pages else context.new_page()
|
|
except Exception as e:
|
|
logger.warning("Failed to launch Chrome channel: {}. Falling back to chromium bundle.", e)
|
|
context = p.chromium.launch_persistent_context(user_data_dir=args.profile, headless=args.headless)
|
|
page = context.pages[0] if context.pages else context.new_page()
|
|
|
|
# attach CDP session to the page
|
|
try:
|
|
session = context.new_cdp_session(page)
|
|
except Exception as e:
|
|
logger.error("Failed to create CDP session: {}", e)
|
|
try:
|
|
context.close()
|
|
except Exception:
|
|
pass
|
|
return
|
|
|
|
logger.info("Enabling Network domain on CDP")
|
|
session.send("Network.enable")
|
|
# Disable browser cache to force network requests (helps capture cached items)
|
|
try:
|
|
session.send("Network.setCacheDisabled", {"cacheDisabled": True})
|
|
except Exception:
|
|
logger.debug("Could not disable cache via CDP; continuing without cache disable")
|
|
|
|
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; sometimes responseReceived doesn't allow
|
|
# getResponseBody immediately, so we'll try but also rely on
|
|
# a loadingFinished handler below.
|
|
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 (responseReceived) 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)
|
|
# For images, avoid per-capture timestamps: use stable name with hash
|
|
if mime and mime.startswith('image/'):
|
|
body_filename = f"{safe}_{h}.{ext}"
|
|
else:
|
|
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)
|
|
|
|
# loadingFinished is a good fallback: some responses only have bodies
|
|
# available after loadingFinished. We'll attempt to fetch body there
|
|
# if not previously saved for the same requestId.
|
|
def on_loading_finished(params):
|
|
try:
|
|
requestId = params.get("requestId")
|
|
# attempt to get the body now
|
|
try:
|
|
result = session.send("Network.getResponseBody", {"requestId": requestId})
|
|
except Exception:
|
|
result = None
|
|
if result:
|
|
# build a minimal params for saving using response info from CDP
|
|
# We don't have the URL here; try to fetch it via CDP - getRequestPostData isn't available.
|
|
# Instead, reuse on_response_received's behavior by calling getResponseBody and writing
|
|
# files directly here when possible.
|
|
try:
|
|
body = result.get("body")
|
|
base64_encoded = bool(result.get("base64Encoded"))
|
|
if body is None:
|
|
# no body to save
|
|
return
|
|
# fetch additional response info via Network.getResponseBody? we already have body.
|
|
# write a minimal metadata file
|
|
ts = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ")
|
|
safe = sanitize_fn(requestId)
|
|
h = hashlib.sha1((requestId + str(time.time())).encode("utf-8")).hexdigest()[:8]
|
|
ext = "bin"
|
|
try:
|
|
data = base64.b64decode(body) if base64_encoded else body.encode("utf-8")
|
|
# requestId-based captures: use stable name without timestamp for binary/image-like items
|
|
body_filename = f"{safe}_{h}.{ext}"
|
|
write_file(out_dir / body_filename, data)
|
|
meta = {"requestId": requestId, "timestamp": ts, "response_body_file": str(out_dir / body_filename)}
|
|
meta_filename = f"{safe}_{h}_{ts}.json"
|
|
with open(out_dir / meta_filename, "w", encoding="utf-8") as mf:
|
|
json.dump(meta, mf, ensure_ascii=False, indent=2)
|
|
logger.info("Saved body via loadingFinished: {}", body_filename)
|
|
except Exception:
|
|
pass
|
|
except Exception:
|
|
pass
|
|
except Exception:
|
|
logger.exception("Error in loadingFinished handler")
|
|
|
|
session.on("Network.loadingFinished", on_loading_finished)
|
|
|
|
logger.info("Opening page: {}.", args.url)
|
|
page.goto(args.url)
|
|
|
|
# Automatic login if credentials provided
|
|
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')
|
|
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 user_sel and pass_sel:
|
|
try:
|
|
page.fill(user_sel, username)
|
|
time.sleep(0.1)
|
|
page.fill(pass_sel, password)
|
|
time.sleep(0.1)
|
|
# try submit methods
|
|
submit_selectors = ['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('Clicked submit button using selector: %s', s)
|
|
break
|
|
except Exception:
|
|
continue
|
|
if not clicked:
|
|
page.press(pass_sel, 'Enter')
|
|
|
|
# wait for post-login selector
|
|
try:
|
|
if args.post_login_selector:
|
|
page.wait_for_selector(args.post_login_selector, timeout=10000)
|
|
except Exception:
|
|
logger.info('Post-login selector not found within timeout')
|
|
except Exception:
|
|
logger.exception('Automatic login attempt failed')
|
|
|
|
# Start capture: either prompt the user or begin immediately depending on flags
|
|
try:
|
|
if not args.no_prompt and not args.continuous:
|
|
try:
|
|
input("After logging in and navigating to the page you want to capture, press Enter to begin capture...\n")
|
|
except Exception:
|
|
# non-interactive; proceed
|
|
pass
|
|
|
|
logger.info("Reloading page to trigger XHRs and network activity")
|
|
try:
|
|
page.reload()
|
|
except Exception:
|
|
logger.exception('Failed to reload page; continuing capture')
|
|
|
|
# wait a bit to let XHRs fire and be processed by CDP
|
|
wait = args.capture_wait
|
|
logger.info("Waiting {} seconds to collect initial responses...", wait)
|
|
time.sleep(wait)
|
|
|
|
logger.info("Initial capture pass complete. Metadata and bodies written to {}", out_dir)
|
|
|
|
if args.continuous:
|
|
logger.info("Continuous capture enabled: saving responses as you browse. Press Ctrl+C to stop.")
|
|
try:
|
|
while True:
|
|
time.sleep(1)
|
|
except KeyboardInterrupt:
|
|
logger.info("Interrupted by user; closing browser and exiting")
|
|
else:
|
|
logger.info("One-shot capture complete.")
|
|
|
|
finally:
|
|
try:
|
|
context.close()
|
|
except Exception:
|
|
try:
|
|
context.close()
|
|
except Exception:
|
|
pass
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|