File indexing completed on 2026-11-09 10:47:22
0001
0002 """Check a deployed ES page and capture its first core plot in Chromium."""
0003 import argparse
0004 import json
0005 import os
0006 from pathlib import Path
0007 import re
0008 import shlex
0009 import subprocess
0010 import tempfile
0011 import time
0012
0013 import requests
0014 import websocket
0015
0016
0017 def credentials():
0018 wanted = ('SWF_REMOTE_CLAUDE_USER', 'SWF_REMOTE_CLAUDE_PASSWORD')
0019 values = {key: os.environ.get(key) for key in wanted}
0020 for line in (Path.home() / '.env').read_text().splitlines():
0021 key, separator, value = line.strip().removeprefix('export ').partition('=')
0022 if separator and key in wanted and not values[key]:
0023 parts = shlex.split(value, comments=True)
0024 if len(parts) != 1:
0025 raise RuntimeError('Cannot parse the configured login field ' + key)
0026 values[key] = parts[0]
0027 if not all(values.values()):
0028 raise RuntimeError('The configured read-only browser login is incomplete')
0029 return values
0030
0031
0032 def main():
0033 parser = argparse.ArgumentParser(description=__doc__)
0034 parser.add_argument('--url', required=True)
0035 parser.add_argument('--output', type=Path, required=True)
0036 parser.add_argument('--expect', default='Unit occupancy')
0037 args = parser.parse_args()
0038 session = requests.Session()
0039 response = session.get(args.url, timeout=40)
0040 response.raise_for_status()
0041 if args.expect not in response.text:
0042 token = re.search(r'name=["\']csrfmiddlewaretoken["\']\s+value=["\']([^"\']+)', response.text)
0043 if token is None:
0044 raise RuntimeError('The requested page has neither the expected marker nor a login form')
0045 auth = credentials()
0046 response = session.post(response.url, data={
0047 'username': auth['SWF_REMOTE_CLAUDE_USER'],
0048 'password': auth['SWF_REMOTE_CLAUDE_PASSWORD'],
0049 'csrfmiddlewaretoken': token.group(1), 'next': args.url},
0050 headers={'Referer': response.url}, timeout=40)
0051 response.raise_for_status()
0052 if args.expect not in response.text:
0053 raise RuntimeError('The deployed page did not contain the expected marker after login')
0054 args.output.parent.mkdir(parents=True, exist_ok=True)
0055 executable = sorted((Path.home() / '.cache/ms-playwright').glob(
0056 'chromium_headless_shell-*/chrome-headless-shell-linux64/chrome-headless-shell'))[-1]
0057 with tempfile.TemporaryDirectory(prefix='es-browser-', dir='/data/swf-tmp') as profile:
0058 process = subprocess.Popen([str(executable), '--no-sandbox', '--disable-gpu',
0059 '--remote-debugging-port=0', '--remote-allow-origins=*',
0060 '--user-data-dir=' + profile, 'about:blank'],
0061 stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
0062 try:
0063 active = Path(profile) / 'DevToolsActivePort'
0064 deadline = time.monotonic() + 15
0065 while not active.exists():
0066 if process.poll() is not None or time.monotonic() > deadline:
0067 raise RuntimeError('Chromium did not expose its local inspection endpoint')
0068 time.sleep(.1)
0069 port = active.read_text().splitlines()[0]
0070 page = requests.get('http://127.0.0.1:' + port + '/json/list', timeout=10).json()[0]
0071 connection = websocket.create_connection(page['webSocketDebuggerUrl'], timeout=30)
0072 sequence = 0
0073
0074 def call(method, params=None):
0075 nonlocal sequence
0076 sequence += 1
0077 connection.send(json.dumps({'id': sequence, 'method': method, 'params': params or {}}))
0078 while True:
0079 answer = json.loads(connection.recv())
0080 if answer.get('id') == sequence:
0081 if answer.get('error'):
0082 raise RuntimeError('Browser inspection failed for ' + method)
0083 return answer.get('result', {})
0084
0085 call('Network.enable')
0086 call('Network.setCookies', {'cookies': [
0087 {'name': c.name, 'value': c.value, 'domain': c.domain,
0088 'path': c.path, 'secure': c.secure} for c in session.cookies]})
0089 call('Page.enable')
0090 call('Emulation.setDeviceMetricsOverride',
0091 {'width': 1600, 'height': 1200, 'deviceScaleFactor': 1, 'mobile': False})
0092 call('Page.navigate', {'url': args.url})
0093 deadline = time.monotonic() + 30
0094 expression = ('document.readyState === "complete" && document.body.innerText.includes('
0095 + json.dumps(args.expect) + ')')
0096 while not call('Runtime.evaluate', {'expression': expression})['result'].get('value'):
0097 if time.monotonic() > deadline:
0098 raise RuntimeError('The authenticated browser did not render the expected page')
0099 time.sleep(.25)
0100 box = call('Runtime.evaluate', {'returnByValue': True, 'expression':
0101 '(()=>{const e=document.querySelector(\'svg[aria-label="core occupancy"]\');'
0102 'if(!e)return null;const r=e.getBoundingClientRect();return {x:Math.max(0,r.x-20),'
0103 'y:Math.max(0,r.y+window.scrollY-145),width:r.width+40,height:r.height+420,scale:1}})()'
0104 })['result'].get('value')
0105 if not box:
0106 raise RuntimeError('The deployed page has no rendered core plot')
0107 import base64
0108 capture = call('Page.captureScreenshot', {'format': 'png', 'captureBeyondViewport': True, 'clip': box})
0109 args.output.write_bytes(base64.b64decode(capture['data']))
0110 connection.close()
0111 print(json.dumps({'url': args.url, 'expected_marker': args.expect,
0112 'screenshot': str(args.output), 'status': 'passed'}))
0113 finally:
0114 process.terminate()
0115 process.wait(timeout=10)
0116
0117
0118 if __name__ == '__main__':
0119 main()
