Fix modern ICICI template parsing and parser fallback
This commit is contained in:
@@ -10,10 +10,15 @@ from .base import BaseParser, StatementMeta, amount, extract_text, finalize, nor
|
||||
from .common import date_iso, find, infer_mode
|
||||
|
||||
|
||||
_DATE_RE = re.compile(r"^\d{2}-\d{2}-\d{4}$")
|
||||
_DATE_RE = re.compile(r"^(?:\d{2}-\d{2}-\d{4}|\d{2}-[A-Za-z]{3}-\d{4})$")
|
||||
_MONEY_RE = re.compile(r"^-?(?:\d{1,3}(?:,\d{2,3})+|\d+)\.\d{2}$")
|
||||
|
||||
|
||||
def _clean_date_cell(value: str) -> str:
|
||||
text = norm(value)
|
||||
return re.sub(r"\s*([/-])\s*", r"\1", text)
|
||||
|
||||
|
||||
class ICICIParser(BaseParser):
|
||||
"""Parser for ICICI Bank retail/Privilege PDF account statements.
|
||||
|
||||
@@ -34,12 +39,83 @@ class ICICIParser(BaseParser):
|
||||
score += 0.55
|
||||
if "STATEMENT OF TRANSACTIONS IN SAVINGS ACCOUNT" in upper:
|
||||
score += 0.20
|
||||
if "ACCOUNT STATEMENT" in upper and "TRANSACTION ID" in upper:
|
||||
score += 0.20
|
||||
if "AVAILABLE BALANCE" in upper and "WITHDRAWAL" in upper and "DEPOSIT" in upper:
|
||||
score += 0.15
|
||||
if all(token in upper for token in ("DEPOSITS", "WITHDRAWALS", "BALANCE")):
|
||||
score += 0.20
|
||||
if "ACCOUNT RELATED OTHER INFORMATION" in upper:
|
||||
score += 0.05
|
||||
return min(score, 0.99)
|
||||
|
||||
@staticmethod
|
||||
def _modern_header_mapping(row: list) -> dict[str, int] | None:
|
||||
cells = [norm(cell).upper() for cell in (row or [])]
|
||||
aliases = {
|
||||
"serial": ("S.NO", "S NO", "SERIAL NO"),
|
||||
"transaction_id": ("TRANSACTION ID", "TXN ID"),
|
||||
"transaction_date": ("TRANSACTION DATE", "TXN DATE"),
|
||||
"cheque_no": ("CHEQUE NO", "CHQ NO"),
|
||||
"description": ("DESCRIPTION", "PARTICULARS"),
|
||||
"debit": ("WITHDRAWAL (DR)", "WITHDRAWAL", "DEBIT"),
|
||||
"credit": ("DEPOSIT (CR)", "DEPOSIT", "CREDIT"),
|
||||
"balance": ("AVAILABLE BALANCE", "BALANCE"),
|
||||
}
|
||||
mapping: dict[str, int] = {}
|
||||
for field, names in aliases.items():
|
||||
for index, cell in enumerate(cells):
|
||||
if any(name == cell or name in cell for name in names):
|
||||
mapping[field] = index
|
||||
break
|
||||
required = {"transaction_date", "description", "debit", "credit", "balance"}
|
||||
return mapping if required.issubset(mapping) else None
|
||||
|
||||
def _parse_modern_tables(self, pdf) -> list[dict]:
|
||||
rows: list[dict] = []
|
||||
active_mapping: dict[str, int] | None = None
|
||||
for page_number, page in enumerate(pdf.pages, start=1):
|
||||
for table in page.extract_tables() or []:
|
||||
if not table:
|
||||
continue
|
||||
start_index = 0
|
||||
header_mapping = self._modern_header_mapping(table[0])
|
||||
if header_mapping:
|
||||
active_mapping = header_mapping
|
||||
start_index = 1
|
||||
if active_mapping is None:
|
||||
continue
|
||||
mapping = active_mapping
|
||||
for raw in table[start_index:]:
|
||||
cells = list(raw or [])
|
||||
max_index = max(mapping.values())
|
||||
if len(cells) <= max_index:
|
||||
cells.extend([None] * (max_index + 1 - len(cells)))
|
||||
transaction_date = _clean_date_cell(cells[mapping["transaction_date"]])
|
||||
if not _DATE_RE.fullmatch(transaction_date):
|
||||
continue
|
||||
narration = norm(cells[mapping["description"]])
|
||||
transaction_id = norm(cells[mapping.get("transaction_id", -1)]) if "transaction_id" in mapping else ""
|
||||
cheque_no = norm(cells[mapping.get("cheque_no", -1)]) if "cheque_no" in mapping else ""
|
||||
debit = amount(cells[mapping["debit"]])
|
||||
credit = amount(cells[mapping["credit"]])
|
||||
balance_value = amount(cells[mapping["balance"]])
|
||||
if balance_value is None or (debit is None and credit is None):
|
||||
continue
|
||||
reference = cheque_no if cheque_no and cheque_no != "-" else transaction_id
|
||||
rows.append({
|
||||
"transaction_date": date_iso(transaction_date),
|
||||
"value_date": date_iso(transaction_date),
|
||||
"narration": narration,
|
||||
"reference_no": reference,
|
||||
"debit": debit,
|
||||
"credit": credit,
|
||||
"balance": balance_value,
|
||||
"source_page": page_number,
|
||||
"mode": infer_mode(narration),
|
||||
})
|
||||
return rows
|
||||
|
||||
@staticmethod
|
||||
def _table_geometry(page, words: list[dict]) -> tuple[float, list[float]] | None:
|
||||
required = {"DATE", "PARTICULARS", "DEPOSITS", "WITHDRAWALS", "BALANCE"}
|
||||
@@ -188,10 +264,14 @@ class ICICIParser(BaseParser):
|
||||
|
||||
name_match = re.search(r"(?m)^\s*((?:MS|MR|MRS)\.?\s*[^\n]+)$", text, re.I)
|
||||
meta.customer_name = norm(re.split(r"\s{2,}", name_match.group(1).strip())[0]) if name_match else ""
|
||||
meta.customer_id = find(r"Cust ID\s*:\s*([0-9]+)", text)
|
||||
meta.account_number = find(r"Savings Account Number\s*:\s*([0-9]+)", text)
|
||||
meta.customer_id = find(r"(?:Cust ID|Customer ID)\s*:\s*([0-9]+)", text)
|
||||
meta.account_number = find(r"(?:Savings Account Number|Account number)\s*:\s*([0-9]+)", text, flags=re.I)
|
||||
if not meta.account_number:
|
||||
meta.account_number = find(r"Savings\s+A/c\s+([0-9]+)", text)
|
||||
if not meta.customer_name:
|
||||
meta.customer_name = norm(find(r"Account name\s*:\s*([^\n]+)", text, flags=re.I))
|
||||
if not meta.ifsc:
|
||||
meta.ifsc = find(r"IFSC code\s*:\s*([A-Z0-9]+)", text, flags=re.I)
|
||||
meta.ifsc = find(r"IFSC CODE\s+NAME OF NOMINEE.*?Savings\s+[0-9]+\s+[0-9]+\s+([A-Z0-9]+)", text, flags=re.I | re.S)
|
||||
|
||||
period = re.search(
|
||||
@@ -205,11 +285,13 @@ class ICICIParser(BaseParser):
|
||||
|
||||
all_rows: list[dict] = []
|
||||
with pdfplumber.open(str(pdf_path)) as pdf:
|
||||
for page_number, page in enumerate(pdf.pages, start=1):
|
||||
page_rows, page_opening = self._parse_page(page, page_number)
|
||||
if meta.opening_balance is None and page_opening is not None:
|
||||
meta.opening_balance = page_opening
|
||||
all_rows.extend(page_rows)
|
||||
all_rows = self._parse_modern_tables(pdf)
|
||||
if not all_rows:
|
||||
for page_number, page in enumerate(pdf.pages, start=1):
|
||||
page_rows, page_opening = self._parse_page(page, page_number)
|
||||
if meta.opening_balance is None and page_opening is not None:
|
||||
meta.opening_balance = page_opening
|
||||
all_rows.extend(page_rows)
|
||||
|
||||
data = pd.DataFrame(all_rows)
|
||||
if not data.empty:
|
||||
|
||||
Reference in New Issue
Block a user