From f3fee37e80c4936fc483c9fa6fc8619f3328c24a Mon Sep 17 00:00:00 2001 From: Adam Moussa <166072409+amoussa1229@users.noreply.github.com> Date: Fri, 8 May 2026 15:44:35 -0400 Subject: [PATCH] Add CI workflow and apply ruff formatting --- .github/workflows/ci.yaml | 11 ++ edit_pay_dates.py | 59 ++++++-- generate_icon.py | 21 ++- scan_pending.py | 34 +++-- stampli_core.py | 195 ++++++++++++++++++------- stampli_gui.py | 299 ++++++++++++++++++++++++++++---------- 6 files changed, 460 insertions(+), 159 deletions(-) create mode 100644 .github/workflows/ci.yaml diff --git a/.github/workflows/ci.yaml b/.github/workflows/ci.yaml new file mode 100644 index 0000000..5e672a6 --- /dev/null +++ b/.github/workflows/ci.yaml @@ -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 diff --git a/edit_pay_dates.py b/edit_pay_dates.py index cdcf322..dd2b667 100644 --- a/edit_pay_dates.py +++ b/edit_pay_dates.py @@ -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): diff --git a/generate_icon.py b/generate_icon.py index be1a698..618e496 100644 --- a/generate_icon.py +++ b/generate_icon.py @@ -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__": diff --git a/scan_pending.py b/scan_pending.py index 7fab874..45d42dc 100644 --- a/scan_pending.py +++ b/scan_pending.py @@ -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() diff --git a/stampli_core.py b/stampli_core.py index 6d1d0e7..21e15e5 100644 --- a/stampli_core.py +++ b/stampli_core.py @@ -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({ - "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 "", - }) + 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 "", + } + ) 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,16 +410,20 @@ 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({ - "row_id": row.get("Row ID", "").strip(), - "invoice": invoice, - "vendor": row.get("Vendor", "").strip(), - "amount": row.get("Amount", "").strip(), - "due_text": row.get("Due Date", "").strip(), - "pay_text": row.get("Current Pay", "").strip(), - "new_pay": new_pay, - }) + 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(), + "amount": row.get("Amount", "").strip(), + "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) diff --git a/stampli_gui.py b/stampli_gui.py index a52faae..40c8c34 100644 --- a/stampli_gui.py +++ b/stampli_gui.py @@ -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", - f" WRONG: {r['invoicesNumbers']} — " - f"pay={r['sendPaymentOn']} expected={r['expected']}") + self._post( + "log", + f" WRONG: {r['invoicesNumbers']} — " + 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