| 1 | """Zion PM - browser runner (Playwright). The universal hand: drive a real Chromium to read, |
| 2 | fill, click, post, screenshot, extract. Read-only specs run inline (full-auto, for research); |
| 3 | specs that submit/post/buy are queued to the outbox and run here only after Zion approves. |
| 4 | |
| 5 | Spec (JSON via stdin or a file path arg): |
| 6 | {"headless": true, "steps": [ |
| 7 | {"action":"goto","url":"https://..."}, |
| 8 | {"action":"fill","selector":"#email","value":"..."}, |
| 9 | {"action":"click","selector":"button[type=submit]"}, |
| 10 | {"action":"wait","ms":1500}, |
| 11 | {"action":"extract","selector":".result","name":"results"}, |
| 12 | {"action":"screenshot","path":"/tmp/shot.png"} |
| 13 | ]} |
| 14 | Prints JSON: {"ok":bool,"extracted":{...},"error":str|null,"shots":[...]} |
| 15 | |
| 16 | Install on the RUNNER host (local ext4, NOT the noexec share): |
| 17 | pip install --user playwright && python3 -m playwright install chromium |
| 18 | """ |
| 19 | import json, sys |
| 20 | |
| 21 | def load_spec(): |
| 22 | if len(sys.argv) > 1: |
| 23 | return json.load(open(sys.argv[1])) |
| 24 | return json.load(sys.stdin) |
| 25 | |
| 26 | def main(): |
| 27 | spec = load_spec() |
| 28 | out = {"ok": False, "extracted": {}, "shots": [], "error": None} |
| 29 | try: |
| 30 | from playwright.sync_api import sync_playwright |
| 31 | except Exception: |
| 32 | out["error"] = "playwright not installed: pip install --user playwright && python3 -m playwright install chromium" |
| 33 | print(json.dumps(out)); return |
| 34 | try: |
| 35 | with sync_playwright() as p: |
| 36 | browser = p.chromium.launch(headless=spec.get("headless", True)) |
| 37 | page = browser.new_page() |
| 38 | for s in spec.get("steps", []): |
| 39 | a = s.get("action") |
| 40 | if a == "goto": page.goto(s["url"], wait_until="domcontentloaded") |
| 41 | elif a == "fill": page.fill(s["selector"], s["value"]) |
| 42 | elif a == "click": page.click(s["selector"]) |
| 43 | elif a == "wait": page.wait_for_timeout(int(s.get("ms", 1000))) |
| 44 | elif a == "waitfor": page.wait_for_selector(s["selector"], timeout=int(s.get("timeout", 15000))) |
| 45 | elif a == "extract": |
| 46 | els = page.query_selector_all(s["selector"]) |
| 47 | out["extracted"][s.get("name", s["selector"])] = [e.inner_text() for e in els] |
| 48 | elif a == "screenshot": |
| 49 | page.screenshot(path=s["path"]); out["shots"].append(s["path"]) |
| 50 | elif a == "press": page.press(s["selector"], s["key"]) |
| 51 | browser.close() |
| 52 | out["ok"] = True |
| 53 | except Exception as e: |
| 54 | out["error"] = str(e) |
| 55 | print(json.dumps(out)) |
| 56 | |
| 57 | if __name__ == "__main__": |
| 58 | main() |