153 lines
6.3 KiB
Python
153 lines
6.3 KiB
Python
"""Runs the bot in its own process and turns its log file into Run tab events.
|
|
|
|
No Qt in here (ADR-0002): the bot runs as a separate process with no Qt, writes every line
|
|
to a log file, and this module reads that file every 0.2 s. It does not read a stdout pipe,
|
|
because a windowed PyInstaller child's pipe is block-buffered and lines arrive late.
|
|
|
|
Bot lines it understands:
|
|
[CPSTATE] active N / [CPSTATE] done N checkpoint progress (ADR-0001)
|
|
[UI] text, [UI:ok] / [UI:warn] / [UI:error] text lines the operator sees
|
|
Error / exception lines always shown, even untagged
|
|
__RUN_SUMMARY__{"total":n,"ok":n,"failed":[[item, reason], ...]} the outcome
|
|
Everything else stays in the log file only.
|
|
"""
|
|
import json
|
|
import re
|
|
import subprocess
|
|
import sys
|
|
import threading
|
|
import time
|
|
|
|
SUMMARY = "__RUN_SUMMARY__"
|
|
_CP = re.compile(r"^\[CPSTATE\]\s+(active|done)\s+(\d+)")
|
|
_UI = re.compile(r"^\[UI(?::(ok|warn|error))?\]\s?(.*)$")
|
|
# Untagged lines that still reach the screen: the last line of a traceback
|
|
# ("ValueError: ...") and lines starting with ERROR.
|
|
# ponytail: a regex guess, so add a [UI:error] line to the bot when an error shows badly.
|
|
_ERR = re.compile(r"^([A-Za-z_][\w.]*(Error|Exception)\b|ERROR\b)")
|
|
_UI_TYPE = {None: "info", "ok": "success", "warn": "warn", "error": "error"}
|
|
_NO_WINDOW = getattr(subprocess, "CREATE_NO_WINDOW", 0)
|
|
|
|
|
|
def parse(line):
|
|
"""One bot line -> (event, data), or None when it stays off screen."""
|
|
m = _CP.match(line)
|
|
if m:
|
|
return "cpstate", {"kind": m[1], "cp": int(m[2])}
|
|
m = _UI.match(line)
|
|
if m:
|
|
return "log", {"msg": m[2], "type": _UI_TYPE[m[1]]}
|
|
if _ERR.match(line):
|
|
return "log", {"msg": line, "type": "error"}
|
|
return None
|
|
|
|
|
|
def outcome(summary, stuck_name=None):
|
|
"""(state, message) for the page: done (green), partial (amber) or error (red)."""
|
|
if stuck_name:
|
|
return "error", f"Stuck at {stuck_name}, run again"
|
|
if summary is None:
|
|
return "error", "Failed: the bot stopped without a summary (see the run log)"
|
|
total = int(summary.get("total") or 0)
|
|
failed = summary.get("failed") or []
|
|
if failed and len(failed) >= total:
|
|
return "error", f"Failed: all {len(failed)} item(s) failed"
|
|
if failed:
|
|
return "partial", f"Finished: {len(failed)} of {total} failed"
|
|
return "done", f"Completed: {total} item(s)" if total else "Completed"
|
|
|
|
|
|
def kill_tree(proc, kill_also=()):
|
|
"""Kill the bot and everything it started, then any extra .exe names (e.g. a driver)."""
|
|
if proc is not None and proc.poll() is None:
|
|
if sys.platform.startswith("win"):
|
|
subprocess.run(["taskkill", "/PID", str(proc.pid), "/T", "/F"],
|
|
capture_output=True, timeout=15, creationflags=_NO_WINDOW)
|
|
else:
|
|
proc.kill() # ponytail: direct child only off Windows; use a process group if bots spawn children there
|
|
for image in kill_also:
|
|
subprocess.run(["taskkill", "/F", "/IM", image],
|
|
capture_output=True, timeout=10, creationflags=_NO_WINDOW)
|
|
|
|
|
|
class Run:
|
|
"""One bot run. emit(event, data) is called from the run thread."""
|
|
|
|
def __init__(self, cmd, log_path, checkpoints, emit, kill_also=(), cwd=None):
|
|
self.cmd, self.log_path, self.emit = cmd, str(log_path), emit
|
|
self.checkpoints = checkpoints # [(name, max_seconds)], 0 = no limit
|
|
self.kill_also, self.cwd = kill_also, cwd
|
|
self.proc = None
|
|
self.summary = None
|
|
self.active, self.since, self.stuck = 0, 0.0, None
|
|
self.thread = threading.Thread(target=self._run, daemon=True)
|
|
|
|
def start(self):
|
|
self.thread.start()
|
|
return self
|
|
|
|
def kill(self):
|
|
kill_tree(self.proc, self.kill_also)
|
|
|
|
def _line(self, line):
|
|
if line.startswith(SUMMARY):
|
|
try:
|
|
self.summary = json.loads(line[len(SUMMARY):])
|
|
except ValueError:
|
|
self.emit("log", {"msg": "Could not read the bot's summary line", "type": "warn"})
|
|
return
|
|
ev = parse(line)
|
|
if not ev:
|
|
return
|
|
if ev[0] == "cpstate":
|
|
kind, n = ev[1]["kind"], ev[1]["cp"]
|
|
if kind == "active":
|
|
self.active, self.since = n, time.monotonic()
|
|
elif n == self.active:
|
|
self.active = 0
|
|
self.emit(*ev)
|
|
|
|
def _check_stuck(self):
|
|
if not self.active or self.active > len(self.checkpoints):
|
|
return False
|
|
name, limit = self.checkpoints[self.active - 1]
|
|
if limit and time.monotonic() - self.since > limit:
|
|
self.stuck = name
|
|
self.emit("cpstate", {"kind": "stuck", "cp": self.active})
|
|
self.emit("log", {"msg": f"{name} took longer than {limit}s, stopping the run", "type": "error"})
|
|
self.kill()
|
|
return True
|
|
return False
|
|
|
|
def _run(self):
|
|
self.emit("status", {"state": "running", "msg": "Running automation..."})
|
|
open(self.log_path, "w", encoding="utf-8").close()
|
|
try:
|
|
self.proc = subprocess.Popen(self.cmd, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL,
|
|
cwd=self.cwd, creationflags=_NO_WINDOW)
|
|
except OSError as e:
|
|
self.emit("status", {"state": "error", "msg": f"Could not start the bot: {e}"})
|
|
return
|
|
buf = ""
|
|
with open(self.log_path, encoding="utf-8", errors="replace") as f:
|
|
while True:
|
|
exited = self.proc.poll() is not None # check before reading, so the last lines are never lost
|
|
chunk = f.read()
|
|
if chunk:
|
|
buf += chunk
|
|
*lines, buf = buf.split("\n")
|
|
for ln in lines:
|
|
self._line(ln.rstrip("\r"))
|
|
elif exited or self.stuck:
|
|
break
|
|
elif self._check_stuck():
|
|
self.proc.wait(timeout=30)
|
|
else:
|
|
time.sleep(0.2)
|
|
if buf.strip():
|
|
self._line(buf.rstrip("\r"))
|
|
for item, reason in (self.summary or {}).get("failed") or []:
|
|
self.emit("log", {"msg": f"✕ {item}: {reason}", "type": "error"})
|
|
state, msg = outcome(None if self.stuck else self.summary, self.stuck)
|
|
self.emit("status", {"state": state, "msg": msg})
|