#!/usr/bin/env python3 """Stampli Bulk Editor — tkinter GUI.""" import csv import os import sys import queue import threading import time import traceback from datetime import datetime VERSION = "1.0.0" # 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 # noqa: E402 from tkinter import ttk, messagebox # noqa: E402 from playwright.sync_api import sync_playwright # noqa: E402 from stampli_core import ( # noqa: E402 launch_browser, navigate, active_page, wait_for_login, needs_login, collect_edit_rows, collect_scan_rows, compute_changes, compute_scan_results, apply_edits, apply_edits_by_invoice, audit_edits, filter_select_and_pay, load_config, save_config, parse_import_csv, discover_chrome_profiles, READY_TO_PAY_URL, PENDING_APPROVAL_URL, ) 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._stop_requested = 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.stop_btn = ttk.Button( ctrl, text="Stop", command=self.on_stop, state="disabled" ) self.stop_btn.pack(side="left", padx=5) self.export_btn = ttk.Button( ctrl, text="Export CSV", command=self.on_export, state="disabled" ) self.export_btn.pack(side="left", padx=5) self.import_btn = ttk.Button(ctrl, text="Import CSV", command=self.on_import) self.import_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) ttk.Button(ctrl, text="Settings", command=self.on_settings).pack( side="right", 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._stop_requested = False self.run_btn.configure(state="normal") self.stop_btn.configure(state="disabled") self.import_btn.configure(state="normal") self.close_btn.configure(state="normal") if self.tree.get_children(): self.export_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", "amount", "due", "current_pay", "new_pay") self.tree["columns"] = cols self.tree.heading("invoice", text="Invoice") self.tree.heading("vendor", text="Vendor") self.tree.heading("amount", text="Amount") 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=130, minwidth=80) self.tree.column("vendor", width=220, minwidth=120) self.tree.column("amount", width=100, minwidth=70) self.tree.column("due", width=100, minwidth=80) self.tree.column("current_pay", width=110, minwidth=80) self.tree.column("new_pay", width=100, minwidth=80) for c in changes: self.tree.insert( "", "end", iid=c.get("row_id", ""), values=( c["invoice"], c["vendor"], c.get("amount", ""), 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() # ------------------------------------------------------------------ # Settings # ------------------------------------------------------------------ def on_settings(self): config = load_config() dlg = tk.Toplevel(self.root) dlg.title("Settings") dlg.geometry("420x200") dlg.resizable(False, False) dlg.transient(self.root) dlg.grab_set() frame = ttk.Frame(dlg, padding=15) frame.pack(fill="both", expand=True) # Browser mode ttk.Label(frame, text="Browser:").grid(row=0, column=0, sticky="w", pady=(0, 8)) mode_var = tk.StringVar(value=config.get("browser_mode", "chromium")) mode_frame = ttk.Frame(frame) mode_frame.grid(row=0, column=1, sticky="w", pady=(0, 8)) ttk.Radiobutton( mode_frame, text="Chromium (standalone)", variable=mode_var, value="chromium", command=lambda: profile_combo.configure(state="disabled"), ).pack(anchor="w") ttk.Radiobutton( mode_frame, text="Chrome (existing profile)", variable=mode_var, value="chrome", command=lambda: profile_combo.configure(state="readonly"), ).pack(anchor="w") # Chrome profile picker ttk.Label(frame, text="Chrome Profile:").grid( row=1, column=0, sticky="w", pady=(0, 8) ) profiles = discover_chrome_profiles() profile_labels = [] profile_dirs = [] for p in profiles: label = p["email"] or p["name"] if p["email"] and p["name"] != p["dir"]: label = f"{p['email']}" profile_labels.append(f"{p['dir']} — {label}") profile_dirs.append(p["dir"]) profile_var = tk.StringVar() saved_profile = config.get("chrome_profile", "") if saved_profile in profile_dirs: profile_var.set(profile_labels[profile_dirs.index(saved_profile)]) elif profile_labels: profile_var.set(profile_labels[0]) profile_combo = ttk.Combobox( frame, textvariable=profile_var, values=profile_labels, state="readonly", width=35, ) profile_combo.grid(row=1, column=1, sticky="w", pady=(0, 8)) if mode_var.get() != "chrome": profile_combo.configure(state="disabled") def on_save(): config["browser_mode"] = mode_var.get() sel = profile_var.get() for i, label in enumerate(profile_labels): if label == sel: config["chrome_profile"] = profile_dirs[i] break save_config(config) self._append_log( f"Settings saved: browser={config['browser_mode']}" + ( f", profile={config.get('chrome_profile', '')}" if config["browser_mode"] == "chrome" else "" ) ) dlg.destroy() btn_frame = ttk.Frame(frame) btn_frame.grid(row=2, column=0, columnspan=2, pady=(10, 0)) ttk.Button(btn_frame, text="Save", command=on_save).pack(side="left", padx=5) ttk.Button(btn_frame, text="Cancel", command=dlg.destroy).pack( side="left", padx=5 ) # ------------------------------------------------------------------ # Run button # ------------------------------------------------------------------ def on_export(self): from tkinter import filedialog cols = self.tree["columns"] headers = ["Row ID"] + [self.tree.heading(c, "text") for c in cols] iids = self.tree.get_children() rows = [[iid] + list(self.tree.item(iid, "values")) for iid in iids] if not rows: return timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") default_name = f"stampli_export_{timestamp}.csv" path = filedialog.asksaveasfilename( defaultextension=".csv", filetypes=[("CSV files", "*.csv")], initialfile=default_name, ) if not path: return with open(path, "w", newline="", encoding="utf-8-sig") as f: writer = csv.writer(f) writer.writerow(headers) writer.writerows(rows) self._append_log(f"Exported {len(rows)} rows to {path}") self.status_var.set(f"Exported to {os.path.basename(path)}") def on_stop(self): if self._running: self._stop_requested = True self.stop_btn.configure(state="disabled") self._append_log("Stop requested — finishing current edit...") self.status_var.set("Stopping...") def on_run(self): if self._running: return self._running = True self._stop_requested = False self.run_btn.configure(state="disabled") self.stop_btn.configure(state="normal") self.export_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, stop_check=lambda: self._stop_requested, ) self._post( "log", f"Done: {success}/{len(changes)} updated, {len(failed)} failed." ) if success > 0 and not self._stop_requested: self._post("status", "Auditing edits...") self._post("log", "Verifying edits stuck (scrolling through grid)...") edited = [c for c in changes if c not in failed] def on_audit_progress(checked, total, reverted_count): self._post( "status", f"Auditing... {checked}/{total} ({reverted_count} reverted)", ) confirmed, reverted, not_found = audit_edits( page, edited, on_progress=on_audit_progress, stop_check=lambda: self._stop_requested, ) self._post( "log", f"Audit: {len(confirmed)} confirmed, {len(reverted)} reverted, {len(not_found)} not found", ) for r in reverted: self._post( "log", f" REVERTED: {r['invoice']} — expected {r['expected']}, shows {r['actual']}", ) 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']}", ) # ------------------------------------------------------------------ # Import CSV # ------------------------------------------------------------------ def on_import(self): if self._running: return from tkinter import filedialog path = filedialog.askopenfilename( filetypes=[("CSV files", "*.csv")], ) if not path: return try: changes = parse_import_csv(path) except ValueError as e: messagebox.showerror("Import Error", str(e)) return if not changes: messagebox.showinfo("Import", "No rows found in CSV.") return self._imported_changes = changes self._post("table_edit", changes) self._append_log(f"Imported {len(changes)} rows from {os.path.basename(path)}") self.export_btn.configure(state="normal") if not messagebox.askyesno( "Confirm", f"Apply {len(changes)} imported pay dates?" ): self._append_log("Import aborted by user.") self.status_var.set("Import cancelled.") return self._running = True self._stop_requested = False self.run_btn.configure(state="disabled") self.stop_btn.configure(state="normal") self.import_btn.configure(state="disabled") self.export_btn.configure(state="disabled") self.worker_thread = threading.Thread( target=self._run_import, args=(changes,), daemon=True, ) self.worker_thread.start() def _run_import(self, changes): try: if not self.browser_ctx: self._post("status", "Launching browser...") self._post("log", "Launching browser...") self.pw_ctx = sync_playwright().start() self.browser_ctx = launch_browser(self.pw_ctx) navigate(self.browser_ctx, READY_TO_PAY_URL) self._post("log", f"Navigating to {READY_TO_PAY_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) self._post("status", "Scrolling through grid, matching invoices...") self._post( "log", f"Applying {len(changes)} imported pay dates (matching by invoice)...", ) self._post("table_edit", changes) def on_edit_progress(done, total, inv, ok, info=""): status = "Updated" if ok else "MISSED" msg = f"[{done}/{total}] {status}: {inv}" if info: msg += f" ({info})" self._post("log", msg) self._post("status", f"Applying... {done}/{total}") success, failed = apply_edits_by_invoice( page, changes, on_progress=on_edit_progress, stop_check=lambda: self._stop_requested, ) if self._stop_requested: self._post( "log", f"Stopped: {success}/{len(changes)} updated before stop." ) else: self._post( "log", f"Done: {success}/{len(changes)} updated, {len(failed)} failed.", ) if success > 0 and not self._stop_requested: self._post("status", "Auditing edits...") self._post("log", "Verifying edits stuck (scrolling through grid)...") edited = [ c for c in changes if c.get("invoice") not in {f["invoice"] for f in failed} ] def on_audit_progress(checked, total, reverted_count): self._post( "status", f"Auditing... {checked}/{total} ({reverted_count} reverted)", ) confirmed, reverted, not_found = audit_edits( page, edited, on_progress=on_audit_progress, stop_check=lambda: self._stop_requested, ) self._post( "log", f"Audit: {len(confirmed)} confirmed, {len(reverted)} reverted, {len(not_found)} not found", ) for r in reverted: self._post( "log", f" REVERTED: {r['invoice']} — expected {r['expected']}, shows {r['actual']}", ) 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, ) 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}") # ------------------------------------------------------------------ # Browser cleanup # ------------------------------------------------------------------ def on_close_browser(self): # Stop any running operation first if self._running: self._stop_requested = True if self.browser_ctx: try: for pg in self.browser_ctx.pages: try: pg.close() except Exception: pass 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 # Kill any lingering browser processes import subprocess try: subprocess.run( ["pkill", "-f", "chromium.*--remote-debugging"], capture_output=True, timeout=3, ) except Exception: pass self._running = False self._stop_requested = False self.run_btn.configure(state="normal") self.stop_btn.configure(state="disabled") self.import_btn.configure(state="normal") 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()