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

from __future__ import annotations
from dataclasses import dataclass, asdict
from pathlib import Path
from typing import Optional
from concurrent.futures import ThreadPoolExecutor, as_completed
import os, re, subprocess, tempfile
import pandas as pd
import pdfplumber

@dataclass
class StatementMeta:
    statement_id: str = ""
    bank_name: str = ""
    customer_name: str = ""
    account_number: str = ""
    customer_id: str = ""
    ifsc: str = ""
    period_from: str = ""
    period_to: str = ""
    opening_balance: Optional[float] = None
    total_debit: Optional[float] = None
    total_credit: Optional[float] = None
    closing_balance: Optional[float] = None
    source_file: str = ""
    parser_name: str = ""
    confidence: str = "Medium"

    def to_dict(self):
        return asdict(self)

STANDARD_COLUMNS = [
    "statement_id", "transaction_date", "value_date", "narration", "reference_no",
    "debit", "credit", "balance", "bank_name", "customer_name",
    "account_number", "source_file", "source_page", "parser_name",
    # Internal extraction-audit fields. These are retained for diagnostics but
    # are intentionally omitted from the client-facing workbook.
    "printed_debit", "printed_credit", "balance_delta", "movement_difference",
    "correction_applied", "correction_reason", "extraction_confidence",
]

def amount(v):
    if v is None: return None
    s=str(v).strip().replace('INR','').replace('Rs.','').replace('₹','').replace(',','').replace('+','')
    s=s.replace('CR','').replace('DR','').strip()
    if s in ('','-'): return None
    neg=s.startswith('-')
    s=s.lstrip('-')
    try:
        x=float(s)
        return -x if neg else x
    except: return None

def norm(s): return re.sub(r'\s+',' ',str(s or '')).strip()

_MIN_DIGITAL_TEXT_CHARS = 120
_MIN_DIGITAL_ALPHA_CHARS = 30
_OCR_DPI = max(120, min(int(os.getenv("BANK_ANALYZER_OCR_DPI", "180")), 300))
_OCR_WORKERS = max(1, min(int(os.getenv("BANK_ANALYZER_OCR_WORKERS", "2")), 4))


def _has_meaningful_text(value: str) -> bool:
    text = str(value or "").strip()
    if len(text) < _MIN_DIGITAL_TEXT_CHARS:
        return False
    return sum(character.isalpha() for character in text) >= _MIN_DIGITAL_ALPHA_CHARS


def _run_command(arguments: list[str], timeout: int) -> subprocess.CompletedProcess[str]:
    try:
        return subprocess.run(
            arguments,
            capture_output=True,
            text=True,
            timeout=timeout,
            check=False,
        )
    except FileNotFoundError as exc:
        raise RuntimeError(
            f"Required bank-statement extraction command is unavailable: {arguments[0]}"
        ) from exc


def _ocr_pdf_page(path: Path, page_number: int, work_dir: Path) -> tuple[int, str]:
    prefix = work_dir / f"page_{page_number:05d}"
    rendered_image = prefix.with_suffix(".png")

    render = _run_command(
        [
            "pdftoppm",
            "-f", str(page_number),
            "-l", str(page_number),
            "-r", str(_OCR_DPI),
            "-png",
            "-singlefile",
            str(path),
            str(prefix),
        ],
        timeout=180,
    )
    if render.returncode != 0 or not rendered_image.exists():
        message = render.stderr.strip() or f"Unable to render page {page_number} for OCR."
        raise RuntimeError(message)

    try:
        environment = os.environ.copy()
        environment.setdefault("OMP_THREAD_LIMIT", "1")
        ocr = subprocess.run(
            [
                "tesseract",
                str(rendered_image),
                "stdout",
                "-l", "eng",
                "--psm", "4",
                "-c", "preserve_interword_spaces=1",
            ],
            capture_output=True,
            text=True,
            timeout=180,
            check=False,
            env=environment,
        )
        if ocr.returncode != 0:
            raise RuntimeError(ocr.stderr.strip() or f"OCR failed for page {page_number}.")
        return page_number, ocr.stdout or ""
    finally:
        rendered_image.unlink(missing_ok=True)


def _ocr_selected_pages(path: Path, page_numbers: list[int]) -> dict[int, str]:
    if not page_numbers:
        return {}

    extracted: dict[int, str] = {}
    with tempfile.TemporaryDirectory(prefix="bank_statement_ocr_") as temporary:
        work_dir = Path(temporary)
        with ThreadPoolExecutor(max_workers=min(_OCR_WORKERS, len(page_numbers))) as executor:
            futures = {
                executor.submit(_ocr_pdf_page, path, page_number, work_dir): page_number
                for page_number in page_numbers
            }
            for future in as_completed(futures):
                page_number, text = future.result()
                extracted[page_number] = text
    return extracted


def extract_text(path: str | Path) -> str:
    """Extract PDF text and transparently OCR image-only or mixed pages.

    Existing digitally-generated statements continue to use the original fast
    ``pdftotext``/``pdfplumber`` path. OCR is invoked only when a page has no
    meaningful embedded text, so existing parsers and workbook behaviour remain
    unchanged for supported digital statements.
    """
    pdf_path = Path(path)

    try:
        result = _run_command(
            ["pdftotext", "-layout", str(pdf_path), "-"],
            timeout=120,
        )
        if result.returncode == 0 and _has_meaningful_text(result.stdout):
            return result.stdout
    except RuntimeError:
        # pdfplumber remains the normal fallback when pdftotext is unavailable.
        pass

    digital_pages: list[str] = []
    pages_requiring_ocr: list[int] = []
    with pdfplumber.open(str(pdf_path)) as pdf:
        for page_number, page in enumerate(pdf.pages, start=1):
            page_text = page.extract_text(
                x_tolerance=1,
                y_tolerance=3,
                layout=True,
            ) or ""
            digital_pages.append(page_text)
            if not _has_meaningful_text(page_text):
                pages_requiring_ocr.append(page_number)

    if not pages_requiring_ocr:
        return "\n\f\n".join(digital_pages)

    try:
        ocr_pages = _ocr_selected_pages(pdf_path, pages_requiring_ocr)
    except RuntimeError as exc:
        if any(_has_meaningful_text(page) for page in digital_pages):
            # Preserve the former mixed-PDF behaviour when OCR tooling is absent.
            return "\n\f\n".join(digital_pages)
        raise ValueError(
            "This bank statement is image-only and OCR processing is unavailable. "
            "Install poppler-utils and tesseract-ocr in the application image."
        ) from exc

    merged_pages = [
        ocr_pages.get(page_number, digital_pages[page_number - 1])
        if page_number in pages_requiring_ocr
        else digital_pages[page_number - 1]
        for page_number in range(1, len(digital_pages) + 1)
    ]
    return "\n\f\n".join(merged_pages)

def page_of_line(text: str, position: int) -> int:
    return text[:position].count('\f')+1

def _apply_balance_delta_validation(df: pd.DataFrame, meta: StatementMeta) -> pd.DataFrame:
    """Validate and, when necessary, correct debit/credit using balance movement.

    Printed PDF columns remain the first extraction source.  The running-balance
    delta is the independent accounting control.  Where the printed movement and
    the balance delta disagree, the delta determines the corrected side and
    amount.  This common routine is used by every bank parser through ``finalize``.
    """
    if df.empty:
        return df

    tolerance = 0.01
    x = df.copy()
    x["printed_debit"] = pd.to_numeric(x.get("debit"), errors="coerce")
    x["printed_credit"] = pd.to_numeric(x.get("credit"), errors="coerce")
    x["balance_delta"] = pd.NA
    x["movement_difference"] = pd.NA
    x["correction_applied"] = False
    x["correction_reason"] = ""
    x["extraction_confidence"] = "Printed columns"

    dated = pd.to_datetime(x.get("transaction_date"), errors="coerce")
    valid_dates = dated.dropna()
    descending = len(valid_dates) >= 2 and valid_dates.iloc[0] > valid_dates.iloc[-1]
    order = list(reversed(x.index.tolist())) if descending else x.index.tolist()

    previous_balance = meta.opening_balance
    for idx in order:
        current_balance = x.at[idx, "balance"]
        if pd.isna(current_balance):
            x.at[idx, "extraction_confidence"] = "Review - balance unavailable"
            continue

        current_balance = float(current_balance)
        if previous_balance is None or pd.isna(previous_balance):
            previous_balance = current_balance
            x.at[idx, "extraction_confidence"] = "Printed columns - no opening delta"
            continue

        delta = round(current_balance - float(previous_balance), 2)
        debit = float(x.at[idx, "debit"]) if pd.notna(x.at[idx, "debit"]) else 0.0
        credit = float(x.at[idx, "credit"]) if pd.notna(x.at[idx, "credit"]) else 0.0
        printed_movement = round(credit - debit, 2)
        movement_difference = round(delta - printed_movement, 2)
        x.at[idx, "balance_delta"] = delta
        x.at[idx, "movement_difference"] = movement_difference

        if abs(movement_difference) <= tolerance:
            x.at[idx, "extraction_confidence"] = "100% - printed movement matches delta"
        elif abs(delta) > tolerance:
            corrected_debit = round(abs(delta), 2) if delta < 0 else 0.0
            corrected_credit = round(delta, 2) if delta > 0 else 0.0
            x.at[idx, "debit"] = corrected_debit
            x.at[idx, "credit"] = corrected_credit
            x.at[idx, "correction_applied"] = True
            x.at[idx, "correction_reason"] = "Debit/credit corrected from running-balance delta"
            if abs(abs(printed_movement) - abs(delta)) <= tolerance:
                x.at[idx, "extraction_confidence"] = "99% - amount matched, side corrected by delta"
            elif debit == 0.0 and credit == 0.0:
                x.at[idx, "extraction_confidence"] = "98% - missing movement derived from delta"
            else:
                x.at[idx, "extraction_confidence"] = "Review - printed movement replaced by delta"
        else:
            x.at[idx, "extraction_confidence"] = "Review - zero balance movement"

        previous_balance = current_balance

    return x


def finalize(df: pd.DataFrame, meta: StatementMeta) -> pd.DataFrame:
    if df is None or df.empty:
        return pd.DataFrame(columns=STANDARD_COLUMNS)
    for c in ['debit','credit','balance']:
        df[c]=pd.to_numeric(df.get(c),errors='coerce')
    from .common import parse_flexible_date
    for c in ['transaction_date','value_date']:
        values = df.get(c)
        if values is None:
            df[c] = pd.NaT
        else:
            df[c] = values.map(parse_flexible_date)
    df = _apply_balance_delta_validation(df, meta)
    narration_values = df['narration'] if 'narration' in df.columns else pd.Series('', index=df.index)
    reference_values = df['reference_no'] if 'reference_no' in df.columns else pd.Series('', index=df.index)
    df['narration']=narration_values.fillna('').map(norm)
    df['reference_no']=reference_values.fillna('').map(norm)
    df['bank_name']=meta.bank_name
    df['customer_name']=meta.customer_name
    df['account_number']=meta.account_number
    df['source_file']=meta.source_file
    df['parser_name']=meta.parser_name
    if 'source_page' not in df: df['source_page']=None
    for c in STANDARD_COLUMNS:
        if c not in df: df[c]=None
    return df[STANDARD_COLUMNS]

class BaseParser:
    bank_name='Unknown'
    parser_name='BaseParser'
    @classmethod
    def detect(cls,text:str)->float: return 0.0
    def parse(self,path:str|Path,text:str|None=None): raise NotImplementedError