Add rpa-run-tab skill
A Tebra-style Run tab for desktop PyQt6 bots, built in about 5 minutes: Trigger button, Excel input card, progress with timer, live log, checkpoint boxes with time limits, and green/amber/red endings. It is checked with a fake bot (run_tab.py --selfcheck). The design decisions are in rpa-run-tab/docs/adr. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
VasanthandClaude Opus 5.5 committed
commit
a57b32e8cb
12 files changed
+1435
No files matched your search
@@ -0,0 +1,345 @@
|
||||
"""Run tab: PyQt6 window + bridge for the page in run_tab.html.
|
||||
|
||||
python run_tab/run_tab.py open the Run tab
|
||||
python run_tab/run_tab.py --selfcheck run the fake bot through every ending, save screenshots
|
||||
|
||||
The same file is also the bot's process: the window starts `run_tab.py --run-bot ...`
|
||||
(or `<app>.exe --run-bot ...` when frozen). That branch loads no Qt (ADR-0002).
|
||||
"""
|
||||
import json
|
||||
import os
|
||||
import sys
|
||||
import threading
|
||||
from datetime import datetime
|
||||
from pathlib import Path
|
||||
|
||||
# ---- CONFIG: the rpa-run-tab skill fills these in ------------------------------------
|
||||
TITLE = "Run automation"
|
||||
SUBTITLE = "Upload the input Excel and start the bot"
|
||||
BOT = "bot.py" # "path/to/bot.py" (from the project root) or "package.module:function"
|
||||
HAS_INPUT = True # False hides the input card; Trigger is ready straight away
|
||||
INPUT_HINT = "Excel workbook with the input sheets"
|
||||
CHECKPOINTS = [ # (name, seconds before it counts as stuck; 0 = no limit)
|
||||
("Sign in", 300),
|
||||
("Process items", 0),
|
||||
("Finish up", 300),
|
||||
]
|
||||
SAMPLE_SHEETS = {} # {"Sheet name": ["Column", ...]}; empty hides the Sample input button
|
||||
KILL_ALSO = [] # extra .exe names to kill with the bot, e.g. ["chromedriver.exe"]
|
||||
APP_ID = "RunTab" # QSettings name and the page's dark-mode key
|
||||
# ---------------------------------------------------------------------------------------
|
||||
|
||||
FROZEN = getattr(sys, "frozen", False)
|
||||
HERE = Path(getattr(sys, "_MEIPASS", Path(__file__).resolve().parent))
|
||||
ROOT = Path(sys.executable).parent if FROZEN else Path(__file__).resolve().parent.parent
|
||||
LOGS = ROOT / "logs"
|
||||
|
||||
|
||||
def _arg(flag, default=""):
|
||||
return sys.argv[sys.argv.index(flag) + 1] if flag in sys.argv else default
|
||||
|
||||
|
||||
def _watch_parent(ppid):
|
||||
"""Exit the bot the moment the window's process dies, so it never runs orphaned."""
|
||||
try:
|
||||
if sys.platform.startswith("win"):
|
||||
import ctypes
|
||||
k32 = ctypes.windll.kernel32
|
||||
h = k32.OpenProcess(0x00100000, False, ppid) # SYNCHRONIZE
|
||||
if h:
|
||||
k32.WaitForSingleObject(h, 0xFFFFFFFF)
|
||||
else:
|
||||
import time
|
||||
while os.getppid() == ppid:
|
||||
time.sleep(1)
|
||||
except Exception:
|
||||
return # can't watch: keep running
|
||||
os._exit(1)
|
||||
|
||||
|
||||
def run_bot_child():
|
||||
"""The bot's process: all output goes to --log, the bot runs, its result becomes the summary line."""
|
||||
log = open(_arg("--log"), "a", encoding="utf-8", buffering=1)
|
||||
sys.stdout = sys.stderr = log
|
||||
threading.Thread(target=_watch_parent, args=(int(_arg("--parent-pid", "0")),), daemon=True).start()
|
||||
excel, bot = _arg("--excel"), _arg("--bot", BOT)
|
||||
sys.path.insert(0, str(ROOT))
|
||||
os.chdir(ROOT)
|
||||
if bot.endswith(".py"):
|
||||
import runpy
|
||||
sys.argv = [bot] + ([excel] if excel else [])
|
||||
sys.path.insert(0, str((ROOT / bot).parent))
|
||||
runpy.run_path(str(ROOT / bot), run_name="__main__") # a plain script prints its own summary
|
||||
else:
|
||||
import importlib
|
||||
mod, func = bot.split(":")
|
||||
sys.path.insert(0, str(HERE))
|
||||
result = getattr(importlib.import_module(mod), func)(*([excel] if excel else []))
|
||||
if isinstance(result, dict):
|
||||
print("__RUN_SUMMARY__" + json.dumps(result))
|
||||
log.flush()
|
||||
os._exit(0)
|
||||
|
||||
|
||||
if "--run-bot" in sys.argv:
|
||||
run_bot_child()
|
||||
|
||||
# ---- everything below runs only in the window's process --------------------------------
|
||||
sys.path.insert(0, str(HERE))
|
||||
from PyQt6.QtCore import QObject, QSettings, Qt, QTimer, QUrl, pyqtSignal, pyqtSlot # noqa: E402
|
||||
from PyQt6.QtWebChannel import QWebChannel # noqa: E402
|
||||
from PyQt6.QtWebEngineWidgets import QWebEngineView # noqa: E402
|
||||
from PyQt6.QtWidgets import QApplication, QFileDialog, QMainWindow, QMessageBox # noqa: E402
|
||||
|
||||
import runner # noqa: E402
|
||||
|
||||
SEARCH_DIRS = [ROOT, ROOT / "input", Path.home() / "Downloads", Path.home() / "Desktop", Path.home() / "Documents"]
|
||||
|
||||
|
||||
def page_html():
|
||||
cfg = {"title": TITLE, "sub": SUBTITLE, "input": HAS_INPUT, "inputHint": INPUT_HINT,
|
||||
"sample": bool(SAMPLE_SHEETS), "darkKey": APP_ID + "_dark",
|
||||
"checkpoints": [{"name": n, "max": m} for n, m in CHECKPOINTS]}
|
||||
return (HERE / "run_tab.html").read_text(encoding="utf-8").replace("%%CONFIG%%", json.dumps(cfg))
|
||||
|
||||
|
||||
def bot_cmd(excel, log, bot=BOT):
|
||||
me = [sys.executable] if FROZEN else [sys.executable, str(Path(__file__).resolve())]
|
||||
return me + ["--run-bot", "--bot", bot, "--excel", excel or "", "--log", str(log),
|
||||
"--parent-pid", str(os.getpid())]
|
||||
|
||||
|
||||
def build_sample(path, sheets):
|
||||
"""Headers-only sample workbook with one fake row. Never real data (R13)."""
|
||||
from openpyxl import Workbook
|
||||
wb = Workbook()
|
||||
wb.remove(wb.active)
|
||||
for name, cols in sheets.items():
|
||||
ws = wb.create_sheet(name[:31])
|
||||
ws.append(cols)
|
||||
ws.append([f"Example {c}".strip() for c in cols])
|
||||
wb.save(path)
|
||||
|
||||
|
||||
def read_sheet_rows(path, sheet, cap=200):
|
||||
"""(header, rows, more): first `cap` non-empty rows as text, Excel date serials as m/d/yyyy."""
|
||||
from datetime import date, timedelta
|
||||
from openpyxl import load_workbook
|
||||
|
||||
def cell(c):
|
||||
if c is None:
|
||||
return ""
|
||||
if isinstance(c, datetime):
|
||||
c = c.date()
|
||||
if isinstance(c, date):
|
||||
return f"{c.month}/{c.day}/{c.year}"
|
||||
if isinstance(c, (int, float)) and not isinstance(c, bool) and 36526 <= c <= 73050: # 2000..2099
|
||||
d = date(1899, 12, 30) + timedelta(days=int(c))
|
||||
return f"{d.month}/{d.day}/{d.year}"
|
||||
return str(c)
|
||||
|
||||
wb = load_workbook(path, read_only=True, data_only=True)
|
||||
try:
|
||||
ws = wb[sheet] if sheet in wb.sheetnames else wb.active
|
||||
rows, more = [], False
|
||||
for r in ws.iter_rows(values_only=True):
|
||||
cells = [cell(c) for c in r]
|
||||
if not any(x.strip() for x in cells):
|
||||
continue
|
||||
if len(rows) >= cap:
|
||||
more = True
|
||||
break
|
||||
rows.append(cells)
|
||||
finally:
|
||||
wb.close()
|
||||
return (rows[0] if rows else []), rows[1:], more
|
||||
|
||||
|
||||
class Bridge(QObject):
|
||||
"""Exposed to the page as pyBridge."""
|
||||
js_message = pyqtSignal(str)
|
||||
|
||||
def __init__(self, window, bot=BOT, checkpoints=CHECKPOINTS):
|
||||
super().__init__()
|
||||
self.window, self.bot, self.checkpoints = window, bot, checkpoints
|
||||
self.run = None
|
||||
|
||||
@property
|
||||
def running(self):
|
||||
return self.run is not None and self.run.thread.is_alive()
|
||||
|
||||
def send(self, event, data):
|
||||
self.js_message.emit(json.dumps({"event": event, "data": data}))
|
||||
|
||||
def _file(self, path):
|
||||
sheets = []
|
||||
try:
|
||||
from openpyxl import load_workbook
|
||||
wb = load_workbook(path, read_only=True)
|
||||
sheets = list(wb.sheetnames)
|
||||
wb.close()
|
||||
except Exception as e:
|
||||
self.send("log", {"msg": f"Could not open the workbook: {e}", "type": "warn"})
|
||||
self.send("file_selected", {"path": path, "name": os.path.basename(path), "sheets": sheets})
|
||||
|
||||
@pyqtSlot()
|
||||
def browse_file(self):
|
||||
path, _ = QFileDialog.getOpenFileName(self.window, "Select input Excel file", str(Path.home()),
|
||||
"Excel files (*.xlsx *.xlsm *.xls)")
|
||||
if path:
|
||||
self._file(path)
|
||||
|
||||
@pyqtSlot(str)
|
||||
def process_dropped_path(self, path):
|
||||
# The page only knows the dropped file's name, so look for it in the usual folders.
|
||||
if not os.path.isabs(path):
|
||||
path = next((str(d / path) for d in SEARCH_DIRS if (d / path).is_file()), "")
|
||||
if path and os.path.isfile(path):
|
||||
self._file(path)
|
||||
else:
|
||||
self.send("file_selected", {"path": "", "name": os.path.basename(path) or "file", "sheets": []})
|
||||
|
||||
@pyqtSlot(str, str)
|
||||
def read_sheet(self, path, sheet):
|
||||
try:
|
||||
header, rows, more = read_sheet_rows(path, sheet)
|
||||
self.send("sheet_data", {"sheet": sheet, "header": header, "rows": rows,
|
||||
"total": len(rows), "more": more})
|
||||
except Exception as e:
|
||||
self.send("sheet_data", {"sheet": sheet, "error": str(e), "header": [], "rows": [], "total": 0})
|
||||
|
||||
@pyqtSlot()
|
||||
def download_sample_input(self):
|
||||
path, _ = QFileDialog.getSaveFileName(self.window, "Save sample input",
|
||||
str(Path.home() / f"{APP_ID}_sample_input.xlsx"),
|
||||
"Excel files (*.xlsx)")
|
||||
if path:
|
||||
build_sample(path, SAMPLE_SHEETS)
|
||||
self.send("log", {"msg": f"Sample saved: {path}", "type": "success"})
|
||||
|
||||
@pyqtSlot(str)
|
||||
def start_automation(self, excel):
|
||||
if self.running: # double-run guard
|
||||
return
|
||||
LOGS.mkdir(exist_ok=True)
|
||||
log = LOGS / f"run_{datetime.now():%Y%m%d_%H%M%S}.log"
|
||||
self.run = runner.Run(bot_cmd(excel if HAS_INPUT else "", log, self.bot), log, self.checkpoints,
|
||||
self.send, KILL_ALSO, cwd=str(ROOT)).start()
|
||||
|
||||
@pyqtSlot(bool)
|
||||
def set_dark(self, on):
|
||||
QSettings("RPA", APP_ID).setValue("dark", on)
|
||||
|
||||
|
||||
class RunTabWindow(QMainWindow):
|
||||
def __init__(self, bot=BOT, checkpoints=CHECKPOINTS):
|
||||
super().__init__()
|
||||
self.setWindowTitle(TITLE)
|
||||
self.setMinimumSize(940, 620)
|
||||
self.resize(1320, 820)
|
||||
self.view = QWebEngineView()
|
||||
self.view.setContextMenuPolicy(Qt.ContextMenuPolicy.NoContextMenu) # no Reload mid-run
|
||||
self.bridge = Bridge(self, bot, checkpoints)
|
||||
self.channel = QWebChannel()
|
||||
self.channel.registerObject("pyBridge", self.bridge)
|
||||
self.view.page().setWebChannel(self.channel)
|
||||
self.bridge.js_message.connect(self._push)
|
||||
self.view.loadFinished.connect(self._loaded)
|
||||
self.view.setHtml(page_html(), QUrl("qrc:///"))
|
||||
self.setCentralWidget(self.view)
|
||||
|
||||
def _push(self, s):
|
||||
self.view.page().runJavaScript(f"window.__pyMsg && window.__pyMsg({json.dumps(s)})")
|
||||
|
||||
def _loaded(self, ok):
|
||||
if ok:
|
||||
on = QSettings("RPA", APP_ID).value("dark", False, type=bool)
|
||||
self.bridge.send("apply_dark", {"on": on})
|
||||
|
||||
def closeEvent(self, event):
|
||||
if self.bridge.running:
|
||||
ans = QMessageBox.question(
|
||||
self, f"Close {TITLE}?",
|
||||
"The automation is still running.\n\nClosing will STOP it now. Close anyway?",
|
||||
QMessageBox.StandardButton.Yes | QMessageBox.StandardButton.No, QMessageBox.StandardButton.No)
|
||||
if ans != QMessageBox.StandardButton.Yes:
|
||||
event.ignore()
|
||||
return
|
||||
self.bridge.run.kill()
|
||||
event.accept()
|
||||
|
||||
|
||||
def selfcheck():
|
||||
"""Fake bot through every ending with the real runner, then window screenshots of a green
|
||||
run (light), an amber run (dark) and a stuck run (light), saved to logs/ (R14).
|
||||
Never runs the real bot."""
|
||||
import time
|
||||
fake = str((HERE / "fake_bot.py").relative_to(ROOT))
|
||||
os.environ["FAKE_CPS"] = str(len(CHECKPOINTS))
|
||||
quick = [(n, 2 if i == 1 else 0) for i, (n, _) in enumerate(CHECKPOINTS)] # checkpoint 2 sticks after 2 s
|
||||
want = {"complete": "done", "partial": "partial", "stuck": "error", "crash": "error"}
|
||||
LOGS.mkdir(exist_ok=True)
|
||||
for mode, state in want.items():
|
||||
os.environ["FAKE_MODE"] = mode
|
||||
events = []
|
||||
log = LOGS / f"selfcheck_{mode}.log"
|
||||
r = runner.Run(bot_cmd("", log, fake), log, quick, lambda e, d: events.append((e, d)), cwd=str(ROOT)).start()
|
||||
r.thread.join(60)
|
||||
end = events[-1][1] if events and events[-1][0] == "status" else {}
|
||||
assert end.get("state") == state, f"{mode}: expected {state}, got {end} (log: {log})"
|
||||
if mode == "stuck":
|
||||
assert ("cpstate", {"kind": "stuck", "cp": 2}) in events, f"stuck box missing: {events}"
|
||||
print(f"ok {mode:<8} -> {end['state']}: {end['msg']}")
|
||||
|
||||
# The window: one green run in light, one amber run in dark, saved as PNGs to look at.
|
||||
app = QApplication.instance() or QApplication(sys.argv)
|
||||
w = RunTabWindow(fake, quick)
|
||||
w.resize(1280, 900)
|
||||
w.show()
|
||||
shots = [("complete", False), ("partial", True), ("stuck", False)]
|
||||
|
||||
def wait_for_end(then):
|
||||
def poll():
|
||||
if w.bridge.running or w.bridge.run is None:
|
||||
QTimer.singleShot(200, poll)
|
||||
else:
|
||||
QTimer.singleShot(800, then)
|
||||
poll()
|
||||
|
||||
def next_shot():
|
||||
if not shots:
|
||||
app.quit()
|
||||
return
|
||||
mode, dark = shots.pop(0)
|
||||
os.environ["FAKE_MODE"] = mode
|
||||
w.bridge.run = None
|
||||
w.bridge.send("apply_dark", {"on": dark})
|
||||
w.view.page().runJavaScript("fileReady=true; filePath=''; doRun();")
|
||||
|
||||
def save():
|
||||
out = LOGS / f"selfcheck_{mode}_{'dark' if dark else 'light'}.png"
|
||||
w.view.grab().save(str(out))
|
||||
print(f"ok screenshot {out}")
|
||||
next_shot()
|
||||
|
||||
def scrolled_save():
|
||||
w.view.page().runJavaScript("document.getElementById('prog-wrap').scrollIntoView()")
|
||||
QTimer.singleShot(400, save)
|
||||
QTimer.singleShot(300, lambda: wait_for_end(scrolled_save))
|
||||
|
||||
w.view.loadFinished.connect(lambda ok: QTimer.singleShot(500, next_shot))
|
||||
QTimer.singleShot(90_000, app.quit) # never hang
|
||||
t = time.monotonic()
|
||||
app.exec()
|
||||
print(f"selfcheck done in {time.monotonic() - t:.0f}s")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
if "--selfcheck" in sys.argv:
|
||||
selfcheck()
|
||||
sys.exit(0)
|
||||
app = QApplication(sys.argv)
|
||||
app.setApplicationName(TITLE)
|
||||
w = RunTabWindow()
|
||||
w.show()
|
||||
sys.exit(app.exec())
|
||||
Reference in new issue
Block a user