adam/stampli-bulk-editor
1
0
Fork 0
mirror of https://github.com/Sea-Haven-Industries/stampli-bulk-editor.git synced 2026-09-30 03:23:13 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

Add CI workflow and apply ruff formatting

Browse source
This commit is contained in:
Adam Moussa 2026-05-08 15:44:35 -04:00
parent 0a75d4e7ce
commit f3fee37e80
6 changed files with 460 additions and 159 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

11
.github/workflows/ci.yaml vendored Normal file
View file

@ -0,0 +1,11 @@
name: CI
on:
pull_request:
branches: [main]
jobs:
ci:
uses: Sea-Haven-Industries/.github/.github/workflows/ci-python-sam.yaml@main
with:
source-dirs: "."
run-sam-validate: false

59
edit_pay_dates.py
View file

@ -10,16 +10,22 @@ Usage:
"""
import argparse
import sys
from datetime import datetime
from pathlib import Path
from playwright.sync_api import sync_playwright
from stampli_core import (
load_config, calc_pay_date, active_page, launch_browser, navigate,
wait_for_login, needs_login, collect_edit_rows, compute_changes,
apply_edits, filter_select_and_pay, READY_TO_PAY_URL,
ROW_SELECTOR, EDIT_COLUMNS, SCROLLER_SELECTOR,
load_config,
active_page,
launch_browser,
navigate,
wait_for_login,
needs_login,
collect_edit_rows,
compute_changes,
apply_edits,
filter_select_and_pay,
ROW_SELECTOR,
EDIT_COLUMNS,
)
@ -47,13 +53,25 @@ Type 'snap' to save page HTML, 'scan' to preview rows, 'done' to exit.
for i, row in enumerate(rows[:10]):
due_cell = row.query_selector(EDIT_COLUMNS["due"])
pay_cell = row.query_selector(EDIT_COLUMNS["pay"])
due_text = due_cell.inner_text().strip().split("\n")[0] if due_cell else "?"
pay_text = pay_cell.inner_text().strip().split("\n")[0] if pay_cell else "?"
due_text = (
due_cell.inner_text().strip().split("\n")[0]
if due_cell
else "?"
)
pay_text = (
pay_cell.inner_text().strip().split("\n")[0]
if pay_cell
else "?"
)
vendor_cell = row.query_selector(EDIT_COLUMNS["vendor"])
vendor = vendor_cell.inner_text().strip()[:30] if vendor_cell else "?"
vendor = (
vendor_cell.inner_text().strip()[:30] if vendor_cell else "?"
)
inv_cell = row.query_selector(EDIT_COLUMNS["invoice"])
inv = inv_cell.inner_text().strip()[:20] if inv_cell else "?"
print(f" Row {i}: inv={inv} vendor={vendor} due={due_text} pay={pay_text}")
print(
f" Row {i}: inv={inv} vendor={vendor} due={due_text} pay={pay_text}"
)
elif cmd == "columns":
headers = page.query_selector_all("[role='columnheader']")
for i, h in enumerate(headers):
@ -83,11 +101,15 @@ def bulk_edit(page, config, dry_run=False):
return
print(f"\n{'DRY RUN: ' if dry_run else ''}{len(changes)} rows to update:\n")
print(f" {'Invoice':<20s} {'Vendor':<30s} {'Amount':<12s} {'Due':<12s} {'Current Pay':<16s} {'New Pay':<12s}")
print(f" {'-'*20} {'-'*30} {'-'*12} {'-'*12} {'-'*16} {'-'*12}")
print(
f" {'Invoice':<20s} {'Vendor':<30s} {'Amount':<12s} {'Due':<12s} {'Current Pay':<16s} {'New Pay':<12s}"
)
print(f" {'-' * 20} {'-' * 30} {'-' * 12} {'-' * 12} {'-' * 16} {'-' * 12}")
for c in changes:
print(f" {c['invoice']:<20s} {c['vendor']:<30s} {c.get('amount', ''):<12s} {c['due_text']:<12s} "
f"{c['pay_text'] or '(empty)':<16s} {c['new_pay']:<12s}")
print(
f" {c['invoice']:<20s} {c['vendor']:<30s} {c.get('amount', ''):<12s} {c['due_text']:<12s} "
f"{c['pay_text'] or '(empty)':<16s} {c['new_pay']:<12s}"
)
if dry_run:
return
@ -109,7 +131,13 @@ def bulk_edit(page, config, dry_run=False):
print(f"\nDone: {success}/{len(changes)} rows updated.")
if success > 0:
submit = input("\nFilter to error-free rows, select all, and click Pay Invoices? [y/N]: ").strip().lower()
submit = (
input(
"\nFilter to error-free rows, select all, and click Pay Invoices? [y/N]: "
)
.strip()
.lower()
)
if submit == "y":
filter_select_and_pay(page, on_status=print)
@ -130,6 +158,7 @@ def main():
navigate(browser, url)
import time
time.sleep(3)
page = active_page(browser)
if needs_login(page):

21
generate_icon.py
View file

@ -9,6 +9,7 @@ from PIL import Image, ImageDraw, ImageFont
SIZE = 1024
def draw_icon(size=SIZE):
img = Image.new("RGBA", (size, size), (0, 0, 0, 0))
draw = ImageDraw.Draw(img)
@ -73,8 +74,12 @@ def draw_icon(size=SIZE):
dollar_y = int(660 * s)
dollar_size = int(110 * s)
draw.ellipse(
[dollar_x - dollar_size, dollar_y - dollar_size,
dollar_x + dollar_size, dollar_y + dollar_size],
[
dollar_x - dollar_size,
dollar_y - dollar_size,
dollar_x + dollar_size,
dollar_y + dollar_size,
],
fill=accent,
)
try:
@ -88,8 +93,12 @@ def draw_icon(size=SIZE):
badge_cy = int(750 * s)
badge_r = int(85 * s)
draw.ellipse(
[badge_cx - badge_r, badge_cy - badge_r,
badge_cx + badge_r, badge_cy + badge_r],
[
badge_cx - badge_r,
badge_cy - badge_r,
badge_cx + badge_r,
badge_cy + badge_r,
],
fill=(46, 204, 113),
outline=bg_color,
width=int(12 * s),
@ -133,7 +142,9 @@ def make_icns(img, output_path):
if sz <= 512:
double = img.resize((sz * 2, sz * 2), Image.LANCZOS)
double.save(os.path.join(iconset, f"icon_{sz}x{sz}@2x.png"))
subprocess.run(["iconutil", "-c", "icns", iconset, "-o", output_path], check=True)
subprocess.run(
["iconutil", "-c", "icns", iconset, "-o", output_path], check=True
)
if __name__ == "__main__":

34
scan_pending.py
View file

@ -13,8 +13,14 @@ from pathlib import Path
from playwright.sync_api import sync_playwright
from stampli_core import (
launch_browser, navigate, active_page, wait_for_login, needs_login,
collect_scan_rows, compute_scan_results, PENDING_APPROVAL_URL, DATE_FMT,
launch_browser,
navigate,
active_page,
wait_for_login,
needs_login,
collect_scan_rows,
compute_scan_results,
PENDING_APPROVAL_URL,
)
@ -51,18 +57,26 @@ def main():
results, incorrect = compute_scan_results(all_rows)
print(f" {'Invoice':<20s} {'Vendor':<35s} {'Due':<12s} {'Pay Date':<12s} {'Expected':<12s} {'Status'}")
print(f" {'-'*20} {'-'*35} {'-'*12} {'-'*12} {'-'*12} {'-'*8}")
print(
f" {'Invoice':<20s} {'Vendor':<35s} {'Due':<12s} {'Pay Date':<12s} {'Expected':<12s} {'Status'}"
)
print(f" {'-' * 20} {'-' * 35} {'-' * 12} {'-' * 12} {'-' * 12} {'-' * 8}")
for r in results:
print(f" {r['invoicesNumbers']:<20s} {r['vendorName'][:35]:<35s} "
f"{r['dueDate']:<12s} {r['sendPaymentOn']:<12s} {r['expected']:<12s} {r['status']}")
print(
f" {r['invoicesNumbers']:<20s} {r['vendorName'][:35]:<35s} "
f"{r['dueDate']:<12s} {r['sendPaymentOn']:<12s} {r['expected']:<12s} {r['status']}"
)
print(f"\n{len(incorrect)} rows with incorrect pay dates out of {len(all_rows)} total.")
print(
f"\n{len(incorrect)} rows with incorrect pay dates out of {len(all_rows)} total."
)
if incorrect:
print(f"\nIncorrect rows:")
print("\nIncorrect rows:")
for r in incorrect:
print(f" {r['invoicesNumbers']:<20s} {r['vendorName'][:35]:<35s} "
f"pay={r['sendPaymentOn']} (due={r['dueDate']})")
print(
f" {r['invoicesNumbers']:<20s} {r['vendorName'][:35]:<35s} "
f"pay={r['sendPaymentOn']} (due={r['dueDate']})"
)
input("\nPress Enter to close the browser...")
browser.close()

177
stampli_core.py
View file

@ -11,7 +11,6 @@ from pathlib import Path
import subprocess
from playwright.sync_api import sync_playwright
PROJECT_DIR = Path(__file__).parent
_APP_SUPPORT = Path.home() / "Library" / "Application Support" / "Stampli Bulk Editor"
@ -22,7 +21,9 @@ LOG_PATH = _APP_SUPPORT / "debug.log"
log = logging.getLogger("stampli")
log.setLevel(logging.DEBUG)
_fh = logging.FileHandler(LOG_PATH, mode="w")
_fh.setFormatter(logging.Formatter("%(asctime)s %(levelname)s %(message)s", datefmt="%H:%M:%S"))
_fh.setFormatter(
logging.Formatter("%(asctime)s %(levelname)s %(message)s", datefmt="%H:%M:%S")
)
log.addHandler(_fh)
CHROME_USER_DATA = Path.home() / "Library" / "Application Support" / "Google" / "Chrome"
@ -34,7 +35,9 @@ if _pw_cache.is_dir():
os.environ.setdefault("PLAYWRIGHT_BROWSERS_PATH", str(_pw_cache))
READY_TO_PAY_URL = "https://app.stampli.com/v265n2/dashboard.html#t=ready_to_pay"
PENDING_APPROVAL_URL = "https://app.stampli.com/v265n2/dashboard.html#t=payments_to_approve"
PENDING_APPROVAL_URL = (
"https://app.stampli.com/v265n2/dashboard.html#t=payments_to_approve"
)
DATE_FMT = "%m/%d/%Y"
DAYS_BEFORE_DUE = 5
@ -50,8 +53,15 @@ EDIT_COLUMNS = {
"amount": "div[data-field='amountDue']",
}
SCAN_COLUMNS = ["invoicesNumbers", "vendorName", "dueDate", "sendPaymentOn",
"paymentMethod", "amountDue", "amount"]
SCAN_COLUMNS = [
"invoicesNumbers",
"vendorName",
"dueDate",
"sendPaymentOn",
"paymentMethod",
"amountDue",
"amount",
]
DEFAULT_CONFIG = {
"stampli_url": READY_TO_PAY_URL,
@ -66,6 +76,7 @@ DEFAULT_CONFIG = {
# Config
# ---------------------------------------------------------------------------
def load_config():
if CONFIG_PATH.exists():
with open(CONFIG_PATH) as f:
@ -83,6 +94,7 @@ def save_config(config):
# Chrome profile discovery
# ---------------------------------------------------------------------------
def discover_chrome_profiles():
"""Return list of {'dir': str, 'name': str, 'email': str} for each Chrome profile."""
profiles = []
@ -110,6 +122,7 @@ def discover_chrome_profiles():
# Date helpers
# ---------------------------------------------------------------------------
def next_business_day(d: datetime) -> datetime:
weekday = d.weekday()
if weekday == 5:
@ -136,11 +149,13 @@ def clean_amount(text: str) -> str:
# Browser helpers
# ---------------------------------------------------------------------------
def _chrome_is_running():
try:
result = subprocess.run(
["pgrep", "-x", "Google Chrome"],
capture_output=True, text=True,
capture_output=True,
text=True,
)
return result.returncode == 0
except OSError:
@ -239,6 +254,7 @@ def needs_login(page):
# Virtual-scroll grid scanning
# ---------------------------------------------------------------------------
def collect_edit_rows(page, on_progress=None):
"""Scan the Select to Pay grid. Returns list of dicts with row_id, invoice, vendor, due_text, pay_text."""
scroller = page.query_selector(SCROLLER_SELECTOR)
@ -270,14 +286,24 @@ def collect_edit_rows(page, on_progress=None):
inv_cell = row.query_selector(EDIT_COLUMNS["invoice"])
amt_cell = row.query_selector(EDIT_COLUMNS["amount"])
all_rows.append({
all_rows.append(
{
"row_id": row_id,
"invoice": inv_cell.inner_text().strip() if inv_cell else "",
"vendor": (vendor_cell.inner_text().strip()[:40] if vendor_cell else ""),
"due_text": (due_cell.inner_text().strip().split("\n")[0] if due_cell else ""),
"pay_text": (pay_cell.inner_text().strip().split("\n")[0] if pay_cell else ""),
"amount": clean_amount(amt_cell.inner_text().strip().split("\n")[0]) if amt_cell else "",
})
"vendor": (
vendor_cell.inner_text().strip()[:40] if vendor_cell else ""
),
"due_text": (
due_cell.inner_text().strip().split("\n")[0] if due_cell else ""
),
"pay_text": (
pay_cell.inner_text().strip().split("\n")[0] if pay_cell else ""
),
"amount": clean_amount(amt_cell.inner_text().strip().split("\n")[0])
if amt_cell
else "",
}
)
if new_this_scroll == 0:
stale_count += 1
@ -344,6 +370,7 @@ def collect_scan_rows(page, on_progress=None):
# Change computation
# ---------------------------------------------------------------------------
def compute_changes(all_rows, date_fmt=DATE_FMT, days_before=DAYS_BEFORE_DUE):
"""Pure function: given collected rows, return list of rows that need updating."""
changes = []
@ -383,8 +410,11 @@ def parse_import_csv(path, date_fmt=DATE_FMT):
try:
datetime.strptime(new_pay, date_fmt)
except ValueError:
raise ValueError(f"Row {i}: invalid date '{new_pay}' (expected {date_fmt})")
changes.append({
raise ValueError(
f"Row {i}: invalid date '{new_pay}' (expected {date_fmt})"
)
changes.append(
{
"row_id": row.get("Row ID", "").strip(),
"invoice": invoice,
"vendor": row.get("Vendor", "").strip(),
@ -392,7 +422,8 @@ def parse_import_csv(path, date_fmt=DATE_FMT):
"due_text": row.get("Due Date", "").strip(),
"pay_text": row.get("Current Pay", "").strip(),
"new_pay": new_pay,
})
}
)
return changes
@ -421,6 +452,7 @@ def compute_scan_results(all_rows, date_fmt=DATE_FMT, days_before=DAYS_BEFORE_DU
# Editing
# ---------------------------------------------------------------------------
def find_row(page, scroller, row_id, start_scroll=0):
row = page.query_selector(
f".MuiDataGrid-virtualScrollerRenderZone div.MuiDataGrid-row[data-id='{row_id}']"
@ -462,7 +494,9 @@ def try_edit_cell(page, row, new_date_str):
pay_cell.scroll_into_view_if_needed()
time.sleep(0.15)
calendar_icon = pay_cell.query_selector("i.fa-calendar, [data-test-id='test-icon-calendar']")
calendar_icon = pay_cell.query_selector(
"i.fa-calendar, [data-test-id='test-icon-calendar']"
)
if calendar_icon:
log.debug(" clicking calendar icon")
calendar_icon.click()
@ -535,7 +569,11 @@ def try_edit_cell(page, row, new_date_str):
pay_text_after = pay_cell_after.inner_text().strip().split("\n")[0]
log.debug(" after edit: %r (wanted %r)", pay_text_after, new_date_str)
if pay_text_after != new_date_str:
log.warning(" MISMATCH: cell shows %r but wanted %r", pay_text_after, new_date_str)
log.warning(
" MISMATCH: cell shows %r but wanted %r",
pay_text_after,
new_date_str,
)
else:
log.debug(" row detached after edit (expected with virtual scroll)")
except Exception:
@ -571,7 +609,9 @@ def apply_edits(page, changes, on_progress=None, stop_check=None):
failed.extend(changes[i:])
break
try:
ok, last_scroll = edit_single_row(page, c["row_id"], c["new_pay"], last_scroll)
ok, last_scroll = edit_single_row(
page, c["row_id"], c["new_pay"], last_scroll
)
if ok:
success += 1
if on_progress:
@ -589,7 +629,7 @@ def apply_edits(page, changes, on_progress=None, stop_check=None):
retry_failed = []
for c in failed:
if stop_check and stop_check():
retry_failed.extend(failed[failed.index(c):])
retry_failed.extend(failed[failed.index(c) :])
break
try:
ok, _ = edit_single_row(page, c["row_id"], c["new_pay"], last_scroll=0)
@ -600,7 +640,9 @@ def apply_edits(page, changes, on_progress=None, stop_check=None):
else:
retry_failed.append(c)
if on_progress:
on_progress(-1, len(failed), c["invoice"], False, "retry failed")
on_progress(
-1, len(failed), c["invoice"], False, "retry failed"
)
except Exception as e:
retry_failed.append(c)
if on_progress:
@ -642,8 +684,13 @@ def apply_edits_by_invoice(page, changes, on_progress=None, stop_check=None):
time.sleep(0.3)
visible_rows = page.query_selector_all(ROW_SELECTOR)
log.debug("scroll_top=%d, %d visible rows, %d pending, stale_count=%d",
scroll_top, len(visible_rows), len(pending), stale_count)
log.debug(
"scroll_top=%d, %d visible rows, %d pending, stale_count=%d",
scroll_top,
len(visible_rows),
len(pending),
stale_count,
)
edited_this_position = True
while edited_this_position:
@ -684,7 +731,12 @@ def apply_edits_by_invoice(page, changes, on_progress=None, stop_check=None):
if on_progress:
on_progress(success, total, invoice, True)
edited_this_position = True
log.info(" edit OK (%d/%d done, %d remaining)", success, total, len(pending))
log.info(
" edit OK (%d/%d done, %d remaining)",
success,
total,
len(pending),
)
break
else:
seen_invoices.add(invoice)
@ -701,8 +753,11 @@ def apply_edits_by_invoice(page, changes, on_progress=None, stop_check=None):
break
if not edited_this_position:
log.debug(" no edits at scroll_top=%d, visible invoices: %s",
scroll_top, visible_invoices[:5])
log.debug(
" no edits at scroll_top=%d, visible invoices: %s",
scroll_top,
visible_invoices[:5],
)
found_new = False
for row in page.query_selector_all(ROW_SELECTOR):
@ -720,7 +775,11 @@ def apply_edits_by_invoice(page, changes, on_progress=None, stop_check=None):
stale_count = 0
else:
stale_count += 1
log.debug(" no new invoices at scroll_top=%d (stale_count=%d)", scroll_top, stale_count)
log.debug(
" no new invoices at scroll_top=%d (stale_count=%d)",
scroll_top,
stale_count,
)
if stale_count >= 3:
log.info("End of grid reached (3 consecutive stale scrolls)")
break
@ -738,6 +797,7 @@ def apply_edits_by_invoice(page, changes, on_progress=None, stop_check=None):
# Post-edit audit
# ---------------------------------------------------------------------------
def audit_edits(page, changes, on_progress=None, stop_check=None):
"""Scroll through the grid and verify that edited invoices kept their new pay dates."""
expected = {c["invoice"]: c["new_pay"] for c in changes if c.get("invoice")}
@ -782,15 +842,26 @@ def audit_edits(page, changes, on_progress=None, stop_check=None):
found_new = True
seen.add(invoice)
pay_cell = row.query_selector(EDIT_COLUMNS["pay"])
actual = pay_cell.inner_text().strip().split("\n")[0] if pay_cell else ""
actual = (
pay_cell.inner_text().strip().split("\n")[0] if pay_cell else ""
)
want = expected[invoice]
if actual == want:
confirmed.append({"invoice": invoice, "expected": want, "actual": actual})
confirmed.append(
{"invoice": invoice, "expected": want, "actual": actual}
)
log.debug(" AUDIT OK: %s = %s", invoice, actual)
else:
reverted.append({"invoice": invoice, "expected": want, "actual": actual})
log.warning(" AUDIT REVERTED: %s expected=%s actual=%s", invoice, want, actual)
reverted.append(
{"invoice": invoice, "expected": want, "actual": actual}
)
log.warning(
" AUDIT REVERTED: %s expected=%s actual=%s",
invoice,
want,
actual,
)
if on_progress:
on_progress(len(seen), len(expected), len(reverted))
@ -810,10 +881,16 @@ def audit_edits(page, changes, on_progress=None, stop_check=None):
not_found = [inv for inv in expected if inv not in seen]
if not_found:
log.warning("Audit: %d invoices not found in grid: %s", len(not_found), not_found[:10])
log.warning(
"Audit: %d invoices not found in grid: %s", len(not_found), not_found[:10]
)
log.info("Audit complete: %d confirmed, %d reverted, %d not found",
len(confirmed), len(reverted), len(not_found))
log.info(
"Audit complete: %d confirmed, %d reverted, %d not found",
len(confirmed),
len(reverted),
len(not_found),
)
return confirmed, reverted, not_found
@ -821,14 +898,18 @@ def audit_edits(page, changes, on_progress=None, stop_check=None):
# Filter / Select / Pay
# ---------------------------------------------------------------------------
def filter_select_and_pay(page, on_status=None, on_manual_fallback=None):
"""Apply 'No errors' filter, select all, click Pay Invoices, click Google auth."""
def status(msg):
if on_status:
on_status(msg)
status("Opening filter drawer...")
filter_toggle = page.query_selector("button[data-test-id='selectToPay-filter-toggle']")
filter_toggle = page.query_selector(
"button[data-test-id='selectToPay-filter-toggle']"
)
if not filter_toggle:
status("ERROR: Could not find filter toggle button.")
return
@ -852,14 +933,20 @@ def filter_select_and_pay(page, on_status=None, on_manual_fallback=None):
)
if status_input:
status_input.evaluate("el => el.scrollIntoView({block: 'center', behavior: 'instant'})")
status_input.evaluate(
"el => el.scrollIntoView({block: 'center', behavior: 'instant'})"
)
time.sleep(0.3)
status_input.click(force=True)
time.sleep(0.8)
no_errors = page.query_selector(".MuiAutocomplete-listbox >> text=No errors")
no_errors = page.query_selector(
".MuiAutocomplete-listbox >> text=No errors"
)
if not no_errors:
no_errors = page.query_selector(".MuiAutocomplete-listbox >> text=No Errors")
no_errors = page.query_selector(
".MuiAutocomplete-listbox >> text=No Errors"
)
if not no_errors:
options = page.query_selector_all(".MuiAutocomplete-option")
for opt in options:
@ -878,7 +965,9 @@ def filter_select_and_pay(page, on_status=None, on_manual_fallback=None):
if not applied:
status("Could not auto-apply filter.")
if on_manual_fallback:
on_manual_fallback("Please apply Status -> 'No errors' manually in the browser, then click OK.")
on_manual_fallback(
"Please apply Status -> 'No errors' manually in the browser, then click OK."
)
else:
input(" Apply the filter manually, then press Enter: ")
@ -897,7 +986,9 @@ def filter_select_and_pay(page, on_status=None, on_manual_fallback=None):
status("Selecting all rows...")
select_all = page.query_selector("input[aria-label='Select all rows']")
if not select_all:
select_all = page.query_selector(".MuiDataGrid-columnHeaderCheckbox .MuiCheckbox-root")
select_all = page.query_selector(
".MuiDataGrid-columnHeaderCheckbox .MuiCheckbox-root"
)
if select_all:
select_all.click()
time.sleep(1)
@ -907,7 +998,9 @@ def filter_select_and_pay(page, on_status=None, on_manual_fallback=None):
return
status("Clicking Pay Invoices...")
pay_button = page.query_selector("button[data-test-id='selectToPay-action-payInvoices']")
pay_button = page.query_selector(
"button[data-test-id='selectToPay-action-payInvoices']"
)
if not pay_button:
status("ERROR: Could not find Pay Invoices button.")
return
@ -919,7 +1012,9 @@ def filter_select_and_pay(page, on_status=None, on_manual_fallback=None):
on_manual_fallback("Please select rows manually, then click OK.")
else:
input(" Select rows manually, then press Enter: ")
pay_button = page.query_selector("button[data-test-id='selectToPay-action-payInvoices']")
pay_button = page.query_selector(
"button[data-test-id='selectToPay-action-payInvoices']"
)
pay_button.click()
time.sleep(2)

297
stampli_gui.py
View file

@ -1,8 +1,6 @@
#!/usr/bin/env python3
"""Stampli Bulk Editor — tkinter GUI."""
VERSION = "1.0.0"
import csv
import os
import sys
@ -12,6 +10,8 @@ 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")]:
@ -19,17 +19,31 @@ for _var, _subdir in [("TCL_LIBRARY", "tcl8.6"), ("TK_LIBRARY", "tk8.6")]:
if os.path.isdir(_path) and _var not in os.environ:
os.environ[_var] = _path
import tkinter as tk
from tkinter import ttk, messagebox
import tkinter as tk # noqa: E402
from tkinter import ttk, messagebox # noqa: E402
from playwright.sync_api import sync_playwright
from playwright.sync_api import sync_playwright # noqa: E402
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, 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, DATE_FMT,
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,
)
@ -63,38 +77,49 @@ class StampliApp:
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))
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 = 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 = 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 = 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 = 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)
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 = 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 ---
@ -106,9 +131,15 @@ class StampliApp:
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 = 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)
@ -120,9 +151,12 @@ class StampliApp:
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 = 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)
@ -206,10 +240,19 @@ class StampliApp:
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"],
))
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())
@ -230,10 +273,19 @@ class StampliApp:
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,))
self.tree.insert(
"",
"end",
values=(
r["invoicesNumbers"],
r["vendorName"][:35],
r["dueDate"],
r["sendPaymentOn"],
r["expected"],
r["status"],
),
tags=(tag,),
)
# ------------------------------------------------------------------
# Worker helpers
@ -274,15 +326,25 @@ class StampliApp:
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")
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))
ttk.Label(frame, text="Chrome Profile:").grid(
row=1, column=0, sticky="w", pady=(0, 8)
)
profiles = discover_chrome_profiles()
profile_labels = []
profile_dirs = []
@ -300,8 +362,13 @@ class StampliApp:
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 = 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")
@ -314,15 +381,22 @@ class StampliApp:
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 ""))
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)
ttk.Button(btn_frame, text="Cancel", command=dlg.destroy).pack(
side="left", padx=5
)
# ------------------------------------------------------------------
# Run button
@ -330,6 +404,7 @@ class StampliApp:
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()
@ -403,7 +478,9 @@ class StampliApp:
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)):
if not wait_for_login(
self.browser_ctx, on_status=lambda m: self._post("log", m)
):
self._post("done", "Login timed out.")
return
@ -416,7 +493,10 @@ class StampliApp:
else:
self._run_scan(page)
self._post("done", "Complete — browser still open. Click 'Close Browser' when done.")
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:
@ -465,9 +545,15 @@ class StampliApp:
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.")
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...")
@ -475,18 +561,30 @@ class StampliApp:
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)")
self._post(
"status",
f"Auditing... {checked}/{total} ({reverted_count} reverted)",
)
confirmed, reverted, not_found = audit_edits(
page, edited,
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")
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']}")
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?"):
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,
@ -509,9 +607,11 @@ class StampliApp:
self._post("log", f"{len(incorrect)} incorrect out of {len(all_rows)} total.")
for r in incorrect:
self._post("log",
self._post(
"log",
f" WRONG: {r['invoicesNumbers']} — "
f"pay={r['sendPaymentOn']} expected={r['expected']}")
f"pay={r['sendPaymentOn']} expected={r['expected']}",
)
# ------------------------------------------------------------------
# Import CSV
@ -521,6 +621,7 @@ class StampliApp:
if self._running:
return
from tkinter import filedialog
path = filedialog.askopenfilename(
filetypes=[("CSV files", "*.csv")],
)
@ -542,7 +643,9 @@ class StampliApp:
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?"):
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
@ -576,7 +679,9 @@ class StampliApp:
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)):
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...")
@ -584,7 +689,10 @@ class StampliApp:
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(
"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=""):
@ -596,41 +704,69 @@ class StampliApp:
self._post("status", f"Applying... {done}/{total}")
success, failed = apply_edits_by_invoice(
page, changes,
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.")
self._post(
"log", f"Stopped: {success}/{len(changes)} updated before stop."
)
else:
self._post("log", f"Done: {success}/{len(changes)} updated, {len(failed)} failed.")
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}]
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)")
self._post(
"status",
f"Auditing... {checked}/{total} ({reverted_count} reverted)",
)
confirmed, reverted, not_found = audit_edits(
page, edited,
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")
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']}")
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?"):
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_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._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:
@ -667,8 +803,13 @@ class StampliApp:
# Kill any lingering browser processes
import subprocess
try:
subprocess.run(["pkill", "-f", "chromium.*--remote-debugging"], capture_output=True, timeout=3)
subprocess.run(
["pkill", "-f", "chromium.*--remote-debugging"],
capture_output=True,
timeout=3,
)
except Exception:
pass