import re from typing import List, Dict, Optional from .schemas import TransactionRow AMOUNT_RE = re.compile(r"(\d{1,3}(?:,\d{3})*(?:\.\d{2})?)") def normalize_spaces(text: str) -> str: return " ".join(text.split()) def parse_amount(value: str) -> Optional[float]: try: return float(value.replace(",", "")) except: return None def looks_like_bank_statement(text: str) -> bool: if not text or len(text.strip()) < 50: return False return True def is_datetime_line(line: str) -> bool: return bool(re.match(r"^\d{4}-\d{2}-\d{2}T\d{2}:", line.strip())) def extract_amounts(text: str) -> List[float]: matches = AMOUNT_RE.findall(text) values = [parse_amount(m) for m in matches if parse_amount(m) is not None] return values def parse_statement_pages(page_texts: List[Dict]) -> List[TransactionRow]: rows: List[TransactionRow] = [] for page in page_texts: page_no = page["page"] text = page["text"] or "" lines = [normalize_spaces(l) for l in text.splitlines() if l.strip()] i = 0 while i < len(lines): line = lines[i] if not is_datetime_line(line): i += 1 continue current_date = line.split("T")[0] block_lines = [line] i += 1 while i < len(lines) and not is_datetime_line(lines[i]): block_lines.append(lines[i]) i += 1 block_text = " ".join(block_lines) amounts = extract_amounts(block_text) if len(amounts) < 3: continue debit = amounts[-3] credit = amounts[-2] balance = amounts[-1] description = re.sub(r"^\d{4}-\d{2}-\d{2}T\d{2}:", "", block_text) raw_amount_strings = AMOUNT_RE.findall(block_text) for amt in raw_amount_strings[-3:]: description = description.replace(amt, " ") description = normalize_spaces(description) if not description or len(description) < 3: continue row = TransactionRow( date=current_date, description=description, debit=debit if debit != 0 else None, credit=credit if credit != 0 else None, balance=balance, source_page=page_no, confidence="high", ) rows.append(row) return rows def dedupe_rows(rows: List[TransactionRow]) -> List[TransactionRow]: seen = set() deduped: List[TransactionRow] = [] for row in rows: key = ( row.date, row.description, row.debit, row.credit, row.balance, row.source_page, ) if key in seen: continue seen.add(key) deduped.append(row) return deduped