arrr-erp/app/modules/bank_statement_analyzer/parsers/registry.py

from __future__ import annotations

import logging

from .idfc import IDFCFirstParser
from .axis import AxisParser
from .hdfc import HDFCParser
from .icici import ICICIParser
from .hsbc import HSBCParser
from .indian_bank import IndianBankModernParser, IndianBankLegacyParser
from .indusind import IndusIndParser
from .kotak import KotakParser
from .sbi import SBIModernParser, SBIOtherParser
from .central_bank_of_india import CentralBankOfIndiaParser
from .yes_bank import YesBankParser
from .city_union_bank import CityUnionBankParser
from .bank_of_baroda import BankOfBarodaParser
from .rbl_bank import RBLBankParser
from .andhra_pradesh_grameena_bank import AndhraPradeshGrameenaBankParser
from .bank_identity import apply_detected_bank_identity
from .base import extract_text
from .template_engine import parse_with_template

logger = logging.getLogger(__name__)

PARSERS = [
    IDFCFirstParser,
    AxisParser,
    AndhraPradeshGrameenaBankParser,
    CentralBankOfIndiaParser,
    YesBankParser,
    CityUnionBankParser,
    BankOfBarodaParser,
    RBLBankParser,
    HDFCParser,
    ICICIParser,
    HSBCParser,
    IndianBankModernParser,
    IndianBankLegacyParser,
    IndusIndParser,
    KotakParser,
    SBIOtherParser,
    SBIModernParser,
]

BANK_OPTIONS = [
    ("auto", "Auto Detect"),
    ("axis", "Axis Bank"),
    ("andhra_pradesh_grameena_bank", "Andhra Pradesh Grameena Bank"),
    ("central_bank_of_india", "Central Bank of India"),
    ("yes_bank", "YES Bank"),
    ("city_union_bank", "City Union Bank"),
    ("bank_of_baroda", "Bank of Baroda"),
    ("rbl_bank", "RBL Bank"),
    ("hdfc", "HDFC Bank"),
    ("icici", "ICICI Bank"),
    ("hsbc", "HSBC Bank"),
    ("idfc", "IDFC FIRST Bank"),
    ("indian_bank", "Indian Bank"),
    ("indusind", "IndusInd Bank"),
    ("kotak", "Kotak Mahindra Bank"),
    ("sbi", "State Bank of India"),
]

BANK_PARSERS = {
    "axis": [AxisParser],
    "andhra_pradesh_grameena_bank": [AndhraPradeshGrameenaBankParser],
    "central_bank_of_india": [CentralBankOfIndiaParser],
    "yes_bank": [YesBankParser],
    "city_union_bank": [CityUnionBankParser],
    "bank_of_baroda": [BankOfBarodaParser],
    "rbl_bank": [RBLBankParser],
    "hdfc": [HDFCParser],
    "icici": [ICICIParser],
    "hsbc": [HSBCParser],
    "idfc": [IDFCFirstParser],
    "indian_bank": [IndianBankModernParser, IndianBankLegacyParser],
    "indusind": [IndusIndParser],
    "kotak": [KotakParser],
    "sbi": [SBIOtherParser, SBIModernParser],
}


def detect_parser(text):
    scored = sorted(
        ((parser.detect(text), parser) for parser in PARSERS),
        key=lambda item: item[0],
        reverse=True,
    )
    if not scored or scored[0][0] <= 0:
        return None, 0
    return scored[0][1](), scored[0][0]


def _usable_result(result) -> bool:
    if result is None:
        return False
    try:
        _meta, frame = result
        return frame is not None and not frame.empty
    except Exception:
        return False


def _identified(result, text: str):
    if not _usable_result(result):
        return None
    meta, frame = result
    return apply_detected_bank_identity(meta, frame, text)


def _attempt(parser, path, text: str, route: str):
    try:
        result = parser.parse(path, text)
        result = _identified(result, text)
        if result is not None:
            meta, frame = result
            logger.info(
                "Bank analyzer route accepted: route=%s parser=%s rows=%s file=%s",
                route,
                getattr(meta, "parser_name", parser.__class__.__name__),
                len(frame),
                path,
            )
            return result
        logger.warning(
            "Bank analyzer route returned no rows: route=%s parser=%s file=%s",
            route,
            parser.__class__.__name__,
            path,
        )
    except Exception as exc:
        logger.warning(
            "Bank analyzer route failed: route=%s parser=%s file=%s error=%s",
            route,
            parser.__class__.__name__,
            path,
            exc,
            exc_info=True,
        )
    return None


def _template_first(path, text: str, hint: str):
    try:
        result = parse_with_template(path, text=text, bank_hint=hint)
        result = _identified(result, text)
        if result is not None:
            meta, frame = result
            logger.info(
                "Bank analyzer template accepted: parser=%s rows=%s file=%s",
                getattr(meta, "parser_name", "TemplateBasedStatementParser"),
                len(frame),
                path,
            )
            return result
        logger.warning("Bank analyzer template returned no rows: file=%s", path)
    except Exception as exc:
        logger.warning(
            "Bank analyzer template rejected: file=%s error=%s",
            path,
            exc,
            exc_info=True,
        )
    return None


def _ordered_bank_candidates(text: str):
    scored = sorted(
        ((parser.detect(text), parser) for parser in PARSERS),
        key=lambda item: item[0],
        reverse=True,
    )
    return [parser for score, parser in scored if score > 0]


def parse_pdf(path, bank_hint: str | None = None):
    """Parse a statement using a validated template first, then bank parsers.

    Bank identity is resolved independently after extraction, so using a shared
    layout never labels a statement as the bank whose parser happens to have a
    similar table structure.
    """
    hint = (bank_hint or "auto").strip().lower()
    text = extract_text(path)

    template_result = _template_first(path, text, hint)
    if template_result is not None:
        return template_result

    if hint and hint != "auto":
        candidates = BANK_PARSERS.get(hint, [])
        for parser_class in sorted(candidates, key=lambda cls: cls.detect(text), reverse=True):
            result = _attempt(parser_class(), path, text, f"selected:{hint}")
            if result is not None:
                return result
        label = dict(BANK_OPTIONS).get(hint, "the selected bank")
        raise ValueError(
            f"The uploaded statement could not be parsed by a validated layout template or the "
            f"selected {label} parser. Please verify that the PDF is readable and supported."
        )

    attempted: set[type] = set()
    for parser_class in _ordered_bank_candidates(text):
        attempted.add(parser_class)
        result = _attempt(parser_class(), path, text, "auto-detected")
        if result is not None:
            return result

    if not (text or "").strip():
        result = _attempt(HSBCParser(), path, text, "image-only-ocr")
        if result is not None:
            return result

    # A detector may miss a new bank layout even though an existing dedicated
    # parser can still extract and reconcile it. Keep this compatibility pass.
    for parser_class in PARSERS:
        if parser_class in attempted:
            continue
        result = _attempt(parser_class(), path, text, "compatibility-fallback")
        if result is not None:
            return result

    raise ValueError(
        "The bank statement format was identified, but no transaction rows could be extracted. "
        "The validated template engine and all bank-specific parsers were attempted. "
        "Please verify that the PDF is text-readable and that this statement layout is supported."
    )