genichin
d9182f28f2
Merge BAT-017 / REQ-021 implementation after delivery-runner implementation, independent QC, and passing GitHub Actions CI.
229 lines
6.7 KiB
Python
229 lines
6.7 KiB
Python
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import re
|
|
from dataclasses import dataclass
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
import fitz
|
|
|
|
from .provenance_policy import allowlisted_json_payload
|
|
|
|
|
|
DATETIME_RE = re.compile(r"^\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}$")
|
|
AMOUNT_RE = re.compile(r"^[+-]?\d{1,3}(?:,\d{3})*$")
|
|
ACCOUNT_NUMBER_RE = re.compile(r"\d{3,}-\d{4}-\d{4}")
|
|
|
|
|
|
@dataclass
|
|
class TossBankTransaction:
|
|
row_index: int
|
|
transaction_datetime: str
|
|
transaction_date: str
|
|
action: str
|
|
counterparty: str
|
|
amount: int
|
|
type: str
|
|
balance_after: int
|
|
raw_summary: str
|
|
raw_lines: list[str]
|
|
|
|
def external_id(self, source_hash: str) -> str:
|
|
return f"toss_bank_pdf:{source_hash}:{self.row_index}:{self.transaction_datetime}:{self.amount}"
|
|
|
|
def raw_payload(self) -> str:
|
|
return allowlisted_json_payload(
|
|
{
|
|
"row_index": self.row_index,
|
|
"transaction_datetime": self.transaction_datetime,
|
|
"action": self.action,
|
|
"counterparty": self.counterparty,
|
|
"amount": self.amount,
|
|
"type": self.type,
|
|
"balance_after": self.balance_after,
|
|
"raw_summary": self.raw_summary,
|
|
"raw_lines": self.raw_lines,
|
|
},
|
|
allowed_keys=("row_index",),
|
|
)
|
|
|
|
|
|
def _normalize_lines(text: str) -> list[str]:
|
|
lines: list[str] = []
|
|
for line in text.splitlines():
|
|
line = re.sub(r"\s+", " ", line).strip()
|
|
if not line:
|
|
continue
|
|
# Toss statements include page markers like '1 / 70'
|
|
if re.fullmatch(r"\d+\s*/\s*\d+", line):
|
|
continue
|
|
lines.append(line)
|
|
return lines
|
|
|
|
|
|
def _parse_amount_token(text: str) -> int | None:
|
|
token = text.strip()
|
|
if not AMOUNT_RE.fullmatch(token):
|
|
return None
|
|
try:
|
|
return int(token.replace(",", ""))
|
|
except ValueError:
|
|
return None
|
|
|
|
|
|
def _extract_metadata(text: str) -> dict[str, str]:
|
|
lines = _normalize_lines(text)
|
|
holder_name = ""
|
|
account_name = ""
|
|
account_number_masked = ""
|
|
period = ""
|
|
|
|
for idx, line in enumerate(lines):
|
|
if line.endswith("님의 계좌 거래내역을"):
|
|
holder_name = line.replace("님의 계좌 거래내역을", "").strip()
|
|
if line == "예금주" and idx + 1 < len(lines):
|
|
if not holder_name:
|
|
holder_name = lines[idx + 1]
|
|
continue
|
|
if line == "계좌번호" and idx + 1 < len(lines):
|
|
next_line = lines[idx + 1]
|
|
if ACCOUNT_NUMBER_RE.search(next_line):
|
|
account_number_masked = next_line
|
|
continue
|
|
if line == "예금명" and idx + 1 < len(lines):
|
|
if not account_name:
|
|
account_name = lines[idx + 1]
|
|
continue
|
|
if line == "계좌종류" and idx + 1 < len(lines):
|
|
if not account_name:
|
|
account_name = lines[idx + 1]
|
|
continue
|
|
if line == "조회기간" and idx + 1 < len(lines):
|
|
period = lines[idx + 1]
|
|
|
|
if not account_name:
|
|
account_name = "토스뱅크 통장"
|
|
|
|
return {
|
|
"holder_name": holder_name,
|
|
"account_name": account_name,
|
|
"account_number_masked": account_number_masked,
|
|
"period": period,
|
|
"institution": "토스뱅크",
|
|
"source": "toss_bank_pdf",
|
|
}
|
|
|
|
|
|
def _transaction_from_rows(row_index: int, lines: list[str]) -> TossBankTransaction | None:
|
|
if len(lines) < 4:
|
|
return None
|
|
transaction_datetime = lines[0].strip()
|
|
if not DATETIME_RE.match(transaction_datetime):
|
|
return None
|
|
|
|
action = lines[1].strip() if len(lines) > 1 else ""
|
|
amount = None
|
|
balance_after = None
|
|
memo_lines: list[str] = []
|
|
|
|
for item in lines[2:]:
|
|
value = item.strip()
|
|
parsed_amount = _parse_amount_token(value)
|
|
if amount is None and parsed_amount is not None:
|
|
amount = parsed_amount
|
|
continue
|
|
if balance_after is None and parsed_amount is not None:
|
|
balance_after = parsed_amount
|
|
continue
|
|
memo_lines.append(value)
|
|
|
|
if amount is None or balance_after is None:
|
|
return None
|
|
|
|
counterparty = " ".join(memo_lines).strip()
|
|
tx_type = "expense" if amount < 0 else "income"
|
|
|
|
return TossBankTransaction(
|
|
row_index=row_index,
|
|
transaction_datetime=transaction_datetime,
|
|
transaction_date=transaction_datetime[:10],
|
|
action=action,
|
|
counterparty=counterparty,
|
|
amount=abs(amount),
|
|
type=tx_type,
|
|
balance_after=balance_after,
|
|
raw_summary=counterparty,
|
|
raw_lines=lines,
|
|
)
|
|
|
|
|
|
def _parse_rows(lines: list[str]) -> list[TossBankTransaction]:
|
|
rows: list[list[str]] = []
|
|
current: list[str] = []
|
|
|
|
for line in lines:
|
|
if DATETIME_RE.match(line):
|
|
if current:
|
|
rows.append(current)
|
|
current = [line]
|
|
elif current:
|
|
current.append(line)
|
|
|
|
if current:
|
|
rows.append(current)
|
|
|
|
parsed: list[TossBankTransaction] = []
|
|
for idx, row_lines in enumerate(rows, start=1):
|
|
tx = _transaction_from_rows(idx, row_lines)
|
|
if tx is not None:
|
|
parsed.append(tx)
|
|
return parsed
|
|
|
|
|
|
def parse_toss_bank_pdf(file_path: str) -> dict[str, Any]:
|
|
path = Path(file_path).expanduser()
|
|
if not path.exists():
|
|
raise FileNotFoundError(f"Toss Bank PDF not found: {path}")
|
|
|
|
data = path.read_bytes()
|
|
source_hash = hashlib.sha256(data).hexdigest()
|
|
doc = fitz.open(stream=data, filetype="pdf")
|
|
if doc.is_encrypted:
|
|
raise ValueError("encrypted Toss Bank PDF is not supported yet")
|
|
|
|
lines: list[str] = []
|
|
for page in doc:
|
|
lines.extend(_normalize_lines(page.get_text("text")))
|
|
all_text = "\n".join(lines)
|
|
metadata = _extract_metadata(all_text)
|
|
|
|
transactions = _parse_rows(lines)
|
|
validation = {
|
|
"ok": bool(transactions),
|
|
"parsed_rows": len(transactions),
|
|
"errors": [] if transactions else ["no valid rows parsed"],
|
|
}
|
|
return {
|
|
"file_path": str(path),
|
|
"source_hash": source_hash,
|
|
"page_count": doc.page_count,
|
|
"metadata": metadata,
|
|
"transactions": transactions,
|
|
"validation": validation,
|
|
}
|
|
|
|
|
|
def transaction_preview(tx: TossBankTransaction, source_hash: str) -> dict[str, Any]:
|
|
return {
|
|
"transaction_datetime": tx.transaction_datetime,
|
|
"transaction_date": tx.transaction_date,
|
|
"type": tx.type,
|
|
"action": tx.action,
|
|
"counterparty": tx.counterparty,
|
|
"amount": tx.amount,
|
|
"balance_after": tx.balance_after,
|
|
"raw_summary": tx.raw_summary,
|
|
"external_id": tx.external_id(source_hash),
|
|
}
|