"""PDF extraction for accounting calculations. Accuracy policy: - Digital text PDFs → extract tables, validate, may mark trusted - Scanned PDFs → OCR draft only; calculations blocked until user confirms - Calculations never run on unconfirmed OCR data """ from __future__ import annotations import re import uuid from dataclasses import asdict, dataclass, field from pathlib import Path from typing import Any, Literal import pdfplumber SourceType = Literal["digital", "scanned", "mixed", "empty"] DocumentKind = Literal["bank_statement", "balance_sheet", "unknown"] @dataclass class ExtractedRow: description: str = "" debit: float | None = None credit: float | None = None amount: float | None = None date: str | None = None account: str | None = None raw: str = "" @dataclass class ExtractionResult: job_id: str filename: str source_type: SourceType trusted: bool needs_confirm: bool rows: list[ExtractedRow] = field(default_factory=list) text_preview: str = "" document_text: str = "" warnings: list[str] = field(default_factory=list) validation: dict[str, Any] = field(default_factory=dict) ocr_used: bool = False document_kind: DocumentKind = "unknown" statement_totals: dict[str, Any] = field(default_factory=dict) def to_dict(self) -> dict[str, Any]: data = asdict(self) return data _MONEY_RE = re.compile( r"(? float | None: """Parse currency amounts, including signed values ('- Rs.381.93', '+ $528').""" if value is None: return None if isinstance(value, (int, float)): return float(value) text = str(value).strip() if not text or text in {"-", "—", "–", "+", "−"}: return None # accounting negatives: (123.45) neg = False if text.startswith("(") and text.endswith(")"): neg = True text = text[1:-1].strip() # Leading sign before or after currency: - Rs.381.93 / Rs. -381.93 / +Rs.528 sign = 1.0 sign_m = re.match(r"^([+-]|−)\s*", text) if sign_m: if sign_m.group(1) in {"-", "−"}: sign = -1.0 text = text[sign_m.end() :] text = text.replace(",", "").replace("₹", "").replace("$", "").replace("€", "").replace("£", "") text = re.sub(r"(?i)^(rs\.?|inr|usd|eur|gbp)\s*", "", text).strip() # Sign may sit after currency token if text.startswith("-") or text.startswith("−"): sign = -1.0 text = text[1:].strip() elif text.startswith("+"): text = text[1:].strip() try: num = float(text) num = -abs(num) if neg else num return sign * num except ValueError: return None def parse_statement_header_totals(text: str) -> dict[str, float | int | None] | None: """Extract labeled money-in / money-out totals from statement text when present. Works across bank, card, and wallet statements — values always come from the PDF text, never from hardcoded amounts. """ if not text: return None head = text[:4000] paid = received = None out_m = _LABELED_OUT_RE.search(head) in_m = _LABELED_IN_RE.search(head) if out_m: paid = _parse_money(out_m.group(1)) if in_m: received = _parse_money(in_m.group(1)) # Fallback: signed pair near the top (paid -, received +) if paid is None or received is None: pair = _SIGNED_PAIR_RE.search(head) if pair: if paid is None: paid = _parse_money(pair.group(1)) if received is None: received = _parse_money(pair.group(2)) if paid is not None: paid = abs(paid) if received is not None: received = abs(received) counts = _PAYMENT_COUNT_RE.search(head) paid_n = int(counts.group(1)) if counts else None recv_n = int(counts.group(2)) if counts else None if paid is None and received is None: return None return { "money_out": paid, "money_in": received, "payments_made": paid_n, "payments_received": recv_n, "source": "statement_header", } def _page_has_text(page) -> bool: text = (page.extract_text() or "").strip() return len(text) >= 40 def detect_source_type(pdf_path: Path) -> SourceType: with pdfplumber.open(str(pdf_path)) as pdf: if not pdf.pages: return "empty" text_pages = sum(1 for p in pdf.pages if _page_has_text(p)) total = len(pdf.pages) if text_pages == 0: return "scanned" if text_pages == total: return "digital" return "mixed" def _normalize_header(cell: Any) -> str: return re.sub(r"\s+", " ", str(cell or "").strip().lower()) def _map_table_headers(header_row: list[Any]) -> dict[str, int]: mapping: dict[str, int] = {} for idx, cell in enumerate(header_row): h = _normalize_header(cell) if not h: continue # Bank statement money-out columns (avoid bare "payment" — matches Paytm "Payment received") if any( k in h for k in ( "withdrawal", "withdrawals", "withdrawn", "debit", "dr amount", "money out", "paid out", "payment made", "money paid", "amount paid", "amount debited", ) ) or h in {"dr", "wdl", "withdrawal amt"}: mapping["debit"] = idx # Bank statement money-in columns elif any( k in h for k in ( "deposit", "deposits", "credit", "cr amount", "money in", "paid in", "receipt", "payment received", "money received", "amount credited", ) ) or h in {"cr"}: mapping["credit"] = idx elif any(k in h for k in ("balance", "closing bal", "running bal")): mapping["balance"] = idx # tracked but not required for totals elif h in {"amount", "amt", "value", "total"} or ( "amount" in h and "dr" not in h and "cr" not in h ): mapping["amount"] = idx elif "date" in h or "value date" in h or "txn date" in h or "posting" in h: mapping["date"] = idx elif "account" in h or "ledger" in h or "a/c" in h or "your account" in h: mapping["account"] = idx elif any( k in h for k in ( "desc", "particular", "narration", "details", "item", "remark", "transaction", "narration / particulars", ) ): mapping["description"] = idx return mapping def detect_document_kind( *, filename: str = "", text: str = "", header_hints: str = "", ) -> DocumentKind: """Heuristic: bank statement vs balance sheet vs unknown.""" blob = f"{filename}\n{text[:8000]}\n{header_hints}".lower() bank_hits = sum( 1 for k in ( "bank statement", "account statement", "statement of account", "opening balance", "closing balance", "withdrawal", "deposit", "narration", "transaction date", "value date", "available balance", "a/c no", "account number", "ifsc", "upi", "neft", "imps", "cheque", "chq", "paytm", "phonepe", "gpay", "google pay", "total money paid", "total money received", "payments made", "passbook payments", ) if k in blob ) sheet_hits = sum( 1 for k in ( "balance sheet", "statement of financial position", "total assets", "total liabilities", "shareholders' equity", "stockholders' equity", "retained earnings", "current assets", "non-current assets", ) if k in blob ) name = filename.lower() if any( k in name for k in ( "bank", "statement", "passbook", "txn", "transaction", "upi", "wallet", "card", "account", ) ): bank_hits += 2 if "balance_sheet" in name or "balancesheet" in name or "balance-sheet" in name: sheet_hits += 3 if bank_hits >= 2 and bank_hits > sheet_hits: return "bank_statement" if sheet_hits >= 1 and sheet_hits >= bank_hits: return "balance_sheet" # Many dated debit/credit rows without BS keywords → treat as bank-like statement if bank_hits >= 1: return "bank_statement" return "unknown" def _infer_direction_from_text(description: str, raw: str = "") -> str | None: """Return 'out', 'in', or None from Paytm/UPI/bank narration cues.""" blob = f"{description} {raw}".lower() # Bank credit markers first (NEFT CR / IMPS-CR / UPI-CR) if re.search(r"\b(?:neft|imps|rtgs)\s*cr\b|\bupi[-\s]?cr\b|\bcredit\s+interest\b", blob): return "in" if re.search(r"\b(?:neft|imps|rtgs)\s*dr\b|\bupi[-\s]?dr\b", blob): return "out" out_keys = ( "paid to", "money sent", "transferred to", "sent to", "debit", "payment made", "recharge of", "withdraw", "paid via", ) in_keys = ( "received from", "money received", "deposit", "payment received", "refund", "salary", ) if any(k in blob for k in out_keys): return "out" if any(k in blob for k in in_keys): return "in" # Bare "credit" is noisy on bank PDFs (column headers); only treat as in with word boundary. if re.search(r"\bcredits?\b", blob) and "credit card" not in blob: return "in" return None # HDFC / similar Indian bank text: "DD/MM/YY amount closing_balance" # Optional leading cheque/UPI ref (12–16 digits) on the same line. _HDFC_AMT_BAL_RE = re.compile( r"(?i)^(?:(?P\d{10,18})\s+)?" r"(?P\d{2}/\d{2}/\d{2})\s+" r"(?P[\d,]+\.\d{2})\s+" r"(?P[\d,]+\.\d{2})\s*$" ) _HDFC_DATE_START_RE = re.compile(r"^\d{2}/\d{2}/\d{2}\b") _PLAUSIBLE_TXN_MAX = 50_000_000.0 # ₹5 crore — reject UPI RRNs / account numbers def _is_plausible_txn_amount(amount: float | None) -> bool: """Reject ID-like integers (account nos, UPI RRNs) mistaken for money.""" if amount is None: return False abs_amt = abs(float(amount)) if abs_amt <= 0: return False if abs_amt > _PLAUSIBLE_TXN_MAX: return False # Bare integers with ≥10 digits are almost always refs, not rupees if abs_amt >= 1e9 and abs_amt == int(abs_amt): return False return True def _extract_rows_from_hdfc_style_text(text: str) -> list[ExtractedRow]: """Parse HDFC-style multi-line narrations ending in `date amount balance`.""" if not text: return [] # Strong signal this is an Indian bank statement layout low = text.lower() if not any( k in low for k in ( "withdrawal amt", "closing balance", "statement of account", "chq./ref", "value dt", "deposit amt", ) ): return [] lines = [ln.strip() for ln in text.splitlines() if ln.strip()] rows: list[ExtractedRow] = [] narr_buf: list[str] = [] prev_bal: float | None = None seen: set[str] = set() # After an amount/balance line, HDFC often prints leftover narration wraps — # skip those until the next dated narration starts. skip_wraps = False skip_prefixes = ( "page ", "hdfc bank", "statement of", "--------", "date narration", "deposit amt", "for any queries", "**continue", "account branch", "joint holders", "nomination", "currency", "cust id", "od limit", "branch code", "product code", "rtgs/neft", "a/c open", "statement from", "address", "phone no", "email", "city :", "state :", ) for ln in lines: low_ln = ln.lower() if any(low_ln.startswith(p) for p in skip_prefixes): continue if re.fullmatch(r"-+", ln.replace(" ", "")): continue m = _HDFC_AMT_BAL_RE.match(ln) if m: amount = _parse_money(m.group("amt")) bal = _parse_money(m.group("bal")) vdate = m.group("vdate") if amount is None or bal is None or not _is_plausible_txn_amount(amount): narr_buf = [] skip_wraps = True continue narration = " ".join(narr_buf).strip() narr_buf = [] # Drop leading txn date from narration if present narration = re.sub(r"^\d{2}/\d{2}/\d{2}\s+", "", narration).strip() narration = re.sub(r"\s+", " ", narration) direction: str | None = None if prev_bal is not None: as_out = abs((prev_bal - amount) - bal) < 0.05 as_in = abs((prev_bal + amount) - bal) < 0.05 if as_out and not as_in: direction = "out" elif as_in and not as_out: direction = "in" if direction is None: direction = _infer_direction_from_text(narration, ln) or "out" key = f"{vdate}|{amount:.2f}|{bal:.2f}|{narration[:80]}".lower() if key in seen: prev_bal = bal skip_wraps = True continue seen.add(key) rows.append( ExtractedRow( description=narration or "Transaction", debit=amount if direction == "out" else None, credit=amount if direction == "in" else None, amount=None, date=vdate, raw=f"{narration} | {ln}".strip(" |"), ) ) prev_bal = bal skip_wraps = True continue if skip_wraps: if _HDFC_DATE_START_RE.match(ln): skip_wraps = False narr_buf = [ln] # else: leftover wrap of previous narration — ignore continue # Accumulate narration / wrapped UPI lines until the amount-balance line if _HDFC_DATE_START_RE.match(ln): narr_buf = [ln] elif narr_buf: if len(narr_buf) > 12: narr_buf = narr_buf[-6:] narr_buf.append(ln) elif re.search(r"(?i)\b(?:upi|neft|imps|rtgs|ach|nach|pos|atm)\b", ln): narr_buf = [ln] return rows def _row_from_cells(cells: list[Any], mapping: dict[str, int]) -> ExtractedRow | None: def get(key: str) -> Any: idx = mapping.get(key) if idx is None or idx >= len(cells): return None return cells[idx] description = str(get("description") or "").strip() account = str(get("account") or "").strip() or None date = str(get("date") or "").strip() or None debit = _parse_money(get("debit")) credit = _parse_money(get("credit")) amount = _parse_money(get("amount")) raw = " | ".join(str(c) for c in cells if c is not None) if amount is None and debit is None and credit is None: # try to find any money in the row (prefer last signed match) joined = " ".join(str(c) for c in cells if c) money: list[float] = [] for m in _MONEY_RE.finditer(joined): sign_prefix = m.group(1) or "" num = _parse_money(m.group(2)) if num is None: continue if sign_prefix == "-": num = -abs(num) elif sign_prefix == "+": num = abs(num) money.append(num) if money: amount = money[-1] else: return None # Signed Amount column → debit (out) / credit (in) if amount is not None and debit is None and credit is None: direction = _infer_direction_from_text(description, raw) if amount < 0 or direction == "out": debit = abs(amount) amount = None elif amount > 0 and direction == "in": credit = abs(amount) amount = None elif amount < 0: debit = abs(amount) amount = None if not description: description = account or "Line item" # Skip summary / header junk rows desc_l = description.lower() if any( k in desc_l for k in ( "passbook payments", "payment made", "payment received", "total money paid", "page ", "for any queries", ) ): return None if re.search(r"\(\d+\s+payments?\)", raw, re.IGNORECASE) and debit is None and credit is None: # Skip account-summary rows that only show a balance + payment count. return None return ExtractedRow( description=description, debit=debit, credit=credit, amount=amount, date=date, account=account, raw=raw, ) def _find_table_header_index(table: list[list[Any]]) -> tuple[int, dict[str, int]]: """Locate the best header row in a pdfplumber table (title rows may sit above headers).""" best_i = 0 best_map: dict[str, int] = {} scan = min(4, len(table)) for i in range(scan): mapping = _map_table_headers(table[i]) if len(mapping) > len(best_map): best_map = mapping best_i = i return best_i, best_map def extract_document_text(pdf_path: Path, *, max_chars: int = 12000) -> str: """Pull readable text from the PDF pages for the LLM to read.""" bits: list[str] = [] with pdfplumber.open(str(pdf_path)) as pdf: for i, page in enumerate(pdf.pages, start=1): page_text = (page.extract_text() or "").strip() if page_text: bits.append(f"--- Page {i} ---\n{page_text}") text = "\n\n".join(bits).strip() if len(text) > max_chars: # Keep start (account header / opening) and end (closing / recent) head = max_chars // 2 tail = max_chars - head - 40 return text[:head].rstrip() + "\n\n…[middle pages truncated]…\n\n" + text[-tail:].lstrip() return text def extract_digital_tables(pdf_path: Path) -> tuple[list[ExtractedRow], str]: rows: list[ExtractedRow] = [] text_bits: list[str] = [] seen_raw: set[str] = set() with pdfplumber.open(str(pdf_path)) as pdf: for page in pdf.pages: page_text = (page.extract_text() or "").strip() if page_text: text_bits.append(page_text) tables = page.extract_tables() or [] for table in tables: if not table or len(table) < 1: continue header_i, mapping = _find_table_header_index(table) start = header_i + 1 if mapping else 0 if not mapping: # no clear header — treat first column as description, last as amount mapping = {"description": 0, "amount": max(0, len(table[0]) - 1)} # Prefer middle column as description when 5-col Paytm layout without header if len(table[0]) >= 5: mapping = { "date": 0, "description": 1, "account": 3, "amount": 4, } start = 0 for raw_row in table[start:]: if not raw_row or all(c is None or str(c).strip() == "" for c in raw_row): continue # skip repeated header-looking rows row_map = _map_table_headers(raw_row) if row_map and len(row_map) >= 2: continue parsed = _row_from_cells(list(raw_row), mapping) if not parsed: continue key = re.sub(r"\s+", " ", parsed.raw).strip().lower() if key in seen_raw: continue seen_raw.add(key) rows.append(parsed) full_text = "\n\n".join(text_bits) # Indian bank (HDFC-style) text layout — prefer over generic number scraping if len(rows) < 15 and full_text: hdfc_rows = _extract_rows_from_hdfc_style_text(full_text) if len(hdfc_rows) > len(rows): rows = hdfc_rows # Text-line fallback when tables under-extract (common on multi-page wallet/bank PDFs) if len(rows) < 15 and full_text: text_rows = _extract_rows_from_statement_text(full_text) if len(text_rows) > len(rows): rows = text_rows # Last-resort line parse — only currency-marked or decimal money, never bare IDs if not rows and text_bits: for line in full_text.splitlines(): line = line.strip() if len(line) < 4: continue low = line.lower() if any( k in low for k in ( "account no", "cust id", "phone", "micr", "ifsc", "pin code", "joint holders", "page no", ) ): continue # Prefer explicit currency markers cur_matches = list( re.finditer( r"(?i)([+-])?\s*(?:₹|rs\.?|inr)\s*(-?\d{1,3}(?:,\d{3})*(?:\.\d{1,2})?|-?\d+\.\d{1,2})", line, ) ) money = cur_matches or [ m for m in _MONEY_RE.finditer(line) if "." in (m.group(2) or "") # require decimals when no currency ] if not money: continue m = money[-1] amount = _parse_money((m.group(1) or "") + (m.group(2) or "")) if amount is None: amount = _parse_money(m.group(2)) if not _is_plausible_txn_amount(amount): continue desc = _MONEY_RE.sub("", line).strip(" -–—|\t") rows.append( ExtractedRow( description=desc or "Line item", amount=amount, raw=line, ) ) preview = full_text[:12000] return rows, preview def _extract_rows_from_statement_text(text: str) -> list[ExtractedRow]: """Parse statement lines from plain text when tables are fragmented.""" rows: list[ExtractedRow] = [] # Flatten soft line wraps: keep scanning with a rolling window lines = [ln.strip() for ln in text.splitlines() if ln.strip()] # Rebuild candidate strings that look like transactions buf = "" candidates: list[str] = [] for ln in lines: low = ln.lower() if low.startswith("page ") or low.startswith("for any queries") or low.startswith("note:"): if buf: candidates.append(buf) buf = "" continue if re.match(r"^\d{1,2}\s+[A-Za-z]{3}\b", ln) or re.match(r"^\d{1,2}:\d{2}\s*(AM|PM)\b", ln, re.I): if buf: candidates.append(buf) buf = ln continue if buf: buf = f"{buf} {ln}" if any( k in low for k in ("upi ref", "ref no", "txn id", "transaction id", "neft", "imps", "rtgs") ) or re.search(r"[+-]\s*(?:₹|rs\.?|\$)\s*[\d,]+", ln, re.I): if buf: candidates.append(buf) buf = "" if buf: candidates.append(buf) seen: set[str] = set() for cand in candidates: m = re.search( r"(?i)(?P[+-])\s*(?:₹|rs\.?|\$)\s*(?P[\d,]+\.?\d*)\s*$", cand.strip(), ) if not m: m = re.search( r"(?i)(?P[+-])\s*(?:₹|rs\.?|\$)\s*(?P[\d,]+\.?\d*)", cand, ) if not m: continue amount = _parse_money(m.group("amt")) if amount is None: continue amount = abs(amount) signed = m.group("signed") body = cand[: m.start()].strip() date = None dm = re.match(r"^(\d{1,2}\s+[A-Za-z]{3}(?:\s+\d{1,2}:\d{2}\s*(?:AM|PM))?)", body, re.I) if dm: date = dm.group(1) body = body[dm.end() :].strip() direction = "out" if signed == "-" else "in" narr_dir = _infer_direction_from_text(body) if narr_dir: direction = narr_dir key = re.sub(r"\s+", " ", cand).lower() if key in seen: continue seen.add(key) rows.append( ExtractedRow( description=body or "Transaction", debit=amount if direction == "out" else None, credit=amount if direction == "in" else None, amount=None, date=date, raw=cand, ) ) return rows def extract_ocr_draft(pdf_path: Path, max_pages: int = 5) -> tuple[list[ExtractedRow], str, list[str]]: """OCR draft for scanned PDFs. Never trusted for calculations.""" warnings: list[str] = [] try: from pdf2image import convert_from_path # type: ignore except ImportError: return [], "", [ "OCR packages not installed. Install: pip install pdf2image pytesseract " "and system packages poppler-utils + tesseract-ocr. " "Or enter numbers manually after upload." ] try: import pytesseract # type: ignore except ImportError: return [], "", [ "pytesseract not installed. Run: pip install pytesseract " "and install system package tesseract-ocr." ] try: images = convert_from_path(str(pdf_path), dpi=200, first_page=1, last_page=max_pages) except Exception as exc: return [], "", [f"Could not render PDF for OCR (is poppler installed?): {exc}"] text_bits: list[str] = [] rows: list[ExtractedRow] = [] for img in images: try: text = pytesseract.image_to_string(img) or "" except Exception as exc: warnings.append(f"Tesseract failed: {exc}") continue text = text.strip() if text: text_bits.append(text) for line in text.splitlines(): line = line.strip() money = list(_MONEY_RE.finditer(line)) if not money: continue m = money[-1] amount = _parse_money((m.group(1) or "") + (m.group(2) or "")) if amount is None: amount = _parse_money(m.group(2)) if amount is None: continue desc = _MONEY_RE.sub("", line).strip(" -–—|\t") rows.append(ExtractedRow(description=desc or "OCR line", amount=amount, raw=line)) warnings.append( "OCR output is a draft only. Confirm or edit every amount before calculating." ) if len(images) >= max_pages: warnings.append(f"OCR limited to first {max_pages} pages for speed.") return rows, "\n\n".join(text_bits)[:12000], warnings def extract_pdf(pdf_path: Path, filename: str | None = None) -> ExtractionResult: job_id = str(uuid.uuid4()) name = filename or pdf_path.name source = detect_source_type(pdf_path) warnings: list[str] = [] rows: list[ExtractedRow] = [] preview = "" ocr_used = False if source in {"digital", "mixed"}: rows, preview = extract_digital_tables(pdf_path) if source == "mixed": warnings.append("PDF has mixed text/image pages. Prefer confirming extracted amounts.") if source == "scanned" or (source == "mixed" and not rows): ocr_rows, ocr_preview, ocr_warnings = extract_ocr_draft(pdf_path) warnings.extend(ocr_warnings) if ocr_rows: rows = ocr_rows ocr_used = True if not preview: preview = ocr_preview elif source == "scanned": warnings.append( "No amounts extracted. Enter line items manually, then confirm to calculate." ) if source == "empty": warnings.append("PDF appears empty.") # Prefer full page text for the model to read (digital); fall back to preview/OCR text document_text = "" if source in {"digital", "mixed"}: document_text = extract_document_text(pdf_path, max_chars=24000) if not document_text: document_text = preview kind = detect_document_kind( filename=name, text=document_text or preview, header_hints=" ".join((r.description or "")[:40] for r in rows[:30]), ) statement_totals = parse_statement_header_totals(document_text or preview) or {} if statement_totals and kind == "unknown": kind = "bank_statement" warnings.append( "Detected wallet/UPI statement totals (money paid / money received)." ) if kind == "bank_statement": warnings.append( "Detected as a bank / account statement. Totals use money-in vs money-out " "(not balance-sheet assets/liabilities)." ) elif kind == "unknown" and len(rows) >= 20: dated = sum(1 for r in rows if r.date) if dated >= max(5, len(rows) // 5): kind = "bank_statement" warnings.append( "Treating this PDF as a transaction statement based on dated line items." ) # Trust only digital extracted with usable numeric rows and no OCR trusted = source == "digital" and bool(rows) and not ocr_used needs_confirm = not trusted if not rows: needs_confirm = True trusted = False # Header totals alone are enough to answer spend/received questions if not rows and statement_totals: trusted = source == "digital" and not ocr_used needs_confirm = not trusted result = ExtractionResult( job_id=job_id, filename=name, source_type=source, trusted=trusted, needs_confirm=needs_confirm, rows=rows, text_preview=(preview or document_text)[:3000], document_text=document_text or preview, warnings=warnings, ocr_used=ocr_used, document_kind=kind, statement_totals=statement_totals, ) return result