184 lines
6.5 KiB
Python
184 lines
6.5 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
parse_text_pdf.py - DETERMINISTISCHER Parser fuer die Text-PDFs (kein LLM).
|
||
|
||
Nutzt die feste Struktur digital ausgefuellter BizMatch-Formulare (pypdf):
|
||
- Info-Seite und CA-Seite werden ueber ANKER-Texte gefunden (positionsunabh.).
|
||
- Auf der Info-Seite folgt nach dem Template ein WERTE-BLOCK in fester
|
||
Feld-Reihenfolge.
|
||
- Zuordnung von VORNE (Name..Types, einzeilig, fix) und von HINTEN
|
||
(letzte 6 Felder: Price, DownPay, Income, Accountant, Attorney, Bank).
|
||
Dazwischen = mehrzeiliger Background.
|
||
|
||
Checkbox 'interested_in_updates' steht NICHT im Text (eingebettetes Bild)
|
||
-> null, Marker _checkbox_pending=true fuer spaeteres Crop-Vision.
|
||
|
||
Aufruf:
|
||
python3 parse_text_pdf.py --pdf ~/data/S/"Sturgill, Garett 040126.pdf" [--debug]
|
||
"""
|
||
import argparse, re, json, os
|
||
from pypdf import PdfReader
|
||
|
||
ANCHOR_INFO = "BUYER INFORMATION SHEET"
|
||
ANCHOR_CA = "PROSPECTIVE BUYER AGREES TO KEEP AND HOLD CONFIDENTIAL"
|
||
|
||
# Leer-Marker: n, na, n/a (case-insensitiv), leer
|
||
_EMPTY_RE = re.compile(r"^(n|na|n/a|n\.a\.?)$", re.IGNORECASE)
|
||
|
||
def is_empty(s):
|
||
if s is None:
|
||
return True
|
||
t = s.strip()
|
||
return (not t) or bool(_EMPTY_RE.match(t))
|
||
|
||
def clean(s):
|
||
return None if is_empty(s) else s.strip()
|
||
|
||
def find_pages(reader):
|
||
info_txt = ca_txt = None
|
||
info_pageno = ca_pageno = None
|
||
for i, page in enumerate(reader.pages):
|
||
t = page.extract_text() or ""
|
||
flat = t.replace("\n", " ")
|
||
if info_txt is None and ANCHOR_INFO in t:
|
||
info_txt, info_pageno = t, i + 1
|
||
if ca_txt is None and ANCHOR_CA in flat:
|
||
ca_txt, ca_pageno = t, i + 1
|
||
return info_txt, ca_txt, info_pageno, ca_pageno
|
||
|
||
def value_block(info_txt):
|
||
"""Zeilen des Werte-Blocks: nach 'BANK:' bis vor 'Doc ID'."""
|
||
lines = [l.rstrip() for l in info_txt.splitlines()]
|
||
bank_idx = None
|
||
for i, l in enumerate(lines):
|
||
if l.strip().upper().startswith("BANK:"):
|
||
bank_idx = i
|
||
if bank_idx is None:
|
||
return None
|
||
block = []
|
||
for l in lines[bank_idx + 1:]:
|
||
if l.strip().lower().startswith("doc id"):
|
||
break
|
||
if l.strip():
|
||
block.append(l.strip())
|
||
return block
|
||
|
||
def parse_down_payment(raw):
|
||
"""
|
||
Gibt (zahl_oder_none, rohwert) zurueck.
|
||
Eindeutige Zahl: '350000' -> '350000'; '$350,000' -> '350000';
|
||
'1.5M'/'1.5 mil' -> '1500000'; '500k' -> '500000'.
|
||
Spanne ('50-200k'), Text ('Depends on deal') -> None, Rohwert behalten.
|
||
"""
|
||
if is_empty(raw):
|
||
return None, None
|
||
s = raw.strip()
|
||
low = s.lower().replace(",", "").replace("$", "").replace(" ", "")
|
||
# Spanne (enthaelt Bindestrich zwischen Zahlen) -> nicht eindeutig
|
||
if re.search(r"\d\s*[-–]\s*\d", low):
|
||
return None, s
|
||
m = re.fullmatch(r"(\d+(?:\.\d+)?)(k|m|mil|million)?", low)
|
||
if m:
|
||
num = float(m.group(1))
|
||
suffix = m.group(2)
|
||
if suffix == "k":
|
||
num *= 1_000
|
||
elif suffix in ("m", "mil", "million"):
|
||
num *= 1_000_000
|
||
return str(int(num)), s
|
||
return None, s # Text -> null, Rohwert behalten
|
||
|
||
def parse_info(info_txt, debug=False):
|
||
block = value_block(info_txt)
|
||
out = {}
|
||
if not block:
|
||
return out, None
|
||
if debug:
|
||
print(" --- Werte-Block ---")
|
||
for j, l in enumerate(block):
|
||
print(f" [{j}] {l!r}")
|
||
n = len(block)
|
||
# Von HINTEN: letzte 6 Felder
|
||
# Reihenfolge: Price, DownPay, Income, Accountant, Attorney, Bank
|
||
if n >= 6:
|
||
last6 = block[-6:]
|
||
out["total_purchase_price"] = clean(last6[0])
|
||
dp_num, dp_raw = parse_down_payment(last6[1])
|
||
out["down_payment"] = dp_num
|
||
out["down_payment_raw"] = dp_raw
|
||
# income/accountant/attorney/bank interessieren uns nicht als Zielfelder
|
||
front_end = n - 6
|
||
else:
|
||
front_end = n
|
||
|
||
# Von VORNE: feste einzeilige Felder
|
||
def g(i):
|
||
return clean(block[i]) if i < front_end else None
|
||
out["name_company"] = g(0)
|
||
# Phone-Zeile: "{PHONE} FAX CELL" bzw. mit Leer-Markern dazwischen
|
||
if 1 < front_end:
|
||
phone_line = block[1]
|
||
# grob: erstes Token=Phone, evtl. weitere; wir nehmen die Zeile roh und
|
||
# bereinigen Leer-Marker. Feinsplit spaeter falls noetig.
|
||
toks = phone_line.split()
|
||
toks = [t for t in toks if not is_empty(t)]
|
||
out["phone"] = toks[0] if toks else None
|
||
out["cell"] = toks[-1] if len(toks) > 1 else None
|
||
out["address"] = g(2)
|
||
out["email"] = g(3)
|
||
out["how_did_you_hear"] = g(4)
|
||
out["types_of_business_raw"] = g(5)
|
||
# Background = alles zwischen Index 6 und front_end (mehrzeilig)
|
||
if front_end > 6:
|
||
bg = " ".join(block[6:front_end]).strip()
|
||
out["background_experience"] = bg or None
|
||
else:
|
||
out["background_experience"] = None
|
||
return out, block
|
||
|
||
def parse_ca(ca_txt):
|
||
out = {}
|
||
lines = [l.strip() for l in ca_txt.splitlines() if l.strip()]
|
||
# Datum
|
||
for l in lines:
|
||
m = re.search(r"(\d{1,2})\s*/\s*(\d{1,2})\s*/\s*(\d{4})", l)
|
||
if m:
|
||
mm, dd, yy = m.groups()
|
||
out["date_of_introduction"] = f"{yy}-{int(mm):02d}-{int(dd):02d}"
|
||
break
|
||
# Prospective Buyer: erste Zeile nach dem Anker-Block (Zeile 1 ist der
|
||
# lange Vertragstext; Zeile 2 ist der Name)
|
||
if len(lines) >= 2 and ANCHOR_CA in lines[0].replace(" ", " "):
|
||
out["prospective_buyer"] = clean(lines[1])
|
||
return out
|
||
|
||
def main():
|
||
ap = argparse.ArgumentParser()
|
||
ap.add_argument("--pdf", required=True)
|
||
ap.add_argument("--debug", action="store_true")
|
||
args = ap.parse_args()
|
||
|
||
reader = PdfReader(args.pdf)
|
||
info_txt, ca_txt, info_no, ca_no = find_pages(reader)
|
||
|
||
result = {
|
||
"is_buyer_sheet": info_txt is not None,
|
||
"name_company": None, "prospective_buyer": None, "company": None,
|
||
"phone": None, "cell": None, "email": None, "address": None,
|
||
"state": None, "how_did_you_hear": None, "interested_in_updates": None,
|
||
"types_of_business_raw": None, "background_experience": None,
|
||
"total_purchase_price": None, "down_payment": None, "down_payment_raw": None,
|
||
"date_of_introduction": None,
|
||
"_checkbox_pending": True, "_parser": "deterministic",
|
||
"_info_page": info_no, "_ca_page": ca_no,
|
||
}
|
||
if info_txt:
|
||
info_fields, _ = parse_info(info_txt, debug=args.debug)
|
||
result.update(info_fields)
|
||
if ca_txt:
|
||
result.update(parse_ca(ca_txt))
|
||
|
||
print(json.dumps(result, indent=2, ensure_ascii=False))
|
||
|
||
if __name__ == "__main__":
|
||
main() |