jarca-ledger-mcp/jarca_ledger_mcp/toss_bank_pdf.py

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),
    }