mirror of
https://github.com/Sea-Haven-Industries/stampli-bulk-editor.git
synced 2026-09-30 06:53:13 +00:00
Add CSV import/export, Chrome profile support, settings, stop button, and post-edit audit
- CSV export with Row ID and Amount columns (UTF-8 BOM for Excel) - CSV import: reimport edited pay dates, apply by invoice matching - Chrome profile support alongside Chromium persistent context - Settings dialog for browser mode and Chrome profile selection - Stop button to halt mid-run (skips filter/pay prompt) - Sequential invoice-matching edit flow for virtual scroll grid - Triple-click + type for Pikaday text date inputs - Post-edit audit pass to catch silently reverted edits - Debug logging to ~/Library/Application Support/Stampli Bulk Editor/ - Close Browser fully kills all pages and lingering processes - Optimized edit timing (~3 sec/edit, down from ~6 sec)
This commit is contained in:
Adam Moussa
parent
215f16a150
commit
b737a5aa80
3 changed files with 743 additions and 37 deletions
|
|
@ -83,10 +83,10 @@ 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} {'Due':<12s} {'Current Pay':<16s} {'New Pay':<12s}")
|
||||
print(f" {'-'*20} {'-'*30} {'-'*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['due_text']:<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:
|
||||
|
|
|
|||
448
stampli_core.py
448
stampli_core.py
|
|
@ -1,16 +1,15 @@
|
|||
"""Shared logic for Stampli bulk editor — used by both CLI scripts and the GUI."""
|
||||
|
||||
import csv
|
||||
import json
|
||||
import logging
|
||||
import os
|
||||
import re
|
||||
import time
|
||||
from datetime import datetime, timedelta
|
||||
from pathlib import Path
|
||||
|
||||
# Use system-wide Playwright browser cache so the bundled app doesn't
|
||||
# look for browsers inside its own .app bundle.
|
||||
_pw_cache = Path.home() / "Library" / "Caches" / "ms-playwright"
|
||||
if _pw_cache.is_dir():
|
||||
os.environ.setdefault("PLAYWRIGHT_BROWSERS_PATH", str(_pw_cache))
|
||||
import subprocess
|
||||
|
||||
from playwright.sync_api import sync_playwright
|
||||
|
||||
|
|
@ -18,8 +17,22 @@ PROJECT_DIR = Path(__file__).parent
|
|||
_APP_SUPPORT = Path.home() / "Library" / "Application Support" / "Stampli Bulk Editor"
|
||||
_APP_SUPPORT.mkdir(parents=True, exist_ok=True)
|
||||
CONFIG_PATH = _APP_SUPPORT / "config.json"
|
||||
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"))
|
||||
log.addHandler(_fh)
|
||||
|
||||
CHROME_USER_DATA = Path.home() / "Library" / "Application Support" / "Google" / "Chrome"
|
||||
BROWSER_DATA_DIR = _APP_SUPPORT / "browser_data"
|
||||
|
||||
# Playwright Chromium cache for standalone mode
|
||||
_pw_cache = Path.home() / "Library" / "Caches" / "ms-playwright"
|
||||
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"
|
||||
|
||||
|
|
@ -34,6 +47,7 @@ EDIT_COLUMNS = {
|
|||
"pay": "div[data-field='invoiceRequestedDate']",
|
||||
"vendor": "div[data-field='vendorName']",
|
||||
"invoice": "div[data-field='invoiceNumber']",
|
||||
"amount": "div[data-field='amountDue']",
|
||||
}
|
||||
|
||||
SCAN_COLUMNS = ["invoicesNumbers", "vendorName", "dueDate", "sendPaymentOn",
|
||||
|
|
@ -43,6 +57,8 @@ DEFAULT_CONFIG = {
|
|||
"stampli_url": READY_TO_PAY_URL,
|
||||
"date_format": DATE_FMT,
|
||||
"days_before_due": DAYS_BEFORE_DUE,
|
||||
"browser_mode": "chromium",
|
||||
"chrome_profile": "",
|
||||
}
|
||||
|
||||
|
||||
|
|
@ -58,6 +74,38 @@ def load_config():
|
|||
return dict(DEFAULT_CONFIG)
|
||||
|
||||
|
||||
def save_config(config):
|
||||
with open(CONFIG_PATH, "w") as f:
|
||||
json.dump(config, f, indent=2)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Chrome profile discovery
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def discover_chrome_profiles():
|
||||
"""Return list of {'dir': str, 'name': str, 'email': str} for each Chrome profile."""
|
||||
profiles = []
|
||||
if not CHROME_USER_DATA.is_dir():
|
||||
return profiles
|
||||
for entry in sorted(CHROME_USER_DATA.iterdir()):
|
||||
prefs_file = entry / "Preferences"
|
||||
if not prefs_file.exists():
|
||||
continue
|
||||
dirname = entry.name
|
||||
if dirname != "Default" and not dirname.startswith("Profile "):
|
||||
continue
|
||||
try:
|
||||
prefs = json.loads(prefs_file.read_text())
|
||||
name = prefs.get("profile", {}).get("name", dirname)
|
||||
accounts = prefs.get("account_info", [])
|
||||
email = accounts[0].get("email", "") if accounts else ""
|
||||
profiles.append({"dir": dirname, "name": name, "email": email})
|
||||
except (json.JSONDecodeError, OSError):
|
||||
profiles.append({"dir": dirname, "name": dirname, "email": ""})
|
||||
return profiles
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Date helpers
|
||||
# ---------------------------------------------------------------------------
|
||||
|
|
@ -78,11 +126,73 @@ def calc_pay_date(due_date: datetime, days_before: int = DAYS_BEFORE_DUE) -> dat
|
|||
return next_business_day(due_date - timedelta(days=days_before))
|
||||
|
||||
|
||||
def clean_amount(text: str) -> str:
|
||||
"""Strip unicode currency formatting down to plain number + currency code."""
|
||||
cleaned = re.sub(r"[^\d.,A-Za-z\s-]", "", text)
|
||||
return re.sub(r"\s+", " ", cleaned).strip()
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Browser helpers
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def _chrome_is_running():
|
||||
try:
|
||||
result = subprocess.run(
|
||||
["pgrep", "-x", "Google Chrome"],
|
||||
capture_output=True, text=True,
|
||||
)
|
||||
return result.returncode == 0
|
||||
except OSError:
|
||||
return False
|
||||
|
||||
|
||||
def _clean_browser_profile(data_dir, profile_subdir=None):
|
||||
"""Remove stale lock files and crash markers left by unclean shutdowns."""
|
||||
base = Path(data_dir)
|
||||
for name in ("SingletonLock", "SingletonCookie", "SingletonSocket"):
|
||||
lock = base / name
|
||||
if lock.exists() or lock.is_symlink():
|
||||
lock.unlink(missing_ok=True)
|
||||
|
||||
prefs_dir = base / profile_subdir if profile_subdir else base / "Default"
|
||||
prefs_file = prefs_dir / "Preferences"
|
||||
if prefs_file.exists():
|
||||
try:
|
||||
prefs = json.loads(prefs_file.read_text())
|
||||
profile = prefs.get("profile", {})
|
||||
if profile.get("exit_type", "") != "Normal":
|
||||
profile["exit_type"] = "Normal"
|
||||
profile["exited_cleanly"] = True
|
||||
prefs["profile"] = profile
|
||||
prefs_file.write_text(json.dumps(prefs))
|
||||
except (json.JSONDecodeError, OSError):
|
||||
pass
|
||||
|
||||
|
||||
def launch_browser(pw):
|
||||
config = load_config()
|
||||
mode = config.get("browser_mode", "chromium")
|
||||
log.info("Launching browser: mode=%s", mode)
|
||||
|
||||
if mode == "chrome":
|
||||
chrome_profile = config.get("chrome_profile", "Default")
|
||||
log.info("Chrome profile: %s", chrome_profile)
|
||||
if _chrome_is_running():
|
||||
raise RuntimeError(
|
||||
"Google Chrome is currently running. "
|
||||
"Please close Chrome before launching the editor."
|
||||
)
|
||||
_clean_browser_profile(CHROME_USER_DATA, chrome_profile)
|
||||
return pw.chromium.launch_persistent_context(
|
||||
channel="chrome",
|
||||
user_data_dir=str(CHROME_USER_DATA),
|
||||
headless=False,
|
||||
viewport={"width": 1400, "height": 900},
|
||||
args=["--disable-gpu", f"--profile-directory={chrome_profile}"],
|
||||
)
|
||||
else:
|
||||
_clean_browser_profile(BROWSER_DATA_DIR)
|
||||
return pw.chromium.launch_persistent_context(
|
||||
user_data_dir=str(BROWSER_DATA_DIR),
|
||||
headless=False,
|
||||
|
|
@ -158,6 +268,7 @@ def collect_edit_rows(page, on_progress=None):
|
|||
pay_cell = row.query_selector(EDIT_COLUMNS["pay"])
|
||||
vendor_cell = row.query_selector(EDIT_COLUMNS["vendor"])
|
||||
inv_cell = row.query_selector(EDIT_COLUMNS["invoice"])
|
||||
amt_cell = row.query_selector(EDIT_COLUMNS["amount"])
|
||||
|
||||
all_rows.append({
|
||||
"row_id": row_id,
|
||||
|
|
@ -165,6 +276,7 @@ def collect_edit_rows(page, on_progress=None):
|
|||
"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:
|
||||
|
|
@ -252,6 +364,38 @@ def compute_changes(all_rows, date_fmt=DATE_FMT, days_before=DAYS_BEFORE_DUE):
|
|||
return changes
|
||||
|
||||
|
||||
def parse_import_csv(path, date_fmt=DATE_FMT):
|
||||
"""Read an exported CSV and return a changes list matched by invoice number."""
|
||||
with open(path, newline="", encoding="utf-8-sig") as f:
|
||||
reader = csv.DictReader(f)
|
||||
headers = reader.fieldnames or []
|
||||
|
||||
for required in ("Invoice", "New Pay"):
|
||||
if required not in headers:
|
||||
raise ValueError(f"CSV missing required column: {required}")
|
||||
|
||||
changes = []
|
||||
for i, row in enumerate(reader, start=2):
|
||||
invoice = row["Invoice"].strip()
|
||||
new_pay = row["New Pay"].strip()
|
||||
if not invoice or not new_pay:
|
||||
continue
|
||||
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,
|
||||
})
|
||||
return changes
|
||||
|
||||
|
||||
def compute_scan_results(all_rows, date_fmt=DATE_FMT, days_before=DAYS_BEFORE_DUE):
|
||||
"""Pure function: compare actual vs expected pay dates. Returns (results, incorrect)."""
|
||||
results = []
|
||||
|
|
@ -300,18 +444,46 @@ def find_row(page, scroller, row_id, start_scroll=0):
|
|||
return None, scroll_top
|
||||
|
||||
|
||||
def _to_iso_date(date_str, date_fmt=DATE_FMT):
|
||||
"""Convert MM/DD/YYYY to YYYY-MM-DD for native date input .fill()."""
|
||||
dt = datetime.strptime(date_str, date_fmt)
|
||||
return dt.strftime("%Y-%m-%d")
|
||||
|
||||
|
||||
def try_edit_cell(page, row, new_date_str):
|
||||
pay_cell = row.query_selector(EDIT_COLUMNS["pay"])
|
||||
if not pay_cell:
|
||||
log.warning(" pay cell not found on row")
|
||||
return False
|
||||
|
||||
pay_text_before = pay_cell.inner_text().strip().split("\n")[0]
|
||||
log.debug(" pay cell found, current value: %r", pay_text_before)
|
||||
|
||||
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']")
|
||||
if calendar_icon:
|
||||
log.debug(" clicking calendar icon")
|
||||
calendar_icon.click()
|
||||
else:
|
||||
log.debug(" no calendar icon, double-clicking cell")
|
||||
pay_cell.dblclick()
|
||||
|
||||
time.sleep(0.5)
|
||||
time.sleep(0.4)
|
||||
|
||||
# Log what the editing cell looks like
|
||||
editing_cell = page.query_selector(".MuiDataGrid-cell--editing")
|
||||
if editing_cell:
|
||||
all_inputs = editing_cell.query_selector_all("input")
|
||||
log.debug(" editing cell has %d input(s)", len(all_inputs))
|
||||
for idx, inp in enumerate(all_inputs):
|
||||
inp_info = inp.evaluate(
|
||||
"el => JSON.stringify({type: el.type, value: el.value, placeholder: el.placeholder})"
|
||||
)
|
||||
log.debug(" input[%d]: %s", idx, inp_info)
|
||||
else:
|
||||
log.warning(" no .MuiDataGrid-cell--editing found")
|
||||
|
||||
date_input = page.query_selector(
|
||||
".MuiDataGrid-cell--editing input, "
|
||||
|
|
@ -324,23 +496,50 @@ def try_edit_cell(page, row, new_date_str):
|
|||
)
|
||||
|
||||
if date_input:
|
||||
date_input.fill("")
|
||||
date_input.fill(new_date_str)
|
||||
time.sleep(0.3)
|
||||
date_input.press("Enter")
|
||||
input_type = date_input.evaluate("el => el.type")
|
||||
log.info(" date input type=%s", input_type)
|
||||
|
||||
if input_type == "date":
|
||||
# Native date input: .fill() requires ISO format YYYY-MM-DD
|
||||
iso_date = _to_iso_date(new_date_str)
|
||||
log.info(" using .fill() with ISO date: %s", iso_date)
|
||||
date_input.fill(iso_date)
|
||||
else:
|
||||
page.keyboard.type(new_date_str)
|
||||
# Text input (Pikaday or similar): clear and type MM/DD/YYYY
|
||||
log.info(" using triple-click + type for text input")
|
||||
date_input.click(click_count=3)
|
||||
time.sleep(0.15)
|
||||
page.keyboard.type(new_date_str, delay=30)
|
||||
|
||||
time.sleep(0.2)
|
||||
log.debug(" pressing Enter")
|
||||
page.keyboard.press("Enter")
|
||||
else:
|
||||
log.warning(" no date input found — trying keyboard type as fallback")
|
||||
page.keyboard.type(new_date_str, delay=30)
|
||||
time.sleep(0.15)
|
||||
page.keyboard.press("Enter")
|
||||
|
||||
time.sleep(0.5)
|
||||
time.sleep(0.4)
|
||||
|
||||
# Dismiss any date picker popup that may still be open
|
||||
page.keyboard.press("Escape")
|
||||
time.sleep(0.3)
|
||||
time.sleep(0.2)
|
||||
|
||||
# Click away from the cell to deselect it
|
||||
page.mouse.click(0, 0)
|
||||
time.sleep(0.3)
|
||||
time.sleep(0.2)
|
||||
|
||||
# Verify the edit took effect
|
||||
try:
|
||||
pay_cell_after = row.query_selector(EDIT_COLUMNS["pay"])
|
||||
if pay_cell_after:
|
||||
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)
|
||||
else:
|
||||
log.debug(" row detached after edit (expected with virtual scroll)")
|
||||
except Exception:
|
||||
log.debug(" row detached after edit (expected with virtual scroll)")
|
||||
|
||||
return True
|
||||
|
||||
|
|
@ -361,13 +560,16 @@ def edit_single_row(page, row_id, new_date_str, last_scroll=0, max_attempts=3):
|
|||
return False, scroll_pos
|
||||
|
||||
|
||||
def apply_edits(page, changes, on_progress=None):
|
||||
def apply_edits(page, changes, on_progress=None, stop_check=None):
|
||||
"""Edit rows and retry failures. Returns (success_count, failed_list)."""
|
||||
success = 0
|
||||
failed = []
|
||||
last_scroll = 0
|
||||
|
||||
for i, c in enumerate(changes):
|
||||
if stop_check and stop_check():
|
||||
failed.extend(changes[i:])
|
||||
break
|
||||
try:
|
||||
ok, last_scroll = edit_single_row(page, c["row_id"], c["new_pay"], last_scroll)
|
||||
if ok:
|
||||
|
|
@ -383,9 +585,12 @@ def apply_edits(page, changes, on_progress=None):
|
|||
if on_progress:
|
||||
on_progress(i + 1, len(changes), c["invoice"], False, str(e))
|
||||
|
||||
if failed:
|
||||
if failed and not (stop_check and stop_check()):
|
||||
retry_failed = []
|
||||
for c in failed:
|
||||
if stop_check and stop_check():
|
||||
retry_failed.extend(failed[failed.index(c):])
|
||||
break
|
||||
try:
|
||||
ok, _ = edit_single_row(page, c["row_id"], c["new_pay"], last_scroll=0)
|
||||
if ok:
|
||||
|
|
@ -405,6 +610,213 @@ def apply_edits(page, changes, on_progress=None):
|
|||
return success, failed
|
||||
|
||||
|
||||
def apply_edits_by_invoice(page, changes, on_progress=None, stop_check=None):
|
||||
"""Scroll through the grid sequentially, matching rows by invoice number.
|
||||
|
||||
After each edit the grid re-renders, so we re-query rows at the current
|
||||
scroll position instead of continuing with stale element handles.
|
||||
stop_check: callable returning True if the user requested a stop.
|
||||
"""
|
||||
pending = {c["invoice"]: c["new_pay"] for c in changes if c.get("invoice")}
|
||||
if not pending:
|
||||
log.info("No invoices to process")
|
||||
return 0, list(changes)
|
||||
|
||||
scroller = page.query_selector(SCROLLER_SELECTOR)
|
||||
if not scroller:
|
||||
log.error("Scroller element not found")
|
||||
return 0, list(changes)
|
||||
|
||||
success = 0
|
||||
seen_invoices = set()
|
||||
scroll_top = 0
|
||||
stale_count = 0
|
||||
total = len(pending)
|
||||
log.info("Starting invoice-match edit: %d invoices to process", total)
|
||||
|
||||
while pending:
|
||||
if stop_check and stop_check():
|
||||
log.info("Stop requested by user")
|
||||
break
|
||||
scroller.evaluate("(el, top) => el.scrollTop = top", scroll_top)
|
||||
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)
|
||||
|
||||
edited_this_position = True
|
||||
while edited_this_position:
|
||||
if stop_check and stop_check():
|
||||
log.info("Stop requested by user (inner loop)")
|
||||
break
|
||||
edited_this_position = False
|
||||
rows = page.query_selector_all(ROW_SELECTOR)
|
||||
log.debug(" re-queried %d rows at scroll_top=%d", len(rows), scroll_top)
|
||||
|
||||
visible_invoices = []
|
||||
for row in rows:
|
||||
try:
|
||||
inv_cell = row.query_selector(EDIT_COLUMNS["invoice"])
|
||||
if not inv_cell:
|
||||
continue
|
||||
invoice = inv_cell.inner_text().strip()
|
||||
visible_invoices.append(invoice)
|
||||
except Exception as e:
|
||||
log.debug(" row read failed (detached?): %s", e)
|
||||
continue
|
||||
|
||||
if invoice in seen_invoices:
|
||||
continue
|
||||
if invoice not in pending:
|
||||
seen_invoices.add(invoice)
|
||||
continue
|
||||
|
||||
new_pay = pending[invoice]
|
||||
log.info("MATCH: invoice=%s, setting pay=%s", invoice, new_pay)
|
||||
time.sleep(0.15)
|
||||
|
||||
try:
|
||||
if try_edit_cell(page, row, new_pay):
|
||||
success += 1
|
||||
del pending[invoice]
|
||||
seen_invoices.add(invoice)
|
||||
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))
|
||||
break
|
||||
else:
|
||||
seen_invoices.add(invoice)
|
||||
log.warning(" edit returned False for %s", invoice)
|
||||
if on_progress:
|
||||
on_progress(success, total, invoice, False, "edit failed")
|
||||
except Exception as e:
|
||||
seen_invoices.add(invoice)
|
||||
log.error(" edit exception for %s: %s", invoice, e)
|
||||
if on_progress:
|
||||
on_progress(success, total, invoice, False, str(e))
|
||||
page.keyboard.press("Escape")
|
||||
time.sleep(0.2)
|
||||
break
|
||||
|
||||
if not edited_this_position:
|
||||
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):
|
||||
try:
|
||||
inv_cell = row.query_selector(EDIT_COLUMNS["invoice"])
|
||||
if inv_cell:
|
||||
inv = inv_cell.inner_text().strip()
|
||||
if inv and inv not in seen_invoices:
|
||||
found_new = True
|
||||
break
|
||||
except Exception:
|
||||
continue
|
||||
|
||||
if found_new:
|
||||
stale_count = 0
|
||||
else:
|
||||
stale_count += 1
|
||||
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
|
||||
|
||||
scroll_top += 200
|
||||
|
||||
failed = [c for c in changes if c.get("invoice") in pending]
|
||||
log.info("Finished: %d/%d succeeded, %d failed", success, total, len(failed))
|
||||
if failed:
|
||||
log.info("Failed invoices: %s", [c["invoice"] for c in failed])
|
||||
return success, failed
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# 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")}
|
||||
if not expected:
|
||||
return [], []
|
||||
|
||||
scroller = page.query_selector(SCROLLER_SELECTOR)
|
||||
if not scroller:
|
||||
log.error("Scroller element not found for audit")
|
||||
return [], list(changes)
|
||||
|
||||
confirmed = []
|
||||
reverted = []
|
||||
seen = set()
|
||||
scroll_top = 0
|
||||
stale_count = 0
|
||||
log.info("Audit: verifying %d invoices", len(expected))
|
||||
|
||||
while len(seen) < len(expected):
|
||||
if stop_check and stop_check():
|
||||
log.info("Audit stopped by user")
|
||||
break
|
||||
|
||||
scroller.evaluate("(el, top) => el.scrollTop = top", scroll_top)
|
||||
time.sleep(0.3)
|
||||
|
||||
rows = page.query_selector_all(ROW_SELECTOR)
|
||||
found_new = False
|
||||
|
||||
for row in rows:
|
||||
try:
|
||||
inv_cell = row.query_selector(EDIT_COLUMNS["invoice"])
|
||||
if not inv_cell:
|
||||
continue
|
||||
invoice = inv_cell.inner_text().strip()
|
||||
if not invoice or invoice in seen:
|
||||
continue
|
||||
|
||||
if invoice not in expected:
|
||||
continue
|
||||
|
||||
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 ""
|
||||
want = expected[invoice]
|
||||
|
||||
if actual == want:
|
||||
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)
|
||||
|
||||
if on_progress:
|
||||
on_progress(len(seen), len(expected), len(reverted))
|
||||
except Exception as e:
|
||||
log.debug(" audit row read failed: %s", e)
|
||||
continue
|
||||
|
||||
if found_new:
|
||||
stale_count = 0
|
||||
else:
|
||||
stale_count += 1
|
||||
if stale_count >= 3:
|
||||
log.info("Audit: end of grid reached")
|
||||
break
|
||||
|
||||
scroll_top += 200
|
||||
|
||||
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.info("Audit complete: %d confirmed, %d reverted, %d not found",
|
||||
len(confirmed), len(reverted), len(not_found))
|
||||
return confirmed, reverted, not_found
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Filter / Select / Pay
|
||||
# ---------------------------------------------------------------------------
|
||||
|
|
|
|||
314
stampli_gui.py
314
stampli_gui.py
|
|
@ -3,6 +3,7 @@
|
|||
|
||||
VERSION = "1.0.0"
|
||||
|
||||
import csv
|
||||
import os
|
||||
import sys
|
||||
import queue
|
||||
|
|
@ -26,7 +27,8 @@ from playwright.sync_api import sync_playwright
|
|||
from stampli_core import (
|
||||
launch_browser, navigate, active_page, wait_for_login, needs_login,
|
||||
collect_edit_rows, collect_scan_rows, compute_changes, compute_scan_results,
|
||||
apply_edits, filter_select_and_pay, load_config,
|
||||
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,
|
||||
)
|
||||
|
||||
|
|
@ -44,6 +46,7 @@ class StampliApp:
|
|||
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)
|
||||
|
|
@ -72,10 +75,22 @@ class StampliApp:
|
|||
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",
|
||||
|
|
@ -159,8 +174,13 @@ class StampliApp:
|
|||
|
||||
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")
|
||||
|
||||
|
|
@ -170,23 +190,25 @@ class StampliApp:
|
|||
|
||||
def _populate_edit_table(self, changes):
|
||||
self.tree.delete(*self.tree.get_children())
|
||||
cols = ("invoice", "vendor", "due", "current_pay", "new_pay")
|
||||
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=140, minwidth=80)
|
||||
self.tree.column("vendor", width=250, minwidth=120)
|
||||
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=120, 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", values=(
|
||||
c["invoice"], c["vendor"], 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):
|
||||
|
|
@ -231,15 +253,123 @@ class StampliApp:
|
|||
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")
|
||||
|
|
@ -335,10 +465,27 @@ class StampliApp:
|
|||
if i > 0:
|
||||
self._post("status", f"Editing... {i}/{total}")
|
||||
|
||||
success, failed = apply_edits(page, changes, on_progress=on_edit_progress)
|
||||
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:
|
||||
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(
|
||||
|
|
@ -366,13 +513,147 @@ class StampliApp:
|
|||
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
|
||||
|
|
@ -383,6 +664,19 @@ class StampliApp:
|
|||
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.")
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue