Fix Gmail statement attribution and reject non-card imports

app/main.py

@@ -867,7 +867,7 @@ async def parse_statements(
 ):
     _limit(request, "parse", rate=12)       # 12/min per IP per worker
     from statement_parser import (parse_statement, detect_cards, detect_issuers,
-                              detect_points_balance_file,
+                              detect_points_balance_file, is_credit_card_statement,
                                   find_pdf_password, detect_credit_limit)
     # Smart unlock: the password field may carry MULTIPLE candidates (newline or
     # comma separated), derived app-side from the user's details the way every
@@ -902,6 +902,16 @@ async def parse_statements(
                 txns = await anyio.to_thread.run_sync(
                     lambda c=content, n=f.filename, p=pw, r=revs: parse_statement(n, c, password=p, reversals=r)
                 )
+                # Parsing rows is not proof that this is a CREDIT-CARD statement:
+                # savings/current-account CSVs share the same Date/Description/
+                # Amount shape. Reject them before any transaction, balance or
+                # inferred card id can reach the device.
+                is_card_statement = await anyio.to_thread.run_sync(
+                    lambda c=content, n=f.filename, p=pw: is_credit_card_statement(n, c, password=p)
+                )
+                if not is_card_statement:
+                    errors.append({"file": f.filename, "error": "NOT_CREDIT_CARD_STATEMENT"})
+                    continue
                 for t in txns:
                     t["source_file"] = f.filename
                 for r in revs:
@@ -992,7 +1002,7 @@ async def email_inbound(request: Request):
     """Inbound-email webhook (provider-agnostic: reads recipient/to plus any file
     fields from the multipart form)."""
     _limit(request, "inbound", rate=30)
-    from statement_parser import parse_statement, detect_cards
+    from statement_parser import parse_statement, detect_cards, is_credit_card_statement
     form = await request.form()
     token = _inbox_token(str(form.get("recipient") or form.get("to") or ""))
     if not token:
@@ -1007,6 +1017,12 @@ async def email_inbound(request: Request):
             revs: list = []
             txns = await anyio.to_thread.run_sync(
                 lambda c=content, n=val.filename, r=revs: parse_statement(n, c, reversals=r))
+            is_card_statement = await anyio.to_thread.run_sync(
+                lambda c=content, n=val.filename: is_credit_card_statement(n, c))
+            if not is_card_statement:
+                results.append({"file": val.filename, "count": 0, "transactions": [],
+                                "detected_cards": [], "error": "NOT_CREDIT_CARD_STATEMENT"})
+                continue
             cards = await anyio.to_thread.run_sync(
                 lambda c=content, n=val.filename: detect_cards(n, c))
             for t in txns:

app/statement_parser.py

@@ -1061,85 +1061,28 @@ def parse_statement(filename: str, content: bytes, password: Optional[str] = Non
         raise RuntimeError("Unsupported file type. Upload a .pdf or .csv statement.")
 
 
-# ---------------------------------------------------------------------------
-# Card detection: which catalogue card does this statement belong to?
-# (card_id, required issuer tokens - ANY must appear - or None when the product
-#  name is unique enough, product tokens - ANY must appear). Short tokens are
-# matched at word boundaries via merchants.keyword_hit so 'ace' cannot fire
-# inside 'place'. Mirrored in app statements.ts (detectCardsFromText).
-# ---------------------------------------------------------------------------
-_CARD_SIGNATURES = [
-    ("hdfc_infinia", None, ["infinia"]),
-    ("hdfc_regalia_gold", None, ["regalia"]),
-    ("hdfc_millennia", ["hdfc"], ["millennia"]),
-    # Before the legacy row: "Swiggy BLCK HDFC" contains "swiggy" and "hdfc" too,
-    # so a bare hdfc+swiggy signature would claim it first.
-    ("hdfc_swiggy_blck", ["hdfc", "blck"], ["swiggy"]),
-    ("hdfc_swiggy_ornge", ["hdfc", "ornge"], ["swiggy"]),
-    ("hdfc_swiggy", ["hdfc"], ["swiggy"]),
-    ("hdfc_diners_black", None, ["diners club black", "diners black"]),
-    ("tataneu_infinity", None, ["tata neu infinity", "neu infinity"]),
-    ("sbi_cashback", ["sbi"], ["cashback"]),
-    ("sbi_elite", ["sbi"], ["elite"]),
-    ("sbi_simplyclick", None, ["simplyclick", "simply click"]),
-    ("sbi_bpcl_octane", None, ["octane"]),
-    ("sbi_rupay_select", ["sbi"], ["rupay select"]),
-    ("icici_amazon_pay", ["icici"], ["amazon pay"]),
-    ("icici_sapphiro", None, ["sapphiro"]),
-    ("icici_emeralde", None, ["emeralde"]),
-    ("axis_magnus", None, ["magnus"]),
-    ("axis_ace", ["axis"], ["ace"]),
-    ("axis_atlas", ["axis"], ["atlas"]),
-    ("axis_flipkart", ["axis"], ["flipkart"]),
-    ("amex_mrcc", ["american express", "amex"], ["membership rewards"]),
-    ("amex_platinum_travel", ["american express", "amex"], ["platinum travel"]),
-    ("kiwi_axis", None, ["kiwi"]),
-    ("idfc_first_select", ["idfc"], ["first select"]),
-    ("idfc_first_wealth", ["idfc"], ["first wealth"]),
-    ("idfc_first_millennia", ["idfc"], ["millennia"]),
-    ("idfc_first_power_plus", ["idfc"], ["power+", "power plus"]),
-    ("au_zenith", ["au bank", "au small finance", "aubank"], ["zenith"]),
-    ("indusind_legend", ["indusind"], ["legend"]),
-    ("rbl_world_safari", None, ["world safari"]),
-    ("onecard_metal", None, ["onecard", "one card"]),
-    ("hsbc_live_plus", ["hsbc"], ["live+", "live plus"]),
-]
-
-
-def detect_cards_from_text(text: str) -> List[str]:
-    """Catalogue card ids this statement text plausibly belongs to, in signature
-    order, deduped. Pure and testable; parity-checked against the app mirror."""
-    from merchants import keyword_hit
-    t = (text or "").lower()
-    out: List[str] = []
-    for card_id, issuers, products in _CARD_SIGNATURES:
-        if issuers is not None and not any(keyword_hit(t, i) for i in issuers):
-            continue
-        if any(keyword_hit(t, p) for p in products):
-            out.append(card_id)
-    return out
-
-
-# Issuer name fragments -> catalogue issuer label. Used as a FALLBACK: when the
-# exact product name doesn't match a signature but the bank is clearly named, we
-# still know the issuer, so the app can ask "which of your <issuer> cards?".
-# Tokens include the email-DOMAIN forms (hdfcbank, axisbank, sbicard...) so a
-# FROM address like alerts@axisbank.com is recognised even without body text.
+# Exact catalogue issuer -> text/domain aliases. Mirrored in app statements.ts.
 _ISSUER_TOKENS = [
     ("HDFC Bank", ["hdfc", "hdfcbank"]),
-    ("SBI", ["sbi card", "sbi cards", "sbicard", "state bank"]),
+    ("SBI Card", ["sbi card", "sbi cards", "sbicard", "state bank"]),
     ("ICICI Bank", ["icici", "icicibank"]),
-    ("Axis Bank", ["axis bank", "axisbank"]),
+    ("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"]),
-    ("Kotak Mahindra Bank", ["kotak"]),
+    ("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"]),
-    ("OneCard", ["onecard", "one card", "getonecard"]),
-    ("Kiwi", ["kiwi"]),
+    ("Standard Chartered", ["standard chartered", "standardchartered", "sc bank"]),
 ]
 
 
@@ -1174,6 +1117,89 @@ def detect_issuers_from_text(text: str) -> List[str]:
     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
@@ -1229,3 +1255,38 @@ def detect_cards(filename: str, content: bytes, password: Optional[str] = None)
 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)