Delete old bot versions in src/old

The bot, tests and .exe build use none of them. They stay in Git
history (902ff93) if the old logic is ever needed.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
VasanthandClaude Opus 5.5 committed 2026-10-21 17:31:33 +05:30
1 parent 230da1af36
commit 48781f35eb
10 files changed
-5021

No files matched your search

-22
View File
@@ -1,22 +0,0 @@
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
-1548
View File
File diff suppressed because it is too large. Load diff
-1351
View File
File diff suppressed because it is too large. Load diff
-38
View File
@@ -1,38 +0,0 @@
# -*- 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,
)
-38
View File
@@ -1,38 +0,0 @@
# -*- 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,
)
-491
View File
@@ -1,491 +0,0 @@
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))
File diff suppressed because it is too large. Load diff
-31
View File
@@ -1,31 +0,0 @@
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()
-60
View File
@@ -1,60 +0,0 @@
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()