automation-tools/Potiential-Automation
8
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Initial commit: Charge Billed Automation bot

Browse Source 
Waystar Charge Billed Report bot. Reads CMS-1500 claim PDFs from the
text layer (src/cms1500_text.py) instead of Google Document AI.
Dependencies managed with uv (Python 3.12). Secrets, PHI and build
output are git-ignored.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
VasanthandClaude Opus 5.5 committed 2026-10-21 17:12:27 +05:30
commit 902ff936cb
20 files changed
+7479
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

.gitignore
+18
View File
@@ -0,0 +1,18 @@
# Secrets: Google key, Waystar login and security answers
.env/
Input/
# PHI: claim PDFs, reports, logs, crash screenshots, payment export
Output/
Screenshots/
output.csv
# Python / tools
.venv*/
__pycache__/
.pytest_cache/
# PyInstaller build output
src/build/
src/dist/
build_log.txt
.python-version
+1
View File
@@ -0,0 +1 @@
3.12
.vscode/launch.json
Vendored
+16
View File
@@ -0,0 +1,16 @@
{
// Use IntelliSense to learn about possible attributes.
// Hover to view descriptions of existing attributes.
// For more information, visit: https://go.microsoft.com/fwlink/?linkid=830387
"version": "0.2.0",
"configurations": [
{
"name": "Python Debugger: Current File",
"type": "debugpy",
"request": "launch",
"program": "${file}",
"console": "integratedTerminal"
}
]
}
pyproject.toml
+22
View File
@@ -0,0 +1,22 @@
[project]
name = "charge-billed-automation"
version = "0.1.0"
requires-python = ">=3.12"
dependencies = [
"openpyxl>=3.1.5",
"pandas>=3.0.6",
"pdfplumber>=0.11.10",
"pywinauto>=0.6.9", # optional; imported inside try/except
"requests>=2.34.2",
"selenium>=4.51.0",
]
[dependency-groups]
dev = [
"pyinstaller==6.22.3",
"pytest>=9.1.1",
]
[tool.pytest.ini_options]
# src/test_docai.py and src/test_pdf.py are old Document AI scripts, not tests
testpaths = ["tests"]
src/ChargeBilledAutomation.py
+844
View File
@@ -0,0 +1,844 @@
import subprocess
import time
import os
import sys
import socket
import re
import pandas as pd
import requests
from pathlib import Path
from datetime import datetime
from selenium import webdriver
from selenium.webdriver.common.by import By
from selenium.webdriver.chrome.options import Options
from selenium.webdriver.support.ui import WebDriverWait
from selenium.webdriver.support import expected_conditions as EC
from selenium.webdriver.common.action_chains import ActionChains
from selenium.common.exceptions import TimeoutException
from selenium.webdriver.chrome.service import Service
from selenium.common.exceptions import WebDriverException
from selenium.common.exceptions import NoSuchElementException
from selenium.common.exceptions import StaleElementReferenceException
from selenium.webdriver.support.ui import Select
from selenium.webdriver.common.keys import Keys
from cms1500_text import extract_claim, to_report_row
try:
from pywinauto import Application
except Exception:
Application = None
# ================= PROJECT ROOT =================
if getattr(sys, "frozen", False):
PROJECT_ROOT = Path(sys.executable).parent
else:
PROJECT_ROOT = Path(__file__).resolve().parent.parent
# ================= GLOBAL CONFIG =================
DEBUG_PORT = 9222
USER_DATA_DIR = os.path.join(
os.environ["USERPROFILE"], "Automation", "ChromeProfile_CF"
)
today_date = datetime.now().strftime("%m-%d-%Y")
DOWNLOADS_DIR = Path.home() / "Downloads"
# ================= CHROME PATH =================
def find_chrome_path():
possible_paths = [
r"C:\Program Files\Google\Chrome\Application\chrome.exe",
r"C:\Program Files (x86)\Google\Chrome\Application\chrome.exe",
os.path.join(
os.environ.get("LOCALAPPDATA", ""),
r"Google\Chrome\Application\chrome.exe"
),
]
for path in possible_paths:
if os.path.exists(path):
return path
return None
# ================= PROXY CONFIG =================
def read_input_config():
config_path = PROJECT_ROOT / "Input" / "input.xlsx"
if not config_path.exists():
print("[ERROR] input.xlsx not found.")
sys.exit(1)
df = pd.read_excel(config_path, sheet_name="Sheet1", header=None)
config = {}
for _, row in df.iterrows():
key = str(row[0]).strip()
value = str(row[1]).strip()
config[key] = value
proxy_ip = config.get("Proxy IP", "")
proxy_port = config.get("Proxy Port", "")
waystar_username = config.get("Waystar Username", "")
waystar_password = config.get("Waystar Password", "")
portal_url = config.get("Waystar URL", "")
first_car_answer = config.get("Waystar SQ First Car", "")
first_job_answer = config.get("Waystar SQ First Job", "")
print(f"[INFO] Proxy → {proxy_ip}:{proxy_port}")
print(f"[INFO] Username → {waystar_username}")
print(f"[INFO] Portal URL → {portal_url}")
return (
proxy_ip, proxy_port, waystar_username, waystar_password,
portal_url, first_car_answer, first_job_answer,
)
# ================= DEBUG PORT CHECK =================
def wait_for_debug_port(port=9222, timeout=40):
start = time.time()
while time.time() - start < timeout:
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
if s.connect_ex(("127.0.0.1", port)) == 0:
return True
time.sleep(1)
return False
# ================= KILL CHROME =================
def kill_all_chrome():
try:
os.system("taskkill /F /IM chrome.exe /T >nul 2>&1")
time.sleep(3)
print("[INFO] All Chrome instances closed")
except Exception as e:
print(f"[WARN] Unable to kill Chrome: {e}")
# ================= START CHROME DEBUG =================
def start_chrome_debug(proxy_ip=None, proxy_port=None):
chrome_path = find_chrome_path()
if not chrome_path:
raise Exception("Google Chrome not found.")
print(f"[INFO] Using Chrome path: {chrome_path}")
os.makedirs(USER_DATA_DIR, exist_ok=True)
cmd = [
chrome_path,
f"--remote-debugging-port={DEBUG_PORT}",
f"--user-data-dir={USER_DATA_DIR}",
"--disable-blink-features=AutomationControlled",
"--ignore-certificate-errors",
"about:blank",
]
if proxy_ip and proxy_port:
proxy = f"{proxy_ip}:{proxy_port}"
cmd.append(f"--proxy-server=http://{proxy}")
print(f"[INFO] Chrome started with proxy → {proxy}")
else:
print("[INFO] Chrome started WITHOUT proxy")
print("[INFO] Launching Chrome...")
subprocess.Popen(cmd, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
if not wait_for_debug_port(DEBUG_PORT, timeout=40):
print("[ERROR] Chrome debug port did not open")
return False
print("[INFO] Chrome debug port is ready")
return True
# ================= SCREENSHOT =================
def capture_screenshot(driver, client_name, step_name):
if not driver:
return None
safe_client = "".join(c for c in client_name if c.isalnum() or c in ("_", "-"))
safe_step = "".join(c for c in step_name if c.isalnum() or c in ("_", "-"))
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
screenshots_dir = PROJECT_ROOT / "Screenshots" / today_date / safe_client
screenshots_dir.mkdir(parents=True, exist_ok=True)
screenshot_path = screenshots_dir / f"{safe_step}_{timestamp}.png"
try:
driver.save_screenshot(str(screenshot_path))
print(f"[INFO] Screenshot saved: {screenshot_path}")
return screenshot_path
except Exception as e:
print(f"[ERROR] Screenshot failed: {e}")
return None
# ================= ATTACH BROWSER =================
def attach_browser(download_path, retries=5):
for attempt in range(1, retries + 1):
try:
print(f"[INFO] Attaching browser (attempt {attempt})")
chrome_options = Options()
chrome_options.add_experimental_option(
"debuggerAddress", f"127.0.0.1:{DEBUG_PORT}"
)
chrome_options.set_capability("webSocketUrl", False)
prefs = {
"download.default_directory": str(download_path),
"download.prompt_for_download": False,
"download.directory_upgrade": True,
"plugins.always_open_pdf_externally": True,
"safebrowsing.enabled": True,
}
chrome_options.add_experimental_option("prefs", prefs)
service = Service()
driver = webdriver.Chrome(service=service, options=chrome_options)
_ = driver.current_url
driver.maximize_window()
print("[INFO] Browser attached successfully")
try:
driver.execute_cdp_cmd(
"Page.setDownloadBehavior",
{"behavior": "allow", "downloadPath": str(download_path)},
)
except Exception as e:
print(f"[WARN] Unable to set download behavior: {e}")
return driver
except WebDriverException as e:
print(f"[WARN] Attach attempt {attempt} failed: {e}")
time.sleep(3)
print("[ERROR] Failed to attach Chrome after retries")
return None
# ================= LOGIN =================
def login_to_portal(driver, waystar_username, waystar_password,
first_car_answer, first_job_answer):
wait = WebDriverWait(driver, 30)
try:
time.sleep(5)
print("[INFO] Starting login process")
username_field = wait.until(EC.presence_of_element_located((By.ID, "loginName")))
username_field.clear()
username_field.send_keys(waystar_username)
time.sleep(2)
password_field = wait.until(EC.presence_of_element_located((By.ID, "password")))
password_field.clear()
password_field.send_keys(waystar_password)
time.sleep(2)
driver.find_element(By.ID, "loginButton").click()
time.sleep(5)
print("[INFO] Login clicked")
try:
alert_page = driver.find_elements(
By.XPATH, "//span[contains(text(),'Proactive Alert')]"
)
if alert_page:
wait.until(
EC.element_to_be_clickable((By.XPATH, "//button[@id='continueButton']"))
).click()
time.sleep(3)
except Exception:
pass
try:
question_exists = driver.find_elements(
By.XPATH, "//*[@id='authDiv']//strong"
)
if question_exists:
question = wait.until(
EC.presence_of_element_located((By.XPATH, "//*[@id='authDiv']//strong"))
).text.lower()
print(f"[INFO] Security Question → {question}")
if "first car" in question:
answer = first_car_answer
elif "first job" in question:
answer = first_job_answer
else:
raise Exception(f"Unknown security question: {question}")
answer_box = wait.until(
EC.presence_of_element_located((By.ID, "verifyAnswer"))
)
answer_box.clear()
answer_box.send_keys(answer)
driver.find_element(By.ID, "VerifyButton").click()
print("[INFO] Security answer submitted")
time.sleep(3)
except Exception:
pass
try:
security_qpage = driver.find_elements(
By.XPATH, "//h2[normalize-space()='Security Information']"
)
if security_qpage:
print("[INFO] Security page appeared, skipping")
wait.until(EC.element_to_be_clickable((By.ID, "NextButton"))).click()
time.sleep(2)
wait.until(
EC.element_to_be_clickable(
(By.XPATH, "//a[normalize-space()='Skip This Step']")
)
).click()
time.sleep(5)
except Exception:
pass
print("[INFO] Login completed successfully")
except Exception as e:
print(f"[ERROR] Login failed: {e}")
raise
# ================= CALCULATE DATE RANGE =================
def get_date_range():
now = datetime.now()
first_of_current = datetime(now.year, now.month, 1)
last_of_prev = first_of_current - pd.Timedelta(days=1)
first_of_prev = datetime(last_of_prev.year, last_of_prev.month, 1)
return (
first_of_prev.strftime("%m/%d/%Y"),
last_of_prev.strftime("%m/%d/%Y"),
)
# ================= APPLY FILTERS =================
def apply_filters(driver):
wait = WebDriverWait(driver, 40)
print("[INFO] Applying filters")
wait.until(EC.presence_of_element_located((By.ID, "ClaimListingSearchButtonBottom")))
wait.until(EC.presence_of_element_located((By.ID, "SearchCriteria_Status")))
status_dropdown = Select(
wait.until(EC.element_to_be_clickable((By.ID, "SearchCriteria_Status")))
)
status_dropdown.select_by_visible_text("All")
date_dropdown = Select(
wait.until(EC.element_to_be_clickable((By.ID, "SearchCriteria_TransDate")))
)
date_dropdown.select_by_visible_text("Custom")
time.sleep(2)
start_date, end_date = get_date_range()
print(f"[INFO] Setting date range: {start_date} to {end_date}")
try:
start_date_field = wait.until(
EC.presence_of_element_located((By.ID, "transactionDateFromDateBox"))
)
start_date_field.clear()
start_date_field.send_keys(start_date)
time.sleep(1)
except Exception as e:
print(f"[WARN] Could not set start date: {e}")
try:
end_date_field = wait.until(
EC.presence_of_element_located((By.ID, "transactionDateToDateBox"))
)
end_date_field.clear()
end_date_field.send_keys(end_date)
time.sleep(1)
except Exception as e:
print(f"[WARN] Could not set end date: {e}")
search_btn = wait.until(
EC.element_to_be_clickable((By.ID, "ClaimListingSearchButtonBottom"))
)
driver.execute_script("arguments[0].click();", search_btn)
wait.until(
EC.presence_of_element_located(
(By.XPATH, "//table[@id='claimsGrid']//tbody//tr")
)
)
time.sleep(3)
print("[INFO] Filters applied successfully")
# ================= CLICK NEXT PAGE =================
def click_next_page(driver):
try:
next_btn = driver.find_element(By.XPATH, "//button[@id='innerPagerRight']")
if "disabled" in next_btn.get_attribute("class"):
return False
driver.execute_script("arguments[0].click();", next_btn)
WebDriverWait(driver, 10).until(EC.staleness_of(next_btn))
time.sleep(2)
return True
except Exception:
return False
# ================= DIRECT DOWNLOAD =================
def download_pdf_direct(driver, pdf_url, save_path):
try:
cookies = {c["name"]: c["value"] for c in driver.get_cookies()}
headers = {
"User-Agent": driver.execute_script("return navigator.userAgent;"),
"Referer": driver.current_url,
"Accept": "application/pdf,*/*",
}
response = requests.get(
pdf_url, headers=headers, cookies=cookies, stream=True, timeout=60
)
if response.status_code != 200:
print(f"[ERROR] HTTP {response.status_code} → {pdf_url}")
return False
content_type = response.headers.get("Content-Type", "")
if "pdf" not in content_type.lower() and "octet-stream" not in content_type.lower():
print(f"[WARN] Unexpected content type → {content_type}")
save_path = Path(save_path)
save_path.parent.mkdir(parents=True, exist_ok=True)
with open(save_path, "wb") as f:
for chunk in response.iter_content(chunk_size=8192):
if chunk:
f.write(chunk)
print(f"[SUCCESS] Downloaded → {save_path.name}")
return True
except Exception as e:
print(f"[ERROR] Direct download failed → {e}")
return False
# ================= PDF EXTRACTION =================
# Waystar claim PDFs are digital, so fields are read from the PDF text layer
# at their CMS-1500 positions (see cms1500_text.py) — exact, offline, no OCR.
def extract_pdf_data_fast(
pdf_path,
claim_id="",
patient_account="",
insurance="",
patient_name="",
total_charges="",
signature_physician="",
):
try:
claim = extract_claim(pdf_path)
except Exception as e:
print(f"[ERROR] PDF extraction failed → {pdf_path}: {e}")
return None
if claim.issues:
print(f"[WARN] Claim {claim_id} needs review → {'; '.join(claim.issues)}")
return to_report_row(
claim,
claim_id=claim_id,
insurance=insurance,
patient_name=patient_name,
signature_physician=signature_physician,
fallback_account=patient_account,
fallback_charges=total_charges,
)
# ================= UPDATE LOG =================
def update_log(log_path, claim_id, status, remarks=""):
now = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
new_row = pd.DataFrame([{
"Claim ID": claim_id,
"Status": status,
"Remarks": remarks,
"Timestamp": now,
}])
try:
existing = pd.read_excel(log_path)
final = pd.concat([existing, new_row], ignore_index=True)
except Exception:
final = new_row
final.to_excel(log_path, index=False)
# ================= INCREMENTAL EXCEL SAVE =================
def _save_batch_to_excel(excel_path, batch_data: list):
if not batch_data:
return
df_new = pd.DataFrame(batch_data)
try:
existing = pd.read_excel(excel_path)
existing = existing.dropna(how="all")
final = pd.concat([existing, df_new], ignore_index=True)
except Exception:
final = df_new
final.to_excel(excel_path, index=False)
print(f"[INFO] Saved {len(batch_data)} records → {excel_path.name} "
f"(total rows: {len(final)})")
def _get_processed_claim_ids(log_path) -> set:
try:
df = pd.read_excel(log_path)
done = df[df["Status"] == "SUCCESS"]["Claim ID"].astype(str).tolist()
return set(done)
except Exception:
return set()
# ================= RUN CLAIM REPORT =================
def run_claim_report(driver, account_name):
SAVE_EVERY = 10
wait = WebDriverWait(driver, 30)
actions = ActionChains(driver)
print(f"\n[INFO] ====== Running report for {account_name} ======")
today = datetime.now().strftime("%m-%d-%Y")
safe_name = "".join(
c for c in account_name if c.isalnum() or c in (" ", "_", "(", ")")
)
base = PROJECT_ROOT / "Output" / today
pdf_folder = base / "PDF Folder" / safe_name
report_dir = base / "reports"
logs_dir = base / "Logs"
pdf_folder.mkdir(parents=True, exist_ok=True)
report_dir.mkdir(parents=True, exist_ok=True)
logs_dir.mkdir(parents=True, exist_ok=True)
excel_path = report_dir / f"{safe_name}.xlsx"
log_path = logs_dir / f"{safe_name}_log.xlsx"
if not excel_path.exists():
pd.DataFrame(columns=[
"Claim ID", "Rendering Provider #", "Signature of Physician",
"Patient's Account No", "Date of Birth", "Insurance Name",
"Insured ID Number", "Patient's Name", "Date of Service",
"Diagnosis Codes", "CPT", "POS", "Modifier", "Units", "Charges",
"Line Charges", "Review Notes",
]).to_excel(excel_path, index=False)
if not log_path.exists():
pd.DataFrame(columns=[
"Claim ID", "Status", "Remarks", "Timestamp",
]).to_excel(log_path, index=False)
already_done = _get_processed_claim_ids(log_path)
print(f"[INFO] {len(already_done)} claims already processed for {account_name}")
# Navigate to Claims
try:
claims_processing = wait.until(
EC.presence_of_element_located(
(By.XPATH, "//a[contains(.,'Claims Processing')]")
)
)
actions.move_to_element(claims_processing).perform()
time.sleep(1)
claims_option = wait.until(
EC.presence_of_element_located(
(By.XPATH, "//a[normalize-space()='Claims']")
)
)
driver.execute_script("arguments[0].click();", claims_option)
time.sleep(3)
except Exception as e:
print(f"[ERROR] Navigation to Claims failed: {e}")
return
try:
apply_filters(driver)
except Exception as e:
print(f"[ERROR] Filter application failed: {e}")
return
page = 1
batch = []
total_saved = 0
consecutive_errors = 0
while True:
print(f"\n[INFO] Processing page {page}")
try:
WebDriverWait(driver, 20).until(
EC.presence_of_element_located(
(By.XPATH, "//table[@id='claimsGrid']//tbody//tr[@data-claimid]")
)
)
rows = driver.find_elements(
By.XPATH, "//table[@id='claimsGrid']//tbody//tr[@data-claimid]"
)
print(f"[INFO] Rows on page {page} → {len(rows)}")
except Exception as e:
print(f"[ERROR] Could not load rows on page {page}: {e}")
consecutive_errors += 1
if consecutive_errors >= 3:
print("[ERROR] 3 consecutive page-load failures — stopping account")
break
time.sleep(5)
continue
consecutive_errors = 0
for i in range(len(rows)):
claim_id = f"row_{page}_{i + 1}"
try:
rows = driver.find_elements(
By.XPATH,
"//table[@id='claimsGrid']//tbody//tr[@data-claimid]",
)
if i >= len(rows):
print(f"[WARN] Row {i + 1} no longer exists — skipping")
continue
row = rows[i]
claim_id = row.get_attribute("data-claimid") or ""
instance_id = row.get_attribute("data-instanceid") or ""
if not claim_id or not instance_id:
print(f"[WARN] Missing IDs row {i + 1} — skipping")
continue
if claim_id in already_done:
print(f"[INFO] [{i + 1}/{len(rows)}] Skipping (already done): {claim_id}")
continue
try:
claim_number = row.find_element(By.XPATH, ".//td[2]").text.strip()
except Exception:
claim_number = ""
try:
sig_raw = row.find_element(By.XPATH, ".//td[10]").text.strip()
if "," in sig_raw:
last, first = sig_raw.split(",", 1)
signature_physician = (
f"{first.strip().title()} {last.strip().title()}"
)
else:
signature_physician = sig_raw.strip().title()
except Exception:
signature_physician = ""
insurance = re.sub(
r"\s*\(.*?\)\s*$", "",
row.get_attribute("data-payername") or "",
).strip()
patient_name = row.get_attribute("data-patientname") or ""
charges = row.get_attribute("data-totalcharges") or ""
print(f"[INFO] [{i + 1}/{len(rows)}] Claim: {claim_id} | {patient_name}")
pdf_url = (
f"https://claims.zirmed.com/Claims/History/ViewClaimPDF"
f"?appId=1&claimId={claim_id}&instanceId={instance_id}"
f"&form=CMS1500_0212"
)
save_path = pdf_folder / f"{claim_id}.pdf"
if save_path.exists():
save_path = pdf_folder / f"{claim_id}_{instance_id}.pdf"
success = download_pdf_direct(driver, pdf_url, save_path)
if not success:
time.sleep(3)
success = download_pdf_direct(driver, pdf_url, save_path)
if not success:
update_log(log_path, claim_id, "FAILED", "Download failed after retry")
continue
pdf_data = extract_pdf_data_fast(
save_path,
claim_id=claim_id,
patient_account=claim_number,
insurance=insurance,
patient_name=patient_name,
total_charges=charges,
signature_physician=signature_physician,
)
if not pdf_data:
update_log(log_path, claim_id, "FAILED", "PDF extraction failed")
continue
batch.append(pdf_data)
already_done.add(claim_id)
update_log(log_path, claim_id, "SUCCESS", pdf_data["Review Notes"])
if len(batch) >= SAVE_EVERY:
_save_batch_to_excel(excel_path, batch)
total_saved += len(batch)
batch = []
except Exception as e:
print(f"[ERROR] Row {i + 1} (claim {claim_id}) failed → {e}")
update_log(log_path, claim_id, "FAILED", str(e)[:200])
if batch:
_save_batch_to_excel(excel_path, batch)
total_saved += len(batch)
batch = []
if not click_next_page(driver):
print(f"[INFO] No more pages for {account_name}")
break
page += 1
print(f"\n[SUCCESS] {account_name} complete — {total_saved} records saved")
print(f"[INFO] ====== Account {account_name} finished ======\n")
# ================= PROCESS ALL ACCOUNTS =================
def process_all_accounts(driver):
wait = WebDriverWait(driver, 30)
print("[INFO] Starting account processing...")
all_accounts = []
try:
dropdown = wait.until(
EC.element_to_be_clickable(
(By.XPATH, "//input[contains(@data-search-url,'AccountSearch')]")
)
)
dropdown.click()
time.sleep(2)
account_elements = driver.find_elements(
By.XPATH, "//ul[contains(@class,'ui-autocomplete')]//li"
)
all_accounts = [
acc.text.strip() for acc in account_elements if acc.text.strip()
]
print(f"[INFO] Total accounts found → {len(all_accounts)}")
driver.find_element(By.TAG_NAME, "body").click()
time.sleep(1)
except Exception as e:
print(f"[ERROR] Failed to get account list: {e}")
return
if not all_accounts:
print("[WARN] No accounts found — exiting")
return
# Deduplicate while preserving order
seen_names = set()
unique_accounts = []
for name in all_accounts:
if name not in seen_names:
seen_names.add(name)
unique_accounts.append(name)
if len(unique_accounts) < len(all_accounts):
print(f"[INFO] Removed {len(all_accounts) - len(unique_accounts)} "
f"duplicate account names")
all_accounts = unique_accounts
print(f"[INFO] Unique accounts to process → {len(all_accounts)}")
portal_home_url = driver.current_url
processed_accounts = set()
for idx, account_name in enumerate(all_accounts, 1):
if account_name in processed_accounts:
print(f"[WARN] '{account_name}' already processed — skipping")
continue
print(f"\n[INFO] ===== Account {idx}/{len(all_accounts)}: {account_name} =====")
try:
driver.get(portal_home_url)
time.sleep(3)
dropdown = wait.until(
EC.element_to_be_clickable(
(By.XPATH, "//input[contains(@data-search-url,'AccountSearch')]")
)
)
dropdown.clear()
dropdown.click()
time.sleep(2)
account_elements = driver.find_elements(
By.XPATH, "//ul[contains(@class,'ui-autocomplete')]//li"
)
clicked = False
for acc in account_elements:
if acc.text.strip() == account_name:
acc.click()
clicked = True
break
if not clicked:
print(f"[WARN] '{account_name}' not found in dropdown — skipping")
continue
try:
wait.until(
EC.invisibility_of_element_located(
(By.XPATH, "//ul[contains(@class,'ui-autocomplete')]")
)
)
except Exception:
pass
time.sleep(3)
processed_accounts.add(account_name)
run_claim_report(driver, account_name)
except Exception as e:
print(f"[ERROR] Account '{account_name}' failed: {e}")
try:
driver.get(portal_home_url)
time.sleep(3)
except Exception:
pass
continue
print(f"\n[INFO] ===== All {len(processed_accounts)} accounts processed =====")
print(f"[INFO] Accounts completed: {sorted(processed_accounts)}")
# ================= SELF-TEST =================
def self_test(pdf_paths):
"""Check a build without Chrome or Waystar: imports load, PDFs extract.
Prints counts only (no patient data). Returns a process exit code."""
failed = 0
for path in pdf_paths:
try:
claim = extract_claim(path)
print(f"[SELFTEST] {Path(path).name}: lines={len(claim.lines)} "
f"dx={len(claim.dx)} review_notes={len(claim.issues)}")
except Exception as e:
failed += 1
print(f"[SELFTEST] {Path(path).name}: ERROR {type(e).__name__}: {e}")
print("[SELFTEST] SELFTEST FAILED" if failed else "[SELFTEST] SELFTEST OK")
return 1 if failed else 0
# ================= MAIN =================
if __name__ == "__main__":
if "--selftest" in sys.argv:
sys.exit(self_test(sys.argv[sys.argv.index("--selftest") + 1:]))
print("=" * 80)
print("🚀 STARTING CHARGE BILLED AUTOMATION - PDF text layer v8.0")
print("=" * 80)
driver = None
try:
print("\n[STEP 1] Closing existing Chrome instances...")
kill_all_chrome()
print("\n[STEP 2] Reading configuration...")
(
proxy_ip, proxy_port, username, waystar_password,
portal_url, first_car_answer, first_job_answer,
) = read_input_config()
print("\n[STEP 3] Launching Chrome in debug mode...")
if not start_chrome_debug(proxy_ip, proxy_port):
raise Exception("Chrome debug failed to start")
time.sleep(5)
print("\n[STEP 4] Attaching Selenium to Chrome...")
driver = attach_browser(DOWNLOADS_DIR)
if not driver:
raise Exception("Driver attach failed")
print("\n[STEP 5] Opening Waystar portal...")
driver.execute_script("window.open('');")
driver.switch_to.window(driver.window_handles[-1])
driver.get(portal_url)
WebDriverWait(driver, 30).until(
EC.presence_of_element_located((By.ID, "loginName"))
)
print("[INFO] Portal loaded successfully")
print("\n[STEP 6] Performing login...")
login_to_portal(
driver, username, waystar_password,
first_car_answer, first_job_answer,
)
print("\n[STEP 7] Starting account processing...")
process_all_accounts(driver)
print("\n" + "=" * 80)
print("✅ AUTOMATION COMPLETED SUCCESSFULLY")
print("=" * 80)
except Exception as e:
print("\n" + "=" * 80)
print(f"❌ FATAL ERROR: {str(e)}")
print("=" * 80)
if driver:
try:
capture_screenshot(driver, "MAIN", "Fatal_Error")
except Exception:
pass
finally:
print("\n[CLEANUP] Closing resources...")
if driver:
try:
driver.quit()
print("[INFO] Browser closed successfully")
except Exception as e:
print(f"[WARN] Error closing browser: {e}")
print("\n🛑 Execution finished\n")
src/ChargeBilledAutomation.spec
+38
View File
@@ -0,0 +1,38 @@
# -*- mode: python ; coding: utf-8 -*-
a = Analysis(
['ChargeBilledAutomation.py'],
pathex=[],
binaries=[],
datas=[],
hiddenimports=['openpyxl', 'pywinauto'],
hookspath=[],
hooksconfig={},
runtime_hooks=[],
excludes=['torch', 'onnx', 'paddle', 'tensorflow', 'scipy', 'cv2'],
noarchive=False,
optimize=0,
)
pyz = PYZ(a.pure)
exe = EXE(
pyz,
a.scripts,
a.binaries,
a.datas,
[],
name='ChargeBilledAutomation',
debug=False,
bootloader_ignore_signals=False,
strip=False,
upx=True,
upx_exclude=[],
runtime_tmpdir=None,
console=True,
disable_windowed_traceback=False,
argv_emulation=False,
target_arch=None,
codesign_identity=None,
entitlements_file=None,
)
src/cms1500_text.py
+261
View File
@@ -0,0 +1,261 @@
"""
CMS-1500 (02/12) reader that uses the PDF's own text layer.
Waystar's ViewClaimPDF returns digital PDFs: every value is real text at a
fixed spot on the form. Reading those words by position is exact and runs
offline, so no OCR and no PHI leaves the machine.
All coordinates are normalized to 0-1 (x from the left, y from the top) and
taken from Waystar's CMS1500_0212 rendering.
"""
import re
from decimal import Decimal
from dataclasses import dataclass
@dataclass(frozen=True)
class Word:
text: str
x: float # horizontal mid-point, 0-1
y: float # vertical mid-point, 0-1
# ================= BOX 21 — DIAGNOSIS CODES =================
_DX_SLOT_LABEL = re.compile(r"^([A-L])\.$")
_ICD10 = re.compile(r"^[A-Z]\d[0-9A-Z]{1,5}$")
_DX_Y_MIN, _DX_Y_MAX = 0.590, 0.640
_DX_X_MAX = 0.615 # box 22/23 starts to the right
_DX_ROW_TOL = 0.010 # codes print a little above their slot label
def parse_diagnosis_codes(words):
"""Return [(slot_letter, code), ...] ordered A..L."""
in_box = [w for w in words if _DX_Y_MIN <= w.y <= _DX_Y_MAX and w.x <= _DX_X_MAX]
labels = [(m.group(1), w) for w in in_box if (m := _DX_SLOT_LABEL.match(w.text))]
found = {}
for w in in_box:
if not _ICD10.match(w.text):
continue
same_row = [(s, lw) for s, lw in labels
if abs(lw.y - w.y) <= _DX_ROW_TOL and lw.x <= w.x]
if not same_row:
continue
slot, _ = max(same_row, key=lambda sl: sl[1].x) # nearest label to the left
found.setdefault(slot, w.text)
return sorted(found.items())
# ================= BOX 24 — SERVICE LINES =================
@dataclass(frozen=True)
class ServiceLine:
line_no: int
dos_from: str
dos_to: str
pos: str
cpt: str
modifiers: tuple
dx_pointer: str
charge: object # Decimal, or None when not printed
units: str
npi: str
_CPT = re.compile(r"^(\d{5}|[A-V]\d{4}|\d{4}[TF])$")
_LINES_Y_MIN, _LINES_Y_MAX = 0.665, 0.848 # below the box-24 header, above box 25
_LINE_ROW_TOL = 0.006 # shaded supplemental row sits ~0.009 above
# Column x-ranges (on the word's mid-point), left to right.
_COLS = {
"from_mm": (0.035, 0.066), "from_dd": (0.066, 0.101), "from_yy": (0.101, 0.137),
"to_mm": (0.137, 0.171), "to_dd": (0.171, 0.205), "to_yy": (0.205, 0.240),
"pos": (0.240, 0.280),
"cpt": (0.315, 0.405),
"mods": (0.405, 0.545),
"pointer": (0.545, 0.615),
"charge": (0.615, 0.712),
"units": (0.712, 0.760),
"npi": (0.825, 0.960),
}
def _in_col(row, col):
lo, hi = _COLS[col]
return [w.text for w in sorted(row, key=lambda w: w.x) if lo <= w.x < hi]
def _date(mm, dd, yy):
if not (mm and dd and yy):
return ""
year = yy if len(yy) == 4 else f"20{yy}"
return f"{mm.zfill(2)}/{dd.zfill(2)}/{year}"
def parse_charge(text):
"""'9000' -> 90.00 (Waystar prints dollars and cents without a point)."""
text = text.replace("$", "").replace(",", "").strip()
if not re.fullmatch(r"\d+(\.\d{1,2})?", text):
return None
return Decimal(text).quantize(Decimal("0.01")) if "." in text else Decimal(text) / 100
def parse_service_lines(words):
"""Return ServiceLine for each box-24 row that has a CPT/HCPCS code, top to bottom."""
section = [w for w in words if _LINES_Y_MIN <= w.y <= _LINES_Y_MAX]
lo, hi = _COLS["cpt"]
anchors = sorted((w for w in section if lo <= w.x < hi and _CPT.match(w.text)),
key=lambda w: w.y)
lines = []
for anchor in anchors:
row = [w for w in section if abs(w.y - anchor.y) <= _LINE_ROW_TOL]
first = lambda col: (_in_col(row, col) or [""])[0]
charge_text = "".join(_in_col(row, "charge"))
lines.append(ServiceLine(
line_no=len(lines) + 1,
dos_from=_date(first("from_mm"), first("from_dd"), first("from_yy")),
dos_to=_date(first("to_mm"), first("to_dd"), first("to_yy")),
pos=first("pos"),
cpt=anchor.text,
modifiers=tuple(_in_col(row, "mods")),
dx_pointer="".join(_in_col(row, "pointer")),
charge=parse_charge(charge_text) if charge_text else None,
units="".join(_in_col(row, "units")),
npi=first("npi"),
))
return lines
# ================= HEADER BOXES (1a, 3, 26, 28) =================
_BOX = { # (x_min, x_max, y_min, y_max)
"insured_id": (0.600, 0.990, 0.135, 0.155),
"dob": (0.380, 0.500, 0.166, 0.182),
"patient_account": (0.300, 0.470, 0.862, 0.876),
"total_charge": (0.615, 0.745, 0.862, 0.876),
}
def _box_words(words, box):
x0, x1, y0, y1 = _BOX[box]
return [w.text for w in sorted(words, key=lambda w: (round(w.y, 3), w.x))
if x0 <= w.x <= x1 and y0 <= w.y <= y1]
def parse_header(words):
dob = [t for t in _box_words(words, "dob") if t.isdigit()]
total = "".join(t for t in _box_words(words, "total_charge") if t != "$")
return {
"insured_id": "".join(_box_words(words, "insured_id")).upper(),
"dob": _date(*dob) if len(dob) == 3 else "",
"patient_account": " ".join(_box_words(words, "patient_account")),
"total_charge": parse_charge(total) if total else None,
}
# ================= WHOLE CLAIM =================
@dataclass
class Claim:
insured_id: str
dob: str
patient_account: str
total_charge: object # Decimal or None
dx: list # [(slot_letter, code), ...]
lines: list # [ServiceLine, ...]
issues: list # human-readable review notes; empty = clean
def build_claim(pages):
"""Build one Claim from the word lists of every page of a claim PDF."""
if not any(pages):
return Claim("", "", "", None, [], [], ["No text layer on PDF (scanned?) - check by hand"])
header = parse_header(pages[0])
lines, totals = [], []
for words in pages:
for ln in parse_service_lines(words):
lines.append(ServiceLine(**{**ln.__dict__, "line_no": len(lines) + 1}))
totals.append(parse_header(words)["total_charge"])
total = None if None in totals else sum(totals, Decimal("0.00"))
claim = Claim(header["insured_id"], header["dob"], header["patient_account"],
total, parse_diagnosis_codes(pages[0]), lines, [])
claim.issues = _review(claim)
return claim
def _review(claim):
issues = []
if not claim.insured_id:
issues.append("No Insured ID (box 1a)")
if not claim.dob:
issues.append("No Date of Birth (box 3)")
if not claim.dx:
issues.append("No diagnosis codes (box 21)")
if not claim.lines:
issues.append("No service lines found")
slots = {s for s, _ in claim.dx}
for ln in claim.lines:
for label, value in (("Date of Service", ln.dos_from), ("POS", ln.pos),
("Units", ln.units), ("Charge", ln.charge),
("Rendering NPI", ln.npi)):
if not value:
issues.append(f"Line {ln.line_no}: no {label}")
for letter in ln.dx_pointer:
if letter not in slots:
issues.append(f"Line {ln.line_no}: Dx pointer {letter} has no diagnosis code")
line_sum = sum((ln.charge or Decimal("0.00") for ln in claim.lines), Decimal("0.00"))
if claim.total_charge is None:
issues.append("No total charge (box 28)")
elif line_sum != claim.total_charge:
issues.append(f"Line charges {line_sum:.2f} != box 28 total {claim.total_charge:.2f}")
return issues
# ================= REPORT ROW =================
def _per_line(values, label):
if len(values) == 1:
return values[0]
return "; ".join(f"{label} {i}: {v}" for i, v in enumerate(values, 1))
def to_report_row(claim, claim_id="", insurance="", patient_name="",
signature_physician="", fallback_account="", fallback_charges=""):
"""One report row per claim: the bot's original 15 columns + Line Charges + Review Notes."""
lines = claim.lines
money = lambda d: f"{d:.2f}" if d is not None else ""
dx_codes = [code for _, code in claim.dx]
return {
"Claim ID": claim_id,
"Rendering Provider #": _per_line([l.npi for l in lines], "RP #"),
"Signature of Physician": signature_physician,
"Patient's Account No": claim.patient_account or fallback_account,
"Date of Birth": claim.dob,
"Insurance Name": insurance,
"Insured ID Number": claim.insured_id,
"Patient's Name": patient_name,
"Date of Service": _per_line([l.dos_from for l in lines], "DOS"),
"Diagnosis Codes": f"{', '.join(dx_codes)} (Total: {len(dx_codes)})" if dx_codes else "",
"CPT": _per_line([l.cpt for l in lines], "CPT"),
"POS": _per_line([l.pos for l in lines], "POS"),
"Modifier": _per_line([" ".join(l.modifiers) for l in lines], "Modifier"),
"Units": _per_line([l.units for l in lines], "Units"),
"Charges": money(claim.total_charge) or fallback_charges,
"Line Charges": _per_line([money(l.charge) for l in lines], "Charge"),
"Review Notes": "; ".join(claim.issues),
}
# ================= PDF ENTRY POINT =================
def read_pdf_pages(pdf_path):
"""Word lists (normalized positions) for every page of the PDF."""
import pdfplumber
pages = []
with pdfplumber.open(str(pdf_path)) as pdf:
for page in pdf.pages:
width, height = float(page.width), float(page.height)
pages.append([
Word(w["text"], (w["x0"] + w["x1"]) / 2 / width, (w["top"] + w["bottom"]) / 2 / height)
for w in page.extract_words()
])
return pages
def extract_claim(pdf_path):
return build_claim(read_pdf_pages(pdf_path))
src/command to run exe.txt
+22
View File
@@ -0,0 +1,22 @@
python -m PyInstaller ^
--onefile ^
--console ^
--name ClaimStatusBot ^
--add-data "..\Config;Config" ^
main.py
python -m PyInstaller --onefile ^
--add-data "..\Input;Input" ^
--add-data "..\Screenshots;Screenshots" ^
--hidden-import=pyotp ^
--hidden-import=pandas ^
--hidden-import=openpyxl ^
main.py
python -m PyInstaller --onefile ^
--add-data "..\Input;Input" ^
--add-data "..\Screenshots;Screenshots" ^
--hidden-import=pandas ^
--hidden-import=openpyxl ^
AR_Workable
src/draft.py
+1548
View File
File diff suppressed because it is too large. Load diff
src/main.py
+1351
View File
File diff suppressed because it is too large. Load diff
src/main.spec
+38
View File
@@ -0,0 +1,38 @@
# -*- mode: python ; coding: utf-8 -*-
a = Analysis(
['main.py'],
pathex=[],
binaries=[],
datas=[('..\\Input', 'Input'), ('..\\Screenshots', 'Screenshots')],
hiddenimports=['pyotp', 'pandas', 'openpyxl'],
hookspath=[],
hooksconfig={},
runtime_hooks=[],
excludes=[],
noarchive=False,
optimize=0,
)
pyz = PYZ(a.pure)
exe = EXE(
pyz,
a.scripts,
a.binaries,
a.datas,
[],
name='main',
debug=False,
bootloader_ignore_signals=False,
strip=False,
upx=True,
upx_exclude=[],
runtime_tmpdir=None,
console=True,
disable_windowed_traceback=False,
argv_emulation=False,
target_arch=None,
codesign_identity=None,
entitlements_file=None,
)
src/main_v1.spec
+38
View File
@@ -0,0 +1,38 @@
# -*- mode: python ; coding: utf-8 -*-
a = Analysis(
['main_v1.py'],
pathex=[],
binaries=[],
datas=[],
hiddenimports=[],
hookspath=[],
hooksconfig={},
runtime_hooks=[],
excludes=[],
noarchive=False,
optimize=0,
)
pyz = PYZ(a.pure)
exe = EXE(
pyz,
a.scripts,
a.binaries,
a.datas,
[],
name='main_v1',
debug=False,
bootloader_ignore_signals=False,
strip=False,
upx=True,
upx_exclude=[],
runtime_tmpdir=None,
console=True,
disable_windowed_traceback=False,
argv_emulation=False,
target_arch=None,
codesign_identity=None,
entitlements_file=None,
)
src/pdf extraction.py
+491
View File
@@ -0,0 +1,491 @@
import re
import pdfplumber
# ===========================================================================
# CMS-1500 PDF EXTRACTOR — Production-Ready v3.0
# ===========================================================================
# Fixes vs prior versions:
# • CPT : strict 5-digit match guarded by known CPT range (10000-99999),
# zip-codes / account-numbers / tax-ids filtered out
# • POS : extracted per service-line row; count always matches CPT count
# • Modifier: column-position aware extraction (not regex on full row text)
# • Diagnosis codes: dual-pass — word-level near field-21 + full-text regex
# • Date of Service: handles split cells (MM | DD | YY) and slash formats
# • Units: column-position aware (column G on CMS-1500)
# • Rendering Provider #: 10-digit NPI per row, column J
# • All multi-value fields formatted as "Field 1: val; Field 2: val …"
# ===========================================================================
# ---------------------------------------------------------------------------
# Helper: is this token a valid CPT code?
# ---------------------------------------------------------------------------
def _is_cpt(token: str) -> bool:
"""Return True only for 5-digit numbers in the plausible CPT range."""
if not re.fullmatch(r"\d{5}", token):
return False
val = int(token)
return 10000 <= val <= 99999
# ---------------------------------------------------------------------------
# Helper: is this token a valid ICD-10 diagnosis code?
# ---------------------------------------------------------------------------
def _is_icd10(token: str) -> bool:
return bool(re.fullmatch(r"[A-Z]\d{2,3}(?:\.\d{1,4})?", token))
# ---------------------------------------------------------------------------
# Helper: clean / normalise a cell value
# ---------------------------------------------------------------------------
def _cell(c) -> str:
return str(c).strip() if c else ""
# ---------------------------------------------------------------------------
# Helper: join multi-value list into labelled string
# ---------------------------------------------------------------------------
def _format_list(values: list, label: str) -> str:
clean = [v for v in values if v]
if not clean:
return ""
if len(clean) == 1:
return clean[0]
return "; ".join(f"{label} {i+1}: {v}" for i, v in enumerate(clean))
# ---------------------------------------------------------------------------
# Known modifier codes (GP, GN, GO …) + numeric modifiers (25, 26, 51, 59 …)
# CMS-1500 column E holds up to 4 modifier slots.
# ---------------------------------------------------------------------------
_ALPHA_MODIFIERS = {
"GP", "GN", "GO", "GQ", "GX", "GY", "GZ",
"CQ", "TC", "26", "LT", "RT", "NU", "RR", "UE",
"KX", "KU", "KV", "KW", "KX",
"JW", "JZ",
"AA", "AB", "AD", "AE", "AF", "AG", "AH", "AI", "AJ", "AK",
"AM", "AO", "AP", "AQ", "AR", "AS", "AT",
"CA", "CB", "CC", "CD", "CE", "CF", "CH", "CI", "CJ", "CK", "CL",
"CM", "CN", "CO", "CP", "CR", "CS",
"DA",
"E1","E2","E3","E4",
"FA","F1","F2","F3","F4","F5","F6","F7","F8","F9",
"FX",
"GA","GB","GC","GD","GE","GF","GH","GJ","GK","GL","GM",
"HA","HB","HC","HD","HE","HF","HG","HH","HI","HJ","HK",
"HL","HM","HN","HO","HP","HQ","HR","HS","HT","HU","HV","HW",
"HX","HY","HZ",
"JA","JB","JC","JD","JE",
"LA","LC","LD","LE","LF","LG","LH","LI","LJ","LK","LL","LM",
"LN","LO","LP","LQ","LR","LS","LT","LU","LV","LW","LX","LY","LZ",
"P1","P2","P3","P4","P5","P6",
"Q0","Q1","Q2","Q3","Q4","Q5","Q6","Q7","Q8","Q9",
"QA","QB","QC","QD","QE","QF","QG","QH","QJ","QK","QL","QM","QN",
"QP","QQ","QR","QS","QT","QU","QV","QW","QX","QY","QZ",
"RA","RB","RC","RD","RE","RF","RG","RH","RI","RJ","RK","RL","RM",
"RN","RO","RP","RQ","RR","RS","RT","RU","RV","RW","RX","RY","RZ",
"SA","SB","SC","SD","SE","SF","SG","SH","SI","SJ","SK","SL","SM",
"SN","SO","SP","SQ","SR","SS","ST","SU","SV","SW","SX","SY",
"TA","TB","TC","TD","TE","TF","TG","TH","TI","TJ","TK","TL","TM",
"TN","TP","TQ","TR","TS","TT","TU","TV","TW","TX","TY","TZ",
"U1","U2","U3","U4","U5","U6","U7","U8","U9",
"UA","UB","UC","UD","UE","UF","UG","UH","UJ","UK","UL","UM","UN",
"UP","UQ","UR","US",
"V5","V6","V7","V8","V9",
"XE","XP","XS","XU",
}
_NUMERIC_MODIFIERS = {str(n) for n in range(22, 100)} # 22-99
_ALL_MODIFIERS = _ALPHA_MODIFIERS | _NUMERIC_MODIFIERS
# Tokens that look like 2-letter codes but are NOT modifiers
_NOT_MODIFIERS = {
"MM","DD","YY","TO","OF","OR","IN","IS","NO","BY","AT","BE",
"DO","GO","HE","HI","IF","ME","MY","OK","ON","SO","UP","US","WE",
"NPI","ABC","ICD","CPT","POS","DOS","EMG","REF","QTY",
"MM","DD","YY",
}
def _extract_modifiers_from_tokens(tokens: list[str]) -> str:
"""
Given the token list of a service-line row, return only legitimate
modifier codes. Tokens are scanned left-to-right; we stop after 4
modifiers (CMS-1500 allows max 4 per line).
"""
found = []
for t in tokens:
t_up = t.upper()
if t_up in _NOT_MODIFIERS:
continue
if t_up in _ALL_MODIFIERS and t_up not in found:
found.append(t_up)
if len(found) == 4:
break
return " ".join(found)
# ---------------------------------------------------------------------------
# CMS-1500 service-line table parser
# ---------------------------------------------------------------------------
def _parse_service_lines(tables: list) -> dict:
"""
Walk every table extracted by pdfplumber and collect per-row data for
columns A-J of CMS-1500 section 24:
A Date(s) of Service (From / To)
B Place of Service
C EMG
D CPT / HCPCS + Modifier(s)
E Diagnosis Pointer
F $ Charges
G Days or Units
H EPSDT
I ID Qual
J Rendering Provider ID
"""
cpts, pos_list, dos_list, mod_list, units_list, rp_list = [], [], [], [], [], []
for table in tables:
for row in table:
if not row:
continue
cells = [_cell(c) for c in row]
row_text = " ".join(cells)
# ---- Quick pre-filter: row must contain a valid CPT code --------
candidate_cpts = re.findall(r"\b(\d{5})\b", row_text)
valid_cpts_in_row = [c for c in candidate_cpts if _is_cpt(c)]
if not valid_cpts_in_row:
continue
# ---- Guard: skip header / label rows ----------------------------
# If the row text contains many uppercase words it's likely a label
upper_words = re.findall(r"\b[A-Z]{3,}\b", row_text)
if len(upper_words) > 4:
continue
# ---- CPT: first valid CPT per row (CMS-1500 = 1 CPT per line) ---
cpt_val = valid_cpts_in_row[0]
cpts.append(cpt_val)
# ---- Date of Service -------------------------------------------
# Format variants: "03 26 26", "03/26/26", "03-26-26"
date_match = re.search(
r"\b(\d{2})[\/\s\-]+(\d{2})[\/\s\-]+(\d{2,4})\b", row_text
)
if date_match:
mm, dd, yy = date_match.group(1), date_match.group(2), date_match.group(3)
yy = yy if len(yy) == 4 else ("20" + yy if int(yy) < 50 else "19" + yy)
dos_list.append(f"{mm}/{dd}/{yy}")
else:
dos_list.append("")
# ---- POS (Place of Service) — 2-digit code, column B -----------
# Strategy: find the 2-digit token that appears BEFORE the CPT code
# in the row text. Common POS values: 11,12,21,22,23,24,31,32,33,
# 41,42,49,50,51,52,53,54,55,56,57,61,62,65,71,72,81,99
pos_val = ""
row_before_cpt = row_text[: row_text.index(cpt_val)]
pos_candidates = re.findall(r"\b(\d{2})\b", row_before_cpt)
# Take the LAST 2-digit number before CPT (closest to CPT = POS)
# but exclude date fragments (must not be 01-31 preceded by another)
valid_pos_codes = {
"11","12","13","14","15","16","17","18","19","20",
"21","22","23","24","25","31","32","33","34","41",
"42","49","50","51","52","53","54","55","56","57",
"60","61","62","65","71","72","81","99",
}
for p in reversed(pos_candidates):
if p in valid_pos_codes:
pos_val = p
break
pos_list.append(pos_val)
# ---- Modifiers — tokens AFTER CPT code, before $-amount --------
# Isolate the segment after CPT to reduce noise
cpt_idx = row_text.index(cpt_val)
segment_after_cpt = row_text[cpt_idx + len(cpt_val):]
tokens_after = segment_after_cpt.split()
mod_val = _extract_modifiers_from_tokens(tokens_after)
mod_list.append(mod_val)
# ---- Units (Days/Units) — column G on CMS-1500 -----------------
# Units is almost always a small integer (1-99).
# We look for a standalone 1-2 digit number that appears AFTER
# any dollar amount (charges column F).
units_val = ""
# Remove currency-looking tokens to isolate units
seg_no_dollars = re.sub(r"\d{1,3}(?:,\d{3})*(?:\.\d{2})?", " MONEY ", segment_after_cpt)
unit_candidates = re.findall(r"\b([1-9]\d?)\b", seg_no_dollars)
# The first small integer after modifiers is usually units
for uc in unit_candidates:
if 1 <= int(uc) <= 99:
units_val = uc
break
units_list.append(units_val)
# ---- Rendering Provider / NPI — 10-digit number, column J ------
npi_candidates = re.findall(r"\b(\d{10})\b", row_text)
rp_list.append(npi_candidates[-1] if npi_candidates else "")
return {
"cpts": cpts,
"pos_list": pos_list,
"dos_list": dos_list,
"mod_list": mod_list,
"units_list": units_list,
"rp_list": rp_list,
}
# ---------------------------------------------------------------------------
# Diagnosis code extraction (field 21 of CMS-1500)
# ---------------------------------------------------------------------------
def _extract_diagnosis_codes(text: str, words: list) -> list:
"""Return a deduplicated, ordered list of ICD-10 codes from field 21."""
codes = []
# --- Pass 1: word-coordinate scan near field "21" label -----------------
for i, w in enumerate(words):
if w["text"] in ("21", "21.", "21:"):
base_y = w["top"]
# Field 21 spans ~80px vertically and the full page width after label
region = [
nxt for nxt in words
if (base_y - 5) <= nxt["top"] <= (base_y + 90)
and nxt["x0"] >= w["x0"] - 5
]
for word in region:
if _is_icd10(word["text"]):
codes.append(word["text"])
if codes:
break
# --- Pass 2: regex scan on full extracted text --------------------------
if not codes:
# Try to isolate the diagnosis section (between field 21 and 22/23/24)
diag_section_match = re.search(
r"(?:21|DIAGNOSIS)[^\n]*\n([\s\S]{0,400})(?:22\.|23\.|24\.|ICD)",
text, re.IGNORECASE
)
target = diag_section_match.group(1) if diag_section_match else text
raw = re.findall(r"\b([A-Z]\d{2,3}(?:\.\d{1,4})?)\b", target)
codes.extend(raw)
# --- Pass 3: broad scan if still empty ----------------------------------
if not codes:
raw = re.findall(r"\b([A-Z]\d{2,3}(?:\.\d{1,4})?)\b", text)
# Filter obvious non-ICD tokens
codes = [c for c in raw if not c.startswith("NP")]
# Deduplicate preserving order
seen = set()
unique = []
for c in codes:
if c not in seen:
seen.add(c)
unique.append(c)
return unique
# ===========================================================================
# MAIN FUNCTION
# ===========================================================================
def extract_pdf_data_fast(
pdf_path: str,
claim_id: str = "",
patient_account: str = "",
insurance: str = "",
patient_name: str = "",
total_charges: str = "",
signature_physician: str = "",
) -> dict | None:
"""
Extract all relevant fields from a CMS-1500 PDF.
Parameters
----------
pdf_path : Path to the PDF file.
claim_id : Pre-populated from upstream data (e.g. spreadsheet).
patient_account : Pre-populated patient account number.
insurance : Pre-populated insurance name.
patient_name : Pre-populated patient name.
total_charges : Pre-populated total charges.
signature_physician : Pre-populated physician signature.
Returns
-------
dict with all extracted fields, or None on critical failure.
"""
data = {
"Claim ID": claim_id,
"Rendering Provider #": "",
"Signature of Physician": signature_physician,
"Patient's Account No": patient_account,
"Date of Birth": "",
"Insurance Name": insurance,
"Insured ID Number": "",
"Patient's Name": patient_name,
"Date of Service": "",
"Diagnosis Codes": "",
"CPT": "",
"POS": "",
"Modifier": "",
"Units": "",
"Charges": total_charges,
}
try:
# ------------------------------------------------------------------ #
# Open once, grab text + words + tables #
# ------------------------------------------------------------------ #
with pdfplumber.open(pdf_path) as pdf:
page = pdf.pages[0]
text = page.extract_text(x_tolerance=2, y_tolerance=2) or ""
words = page.extract_words()
tables = page.extract_tables()
print(f"[DEBUG] Opened PDF: {pdf_path}")
print(f"[DEBUG] Tables found: {len(tables)}")
# ================================================================== #
# 1. DATE OF BIRTH (field 3 — format MM DD YYYY) #
# ================================================================== #
dob_match = re.search(r"\b(\d{2})\s+(\d{2})\s+(\d{4})\b", text)
if dob_match:
data["Date of Birth"] = (
f"{dob_match.group(1)}/{dob_match.group(2)}/{dob_match.group(3)}"
)
# ================================================================== #
# 2. INSURED'S I.D. NUMBER (field 1a) #
# ================================================================== #
insured_id = ""
# Method A: word following "1a." label
for i, w in enumerate(words):
if w["text"].lower().rstrip(".") in ("1a",):
for nxt in words[i + 1 : i + 6]:
t = nxt["text"]
# Must be alphanumeric, ≥8 chars, contain both letters & digits
if (
re.match(r"^[A-Z0-9]{8,20}$", t)
and re.search(r"[A-Z]", t)
and re.search(r"\d", t)
):
insured_id = t
break
if insured_id:
break
# Method B: proximity to "INSURED'S I.D. NUMBER" label
if not insured_id:
for i, w in enumerate(words):
if "INSURED" in w["text"].upper():
nearby = words[i : min(i + 6, len(words))]
if any("NUMBER" in nxt["text"].upper() for nxt in nearby):
base_y = w["top"]
candidates = [
nxt for nxt in words
if abs(nxt["top"] - base_y) < 15 and nxt["x0"] > w["x0"]
]
for c in reversed(candidates):
t = c["text"]
if (
re.match(r"^[A-Z0-9]{8,20}$", t)
and re.search(r"[A-Z]", t)
and re.search(r"\d", t)
):
insured_id = t
break
if insured_id:
break
# Method C: top-right quadrant scan
if not insured_id:
_skip = {"MEDICARE", "MEDICAID", "CHAMPVA", "TRICARE", "CHAMPUS"}
for w in words:
if w["x0"] > 400 and w["top"] < 200:
t = w["text"]
if (
re.match(r"^[A-Z0-9]{8,20}$", t)
and re.search(r"[A-Z]", t)
and re.search(r"\d", t)
and t not in _skip
):
insured_id = t
break
data["Insured ID Number"] = insured_id
print(f"[DEBUG] Insured ID: {insured_id}")
# ================================================================== #
# 3. DIAGNOSIS CODES (field 21) #
# ================================================================== #
dx_codes = _extract_diagnosis_codes(text, words)
if dx_codes:
codes_str = ", ".join(dx_codes)
data["Diagnosis Codes"] = f"{codes_str} (Total: {len(dx_codes)})"
print(f"[DEBUG] Diagnosis codes ({len(dx_codes)}): {dx_codes}")
# ================================================================== #
# 4. SERVICE LINES (section 24 — columns A-J) #
# ================================================================== #
svc = _parse_service_lines(tables)
cpts = svc["cpts"]
pos_list = svc["pos_list"]
dos_list = svc["dos_list"]
mod_list = svc["mod_list"]
units_list= svc["units_list"]
rp_list = svc["rp_list"]
print(f"[DEBUG] Service lines found: {len(cpts)}")
print(f"[DEBUG] CPTs : {cpts}")
print(f"[DEBUG] POS : {pos_list}")
print(f"[DEBUG] DOS : {dos_list}")
print(f"[DEBUG] Modifiers : {mod_list}")
print(f"[DEBUG] Units : {units_list}")
print(f"[DEBUG] Providers : {rp_list}")
# Counts must match CPT count — pad with "" if somehow shorter
n = len(cpts)
def _pad(lst): return lst + [""] * max(0, n - len(lst))
pos_list = _pad(pos_list)
dos_list = _pad(dos_list)
mod_list = _pad(mod_list)
units_list = _pad(units_list)
rp_list = _pad(rp_list)
data["CPT"] = _format_list(cpts, "CPT")
data["POS"] = _format_list(pos_list, "POS")
data["Date of Service"] = _format_list(dos_list, "DOS")
data["Modifier"] = _format_list(mod_list, "Modifier")
data["Units"] = _format_list(units_list, "Units")
data["Rendering Provider #"] = _format_list(rp_list, "RP #")
return data
except Exception as exc:
import traceback
print(f"[ERROR] PDF parsing failed → {pdf_path} → {exc}")
traceback.print_exc()
return None
# ===========================================================================
# Quick smoke-test (remove / guard with __main__ in production)
# ===========================================================================
if __name__ == "__main__":
import sys, json
if len(sys.argv) < 2:
print("Usage: python extract_pdf_data_v3.py <path_to_cms1500.pdf>")
sys.exit(1)
result = extract_pdf_data_fast(sys.argv[1])
print("\n===== EXTRACTED DATA =====")
print(json.dumps(result, indent=2))
src/rough draft.py
+1441
View File
File diff suppressed because it is too large. Load diff
src/test_docai.py
+31
View File
@@ -0,0 +1,31 @@
from google.cloud import documentai_v1 as documentai
from google.oauth2 import service_account
DOCAI_CREDENTIALS_PATH = r"C:\Potential Project G2\.env\pdf-extraction-492807-41531f078f92.json"
DOCAI_PROJECT_ID = "580776411227"
DOCAI_LOCATION = "us"
DOCAI_PROCESSOR_ID = "894f921d904bc9bd"
def test_connection():
try:
credentials = service_account.Credentials.from_service_account_file(
DOCAI_CREDENTIALS_PATH,
scopes=["https://www.googleapis.com/auth/cloud-platform"]
)
client_options = {"api_endpoint": "us-documentai.googleapis.com"}
client = documentai.DocumentProcessorServiceClient(
credentials=credentials,
client_options=client_options
)
processor_name = (
f"projects/{DOCAI_PROJECT_ID}/locations/{DOCAI_LOCATION}"
f"/processors/{DOCAI_PROCESSOR_ID}"
)
processor = client.get_processor(name=processor_name)
print(f"✅ Connection successful!")
print(f" Processor: {processor.display_name}")
print(f" State: {processor.state.name}")
except Exception as e:
print(f"❌ Connection failed: {e}")
test_connection()
src/test_pdf.py
+60
View File
@@ -0,0 +1,60 @@
from google.cloud import documentai_v1 as documentai
from google.oauth2 import service_account
DOCAI_CREDENTIALS_PATH = r"C:\Potential Project G2\.env\pdf-extraction-492807-41531f078f92.json"
DOCAI_PROJECT_ID = "1065395041789"
DOCAI_LOCATION = "us"
DOCAI_PROCESSOR_ID = "2d4c7121ce241b13"
# ← Put path to any CMS-1500 PDF you have
TEST_PDF_PATH = r"C:\Potential Project G2\Output\04-07-2026\PDF Folder\Dynamic Mobile Rehab (238268)\10882239127.pdf"
def test_pdf():
credentials = service_account.Credentials.from_service_account_file(
DOCAI_CREDENTIALS_PATH,
scopes=["https://www.googleapis.com/auth/cloud-platform"]
)
client_options = {"api_endpoint": "us-documentai.googleapis.com"}
client = documentai.DocumentProcessorServiceClient(
credentials=credentials,
client_options=client_options
)
with open(TEST_PDF_PATH, "rb") as f:
pdf_bytes = f.read()
processor_name = (
f"projects/{DOCAI_PROJECT_ID}/locations/{DOCAI_LOCATION}"
f"/processors/{DOCAI_PROCESSOR_ID}"
)
request = documentai.ProcessRequest(
name=processor_name,
raw_document=documentai.RawDocument(
content=pdf_bytes,
mime_type="application/pdf"
)
)
result = client.process_document(request=request)
document = result.document
print(f"\n✅ PDF processed successfully!")
print(f" Pages: {len(document.pages)}")
print(f"\n--- RAW TEXT (first 500 chars) ---")
print(document.text[:500])
print(f"\n--- FORM FIELDS DETECTED ---")
for page in document.pages:
for field in page.form_fields:
key = "".join(
document.text[s.start_index:s.end_index]
for s in field.field_name.text_anchor.text_segments
).strip()
val = "".join(
document.text[s.start_index:s.end_index]
for s in field.field_value.text_anchor.text_segments
).strip()
if key and val:
print(f" {key:30s} → {val}")
test_pdf()
tests/conftest.py
+5
View File
@@ -0,0 +1,5 @@
import sys
from pathlib import Path
# The bot's code lives in src/ (not a package); make it importable.
sys.path.insert(0, str(Path(__file__).resolve().parent.parent / "src"))
tests/test_cms1500_text.py
+197
View File
@@ -0,0 +1,197 @@
"""Unit tests for the CMS-1500 text-layer reader.
Words are synthetic (made-up codes and numbers placed at real CMS-1500
positions), so no patient data lives in this file.
"""
from decimal import Decimal
from cms1500_text import Word, parse_diagnosis_codes, parse_service_lines
def W(text, x, y):
return Word(text=text, x=x, y=y)
# Slot labels as Waystar prints them in box 21 (x, y are normalized 0-1).
DX_LABELS = [
W("A.", 0.047, 0.602), W("B.", 0.199, 0.603), W("C.", 0.352, 0.602), W("D.", 0.505, 0.604),
W("E.", 0.047, 0.617), W("F.", 0.200, 0.618), W("G.", 0.352, 0.618), W("H.", 0.505, 0.619),
W("I.", 0.045, 0.632), W("J.", 0.199, 0.632), W("K.", 0.352, 0.631), W("L.", 0.504, 0.632),
W("21.", 0.046, 0.581), W("DIAGNOSIS", 0.082, 0.581),
]
def test_dx_codes_are_keyed_by_slot_letter_in_slot_order():
words = DX_LABELS + [
# Printed slightly above their labels, with jittery y — the old OCR
# sort by (y, x) scrambled exactly this kind of row.
W("Z9999", 0.248, 0.5975), # B
W("M1111", 0.095, 0.5985), # A
W("R2222", 0.400, 0.5980), # C
W("S81812A", 0.095, 0.6130), # E
]
assert parse_diagnosis_codes(words) == [
("A", "M1111"), ("B", "Z9999"), ("C", "R2222"), ("E", "S81812A"),
]
def line_words(y, mm="01", dd="05", yy="26", pos="12", cpt="97110", mods=("GP",),
ptr="AB", charge="9000", units="2", npi="1234567890", row_no=None):
"""One box-24 service line at Waystar's x positions. None = not printed."""
ws = [W("NPI", 0.801, y)]
if row_no:
ws.append(W(row_no, 0.026, y - 0.009)) # gutter number sits in the shaded row above
for text, x in ((mm, 0.048), (dd, 0.083), (yy, 0.118), (pos, 0.259), (cpt, 0.348),
(ptr, 0.565), (charge, 0.683), (units, 0.724), (npi, 0.883)):
if text is not None:
ws.append(W(text, x, y))
for i, m in enumerate(mods):
ws.append(W(m, 0.424 + 0.035 * i, y))
return ws
BOX_24_HEADER = [W("24.", 0.046, 0.642), W("UNITS", 0.736, 0.661)]
BOX_25_ROW = [W("25.", 0.046, 0.854), W("44680", 0.324, 0.869), W("14439", 0.395, 0.869),
W("63000", 0.701, 0.869)]
def test_service_line_fields_are_read_from_their_columns():
words = BOX_24_HEADER + BOX_25_ROW + line_words(0.689, mods=("GP", "CQ", "59"))
[line] = parse_service_lines(words)
assert line.line_no == 1
assert line.dos_from == "01/05/2026"
assert line.dos_to == ""
assert line.pos == "12"
assert line.cpt == "97110"
assert line.modifiers == ("GP", "CQ", "59")
assert line.dx_pointer == "AB"
assert line.charge == Decimal("90.00")
assert line.units == "2"
assert line.npi == "1234567890"
def test_lines_are_numbered_top_to_bottom_and_ignore_gutter_and_box_26():
words = (BOX_24_HEADER + BOX_25_ROW
+ line_words(0.719, cpt="97140", row_no="2")
+ line_words(0.689, cpt="97110", row_no="1"))
lines = parse_service_lines(words)
assert [(l.line_no, l.cpt) for l in lines] == [(1, "97110"), (2, "97140")]
assert all(l.dos_from == "01/05/2026" for l in lines) # gutter "1"/"2" never read as MM
def test_a_missing_value_stays_blank_on_its_own_line():
words = (BOX_24_HEADER
+ line_words(0.689, units="2")
+ line_words(0.719, units=None)
+ line_words(0.750, units="1"))
assert [l.units for l in parse_service_lines(words)] == ["2", "", "1"]
def test_charge_without_decimal_point_has_implied_cents():
words = (BOX_24_HEADER
+ line_words(0.689, charge="12000")
+ line_words(0.719, charge="45.50"))
assert [l.charge for l in parse_service_lines(words)] == [Decimal("120.00"), Decimal("45.50")]
def test_hcpcs_line_with_dos_to_and_large_units():
words = BOX_24_HEADER + line_words(0.689, cpt="Q4250", mods=(), ptr="A", units="120")
words += [W("01", 0.155, 0.689), W("07", 0.186, 0.689), W("26", 0.224, 0.689)]
[line] = parse_service_lines(words)
assert (line.cpt, line.units, line.modifiers) == ("Q4250", "120", ())
assert line.dos_to == "01/07/2026"
from cms1500_text import Claim, build_claim, parse_header, to_report_row # noqa: E402
HEADER_WORDS = [
W("1a.", 0.619, 0.128), W("NUMBER", 0.723, 0.128),
W("1AB2CD3EF45", 0.677, 0.145),
W("3.", 0.381, 0.158), W("MM", 0.402, 0.165), W("DD", 0.439, 0.165), W("YY", 0.482, 0.165),
W("07", 0.401, 0.174), W("04", 0.436, 0.174), W("1950", 0.483, 0.174), W("M", 0.524, 0.174),
]
def test_header_fields_come_from_their_boxes():
h = parse_header(HEADER_WORDS + BOX_25_ROW)
assert h == {
"insured_id": "1AB2CD3EF45",
"dob": "07/04/1950",
"patient_account": "44680 14439",
"total_charge": Decimal("630.00"),
}
def page(*line_specs, total="18000"):
"""A full synthetic page: header + dx + given lines + box 25/28 row."""
ws = HEADER_WORDS + DX_LABELS + [W("M1111", 0.095, 0.598), W("Z9999", 0.248, 0.598)]
ws += BOX_24_HEADER + [W("25.", 0.046, 0.854), W("44680", 0.324, 0.869),
W("14439", 0.395, 0.869), W(total, 0.701, 0.869)]
for i, kw in enumerate(line_specs):
ws += line_words(0.689 + 0.030 * i, **kw)
return ws
def test_clean_claim_has_no_review_notes():
claim = build_claim([page({}, {"cpt": "97140"})])
assert isinstance(claim, Claim)
assert [l.cpt for l in claim.lines] == ["97110", "97140"]
assert claim.issues == []
def test_line_charges_must_add_up_to_box_28():
claim = build_claim([page({}, {}, total="20000")])
assert claim.issues == ["Line charges 180.00 != box 28 total 200.00"]
def test_missing_line_values_and_bad_pointer_are_flagged():
claim = build_claim([page({"units": None, "ptr": "AC"}, {"npi": None})])
assert claim.issues == [
"Line 1: no Units",
"Line 1: Dx pointer C has no diagnosis code",
"Line 2: no Rendering NPI",
]
def test_claim_with_no_service_lines_or_no_text_is_flagged():
assert build_claim([page()]).issues == ["No service lines found", "Line charges 0.00 != box 28 total 180.00"]
assert build_claim([[]]).issues == ["No text layer on PDF (scanned?) - check by hand"]
def test_lines_continue_across_pages_and_totals_add_up():
claim = build_claim([page({}, {}, {}, {}, {}, {}, total="54000"), page({}, total="9000")])
assert [l.line_no for l in claim.lines] == [1, 2, 3, 4, 5, 6, 7]
assert claim.total_charge == Decimal("630.00")
assert claim.issues == []
def test_report_row_keeps_legacy_columns_and_alignment():
claim = build_claim([page({"units": "2"}, {"cpt": "97140", "units": None, "mods": ("GP", "59")})])
row = to_report_row(claim, claim_id="999", insurance="PAYER", patient_name="DOE, J",
signature_physician="Jo Doe", fallback_account="")
assert list(row)[:15] == [
"Claim ID", "Rendering Provider #", "Signature of Physician", "Patient's Account No",
"Date of Birth", "Insurance Name", "Insured ID Number", "Patient's Name",
"Date of Service", "Diagnosis Codes", "CPT", "POS", "Modifier", "Units", "Charges",
]
assert row["CPT"] == "CPT 1: 97110; CPT 2: 97140"
assert row["Units"] == "Units 1: 2; Units 2: " # blank stays on line 2
assert row["Modifier"] == "Modifier 1: GP; Modifier 2: GP 59"
assert row["Diagnosis Codes"] == "M1111, Z9999 (Total: 2)"
assert row["Patient's Account No"] == "44680 14439"
assert row["Charges"] == "180.00"
assert row["Line Charges"] == "Charge 1: 90.00; Charge 2: 90.00"
assert row["Review Notes"] == "Line 2: no Units"
def test_report_row_for_single_line_claim_uses_plain_values():
row = to_report_row(build_claim([page({}, total="9000")]), claim_id="1")
assert (row["CPT"], row["Units"], row["Rendering Provider #"]) == ("97110", "2", "1234567890")
assert row["Review Notes"] == ""
def test_report_row_falls_back_to_grid_charges_when_box_28_is_blank():
claim = build_claim([page({})])
claim.total_charge = None
row = to_report_row(claim, claim_id="1", fallback_charges="90")
assert row["Charges"] == "90"
tests/test_selftest.py
+18
View File
@@ -0,0 +1,18 @@
"""The --selftest switch must check the build without touching Chrome or Waystar."""
import ChargeBilledAutomation as bot
def test_selftest_without_pdfs_passes_and_says_so(capsys):
assert bot.self_test([]) == 0
assert "SELFTEST OK" in capsys.readouterr().out
def test_selftest_fails_for_unreadable_pdf(capsys, tmp_path):
assert bot.self_test([str(tmp_path / "missing.pdf")]) == 1
assert "SELFTEST FAILED" in capsys.readouterr().out
def test_selftest_never_starts_chrome(monkeypatch):
monkeypatch.setattr(bot, "kill_all_chrome", lambda: (_ for _ in ()).throw(AssertionError("Chrome touched")))
monkeypatch.setattr(bot, "start_chrome_debug", lambda *a: (_ for _ in ()).throw(AssertionError("Chrome touched")))
assert bot.self_test([]) == 0
uv.lock
Generated
+1039
View File
File diff suppressed because it is too large. Load diff