VasanthandClaude Opus 5.5
162edd41df
- templates/bot_runner.py (new, no Qt): runs the bot as a child process (unbuffered, UTF-8), tails its output, keeps raw lines in bot_raw.log, and turns lines into step/log/status events. Step markers and friendly rules use specific substrings, first match wins; unmatched error lines are always shown, other unmatched lines hidden. Dry run: python dashboard/bot_runner.py prints the events the page would get. - bridge.py: delegates to bot_runner, one run at a time, restores the saved dark mode on every page load (fixes sample-data mode in PyQt), adds load_latest_report / download_report slots the page calls. - SKILL.md: skip the stack question for a PyQt project; desktop wiring steps from the bot's real output; verify steps for the host. - tests: assert-based self-check for the line logic and run loop; Seam 1 run replay (status dot, Run button, steps, log) and wired-page mode; fixture b-pyqt-bot and its Seam 2 eval (all 14 checks pass). Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
65 lines
3.0 KiB
Python
65 lines
3.0 KiB
Python
"""Bot wiring for the desktop host: runs the bot and turns its output into page events.
|
|
|
|
No Qt here, so it can be tested and dry-run on its own. Fill BOT_CMD, STEP_MARKERS and
|
|
FRIENDLY_RULES from lines the bot really prints.
|
|
|
|
Dry run, no window: python bot_runner.py (prints each event the page would get)
|
|
"""
|
|
import json, os, subprocess, sys
|
|
from pathlib import Path
|
|
|
|
BOT_CMD = [sys.executable, "-m", "{{BOT_MODULE}}"] # the bot, started as a child process
|
|
BOT_CWD = Path(__file__).resolve().parents[1] # folder BOT_CMD runs in (the project root)
|
|
RAW_LOG = Path(__file__).with_name("bot_raw.log") # every raw bot line, kept for audit
|
|
# (specific substring, step label from STEPS in dashboard.html, pct). First match wins.
|
|
# e.g. ("[step 2]", "Login", 40). Never a bare word like "login": it matches unrelated lines.
|
|
STEP_MARKERS = []
|
|
# (specific substring, plain-English message, type: info|success|warn|error). First match wins.
|
|
# e.g. ("[login] done", "Logged in.", "success"). Unmatched lines are hidden, except error lines.
|
|
FRIENDLY_RULES = []
|
|
# A line holding any of these is an error line. Unmatched error lines are shown raw, never hidden.
|
|
ERROR_WORDS = ("error", "exception", "traceback", "fatal", "failed")
|
|
|
|
|
|
def _first(rules, line):
|
|
low = line.lower()
|
|
return next((r for r in rules if r[0].lower() in low), None)
|
|
|
|
|
|
def events_for(line):
|
|
"""The page events, as (event, data) pairs, for one raw bot line."""
|
|
out = []
|
|
if marker := _first(STEP_MARKERS, line):
|
|
out.append(("step", {"label": marker[1], "pct": marker[2]}))
|
|
if rule := _first(FRIENDLY_RULES, line):
|
|
out.append(("log", {"msg": rule[1], "type": rule[2]}))
|
|
elif any(w in line.lower() for w in ERROR_WORDS): # never hide a real failure
|
|
out.append(("log", {"msg": line, "type": "error"}))
|
|
return out
|
|
|
|
|
|
def run(send):
|
|
"""Run the bot in its own process and send(event, data) for each line it prints, as it prints it."""
|
|
send("status", {"state": "running", "msg": "Running…"})
|
|
env = {**os.environ, "PYTHONUNBUFFERED": "1", "PYTHONIOENCODING": "utf-8"} # live lines; no cp1252 crash
|
|
try:
|
|
raw = open(RAW_LOG, "a", encoding="utf-8")
|
|
p = subprocess.Popen(BOT_CMD, cwd=BOT_CWD, env=env, stdout=subprocess.PIPE, stderr=subprocess.STDOUT,
|
|
text=True, encoding="utf-8", errors="replace")
|
|
except OSError as e:
|
|
send("log", {"msg": f"Could not start the bot: {e}", "type": "error"})
|
|
send("status", {"state": "error", "msg": "Could not start the bot"})
|
|
return
|
|
with p, raw:
|
|
for line in p.stdout:
|
|
raw.write(line)
|
|
for event, data in events_for(line.rstrip()):
|
|
send(event, data)
|
|
ok = p.returncode == 0
|
|
send("status", {"state": "done", "msg": "Completed"} if ok else
|
|
{"state": "error", "msg": f"Failed (exit code {p.returncode})"})
|
|
|
|
|
|
if __name__ == "__main__":
|
|
run(lambda event, data: print(json.dumps({"event": event, "data": data}), flush=True))
|