Ross
256 lines
11 KiB
Python
256 lines
11 KiB
Python
#!/usr/bin/env python3
|
|
"""Launch the site in an actual installed Chrome via Playwright's 'chrome' channel.
|
|
|
|
This can help if the headful Playwright Chromium environment differs from your regular Chrome and the site
|
|
only fully renders in a real browser environment.
|
|
|
|
Usage:
|
|
python scrapers/statdx_play_chrome.py --profile .chrome_profile --url https://app.statdx.com/ --screenshot out.png
|
|
|
|
Notes:
|
|
- Requires Playwright and that the 'chrome' browser is available on your machine.
|
|
- If Playwright cannot find the channel, install Chrome or use the system browser path.
|
|
"""
|
|
|
|
import argparse
|
|
from loguru import logger
|
|
import os
|
|
from time import sleep
|
|
from urllib.parse import urlparse
|
|
import hashlib
|
|
import mimetypes
|
|
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
|
|
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser()
|
|
parser.add_argument('--profile', '-p', default='.chrome_profile')
|
|
parser.add_argument('--url', '-u', default='https://app.statdx.com/')
|
|
parser.add_argument('--domain', '-d', default='app.statdx.com', help='Domain to filter captures')
|
|
parser.add_argument('--screenshot', '-s', default='chrome_page.png')
|
|
parser.add_argument('--wait-for', help='Optional selector to wait for before screenshot')
|
|
parser.add_argument('--headless', action='store_true')
|
|
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 indicating successful login (default: #ds-app)')
|
|
parser.add_argument('--wait-after-login', type=float, default=3.0, help='Seconds to wait after login before snapshot')
|
|
parser.add_argument('--keep-open', action='store_true', help='Leave the browser open after actions')
|
|
parser.add_argument('--capture', action='store_true', help='Capture XHR/fetch requests and responses while the browser is open')
|
|
parser.add_argument('--capture-output', default='xhr_captured', help='Directory to save captured XHRs')
|
|
parser.add_argument('--capture-wait', type=float, default=6.0, help='Seconds to wait after reload for XHRs to complete when capture starts')
|
|
parser.add_argument('--capture-types', default='xhr,fetch,document', help='Comma-separated list of resource types to capture (e.g. xhr,fetch,document,image)')
|
|
args = parser.parse_args()
|
|
|
|
os.makedirs(args.profile, exist_ok=True)
|
|
|
|
with sync_playwright() as p:
|
|
try:
|
|
logger.info('Launching Chrome channel with profile %s (headless=%s)', args.profile, args.headless)
|
|
context = p.chromium.launch_persistent_context(user_data_dir=args.profile, channel='chrome', headless=args.headless)
|
|
except Exception as e:
|
|
logger.error(f'Failed to launch Chrome channel: {e}')
|
|
logger.info('Falling back to default Chromium persistent context')
|
|
context = p.chromium.launch_persistent_context(user_data_dir=args.profile, headless=args.headless)
|
|
|
|
page = context.new_page()
|
|
page.goto(args.url)
|
|
|
|
# Setup capture if requested
|
|
captured = []
|
|
if args.capture:
|
|
os.makedirs(args.capture_output, exist_ok=True)
|
|
|
|
def sanitize_filename(url: str) -> str:
|
|
parsed = urlparse(url)
|
|
base = parsed.netloc + parsed.path
|
|
if parsed.query:
|
|
base += '?' + parsed.query
|
|
base = base.strip('/')
|
|
if not base:
|
|
base = parsed.netloc
|
|
safe = ''.join(c if c.isalnum() or c in ('-', '_', '.') else '_' for c in base)
|
|
h = hashlib.sha1(url.encode('utf-8')).hexdigest()[:8]
|
|
return f"{safe[:200]}_{h}"
|
|
|
|
def save_binary(body: bytes, out_dir: str, url: str, ts: str, ctype: str):
|
|
ext = ''
|
|
try:
|
|
ext = mimetypes.guess_extension(ctype.split(';')[0]) or ''
|
|
except Exception:
|
|
ext = ''
|
|
fname = f"{sanitize_filename(url)}_{ts}{ext}"
|
|
path = os.path.join(out_dir, fname)
|
|
with open(path, 'wb') as fh:
|
|
fh.write(body)
|
|
return path
|
|
|
|
capture_types = {t.strip().lower() for t in args.capture_types.split(',') if t.strip()}
|
|
|
|
def on_response(resp):
|
|
# lightweight debug log
|
|
logger.debug(f'Response: {resp.status} {resp.url}')
|
|
try:
|
|
req = resp.request
|
|
rtype = req.resource_type
|
|
if args.domain not in req.url:
|
|
return
|
|
if rtype not in capture_types:
|
|
return
|
|
|
|
ts = datetime.now(timezone.utc).strftime('%Y%m%dT%H%M%SZ')
|
|
|
|
info = {
|
|
'url': req.url,
|
|
'method': req.method,
|
|
'resource_type': rtype,
|
|
'status': resp.status,
|
|
'request_headers': dict(req.headers),
|
|
'response_headers': dict(resp.headers),
|
|
'timestamp': ts,
|
|
}
|
|
|
|
# Request post data
|
|
try:
|
|
post = req.post_data
|
|
info['request_post_data'] = post
|
|
except Exception:
|
|
try:
|
|
info['request_post_data'] = req.post_data_text()
|
|
except Exception:
|
|
info['request_post_data'] = None
|
|
|
|
# Response body (text or binary)
|
|
ctype = resp.headers.get('content-type', '')
|
|
try:
|
|
if 'application/json' in ctype or ctype.startswith('text/') or 'javascript' in ctype:
|
|
txt = resp.text()
|
|
info['response_text'] = txt
|
|
info['response_body_file'] = None
|
|
else:
|
|
body = resp.body()
|
|
path = save_binary(body, args.capture_output, req.url, ts, ctype)
|
|
info['response_text'] = None
|
|
info['response_body_file'] = path
|
|
except Exception:
|
|
try:
|
|
body = resp.body()
|
|
path = save_binary(body, args.capture_output, req.url, ts, ctype)
|
|
info['response_text'] = None
|
|
info['response_body_file'] = path
|
|
except Exception:
|
|
logger.exception(f'Failed to read response body for {req.url}')
|
|
|
|
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}')
|
|
captured.append(out_path)
|
|
except Exception:
|
|
logger.exception('Error in capture on_response')
|
|
|
|
page.on('response', on_response)
|
|
|
|
# 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')
|
|
# 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')
|
|
|
|
if args.wait_for:
|
|
try:
|
|
logger.info(f'Waiting for selector: {args.wait_for}')
|
|
page.wait_for_selector(args.wait_for, timeout=20000)
|
|
except Exception:
|
|
logger.exception('wait_for selector timed out')
|
|
|
|
# give some time for JS to finish
|
|
page.wait_for_timeout(1500)
|
|
|
|
# Screenshots are disabled per user request
|
|
|
|
if args.keep_open:
|
|
logger.info('Leaving browser open for manual inspection. Close the browser window to finish.')
|
|
try:
|
|
# keep running until the user presses Enter in the terminal
|
|
input('Browser is open. Press Enter here to close and exit...')
|
|
except Exception:
|
|
pass
|
|
|
|
try:
|
|
context.close()
|
|
except Exception:
|
|
logger.exception('Error closing context')
|
|
|
|
|
|
if __name__ == '__main__':
|
|
main()
|