Spaces:
Sleeping
Sleeping
File size: 2,958 Bytes
df77297 3946573 2e7096d a688730 df77297 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 | 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 |