#!/usr/bin/env python3 """Stampli Bulk Editor — tkinter GUI.""" VERSION = "1.0.0" import os import sys import queue import threading import time import traceback from datetime import datetime # Set Tcl/Tk library paths before importing tkinter _py_base = os.path.dirname(os.path.dirname(os.path.realpath(sys.executable))) for _var, _subdir in [("TCL_LIBRARY", "tcl8.6"), ("TK_LIBRARY", "tk8.6")]: _path = os.path.join(_py_base, "lib", _subdir) if os.path.isdir(_path) and _var not in os.environ: os.environ[_var] = _path import tkinter as tk from tkinter import ttk, messagebox from playwright.sync_api import sync_playwright from stampli_core import ( launch_browser, navigate, active_page, wait_for_login, needs_login, collect_edit_rows, collect_scan_rows, compute_changes, compute_scan_results, apply_edits, filter_select_and_pay, load_config, READY_TO_PAY_URL, PENDING_APPROVAL_URL, DATE_FMT, ) class StampliApp: def __init__(self, root): self.root = root self.root.title(f"Stampli Bulk Editor v{VERSION}") self.root.geometry("1000x700") self.root.minsize(800, 500) self.msg_queue = queue.Queue() self.worker_thread = None self.browser_ctx = None self.pw_ctx = None self._confirm_result = False self._running = False self._build_ui() self.root.after(100, self._poll_queue) # Bring window to front on macOS self.root.lift() self.root.attributes("-topmost", True) self.root.after(500, lambda: self.root.attributes("-topmost", False)) def _build_ui(self): # --- Top controls --- ctrl = ttk.Frame(self.root, padding=10) ctrl.pack(fill="x") ttk.Label(ctrl, text="Operation:").pack(side="left") self.mode_var = tk.StringVar(value="edit") ttk.Radiobutton(ctrl, text="Edit Pay Dates", variable=self.mode_var, value="edit").pack(side="left", padx=(5, 10)) ttk.Radiobutton(ctrl, text="Scan Pending", variable=self.mode_var, value="scan").pack(side="left", padx=(0, 20)) self.dry_run_var = tk.BooleanVar(value=False) self.dry_run_cb = ttk.Checkbutton(ctrl, text="Dry Run", variable=self.dry_run_var) self.dry_run_cb.pack(side="left", padx=(0, 20)) self.run_btn = ttk.Button(ctrl, text="Run", command=self.on_run) self.run_btn.pack(side="left", padx=5) self.close_btn = ttk.Button(ctrl, text="Close Browser", command=self.on_close_browser, state="disabled") self.close_btn.pack(side="left", padx=5) # --- Status bar --- self.status_var = tk.StringVar(value="Ready") status_bar = ttk.Label(self.root, textvariable=self.status_var, relief="sunken", padding=(10, 4)) status_bar.pack(fill="x", padx=10) # --- Paned window for table + log --- pane = ttk.PanedWindow(self.root, orient="vertical") pane.pack(fill="both", expand=True, padx=10, pady=(5, 10)) # Table frame table_frame = ttk.Frame(pane) pane.add(table_frame, weight=3) self.tree = ttk.Treeview(table_frame, show="headings", selectmode="none") tree_scroll_y = ttk.Scrollbar(table_frame, orient="vertical", command=self.tree.yview) tree_scroll_x = ttk.Scrollbar(table_frame, orient="horizontal", command=self.tree.xview) self.tree.configure(yscrollcommand=tree_scroll_y.set, xscrollcommand=tree_scroll_x.set) tree_scroll_y.pack(side="right", fill="y") tree_scroll_x.pack(side="bottom", fill="x") self.tree.pack(fill="both", expand=True) self.tree.tag_configure("wrong", foreground="#cc0000") self.tree.tag_configure("ok", foreground="#006600") # Log frame log_frame = ttk.Frame(pane) pane.add(log_frame, weight=1) self.log_text = tk.Text(log_frame, height=8, state="disabled", wrap="word", font=("Menlo", 11)) log_scroll = ttk.Scrollbar(log_frame, orient="vertical", command=self.log_text.yview) self.log_text.configure(yscrollcommand=log_scroll.set) log_scroll.pack(side="right", fill="y") self.log_text.pack(fill="both", expand=True) # ------------------------------------------------------------------ # Queue polling # ------------------------------------------------------------------ def _poll_queue(self): while True: try: msg_type, payload = self.msg_queue.get_nowait() except queue.Empty: break if msg_type == "status": self.status_var.set(payload) elif msg_type == "log": self._append_log(payload) elif msg_type == "table_edit": self._populate_edit_table(payload) elif msg_type == "table_scan": self._populate_scan_table(payload) elif msg_type == "confirm": prompt, event = payload self._show_confirm(prompt, event) elif msg_type == "manual_fallback": prompt, event = payload self._show_manual_fallback(prompt, event) elif msg_type == "done": self._on_done(payload) self.root.after(100, self._poll_queue) def _append_log(self, text): ts = datetime.now().strftime("%H:%M:%S") self.log_text.configure(state="normal") self.log_text.insert("end", f"[{ts}] {text}\n") self.log_text.see("end") self.log_text.configure(state="disabled") def _show_confirm(self, prompt, event): self._confirm_result = messagebox.askyesno("Confirm", prompt) event.set() def _show_manual_fallback(self, prompt, event): messagebox.showinfo("Manual Step Required", prompt) event.set() def _on_done(self, message): self._running = False self.run_btn.configure(state="normal") self.close_btn.configure(state="normal") self.mode_var.set(self.mode_var.get()) # re-enable radios self.status_var.set(message or "Done") # ------------------------------------------------------------------ # Table population # ------------------------------------------------------------------ def _populate_edit_table(self, changes): self.tree.delete(*self.tree.get_children()) cols = ("invoice", "vendor", "due", "current_pay", "new_pay") self.tree["columns"] = cols self.tree.heading("invoice", text="Invoice") self.tree.heading("vendor", text="Vendor") self.tree.heading("due", text="Due Date") self.tree.heading("current_pay", text="Current Pay") self.tree.heading("new_pay", text="New Pay") self.tree.column("invoice", width=140, minwidth=80) self.tree.column("vendor", width=250, minwidth=120) self.tree.column("due", width=100, minwidth=80) self.tree.column("current_pay", width=120, minwidth=80) self.tree.column("new_pay", width=100, minwidth=80) for c in changes: self.tree.insert("", "end", values=( c["invoice"], c["vendor"], c["due_text"], c["pay_text"] or "(empty)", c["new_pay"], )) def _populate_scan_table(self, results): self.tree.delete(*self.tree.get_children()) cols = ("invoice", "vendor", "due", "pay_date", "expected", "status") self.tree["columns"] = cols self.tree.heading("invoice", text="Invoice") self.tree.heading("vendor", text="Vendor") self.tree.heading("due", text="Due Date") self.tree.heading("pay_date", text="Pay Date") self.tree.heading("expected", text="Expected") self.tree.heading("status", text="Status") self.tree.column("invoice", width=140, minwidth=80) self.tree.column("vendor", width=220, minwidth=120) self.tree.column("due", width=90, minwidth=80) self.tree.column("pay_date", width=90, minwidth=80) self.tree.column("expected", width=90, minwidth=80) self.tree.column("status", width=60, minwidth=50) for r in results: tag = "wrong" if r["status"] == "WRONG" else "ok" self.tree.insert("", "end", values=( r["invoicesNumbers"], r["vendorName"][:35], r["dueDate"], r["sendPaymentOn"], r["expected"], r["status"], ), tags=(tag,)) # ------------------------------------------------------------------ # Worker helpers # ------------------------------------------------------------------ def _post(self, msg_type, payload=None): self.msg_queue.put((msg_type, payload)) def _ask_confirm(self, prompt): event = threading.Event() self._post("confirm", (prompt, event)) event.wait() return self._confirm_result def _manual_fallback(self, prompt): event = threading.Event() self._post("manual_fallback", (prompt, event)) event.wait() # ------------------------------------------------------------------ # Run button # ------------------------------------------------------------------ def on_run(self): if self._running: return self._running = True self.run_btn.configure(state="disabled") self.tree.delete(*self.tree.get_children()) self.log_text.configure(state="normal") self.log_text.delete("1.0", "end") self.log_text.configure(state="disabled") mode = self.mode_var.get() dry_run = self.dry_run_var.get() self.worker_thread = threading.Thread( target=self._worker, args=(mode, dry_run), daemon=True, ) self.worker_thread.start() def _worker(self, mode, dry_run): try: self._post("status", "Launching browser...") self._post("log", "Launching browser...") self.pw_ctx = sync_playwright().start() self.browser_ctx = launch_browser(self.pw_ctx) url = READY_TO_PAY_URL if mode == "edit" else PENDING_APPROVAL_URL navigate(self.browser_ctx, url) self._post("log", f"Navigating to {url}") time.sleep(3) page = active_page(self.browser_ctx) if needs_login(page): self._post("status", "Waiting for login...") self._post("log", "Please log in to Stampli in the browser window.") if not wait_for_login(self.browser_ctx, on_status=lambda m: self._post("log", m)): self._post("done", "Login timed out.") return self._post("log", "Waiting for page to settle...") time.sleep(3) page = active_page(self.browser_ctx) if mode == "edit": self._run_edit(page, dry_run) else: self._run_scan(page) self._post("done", "Complete — browser still open. Click 'Close Browser' when done.") self.root.after(0, lambda: self.close_btn.configure(state="normal")) except Exception as e: self._post("log", f"ERROR: {e}") self._post("log", traceback.format_exc()) self._post("done", f"Error: {e}") def _run_edit(self, page, dry_run): config = load_config() date_fmt = config["date_format"] days_before = config["days_before_due"] self._post("status", "Scanning rows...") self._post("log", "Scanning all rows...") all_rows = collect_edit_rows( page, on_progress=lambda n: self._post("status", f"Scanning rows... ({n} found)"), ) self._post("log", f"Found {len(all_rows)} total rows.") changes = compute_changes(all_rows, date_fmt, days_before) if not changes: self._post("log", "All rows already have correct pay dates.") return self._post("table_edit", changes) label = "DRY RUN: " if dry_run else "" self._post("log", f"{label}{len(changes)} rows need updating.") if dry_run: return if not self._ask_confirm(f"Proceed with {len(changes)} updates?"): self._post("log", "Aborted by user.") return self._post("status", "Editing rows...") def on_edit_progress(i, total, inv, ok, info=""): if i == -1: msg = f"RETRY {'OK' if ok else 'FAIL'}: {inv} {info}" else: msg = f"[{i}/{total}] {'Updated' if ok else 'MISSED'}: {inv} {info}" self._post("log", msg) if i > 0: self._post("status", f"Editing... {i}/{total}") success, failed = apply_edits(page, changes, on_progress=on_edit_progress) self._post("log", f"Done: {success}/{len(changes)} updated, {len(failed)} failed.") if success > 0: if self._ask_confirm("Filter to error-free rows, select all, and click Pay Invoices?"): self._post("status", "Running filter/select/pay...") filter_select_and_pay( page, on_status=lambda m: (self._post("log", m), self._post("status", m)), on_manual_fallback=self._manual_fallback, ) def _run_scan(self, page): self._post("status", "Scanning rows...") self._post("log", "Scanning Pending Approval rows...") all_rows = collect_scan_rows( page, on_progress=lambda n: self._post("status", f"Scanning rows... ({n} found)"), ) self._post("log", f"Found {len(all_rows)} rows.") results, incorrect = compute_scan_results(all_rows) self._post("table_scan", results) self._post("log", f"{len(incorrect)} incorrect out of {len(all_rows)} total.") for r in incorrect: self._post("log", f" WRONG: {r['invoicesNumbers']} — " f"pay={r['sendPaymentOn']} expected={r['expected']}") # ------------------------------------------------------------------ # Browser cleanup # ------------------------------------------------------------------ def on_close_browser(self): if self.browser_ctx: try: self.browser_ctx.close() except Exception: pass self.browser_ctx = None if self.pw_ctx: try: self.pw_ctx.stop() except Exception: pass self.pw_ctx = None self.close_btn.configure(state="disabled") self.status_var.set("Browser closed.") self._append_log("Browser closed.") def on_quit(self): self.on_close_browser() self.root.destroy() def main(): root = tk.Tk() app = StampliApp(root) root.protocol("WM_DELETE_WINDOW", app.on_quit) root.mainloop() if __name__ == "__main__": main()