Spaces:
Paused
Paused
| """ | |
| 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. | |
| "(1,200.00)" is the accounting convention for a CREDIT. The regex matched | |
| happily inside the brackets and returned +1200, so a refund imported as a | |
| purchase. _normalise_amount_signs exists for exactly this and was only ever | |
| wired into the PDF path, never the CSV one. | |
| """ | |
| cleaned = re.sub(r"(?:rs\.?|inr|₹)", "", s or "", flags=re.IGNORECASE).replace("/-", "").strip() | |
| negated = bool(re.fullmatch(r"\(\s*[^)]*\)", cleaned)) | |
| m = _AMOUNT_LENIENT_RE.search(cleaned) | |
| if not m: | |
| return None | |
| try: | |
| v = float(m.group(1).replace(",", "")) | |
| except ValueError: | |
| return None | |
| return -abs(v) if negated else v | |
| _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|" | |
| # Axis and SBI print the monthly payment row as "PAYMENT - THANK YOU" with | |
| # no 'received' - it imported as an Rs.8,000 PURCHASE in the layout test | |
| # (2026-08-02) because none of these wordings matched it. | |
| r"payment\s*[-\u2013]*\s*thank\s*you|thank\s*you\s*for\s*your\s*payment|" | |
| r"autopay|auto pay|bbps|nach|e-?mandate|refund|reversal|cashback received)", | |
| re.IGNORECASE, | |
| ) | |
| # Merchant refunds/reversals: credits that CANCEL an earlier purchase (and its | |
| # rewards), as opposed to bill payments. Collected as reversal events so the | |
| # app can net them against the matching spend. | |
| _REFUND_RE = re.compile(r"\b(refund(?:ed)?|revers(?:al|ed)|charge\s*back)\b", re.IGNORECASE) | |
| def _credit_kind(direction: str, desc: str) -> str: | |
| """Classify a row: 'spend' (count it), 'payment' (ignore), 'refund' (net out).""" | |
| d = (direction or "").strip().lower() | |
| text = desc or "" | |
| if _CHARGES_RE.search(text): | |
| return "payment" | |
| if _REFUND_RE.search(text): | |
| return "refund" | |
| if _PAYMENT_RE.search(text): | |
| return "payment" | |
| if d in ("cr", "credit") or d.startswith("cr"): | |
| # a bare credit with merchant-looking text is a refund; else a payment | |
| return "refund" if re.search(r"[A-Za-z]{3,}", text) else "payment" | |
| return "spend" | |
| # 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, reversals: Optional[List[Dict]] = None) -> 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) | |
| # Signed single-amount statements: a NEGATIVE row is a CREDIT. | |
| # _normalize_row does abs(), so without this branch a -899.00 refund | |
| # imported as a +899.00 purchase - a 2x error on that row plus | |
| # phantom rewards. The TS parser has always had this; the Python | |
| # CSV path never did. | |
| if amount < 0: | |
| if _credit_kind("cr", desc) == "refund" and reversals is not None: | |
| reversals.append({"date": iso, "description": desc.strip(), "amount": round(abs(amount), 2)}) | |
| continue | |
| direction = r[ki] if 0 <= ki < len(r) else "" | |
| kind = _credit_kind(direction, desc) | |
| if kind == "refund": | |
| if reversals is not None: | |
| reversals.append({"date": iso, "description": desc.strip(), "amount": round(abs(amount), 2)}) | |
| continue | |
| if kind == "payment": | |
| 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) | |
| kind = _credit_kind("", desc) | |
| if kind == "refund": | |
| if reversals is not None: | |
| reversals.append({"date": _parse_date(date_field), "description": desc.strip(), "amount": round(abs(amount), 2)}) | |
| continue | |
| if kind == "payment": | |
| 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 | |
| # The alphabetic component must be a REAL month. It used to accept any 3-9 letter | |
| # word, so "January 03 SWIGGY 642.00" matched "03 SWIGGY 642" as a date, failed to | |
| # parse as one, and the whole row was discarded as a wrapped merchant name. Any | |
| # merchant of 3-9 letters followed by a 2-4 digit number hit this. | |
| _MONTH_WORD = (r"(?:jan(?:uary)?|feb(?:ruary)?|mar(?:ch)?|apr(?:il)?|may|jun(?:e)?|jul(?:y)?|" | |
| r"aug(?:ust)?|sep(?:t|tember)?|oct(?:ober)?|nov(?:ember)?|dec(?:ember)?)") | |
| _PDF_DATE_RE = re.compile(r"(\d{1,2}[/\-. ](?:\d{1,2}|" + _MONTH_WORD + r")[/\-. ]'?\d{2,4})", re.I) | |
| _SIGNED_AMT_RE = re.compile(r"(-?\d[\d,]*\.\d{2})") | |
| # A row printed WITHOUT paise ("DECATHLON 4,590") matched no amount at all and was | |
| # silently discarded as a wrapped merchant name - real spend vanishing with no | |
| # error and no count. Requires a grouping comma or a decimal so a bare 4-digit | |
| # year can never be read as money. Used only on the tail of a dated line. | |
| _LENIENT_AMT_RE = re.compile(r"(-?(?:\d{1,3}(?:,\d{2,3})+|\d+\.\d{1,2})(?:\.\d{1,2})?)") | |
| # "(9,999.00)" is the accounting convention for a CREDIT. Untouched it parsed as | |
| # +9,999 and imported a refund as a purchase - a 2x error on that line. | |
| _PAREN_AMT_RE = re.compile(r"\((\d[\d,]*(?:\.\d{1,2})?)\)") | |
| def _normalise_amount_signs(line: str) -> str: | |
| """(9,999.00) -> -9,999.00 so the existing negative handling takes over.""" | |
| return _PAREN_AMT_RE.sub(lambda m: "-" + m.group(1), line) | |
| # 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") | |
| # Start of the transaction TABLE. The four literals above only cover issuers that | |
| # print "Transaction Date"; banks whose table header is a plain "Date Description | |
| # Amount" (SBI Card, several PSU layouts) matched NOTHING, so gated parsing found | |
| # no rows and fell through to the ungated fallback, which parses from the top of | |
| # the page and turned the header's own "Statement Date ... Total Outstanding Rs. | |
| # 4,569.50" summary line into a phantom transaction. Match the generic column | |
| # header too, so those layouts gate properly. | |
| _PDF_START_RE = re.compile( | |
| r"\b(?:transaction\s+details|your\s+transactions|transaction\s+date|date\s+transaction)\b" | |
| r"|(?:domestic|international)\s*/?\s*(?:and\s+international\s+)?transactions?\b" | |
| r"|\bdate\b.{0,60}?\b(?:description|details|particulars|narration)\b.{0,60}?\b(?:amount|amt|debit)\b", | |
| re.I, | |
| ) | |
| # Header/summary lines that carry BOTH a date and an amount ("Statement Date: 12 | |
| # Jul 2026 Total Outstanding: Rs. 4,569.50"). _PDF_STOP is only consulted for | |
| # wrapped-name lines (no amount), so these slipped straight through and were | |
| # emitted as transactions. Reject them explicitly, in gated and ungated mode. | |
| # NOTE \s* (not \s+) everywhere: several issuers (Kotak) extract with the spaces | |
| # stripped out - "MinimumAmountDue(MAD) Rs.2,159.25", "TotalAmountDue(TAD)" - so a | |
| # space-requiring pattern misses them and the summary row leaks in as a purchase. | |
| _SUMMARY_RE = re.compile( | |
| r"(?:statement\s*(?:date|period|summary)" | |
| r"|total\s*(?:dues?|outstanding|amount\s*due)" | |
| r"|(?:minimum|min\.?)\s*amount\s*due" | |
| r"|payment\s*due\s*date" | |
| r"|credit\s*limit" | |
| r"|available\s*(?:credit|limit|cash)" | |
| r"|cash\s*limit" | |
| r"|(?:opening|closing|previous)\s*balance" | |
| r"|reward\s*points?\s*(?:summary|balance)" | |
| r"|card\s*(?:number|no\.?)\s*[:\-]" | |
| r"|gst\s*summary)", | |
| re.I, | |
| ) | |
| # 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, period=None) -> 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 | |
| # With a known cycle, pick the year that puts this month/day INSIDE it. A | |
| # December row in a 15-Dec-2025 -> 14-Jan-2026 cycle is 2025, not 2026, and | |
| # only the period says so. | |
| if period: | |
| start, end = period | |
| for cand in (end.year, end.year - 1, start.year): | |
| try: | |
| d = datetime(cand, mo, day).date() | |
| except ValueError: | |
| continue | |
| if start <= d <= end: | |
| return d.isoformat() | |
| 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 | |
| _PERIOD_RE = re.compile( | |
| r"statement\s*(?:period|date|cycle)?\s*[:\-]?\s*" | |
| r"([A-Za-z]{3,9}\s+\d{1,2},?\s+20\d{2}|\d{1,2}[/\-][A-Za-z0-9]{2,9}[/\-]20\d{2})" | |
| r"\s*(?:to|-|–|through)\s*" | |
| r"([A-Za-z]{3,9}\s+\d{1,2},?\s+20\d{2}|\d{1,2}[/\-][A-Za-z0-9]{2,9}[/\-]20\d{2})", re.I) | |
| def _statement_period(blob: str): | |
| """(start, end) dates of the statement cycle, or None. The cycle is what tells | |
| a month-day row which YEAR it belongs to - a December row in a Dec->Jan cycle | |
| is last year's, and nothing else on the page says so.""" | |
| m = _PERIOD_RE.search(blob or "") | |
| if not m: | |
| return None | |
| def _one(t: str): | |
| t = t.replace(",", " ").strip() | |
| for fmt in ("%B %d %Y", "%b %d %Y", "%d/%m/%Y", "%d-%m-%Y", "%d/%b/%Y", "%d-%b-%Y"): | |
| try: | |
| return datetime.strptime(re.sub(r"\s+", " ", t), fmt).date() | |
| except ValueError: | |
| continue | |
| return None | |
| a, b = _one(m.group(1)), _one(m.group(2)) | |
| return (a, b) if a and b and a <= b else None | |
| def _has_balance_column(lines: List[str]) -> bool: | |
| """Does this statement print a RUNNING BALANCE after the amount? | |
| Decided across the whole file rather than row by row, by the only property | |
| that actually identifies a balance column: successive balances differ by | |
| that row's amount. Kotak and several PSU layouts do this; most issuers do | |
| not, and mis-classifying either way corrupts every figure in the import, so | |
| the bar is deliberately high - at least two CONSECUTIVE confirmed pairs and | |
| at least 70% of all pairs consistent. Two independent arithmetic | |
| coincidences are needed for a false positive; a forex sub-amount column | |
| ("USD 25.00 1,200.00") fails immediately because the left figure bears no | |
| relation to the difference between the right ones. | |
| Direction is not assumed: a card statement's balance rises with spend, a | |
| bank account's falls, so |balance_n - balance_(n-1)| == amount_n is the | |
| test. | |
| """ | |
| rows: List[tuple] = [] | |
| for raw in lines: | |
| line = (raw or "").strip() | |
| if not line or _SUMMARY_RE.search(line.lower()): | |
| continue | |
| line = _normalise_amount_signs(line) | |
| dm = _PDF_DATE_RE.search(line) | |
| md = None if dm else _MONTH_DAY_RE.search(line) | |
| if not (dm or md): | |
| continue | |
| amts = list(_SIGNED_AMT_RE.finditer(line)) | |
| if len(amts) < 2: | |
| rows.append(()) # a 1-figure row breaks the balance chain | |
| continue | |
| try: | |
| last = float(amts[-1].group(1).replace(",", "")) | |
| prev = float(amts[-2].group(1).replace(",", "")) | |
| except ValueError: | |
| rows.append(()) | |
| continue | |
| rows.append((prev, last)) | |
| ok = bad = 0 | |
| for i in range(1, len(rows)): | |
| a, b = rows[i - 1], rows[i] | |
| if len(a) != 2 or len(b) != 2: | |
| continue | |
| amount, balance = b | |
| if abs(abs(balance - a[1]) - abs(amount)) < 0.51: | |
| ok += 1 | |
| else: | |
| bad += 1 | |
| return ok >= 2 and ok >= (ok + bad) * 0.7 | |
| def _parse_pdf_lines(lines: List[str], gated: bool = True, reversals: Optional[List[Dict]] = None) -> List[Dict]: | |
| out: List[Dict] = [] | |
| started = not gated # ungated fallback: parse from the top | |
| pending: List[str] = [] # buffered wrapped merchant-name lines (IDFC style) | |
| # Decided ONCE for the whole statement by _has_balance_column below, not | |
| # inferred row by row. The old streaming test asked whether | |
| # (_running + prev) == last, but _running was only ever assigned INSIDE that | |
| # same branch and started at 0.0 - so it could only bootstrap on a statement | |
| # whose opening balance was zero. Credit-card statements always carry a | |
| # previous balance, so the branch never fired and EVERY row on a | |
| # balance-column layout imported the running balance as the spend, with the | |
| # error compounding down the page. | |
| _balance_col = _has_balance_column(lines) | |
| # Year for month-first dates that omit it (Amex "June 03"). | |
| # | |
| # This used to take max() of every 4-digit year printed anywhere, so a | |
| # "Card valid thru 12/2030" or a copyright line hijacked every row, and a | |
| # Dec->Jan cycle stamped December with the JANUARY year - twelve months out. | |
| # Rows dated in the future fall outside every monthly bucket and can never be | |
| # matched by a refund (applyReversals only looks backwards), so the error is | |
| # silent and permanent. Read the statement PERIOD instead and let each row | |
| # take the year that places it inside it. | |
| _blob = "\n".join(lines) | |
| _period = _statement_period(_blob) | |
| _yrs = re.findall(r"\b(20\d{2})\b", _blob) | |
| default_year = (_period[1].year if _period else (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 _PDF_START_RE.search(low): | |
| started = True | |
| continue | |
| # A summary/header row is never a transaction, even though it carries a | |
| # date and an amount. This is what produced hallucinated purchases like | |
| # "Statement Date / Total Outstanding Rs.4,569.50". | |
| if _SUMMARY_RE.search(low): | |
| pending = [] | |
| continue | |
| line = _normalise_amount_signs(line) # "(9,999.00)" is a CREDIT, not a spend | |
| 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 not amts: | |
| # No paise on this row ("DECATHLON BENGALURU 4,590"). Look only AFTER | |
| # the date so the date itself can never be read as money, and require | |
| # a grouping comma or a decimal so a bare year cannot match. | |
| tail_at = (dm.end() if dm else md.end()) | |
| amts = [m for m in _LENIENT_AMT_RE.finditer(line, tail_at)] | |
| if date_txt and amts: | |
| iso = _parse_date(date_txt) if dm else _parse_month_day(date_txt, default_year, _period) | |
| 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 | |
| # On a confirmed balance layout the amount is the SECOND-to-last | |
| # figure; the last one is the running total. Decided for the file, | |
| # so the first data row is classified correctly too. | |
| amt_m = amts[-2] if (_balance_col and len(amts) >= 2) else amts[-1] | |
| amount = float(amt_m.group(1).replace(",", "")) | |
| tail = line[amt_m.end():].strip().upper() | |
| # Axis-style layouts carry the direction as its own COLUMN before | |
| # the amount ("... Debit 5,822.00" / "PAYMENT - THANK YOU Credit | |
| # 8,000.00"). The word immediately left of the amount decides; it | |
| # is also stripped from the description below so 'Debit' never | |
| # reads as part of a merchant name. | |
| _pre_words = line[:amt_m.start()].strip().split() | |
| _dir_word = (_pre_words[-1].strip('.').lower() if _pre_words else "") | |
| # A bare trailing "C"/"D" is HDFC's direction column. The code | |
| # already knew it existed - it scrubs it out of the merchant name a | |
| # few lines below - but is_credit only looked for a sign, a "CR" | |
| # prefix or a Debit/Credit WORD, so a single letter was invisible | |
| # and a merchant credit whose wording lacked "refund" imported as | |
| # spend. The Axis long-word form was handled; this one was not. | |
| _tail_letter = tail.split()[0] if tail.split() else "" | |
| _col_credit = _dir_word in ("credit", "cr") or _tail_letter in ("C", "CR") | |
| _col_debit = _dir_word in ("debit", "dr") or _tail_letter in ("D", "DR") | |
| is_credit = amount < 0 or tail.startswith("CR") or _col_credit or (not _col_debit and _is_credit("", line)) | |
| amount = abs(amount) | |
| if amount == 0: # FX-only / zero rows | |
| pending = [] | |
| continue | |
| desc = line[:amt_m.start()].replace(date_txt, " ") | |
| if _col_credit or _col_debit: | |
| desc = re.sub(r"\b(?:debit|credit|dr|cr)\.?\s*$", " ", desc, flags=re.I) | |
| 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"(?<![A-Za-z])[rR](?=\d)", " ", desc) # ₹ rendered as 'r' | |
| desc = re.sub(r"\s+[CD]\s*$", " ", desc) # HDFC trailing debit/credit column letter | |
| desc = _SPENDS_AREA_RE.sub(" ", desc) # Kotak "... Services"/"... Automotive" MCC column | |
| desc = re.sub(r"\s+in\s*$", " ", desc, flags=re.I) # trailing India country code | |
| desc = re.sub(r"\b(\w+)(\s+\1\b)+", r"\1", desc, flags=re.I) # collapse repeated words ("Gmatclub Gmatclub") | |
| desc = re.sub(r"\s{2,}", " ", desc).strip(" ,|+") | |
| # A real merchant name is one that survives stripping the issuer's EMI-convert | |
| # marker and any "USD 324.50" forex fragment. IDFC prints an international | |
| # purchase as "<merchant wrapped above>\n<date> Convert USD 324.50 <INR>", 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: | |
| # CR / negative rows: a merchant refund becomes a reversal event | |
| # (nets out its spend and rewards); payments stay ignored. | |
| if reversals is not None and _credit_kind("cr", desc) == "refund": | |
| reversals.append({"date": iso, "description": desc.strip(" ,|"), "amount": round(abs(amount), 2)}) | |
| 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". | |
| kind2 = _credit_kind("", desc) | |
| if kind2 == "refund": | |
| if reversals is not None: | |
| reversals.append({"date": iso, "description": desc.strip(" ,|"), "amount": round(abs(amount), 2)}) | |
| continue | |
| if kind2 == "payment" 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, reversals: Optional[List[Dict]] = 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 payments/charges, reports merchant refunds into `reversals` (so the | |
| app can net them against spends), 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")) | |
| rev1: List[Dict] = [] | |
| out = _parse_pdf_lines(lines, gated=True, reversals=rev1) | |
| if not out: # unknown header layout -> retry ungated (best-effort, merchant required) | |
| rev1 = [] | |
| out = _parse_pdf_lines(lines, gated=False, reversals=rev1) | |
| if reversals is not None: | |
| reversals.extend(rev1) | |
| 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 | |
| # --------------------------------------------------------------------------- | |
| # REWARD-POINT BALANCE | |
| # | |
| # The app can already tell you your points are worth more on a transfer partner | |
| # than as statement credit - but only if it knows the BALANCE, and the balance | |
| # was a number the user had to type in by hand. So the panel sat on a nudge and | |
| # most people never saw an answer. | |
| # | |
| # Market check (Aug 2026): on TechnoFino the recurring complaint is that no | |
| # Indian app automates this - people track 17 cards in Google Sheets. The one | |
| # app users report as working is CRED, which reads the emailed statement. That | |
| # is the approach here, and it needs no credentials and no account linking. | |
| # | |
| # EVERY STATEMENT PRINTS IT, because the issuer has to. The wording differs, so | |
| # match the CONTEXT (a reward-points heading or label) and then the closing | |
| # figure, rather than any bare number near the word "points". | |
| # --------------------------------------------------------------------------- | |
| # The label that names the balance we want. Deliberately narrow: "points | |
| # earned", "points redeemed" and "points expiring" are NOT the balance, and | |
| # grabbing one of those would show the user a confidently wrong number. | |
| _PTS_CLOSING_RE = re.compile( | |
| r"(?:closing|total|net|available|balance\s+of)?\s*" | |
| r"(?:reward|edge|membership\s+reward|neu|payback|club\s*vistara|bonus)?\s*" | |
| r"points?\s*(?:balance|as\s+on|closing|total)?\s*[:\-]?\s*" | |
| r"([0-9][0-9,]{0,9})\b", | |
| re.I, | |
| ) | |
| # A reward-points SECTION heading; a "Closing Balance" beneath one is the | |
| # balance even when the word "points" is not repeated on that line. | |
| _PTS_SECTION_RE = re.compile(r"(reward|edge\s+reward|membership\s+reward|neu\s*coin|payback)\s*points?\s*(summary|details|statement)?", re.I) | |
| _CLOSING_RE = re.compile(r"closing\s*(?:balance|points?)?\s*[:\-]?\s*([0-9][0-9,]{0,9})\b", re.I) | |
| # Words that mean this figure is NOT the balance. | |
| _PTS_REJECT = re.compile(r"earn|redeem|expir|lapse|adjust|convert|debit|credit\s*card\s*no|opening", re.I) | |
| # A rupee context: points are a count, never a currency amount, so a figure | |
| # introduced by a currency mark or carrying paise is the wrong figure. | |
| _PTS_MONEY = re.compile(r"(?:rs\.?|inr|₹)\s*$", re.I) | |
| # An integer that is NOT part of a decimal amount - points are whole units, so | |
| # "5,000.00" is a rupee figure and must not be read as a balance. | |
| # The WHOLE digit run, never a fragment of one. With a plain \b boundary, | |
| # "99,99,99,999" backtracked to a valid-looking prefix, finditer then matched | |
| # the trailing "999", and an absurd figure was read as a 999-point balance. | |
| # The lookarounds refuse to start or stop inside a number, so an over-long run | |
| # is rejected by _plausible_points as a whole instead of being salvaged. | |
| _PTS_INT_RE = re.compile(r"(?<![0-9,])([0-9][0-9,]{0,13})(?![0-9,])(?!\s*\.\s*[0-9])") | |
| def _plausible_points(v: float) -> bool: | |
| """A balance a real cardholder can hold. Rejects a stray year (2026), a | |
| card fragment, and an absurd figure that is really a rupee total.""" | |
| return 0 <= v <= 5_000_000 and not (1990 <= v <= 2100 and v == int(v) and len(str(int(v))) == 4) | |
| def detect_points_balance(text: str) -> Optional[int]: | |
| """The reward-point CLOSING balance from statement text, or None. | |
| None is the honest answer whenever the statement is ambiguous: a wrong | |
| balance would feed straight into "your points cover this flight", which is | |
| a claim the user would act on. | |
| """ | |
| if not text: | |
| return None | |
| lines = [l.strip() for l in text.splitlines() if l.strip()] | |
| cands: List[int] = [] | |
| for i, line in enumerate(lines): | |
| low = line.lower() | |
| # (a) an explicit labelled balance on one line. Once the line is | |
| # CONFIRMED to name a balance, take the LAST plausible integer on it | |
| # rather than trying to write one regex wide enough for every issuer's | |
| # wording - Amex prints "Points Balance as on 31 Jul 2026: 2,15,300", | |
| # and a pattern tight enough to reject "Points Earned 3,120" can never | |
| # also span that date clause. Balances are printed last on their line. | |
| if re.search(r"points?", low) and re.search(r"balance|closing|as on|total", low): | |
| if _PTS_REJECT.search(low) and not re.search(r"closing|balance", low): | |
| continue | |
| line_vals: List[int] = [] | |
| for m in _PTS_INT_RE.finditer(line): | |
| pre = line[max(0, m.start() - 6):m.start()] | |
| if _PTS_MONEY.search(pre): | |
| continue | |
| try: | |
| v = int(m.group(1).replace(",", "")) | |
| except ValueError: | |
| continue | |
| if _plausible_points(v): | |
| line_vals.append(v) | |
| if line_vals: | |
| cands.append(line_vals[-1]) | |
| # (b) a "Closing Balance" line inside a reward-points section | |
| if _PTS_SECTION_RE.search(low): | |
| for j in range(i, min(i + 6, len(lines))): | |
| cm = _CLOSING_RE.search(lines[j]) | |
| if cm: | |
| try: | |
| v = int(cm.group(1).replace(",", "")) | |
| except ValueError: | |
| continue | |
| if _plausible_points(v): | |
| cands.append(v) | |
| break | |
| if not cands: | |
| return None | |
| # Several figures and no agreement means we cannot tell which is the | |
| # balance - say nothing rather than pick one. | |
| uniq = sorted(set(cands)) | |
| if len(uniq) > 1 and uniq[-1] != max(cands, key=cands.count): | |
| return None | |
| return max(set(cands), key=cands.count) | |
| 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_points_balance_file(filename: str, content: bytes, password: Optional[str] = None) -> Optional[int]: | |
| """The reward-point balance from a statement FILE. Same tiering discipline | |
| as detect_credit_limit: read the text layer, and return None rather than a | |
| guess. Deliberately does NOT fall through to OCR - a wrong balance would | |
| feed "your points cover this flight", and no balance at all is the safer | |
| failure.""" | |
| 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[:4]) | |
| return detect_points_balance(text) | |
| except Exception: | |
| return None | |
| try: | |
| return detect_points_balance(content.decode("utf-8-sig", errors="ignore")) | |
| except Exception: | |
| return None | |
| 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) | |
| # OCR IS FOR STATEMENTS WITH NO TEXT LAYER, NOT FOR STATEMENTS THAT | |
| # SIMPLY DO NOT PRINT A LIMIT. Falling through to it whenever | |
| # detection found nothing rasterised every ordinary statement at | |
| # 200dpi and ran tesseract over it: measured 28.9s versus 0.56s to | |
| # parse the same file, on every upload, behind a 3-permit semaphore | |
| # with a 20s wait - so three uploads made everyone else 429, and the | |
| # client retried, doubling the load. Most statements just have no | |
| # limit line, and for those the honest answer is None, immediately. | |
| if text.strip(): | |
| return None | |
| 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 find_pdf_password(content: bytes, candidates: List[str]) -> Optional[str]: | |
| """Probe candidate passwords against an encrypted PDF (CRED-style smart | |
| unlock: bank statement passwords are formulaic - name4+DDMM etc - so the | |
| app derives candidates from the user's details and we try them here). | |
| Returns the working password, '' if the PDF isn't encrypted, or None.""" | |
| try: | |
| from pypdf import PdfReader | |
| reader = PdfReader(io.BytesIO(content)) | |
| if not reader.is_encrypted: | |
| return "" | |
| for pw in candidates: | |
| if not pw: | |
| continue | |
| try: | |
| r = PdfReader(io.BytesIO(content)) | |
| if r.decrypt(pw) != 0: | |
| return pw | |
| except Exception: | |
| continue | |
| except Exception: | |
| pass | |
| return None | |
| def parse_statement(filename: str, content: bytes, password: Optional[str] = None, | |
| reversals: Optional[List[Dict]] = None) -> List[Dict]: | |
| name = filename.lower() | |
| if name.endswith(".csv"): | |
| return parse_csv(content, reversals=reversals) | |
| if name.endswith(".pdf"): | |
| return parse_pdf(content, password=password, reversals=reversals) | |
| # try CSV as a fallback | |
| try: | |
| return parse_csv(content, reversals=reversals) | |
| except Exception: | |
| raise RuntimeError("Unsupported file type. Upload a .pdf or .csv statement.") | |
| # Exact catalogue issuer -> text/domain aliases. Mirrored in app statements.ts. | |
| _ISSUER_TOKENS = [ | |
| ("HDFC Bank", ["hdfc", "hdfcbank"]), | |
| ("SBI Card", ["sbi card", "sbi cards", "sbicard", "state bank"]), | |
| ("ICICI Bank", ["icici", "icicibank"]), | |
| ("Axis Bank", ["axis bank", "axisbank", "kiwi"]), | |
| ("American Express", ["american express", "amex", "americanexpress"]), | |
| ("IDFC FIRST Bank", ["idfc", "idfcfirstbank"]), | |
| ("AU Small Finance Bank", ["au small finance", "au bank", "aubank"]), | |
| ("IndusInd Bank", ["indusind"]), | |
| ("IDBI Bank", ["idbi"]), | |
| ("Bandhan Bank", ["bandhan bank", "bandhanbank"]), | |
| ("RBL Bank", ["rbl bank", "rblbank"]), | |
| ("BoB/Federal/SBM", ["onecard", "one card", "getonecard"]), | |
| ("HSBC", ["hsbc"]), | |
| ("Bank of Baroda", ["bobcard", "bank of baroda", "bankofbaroda"]), | |
| ("DBS Bank India", ["dbs bank", "dbs india", "dbsbank"]), | |
| ("Federal Bank", ["federal bank", "federalbank", "scapia"]), | |
| ("Kotak Mahindra Bank", ["kotak", "kotakbank"]), | |
| ("Punjab National Bank", ["punjab national bank", "pnb"]), | |
| ("Yes Bank", ["yes bank", "yesbank"]), | |
| ("Standard Chartered", ["standard chartered", "standardchartered", "sc bank"]), | |
| ] | |
| # Issuer PAN (the 10 characters inside a GSTIN) -> issuer label. | |
| # | |
| # Why this exists: some issuers print their NAME only as a logo IMAGE and put the | |
| # bank name in the terms pages, so the statement header carries no text bank name | |
| # at all - and their text extracts glued ("SALIMRAMZANALISHAIKH"), which defeats | |
| # word-boundary matching anyway. But every Indian statement prints the issuer's | |
| # GSTIN in the header, and the PAN inside it identifies the bank exactly. | |
| # Verified against real statements (Kotak = 27AAACK4409J..., HDFC = 33AAACH2702H...). | |
| _GSTIN_ISSUER = [ | |
| ("aaack4409j", "Kotak Mahindra Bank"), | |
| ("aaach2702h", "HDFC Bank"), | |
| ] | |
| def detect_issuers_from_text(text: str) -> List[str]: | |
| """Issuer labels named in the statement header (fallback when no exact card | |
| signature matches). Reads the same header region, so transaction-row bank | |
| names can't leak in. Falls back to the issuer's GSTIN/PAN, which is present | |
| even when the bank name is only a logo image.""" | |
| from merchants import keyword_hit | |
| t = (text or "").lower() | |
| out: List[str] = [] | |
| for label, toks in _ISSUER_TOKENS: | |
| if any(keyword_hit(t, tok) for tok in toks) and label not in out: | |
| out.append(label) | |
| for pan, label in _GSTIN_ISSUER: | |
| if pan in t and label not in out: | |
| out.append(label) | |
| return out | |
| # --------------------------------------------------------------------------- | |
| # Card detection: catalogue-derived, issuer-required and exclusive. The old | |
| # 32-row ANY-token table made "HDFC Swiggy" match BLCK, ORNGE and legacy at once. | |
| # This mirror covers the whole credit-card catalogue and retains equal top-score | |
| # matches as ambiguous rather than guessing. | |
| # --------------------------------------------------------------------------- | |
| def _normalise_match_text(value: str) -> str: | |
| s = (value or "").lower().replace("&", " and ").replace("+", " plus ") | |
| return re.sub(r"\s+", " ", re.sub(r"[^a-z0-9]+", " ", s)).strip() | |
| def _product_alias_hit(text: str, phrase: str) -> bool: | |
| hay = f" {_normalise_match_text(text)} " | |
| words = [w for w in _normalise_match_text(phrase).split(" ") if w] | |
| return bool(words) and all(f" {word} " in hay for word in words) | |
| _GENERIC_CARD_WORDS = {"bank", "credit", "card"} | |
| _BRAND_ISSUER_ALIASES = {"kiwi", "scapia", "onecard", "one card", "getonecard"} | |
| _CARD_PRODUCT_ALIAS_CACHE = {} | |
| def _card_product_aliases(card) -> List[str]: | |
| cached = _CARD_PRODUCT_ALIAS_CACHE.get(card.id) | |
| if cached is not None: | |
| return cached | |
| name = _normalise_match_text(card.name) | |
| issuer_aliases = next((aliases for issuer, aliases in _ISSUER_TOKENS if issuer == card.issuer), []) | |
| removable = [card.issuer, *issuer_aliases] | |
| removable = [x for x in removable if _normalise_match_text(x) not in _BRAND_ISSUER_ALIASES] | |
| for alias in sorted(removable, key=len, reverse=True): | |
| a = _normalise_match_text(alias) | |
| if a: | |
| name = f" {name} ".replace(f" {a} ", " ").strip() | |
| name_product = " ".join(w for w in name.split() if w not in _GENERIC_CARD_WORDS) | |
| issuer_id_words = { | |
| word | |
| for value in [card.issuer, *issuer_aliases] | |
| for word in _normalise_match_text(value).split() | |
| if len(word) > 1 | |
| } | |
| id_product = " ".join( | |
| word for word in _normalise_match_text(card.id.replace("_", " ")).split() | |
| if word not in issuer_id_words and word != "first" | |
| ) | |
| aliases = [name_product, id_product] | |
| if card.id == "tataneu_infinity": | |
| aliases += ["tata neu infinity", "neu infinity"] | |
| if card.id == "hdfc_tata_neu_plus": | |
| aliases += ["tata neu plus", "neu plus"] | |
| if card.id == "onecard_metal": | |
| aliases += ["onecard", "one card"] | |
| if card.id == "amex_mrcc": | |
| aliases += ["membership rewards", "mrcc"] | |
| result = list(dict.fromkeys(filter(None, (_normalise_match_text(x) for x in aliases)))) | |
| _CARD_PRODUCT_ALIAS_CACHE[card.id] = result | |
| return result | |
| def detect_cards_from_text(text: str) -> List[str]: | |
| from card_catalogue import CATALOGUE | |
| issuers = set(detect_issuers_from_text(text)) | |
| if not issuers: | |
| return [] | |
| matches = [] | |
| for card in CATALOGUE: | |
| if getattr(card, "kind", "credit") == "debit" or card.issuer not in issuers: | |
| continue | |
| best = 0 | |
| for alias in _card_product_aliases(card): | |
| if not _product_alias_hit(text, alias): | |
| continue | |
| words = [w for w in _normalise_match_text(alias).split() if w] | |
| best = max(best, len(words) * 100 + len(_normalise_match_text(alias))) | |
| if best: | |
| matches.append((card.id, best)) | |
| if not matches: | |
| return [] | |
| top = max(score for _, score in matches) | |
| return [card_id for card_id, score in matches if score == top] | |
| # Start of the transaction TABLE (not any stray "transaction" word): a section | |
| # marker or the table's own column header. Matching the table start - rather | |
| # than the first mention of "transaction" - keeps the card name (printed above | |
| # the table) inside the header region even when the page has a "Transaction | |
| # Summary" box up top. | |
| _TXN_SECTION_RE = re.compile( | |
| r"(?:domestic|international)\s*/?\s*(?:and\s*international\s*)?transactions?\b" # "Domestic / International Transactions" | |
| r"|transaction\s+date\b|date\s+of\s+transaction\b|date\s+transaction\b" # transaction-table column header | |
| r"|transaction\s+details\b" # "Transaction Details" section | |
| r"|date\s*[,|~\t].{0,80}?(?:amount|amt|debit|withdrawal)" # delimited Date,...,Amount header (CSV) | |
| r"|\bdate\b.{0,60}?\b(?:description|details|particulars|narration)\b.{0,60}?\b(?:amount|amt)\b", # plain "Date Description Amount" (SBI-style) | |
| re.IGNORECASE, | |
| ) | |
| # Safety cap for the rare statement with no recognisable table marker: the card | |
| # name sits in the first-page header, so stop before the transaction list can | |
| # realistically begin. Deliberately tight to avoid transaction rows leaking in. | |
| _HEADER_MAX_CHARS = 1800 | |
| def _header_region(text: str) -> str: | |
| """Statement text BEFORE the transaction table. Card detection must never | |
| read transaction rows: a Swiggy order inside an HDFC statement is not the | |
| Swiggy HDFC card.""" | |
| m = _TXN_SECTION_RE.search(text or "") | |
| cut = m.start() if m else len(text or "") | |
| return (text or "")[: min(cut, _HEADER_MAX_CHARS)] | |
| def _detection_text(filename: str, content: bytes, password: Optional[str] = None) -> str: | |
| name = (filename or "").lower() | |
| body = "" | |
| if name.endswith(".pdf"): | |
| try: | |
| import pdfplumber | |
| with pdfplumber.open(io.BytesIO(content), password=password or "") as pdf: | |
| body = "\n".join((p.extract_text() or "") for p in pdf.pages[:2]) | |
| except Exception: | |
| pass | |
| else: | |
| try: | |
| body = content[:20000].decode("utf-8-sig", errors="ignore") | |
| except Exception: | |
| pass | |
| return name + "\n" + _header_region(body) | |
| def detect_cards(filename: str, content: bytes, password: Optional[str] = None) -> List[str]: | |
| """Card detection over a statement FILE (filename + pre-transaction header | |
| region of the first PDF pages, or the CSV preamble).""" | |
| return detect_cards_from_text(_detection_text(filename, content, password)) | |
| def detect_issuers(filename: str, content: bytes, password: Optional[str] = None) -> List[str]: | |
| """Issuer fallback over a statement FILE.""" | |
| return detect_issuers_from_text(_detection_text(filename, content, password)) | |
| def is_credit_card_statement_text(filename: str, text: str) -> bool: | |
| """Positive proof that text is from a credit-card statement, not merely a | |
| statement/attachment sent by a bank. Mirrored in app statements.ts.""" | |
| sample = f"{filename or ''}\n{_header_region(text or '')}" | |
| n = _normalise_match_text(sample) | |
| explicit = bool(re.search( | |
| r"\bcredit card (?:account )?(?:e )?statement\b" | |
| r"|\b(?:e )?statement (?:for )?(?:your )?credit card\b", n)) | |
| if explicit: | |
| return True | |
| issuers = detect_issuers_from_text(sample) | |
| cards = detect_cards_from_text(sample) | |
| has_statement = bool(re.search(r"\bstatement\b", n)) | |
| if issuers and re.search(r"\bcredit card\b", n): | |
| return True | |
| if len(cards) == 1 and has_statement: | |
| return True | |
| signals = [ | |
| r"\btotal amount due\b", r"\bminimum amount due\b", r"\bpayment due date\b", | |
| r"\bavailable credit(?: limit)?\b", r"\bcredit limit\b", r"\bcard (?:number|no|ending)\b", | |
| r"\breward points? (?:summary|balance)\b", r"\bstatement balance\b", | |
| r"\btotalamountdue\b", r"\bminimumamountdue\b", r"\bpaymentduedate\b", r"\bcreditlimit\b", | |
| ] | |
| score = sum(1 for pattern in signals if re.search(pattern, n)) | |
| if score >= 2: | |
| return True | |
| return score >= 1 and (len(cards) == 1 or (has_statement and bool(issuers))) | |
| def is_credit_card_statement(filename: str, content: bytes, password: Optional[str] = None) -> bool: | |
| """File-level wrapper used by the API before returning any transactions.""" | |
| detection_text = _detection_text(filename, content, password) | |
| return is_credit_card_statement_text(filename, detection_text) | |