"""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 import subprocess 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" ) DATE_FMT = "%m/%d/%Y" DAYS_BEFORE_DUE = 5 ROW_SELECTOR = ".MuiDataGrid-virtualScrollerRenderZone div.MuiDataGrid-row" SCROLLER_SELECTOR = ".MuiDataGrid-virtualScroller" EDIT_COLUMNS = { "due": "div[data-field='dueDate']", "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", "paymentMethod", "amountDue", "amount", ] DEFAULT_CONFIG = { "stampli_url": READY_TO_PAY_URL, "date_format": DATE_FMT, "days_before_due": DAYS_BEFORE_DUE, "browser_mode": "chromium", "chrome_profile": "", } # --------------------------------------------------------------------------- # Config # --------------------------------------------------------------------------- def load_config(): if CONFIG_PATH.exists(): with open(CONFIG_PATH) as f: saved = json.load(f) return {**DEFAULT_CONFIG, **saved} 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 # --------------------------------------------------------------------------- def next_business_day(d: datetime) -> datetime: weekday = d.weekday() if weekday == 5: d += timedelta(days=2) elif weekday == 6: d += timedelta(days=1) return d def calc_pay_date(due_date: datetime, days_before: int = DAYS_BEFORE_DUE) -> datetime: today = datetime.now().replace(hour=0, minute=0, second=0, microsecond=0) if (due_date - today).days <= 5: return next_business_day(today + timedelta(days=2)) 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, viewport={"width": 1400, "height": 900}, args=["--disable-gpu"], ) def active_page(browser): pages = browser.pages return pages[-1] if pages else browser.new_page() def navigate(browser, url): page = browser.pages[0] if browser.pages else browser.new_page() page.goto(url, wait_until="domcontentloaded") return page def wait_for_login(browser, timeout_minutes=5, on_status=None): if on_status: on_status("Waiting for login...") deadline = time.time() + timeout_minutes * 60 while time.time() < deadline: for pg in browser.pages: url = pg.url.lower() if "login" not in url and "auth" not in url and "signin" not in url: if "stampli.com" in url: if on_status: on_status("Login detected.") return True time.sleep(2) if on_status: on_status("Login timed out.") return False def needs_login(page): url = page.url.lower() return "login" in url or "auth" in url or "signin" in url # --------------------------------------------------------------------------- # 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) if not scroller: return [] seen_ids = set() all_rows = [] scroll_top = 0 stale_count = 0 while True: scroller.evaluate("(el, top) => el.scrollTop = top", scroll_top) time.sleep(0.5) rows = page.query_selector_all(ROW_SELECTOR) new_this_scroll = 0 for row in rows: row_id = row.get_attribute("data-id") if not row_id or row_id in seen_ids: continue seen_ids.add(row_id) new_this_scroll += 1 due_cell = row.query_selector(EDIT_COLUMNS["due"]) 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, "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 if stale_count >= 3: break else: stale_count = 0 if on_progress: on_progress(len(all_rows)) scroll_top += 200 return all_rows def collect_scan_rows(page, on_progress=None): """Scan the Pending Approval grid. Returns list of dicts keyed by SCAN_COLUMNS.""" scroller = page.query_selector(SCROLLER_SELECTOR) if not scroller: return [] seen_ids = set() all_rows = [] scroll_top = 0 stale_count = 0 while True: scroller.evaluate("(el, top) => el.scrollTop = top", scroll_top) time.sleep(0.5) rows = page.query_selector_all(ROW_SELECTOR) new_this_scroll = 0 for row in rows: row_id = row.get_attribute("data-id") if not row_id or row_id in seen_ids: continue seen_ids.add(row_id) new_this_scroll += 1 fields = {} for col in SCAN_COLUMNS: cell = row.query_selector(f"div[data-field='{col}']") fields[col] = cell.inner_text().strip().split("\n")[0] if cell else "" all_rows.append(fields) if new_this_scroll == 0: stale_count += 1 if stale_count >= 3: break else: stale_count = 0 if on_progress: on_progress(len(all_rows)) scroll_top += 200 return all_rows # --------------------------------------------------------------------------- # 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 = [] for row_data in all_rows: due_text = row_data["due_text"] pay_text = row_data["pay_text"] if not due_text: continue try: due_date = datetime.strptime(due_text, date_fmt) except ValueError: continue new_pay_date = calc_pay_date(due_date, days_before) new_pay_str = new_pay_date.strftime(date_fmt) if pay_text == new_pay_str: continue changes.append({**row_data, "new_pay": new_pay_str}) 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 = [] incorrect = [] for r in all_rows: due_text = r["dueDate"] pay_text = r["sendPaymentOn"] try: due_dt = datetime.strptime(due_text, date_fmt) expected = calc_pay_date(due_dt, days_before) expected_str = expected.strftime(date_fmt) except ValueError: expected_str = "???" ok = pay_text == expected_str entry = {**r, "expected": expected_str, "status": "OK" if ok else "WRONG"} results.append(entry) if not ok: incorrect.append(entry) return results, incorrect # --------------------------------------------------------------------------- # 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}']" ) if row: return row, start_scroll scroll_top = 0 max_scroll = scroller.evaluate("el => el.scrollHeight") while scroll_top <= max_scroll: scroller.evaluate("(el, top) => el.scrollTop = top", scroll_top) time.sleep(0.3) row = page.query_selector( f".MuiDataGrid-virtualScrollerRenderZone div.MuiDataGrid-row[data-id='{row_id}']" ) if row: return row, scroll_top scroll_top += 200 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.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, " ".react-datepicker input, " "input[type='date'], " "input.date-input, " ".datepicker input, " "input[placeholder*='date' i], " "input[placeholder*='MM' i]" ) if date_input: 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: # 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.4) page.keyboard.press("Escape") time.sleep(0.2) page.mouse.click(0, 0) 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 def edit_single_row(page, row_id, new_date_str, last_scroll=0, max_attempts=3): scroller = page.query_selector(SCROLLER_SELECTOR) for attempt in range(max_attempts): row, scroll_pos = find_row(page, scroller, row_id, start_scroll=last_scroll) if not row: if attempt < max_attempts - 1: time.sleep(0.5) continue return False, scroll_pos if try_edit_cell(page, row, new_date_str): return True, scroll_pos page.keyboard.press("Escape") time.sleep(0.3) return False, scroll_pos 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: success += 1 if on_progress: on_progress(i + 1, len(changes), c["invoice"], True) else: failed.append(c) if on_progress: on_progress(i + 1, len(changes), c["invoice"], False) except Exception as e: failed.append(c) if on_progress: on_progress(i + 1, len(changes), c["invoice"], False, str(e)) 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: success += 1 if on_progress: on_progress(-1, len(failed), c["invoice"], True, "retry") else: retry_failed.append(c) if on_progress: on_progress( -1, len(failed), c["invoice"], False, "retry failed" ) except Exception as e: retry_failed.append(c) if on_progress: on_progress(-1, len(failed), c["invoice"], False, str(e)) failed = retry_failed 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 # --------------------------------------------------------------------------- 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']" ) if not filter_toggle: status("ERROR: Could not find filter toggle button.") return filter_toggle.click() time.sleep(1) applied = False try: status_input = page.query_selector( "div[data-test-id='selectToPay-filter-filterStatuses'] input[placeholder='Status']" ) if not status_input: view_all = page.query_selector( "div[data-test-id='selectToPay-filter-menu'] >> text=View all" ) if view_all: view_all.click() time.sleep(0.5) status_input = page.query_selector( "div[data-test-id='selectToPay-filter-filterStatuses'] input[placeholder='Status']" ) if status_input: 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" ) if not 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: if "no error" in opt.inner_text().strip().lower(): no_errors = opt break if no_errors: no_errors.click() time.sleep(0.5) applied = True status("Applied 'No errors' filter.") except Exception: pass 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." ) else: input(" Apply the filter manually, then press Enter: ") close_btn = page.query_selector( "div[data-test-id='selectToPay-filter-menu-header'] button:last-child" ) if close_btn: close_btn.click() time.sleep(0.5) time.sleep(2) rows = page.query_selector_all(ROW_SELECTOR) status(f"{len(rows)} rows visible after filtering.") 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" ) if select_all: select_all.click() time.sleep(1) status("Selected all rows.") else: status("ERROR: Could not find Select All checkbox.") return status("Clicking Pay Invoices...") 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 is_disabled = pay_button.get_attribute("disabled") if is_disabled is not None: status("Pay Invoices button disabled — no rows selected.") if on_manual_fallback: 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.click() time.sleep(2) status("Clicked Pay Invoices.") status("Looking for re-authentication modal...") google_btn = page.query_selector("button >> text=Google") if not google_btn: google_btn = page.query_selector("text=Google") if google_btn: google_btn.click() status("Clicked Google login.") time.sleep(3) else: status("No re-auth modal found.")