""" RewardPilot - Statement Parser ============================== Parses uploaded credit card statements (PDF or CSV) into normalized transactions. Real PDF text extraction via pdfplumber; CSV via stdlib. Output transaction shape: {date, description, merchant, category, amount, brand_key} The parser is intentionally tolerant: Indian issuer statements vary widely, so we use a robust line/row heuristic + the merchant resolver to categorize. Password-protected PDFs are supported if the password is supplied. """ import csv import io import re from datetime import datetime from typing import Dict, List, Optional from merchants import resolve_merchant DATE_PATTERNS = [ "%d/%m/%Y", "%d-%m-%Y", "%d/%m/%y", "%d-%b-%Y", "%d %b %Y", "%Y-%m-%d", "%m/%d/%Y", "%d.%m.%Y", "%d-%b-%y", "%d %b %y", ] AMOUNT_RE = re.compile(r"(-?\d[\d,]*\.\d{2})") # strict (decimals) - safe to scan whole rows _AMOUNT_LENIENT_RE = re.compile(r"(-?\d[\d,]*(?:\.\d{1,2})?)") # integer-or-decimal - dedicated amount column only def _parse_amount_cell(s: str): """Parse a value from a known amount column: tolerate Rs/INR/₹, '/-', and missing decimals.""" cleaned = re.sub(r"(?:rs\.?|inr|₹)", "", s or "", flags=re.IGNORECASE).replace("/-", "").strip() m = _AMOUNT_LENIENT_RE.search(cleaned) if not m: return None try: return float(m.group(1).replace(",", "")) except ValueError: return None _DATE_TOKEN_RE = re.compile( r"(\d{1,2}[/\-.]\d{1,2}[/\-.]\d{2,4}|\d{1,2}[ \-]\w{3}[ \-]\d{2,4}|\d{4}-\d{2}-\d{2})" ) # Rows whose description is really a payment/credit, not a purchase. # No trailing \b: HDFC glues the next word onto "PAYMENT" ("CREDIT CARD PAYMENTNet # Banking"), which a trailing word-boundary would miss, letting a bill payment be # counted as spend. _PAYMENT_RE = re.compile( r"\b(payment received|credit card payment|card payment|neft|imps|rtgs|upi[\s/-]*payment|" r"autopay|auto pay|bbps|nach|e-?mandate|refund|reversal|cashback received)", re.IGNORECASE, ) # Non-spend ledger lines: taxes, finance/interest charges, and card fees. These are # not purchases the user chooses a card for, so they're excluded from the analysis. _CHARGES_RE = re.compile( r"(\bigst\b|\bcgst\b|\bsgst\b|\bgst\b|finance charge|interest charge|\binterest\b|" r"late fee|late payment|membership fee|annual fee|joining fee|renewal fee|cash advance fee|" r"over\s?limit fee|fuel surcharge|surcharge|" r"forex markup|fx markup|markup fee|currency conversion|cross[\s-]?currency|" r"reward point|\bfee reversal\b|\bemi (?:principal|interest)\b)", re.IGNORECASE, ) # Foreign-currency codes on international lines, and helpers to drop forex-conversion # detail sub-lines and to render a clean merchant name from a raw descriptor. _CCY_RE = re.compile(r"\b(usd|eur|gbp|aed|sgd|jpy|aud|cad|chf|hkd|thb|myr|sar|qar|cny)\b", re.IGNORECASE) _FOREX_DETAIL_RE = re.compile(r"^(convert|conversion|fx|forex|foreign\s*currency|markup|intl\s*txn)\b", re.IGNORECASE) _CCY_AMT_RE = re.compile(_CCY_RE.pattern + r"\s*[\d.,]+", re.IGNORECASE) _PREFIX_RE = re.compile(r"^\s*(pos|vps|ecom(m)?|imps|neft|rtgs|upi|ach|nach|intl|int'?l)[\s/*:.\-]+", re.IGNORECASE) def _is_forex_detail_line(desc: str) -> bool: d = (desc or "").strip() return bool(_FOREX_DETAIL_RE.match(d) and _CCY_RE.search(d)) def _title(s: str) -> str: return re.sub(r"\bPaypal\b", "PayPal", s.title()) # HDFC glues the city onto the merchant with no space ("Entertainmenthyderabad", # "Kurlamumbai"). Split a trailing known city off, longest names first. _CITIES = tuple(sorted(( "navimumbai", "newdelhi", "bengaluru", "bangalore", "hyderabad", "ahmedabad", "coimbatore", "visakhapatnam", "bhubaneswar", "chandigarh", "gurugram", "gurgaon", "mumbai", "delhi", "chennai", "kolkata", "pune", "noida", "jaipur", "lucknow", "kochi", "indore", "nagpur", "surat", "thane", "vadodara", "bhopal", "patna", "ludhiana", "agra", "nashik", "faridabad", "ghaziabad", "rajkot", "meerut", "amritsar", "mysuru", "mysore", "goa", "kanpur", "varanasi", "guwahati", ), key=len, reverse=True)) def _split_glued_city(s: str) -> str: low = s.lower() for c in _CITIES: if low.endswith(c) and len(s) > len(c) + 2 and s[-len(c) - 1] not in " ,": return s[:-len(c)] + " " + s[-len(c):] return s def _clean_merchant(raw: str) -> str: original = re.sub(r"[\x00-\x1f\x7f]+", " ", (raw or "")).strip() # never show control chars in a name strip = lambda x: re.sub(r"^[\s*.,\-]+|[\s*.,\-]+$", "", x).strip() if re.search(r"paypal", original, re.IGNORECASE): m = re.search(r"paypal\s*\*?\s*([a-z][a-z0-9 &._-]{1,24})", original, re.IGNORECASE) sub = "" if m and m.group(1): sub = strip(re.sub(r"[\d.,]+", "", re.sub(r"\b(convert|conversion|usd|eur|gbp)\b", "", m.group(1), flags=re.IGNORECASE))) sub = sub.upper() if len(sub) <= 4 else _title(sub) return f"PayPal · {sub}" if len(sub) > 1 else "PayPal" s = _PREFIX_RE.sub("", original) s = _CCY_AMT_RE.sub(" ", s) s = re.sub(r"\b(convert|conversion)\b", " ", s, flags=re.IGNORECASE) s = re.sub(r"#\S*", " ", s) s = re.sub(r"\d[\d.,]{3,}", " ", s) s = re.sub(r"\*+", " ", s) s = strip(re.sub(r"\s{2,}", " ", s)) if len(s) < 2: return original[:40].strip() or "Transaction" return _title(_split_glued_city(s))[:40].strip() # trim AFTER truncation so a cut long name has no trailing space def _parse_date(s: str) -> Optional[str]: """Parse a date that may carry a trailing timestamp (e.g. '17/04/2026 22:46:14').""" s = (s or "").strip() if not s: return None # Axis prints the year with a leading apostrophe ("01 Jan '26") - drop it so # strptime's %y matches, and collapse the double space it leaves behind. s = re.sub(r"'\s*(\d)", r"\1", s) s = re.sub(r"\s{2,}", " ", s) # pull just the date token if there's a time or extra text alongside it tok = _DATE_TOKEN_RE.search(s) candidates = [s] if tok: candidates.insert(0, tok.group(1)) for cand in candidates: for fmt in DATE_PATTERNS: try: return datetime.strptime(cand.strip(), fmt).strftime("%Y-%m-%d") except ValueError: continue return None def _detect_delimiter(text: str) -> str: """Issuer CSVs vary: HDFC uses '~|~', others tab/pipe/semicolon/comma.""" if "~|~" in text: return "~|~" sample = next((ln for ln in text.splitlines() if ln.strip()), "") counts = {d: sample.count(d) for d in ["\t", "|", ";", ","]} best = max(counts, key=counts.get) return best if counts[best] > 0 else "," def _split_line(line: str, delim: str) -> List[str]: if delim == ",": try: return [c for c in next(csv.reader([line]))] except StopIteration: return [] return line.split(delim) def _find_col(header: List[str], names) -> int: for i, h in enumerate(header): if any(n in h for n in names): return i return -1 def _is_txn_header(fields: List[str]) -> bool: joined = " ".join(f.lower() for f in fields) has_date = "date" in joined has_amt = any(k in joined for k in ("amt", "amount", "value", "debit", "withdrawal", "credit")) # debit/credit-split statements have no "amount" column has_desc = any(k in joined for k in ("description", "narration", "details", "particular", "merchant")) return has_date and has_amt and has_desc def _normalize_row(date_s: str, desc: str, amount: float) -> Dict: resolved = resolve_merchant(desc) return { "date": _parse_date(date_s) or date_s, "description": desc.strip()[:120], "merchant": resolved["matched_merchant"] or _clean_merchant(desc), "category": resolved["category"], "brand_key": resolved["brand_key"], "amount": round(abs(amount), 2), } def _is_credit(direction: str, desc: str) -> bool: """True for refunds / payments / credits (excluded from spend analysis).""" d = (direction or "").strip().lower() if d in ("cr", "credit") or d.startswith("cr"): return True if _PAYMENT_RE.search(desc or "") or _CHARGES_RE.search(desc or ""): return True return False def parse_csv(content: bytes) -> List[Dict]: """ Parse issuer CSV exports into purchase transactions. Robust to real-world layouts: a metadata preamble, a dedicated transactions section with its own header, non-comma delimiters (HDFC uses '~|~'), timestamped dates ('17/04/2026 22:46:14'), Indian lakh grouping ('1,42,212.00'), and a Debit/Credit column. Credits, payments and refunds are skipped so only real spends feed the optimal-card analysis. """ text = content.decode("utf-8-sig", errors="ignore") text = text.replace("\x00", "") # NUL bytes (corrupt/binary upload) make csv.reader raise delim = _detect_delimiter(text) lines = [ln for ln in text.splitlines() if ln.strip()] rows = [[c.strip() for c in _split_line(ln, delim)] for ln in lines] rows = [r for r in rows if any(r)] if not rows: return [] # locate the transactions header (preferred path) hidx = next((i for i, r in enumerate(rows) if _is_txn_header(r)), -1) out: List[Dict] = [] if hidx >= 0: header = [h.lower() for h in rows[hidx]] di = _find_col(header, ("date",)) ci = _find_col(header, ("description", "narration", "details", "particular", "merchant")) ai = _find_col(header, ("amt", "amount", "value")) if ai < 0: ai = _find_col(header, ("debit", "withdrawal")) # the Debit/Credit indicator column ki = next((i for i, h in enumerate(header) if "debit" in h and "credit" in h), -1) if ki < 0: ki = _find_col(header, ("dr/cr", "cr/dr", "type", "credit")) for r in rows[hidx + 1:]: date_s = r[di] if 0 <= di < len(r) else "" iso = _parse_date(date_s) if not iso: continue # blank line, section break, or summary row amount = _parse_amount_cell(r[ai]) if 0 <= ai < len(r) else None if amount is None and ai < 0: # ONLY when no amount/debit column exists: scan the row. If a debit column # exists but is empty, the row is a credit/refund - do NOT scan (that would # grab the Credit-column value and count it as a spend). m = AMOUNT_RE.search(" ".join(r)) amount = float(m.group(1).replace(",", "")) if m else None if not amount: continue desc = r[ci] if 0 <= ci < len(r) else " ".join(r) direction = r[ki] if 0 <= ki < len(r) else "" if _is_credit(direction, desc): continue if _is_forex_detail_line(desc): # drop forex conversion sub-lines ("Convert USD 324.50") continue out.append(_normalize_row(iso, desc, amount)) return out # header found: trust it (even if every row was a credit/payment) # fallback: no recognizable header - infer per row from date + amount tokens for r in rows: date_field = next((c for c in r if _parse_date(c)), None) if not date_field: continue amts = [c for c in r if AMOUNT_RE.search(c)] if not amts: continue amt_raw = amts[-1] m = AMOUNT_RE.search(amt_raw) amount = float(m.group(1).replace(",", "")) if amount == 0: continue others = [c for c in r if c not in (date_field, amt_raw) and not AMOUNT_RE.fullmatch(c.strip())] desc = max(others, key=len) if others else " ".join(r) if _is_credit("", desc): continue if _is_forex_detail_line(desc): continue out.append(_normalize_row(_parse_date(date_field), desc, amount)) return out # a date anywhere on a line: dd/mm/yyyy, dd-mm-yy, dd Mon yy, dd-MON-yy, dd/Mon/yyyy _PDF_DATE_RE = re.compile(r"(\d{1,2}[/\-. ](?:\d{1,2}|[A-Za-z]{3,9})[/\-. ]'?\d{2,4})") _SIGNED_AMT_RE = re.compile(r"(-?\d[\d,]*\.\d{2})") # lines that are clearly not transactions (summary / headers / footers) _PDF_STOP = ( "total amount due", "minimum amount due", "credit limit", "available", "statement period", "statement date", "reward", "opening balance", "closing balance", "payment due", "card number", "transaction details", "your card", "page ", "apr", "gst summary", "important", ) _PDF_START = ("transaction details", "your transactions", "transaction date", "date transaction") # Amex prints month-first dates with no year on the row ("June 03"); the year comes # from the statement period. Support that with a separate matcher + year injection. _MONTHS = {m: i for i, m in enumerate( ["jan", "feb", "mar", "apr", "may", "jun", "jul", "aug", "sep", "oct", "nov", "dec"], 1)} _MONTH_DAY_RE = re.compile( r"\b((?:jan|feb|mar|apr|may|jun|jul|aug|sep|oct|nov|dec)[a-z]*\.?\s+\d{1,2})(?!\d)", re.I) def _parse_month_day(txt: str, year: int) -> Optional[str]: m = re.match(r"([a-z]{3})[a-z]*\.?\s+(\d{1,2})", (txt or "").strip(), re.I) if not m: return None mo = _MONTHS.get(m.group(1).lower()) day = int(m.group(2)) if not mo or not (1 <= day <= 31): return None return f"{year:04d}-{mo:02d}-{day:02d}" # Descriptor noise seen on Kotak (and others): payment-gateway prefixes, embedded URLs, # brand domains, and the glued "SpendsArea" MCC-group column printed before the amount. _DOMAIN_RE = re.compile(r"\b([a-z][\w-]*)\.(?:com|in|org|net|io|co|edu|gov|sg|us|uk|app)\w*", re.I) _AGG_PREFIX_RE = re.compile(r"\b(raz|pyu|payu|ccbill|billdesk|pinelabs)\s*\*\s*", re.I) _SPENDS_AREA_RE = re.compile( r"\s+(services|education|automotive|apparel(?:\s*&\s*accessories)?|" r"transport\s*&?\s*freight|freight|quasi\s*cash|financial\s*services|" r"government\s*services|professional\s*services|utilities|telecom|insurance)\s*$", re.I) def _strip_descriptor_noise(s: str) -> str: s = re.sub(r"https?[:/]*\S*", " ", s, flags=re.I) # URLs, incl. glued "httpsgmat.." s = re.sub(r"\([^)]*\)", " ", s) # "(*ConverttoEMI)" markers s = re.sub(r"\s*\([^)]*$", " ", s) # a marker truncated at a line wrap "(*Conv" s = _AGG_PREFIX_RE.sub(" ", s) # "RAZ*RAPIDO" -> "RAPIDO" s = _DOMAIN_RE.sub(r"\1", s) # "github.com" -> "github" return re.sub(r"\s{2,}", " ", s).strip() class _PdfRow: pass def _parse_pdf_lines(lines: List[str], gated: bool = True) -> List[Dict]: out: List[Dict] = [] started = not gated # ungated fallback: parse from the top pending: List[str] = [] # buffered wrapped merchant-name lines (IDFC style) # Year for month-first dates that omit it (Amex "June 03"): take the latest 4-digit # year printed anywhere on the statement, else the current year. _yrs = re.findall(r"\b(20\d{2})\b", "\n".join(lines)) default_year = int(max(_yrs)) if _yrs else datetime.now().year for raw in lines: line = (raw or "").strip() if not line: continue low = line.lower() if not started: if any(k in low for k in _PDF_START): started = True continue dm = _PDF_DATE_RE.search(line) md = None if dm else _MONTH_DAY_RE.search(line) # Amex month-first fallback date_txt = dm.group(1) if dm else (md.group(1) if md else None) amts = list(_SIGNED_AMT_RE.finditer(line)) if date_txt and amts: iso = _parse_date(date_txt) if dm else _parse_month_day(date_txt, default_year) if not iso: # date-like but not a real date -> buffer text if re.search(r"[A-Za-z]", line): pending = (pending + [line])[-2:] continue amt_m = amts[-1] # last money figure on the line is the INR amount amount = float(amt_m.group(1).replace(",", "")) tail = line[amt_m.end():].strip().upper() is_credit = amount < 0 or tail.startswith("CR") or _is_credit("", line) amount = abs(amount) if amount == 0: # FX-only / zero rows pending = [] continue desc = line[:amt_m.start()].replace(date_txt, " ") desc = desc.replace("₹", " ") # Axis prints "₹ 980.00" desc = re.sub(r"\b(?:inr|rs)\.?\b", " ", desc, flags=re.I) # RBL "INR 1,899.00" prefix desc = _strip_descriptor_noise(desc) # Kotak URLs / gateway prefix / EMI markers / domains desc = re.sub(r"^[\s|]*\d{1,2}:\d{2}(?::\d{2})?\s+", " ", desc) # HDFC leading time "22:16" desc = re.sub(r"\s\+\s*\d{1,3}\b", " ", desc) # HDFC "+ 15" reward-point marker desc = re.sub(r"\b\d{6,}\b", " ", desc) # strip long reference numbers desc = re.sub(r"#\w+", " ", desc) # Axis "#HJ1S1I6CZYEUZ9" transaction refs desc = re.sub(r"\b0\.00\b", " ", desc) # strip the FX (international) 0.00 column desc = re.sub(r"\b[DC]R\b", " ", desc, flags=re.I) desc = re.sub(r"(?\n Convert USD 324.50 ", so # the amount line itself has no merchant - recover it from the wrapped buffer. core = re.sub(r"\b(convert|conversion|fx|forex)\b", " ", desc, flags=re.I) core = _CCY_AMT_RE.sub(" ", core) core = _CCY_RE.sub(" ", core) core = re.sub(r"[\d.,]+", " ", core).strip(" ,|*") if len(core) < 2: # inline text is only a forex marker desc = " ".join(pending).strip(" ,|") or desc pending = [] if is_credit: continue # Re-run the credit/charge check on the RESOLVED description: on wrapped # layouts (IDFC) the merchant text comes from `pending`, so the raw `line` # check above never sees words like "Interest charges" / "Forex Markup Fee". if _is_credit("", desc) or _CHARGES_RE.search(desc): continue # ungated fallback: require a real merchant name so summary/total lines # (which have no description) aren't mistaken for transactions if not gated and not re.search(r"[A-Za-z]", desc): continue # last resort: an international row whose merchant never resolved - label it # rather than showing the raw "Convert USD 324.50" forex fragment if _is_forex_detail_line(desc): desc = "International transaction" out.append(_normalize_row(iso, desc or "Transaction", amount)) else: # Candidate wrapped merchant name (no amount on this line). Some issuers # (RBL) repeat the date on the continuation line, so strip any leading date # before buffering it for the amount row that follows. nm = (line.replace(date_txt, " ") if date_txt else line).strip(" ,|") if re.search(r"[A-Za-z]", nm) and len(nm) <= 60 and not any(k in low for k in _PDF_STOP): pending = (pending + [nm])[-2:] else: pending = [] return out def parse_pdf(content: bytes, password: Optional[str] = None) -> List[Dict]: """ Extract transactions from a PDF statement (HDFC, IDFC, Axis, ICICI, etc.). Section-gated line parser: starts at the transactions table, keeps debit rows, excludes credits/payments/refunds (CR or negative amount), and reconstructs merchant names that wrap onto separate lines. Password unlocks protected PDFs. """ try: import pdfplumber except ImportError: raise RuntimeError("pdfplumber not installed. Run: pip install pdfplumber") # detect encryption up-front so the app can ask the user for the PDF password try: from pypdf import PdfReader reader = PdfReader(io.BytesIO(content)) if reader.is_encrypted: if not password: raise RuntimeError("PASSWORD_REQUIRED") if reader.decrypt(password) == 0: raise RuntimeError("PASSWORD_INCORRECT") except RuntimeError: raise except Exception: pass # pypdf unavailable or non-fatal; pdfplumber will try below lines: List[str] = [] with pdfplumber.open(io.BytesIO(content), password=password or "") as pdf: for page in pdf.pages: text = page.extract_text() or "" lines.extend(text.split("\n")) out = _parse_pdf_lines(lines, gated=True) if not out: # unknown header layout -> retry ungated (best-effort, merchant required) out = _parse_pdf_lines(lines, gated=False) return out # "Credit Limit Rs. 11,30,000" appears in the statement header. We use it to estimate # income when the user hasn't given one. Excludes "available" / "cash" limits, which are # lower/different. `credit\s*limit` tolerates the glued "TotalCreditLimit" some PDFs render. # Require a currency marker (Rs/INR/₹) right before the figure so we don't grab a stray # year ("Credit Limit ... Jan 2026") when the label and value sit on different lines. _CREDIT_LIMIT_RE = re.compile( r"credit\s*limit\b[^0-9₹]{0,18}(?:rs\.?|inr|₹)\s*([1-9][\d,]{3,}(?:\.\d{1,2})?)", re.I) # A word that is a rupee amount, possibly with an 'r'/Rs/₹ prefix (many PDFs render ₹ as 'r'). _LIMIT_AMT_WORD = re.compile(r"^(?:r|rs\.?|inr|₹)?\s*(\d[\d,]{4,}(?:\.\d{1,2})?)$", re.I) def _plausible_limit(v: float) -> bool: return 25000 <= v <= 1e8 # real card limits; floor rejects years/small numbers def _inline_credit_limit(text: str) -> Optional[float]: """Inline 'Credit Limit Rs 11,30,000' on a single line (Kotak, SBI, many issuers).""" best = None for m in _CREDIT_LIMIT_RE.finditer(text or ""): pre = text[max(0, m.start() - 18):m.start()].lower() if "available" in pre or "cash" in pre: continue try: v = float(m.group(1).replace(",", "")) except ValueError: continue if _plausible_limit(v) and (best is None or v > best): best = v return best def _credit_limit_from_words(words: List[dict]) -> Optional[float]: """Given positioned words ({text,x0,x1,top}) from a page or OCR, find the total credit limit near a (non-available) 'Credit Limit' label: prefer a rupee amount to its right or directly beneath it; failing that, take the LARGEST plausible amount in the summary band around the label - a credit limit is always the biggest figure in its box (dues can never exceed it), which reliably picks it out of statements like HDFC where labels and values don't align in a column.""" best = None for i in range(len(words) - 1): if words[i]["text"].lower() != "credit" or words[i + 1]["text"].lower() != "limit": continue if abs(words[i + 1]["top"] - words[i]["top"]) > 4: continue pre = words[i - 1]["text"].lower() if i > 0 else "" if pre in ("available", "cash") or "avl" in pre: continue lx0, lx1, ltop = words[i]["x0"], words[i + 1]["x1"], words[i]["top"] aligned, band = [], [] for a in words: m = _LIMIT_AMT_WORD.match(a["text"].replace(" ", "")) if not m: continue try: v = float(m.group(1).replace(",", "")) except ValueError: continue if not _plausible_limit(v): continue if (abs(a["top"] - ltop) <= 4 and lx1 < a["x0"] < lx1 + 130) or \ (ltop < a["top"] <= ltop + 70 and a["x0"] < lx1 + 30 and a["x1"] > lx0 - 30): aligned.append(v) if ltop - 55 <= a["top"] <= ltop + 80: band.append(v) cand = max(aligned) if aligned else (max(band) if band else None) if cand is not None and (best is None or cand > best): best = cand return best def _positional_credit_limit(pdf) -> Optional[float]: best = None try: pages = pdf.pages[:3] except Exception: return None for page in pages: try: v = _credit_limit_from_words(page.extract_words()) except Exception: continue if v and (best is None or v > best): best = v return best def _ocr_credit_limit(content: bytes, password: Optional[str] = None) -> Optional[float]: """Last-resort OCR for statements whose summary is a rendered image / has no usable text layer (HDFC). Rasterises the first two pages and reuses the positional logic on the OCR'd words. All heavy deps are lazily imported so a Space without them just skips this.""" try: import io as _io import pdf2image import pytesseract from pypdf import PdfReader, PdfWriter except Exception: return None try: reader = PdfReader(_io.BytesIO(content)) if reader.is_encrypted: reader.decrypt(password or "") writer = PdfWriter() for p in reader.pages[:2]: writer.add_page(p) buf = _io.BytesIO(); writer.write(buf) images = pdf2image.convert_from_bytes(buf.getvalue(), dpi=200) except Exception: return None best = None scale = 72.0 / 200.0 # normalise OCR pixels to PDF points so the same tolerances apply for img in images[:2]: try: d = pytesseract.image_to_data(img, output_type=pytesseract.Output.DICT) except Exception: continue words = [] for j in range(len(d["text"])): t = (d["text"][j] or "").strip() if t: words.append({"text": t, "x0": d["left"][j] * scale, "x1": (d["left"][j] + d["width"][j]) * scale, "top": d["top"][j] * scale}) text = " ".join(w["text"] for w in words) v = _inline_credit_limit(text) or _credit_limit_from_words(words) if v and (best is None or v > best): best = v return best def detect_credit_limit(filename: str, content: bytes, password: Optional[str] = None) -> Optional[float]: """The card's total credit limit from the statement, ignoring available/cash limits. Tiered: inline text, then column-aligned positional, then OCR (only if the text layer yields nothing). Returns None (never a wrong guess) when nothing plausible is found.""" name = (filename or "").lower() if name.endswith(".pdf"): try: import pdfplumber with pdfplumber.open(io.BytesIO(content), password=password or "") as pdf: text = "\n".join((p.extract_text() or "") for p in pdf.pages[:3]) cands = [x for x in (_inline_credit_limit(text), _positional_credit_limit(pdf)) if x] if cands: return max(cands) except Exception: pass return _ocr_credit_limit(content, password) # image-only statements (HDFC) try: return _inline_credit_limit(content.decode("utf-8-sig", errors="ignore")) except Exception: return None def parse_statement(filename: str, content: bytes, password: Optional[str] = None) -> List[Dict]: name = filename.lower() if name.endswith(".csv"): return parse_csv(content) if name.endswith(".pdf"): return parse_pdf(content, password=password) # try CSV as a fallback try: return parse_csv(content) except Exception: raise RuntimeError("Unsupported file type. Upload a .pdf or .csv statement.")