251 lines
8.7 KiB
Python
251 lines
8.7 KiB
Python
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import json
|
|
import re
|
|
from dataclasses import dataclass
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
try:
|
|
import fitz # PyMuPDF
|
|
except ImportError: # pragma: no cover - exercised as runtime config error
|
|
fitz = None # type: ignore[assignment]
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class KbPdfTransaction:
|
|
page: int
|
|
transaction_datetime: str
|
|
transaction_date: str
|
|
raw_summary: str
|
|
counterparty: str
|
|
withdrawal: int
|
|
deposit: int
|
|
balance_after: int
|
|
remittance_memo: str
|
|
branch: str
|
|
|
|
@property
|
|
def type(self) -> str:
|
|
if self.deposit > 0 and self.withdrawal == 0:
|
|
return "income"
|
|
if self.withdrawal > 0 and self.deposit == 0:
|
|
return "expense"
|
|
return "adjustment"
|
|
|
|
@property
|
|
def amount(self) -> int:
|
|
return self.deposit or self.withdrawal
|
|
|
|
def external_id(self, source_hash: str) -> str:
|
|
raw = "|".join(
|
|
[
|
|
"kb_bank_pdf",
|
|
source_hash[:16],
|
|
self.transaction_datetime,
|
|
self.raw_summary,
|
|
self.counterparty,
|
|
str(self.withdrawal),
|
|
str(self.deposit),
|
|
str(self.balance_after),
|
|
]
|
|
)
|
|
return hashlib.sha256(raw.encode("utf-8")).hexdigest()
|
|
|
|
def raw_payload(self) -> str:
|
|
return json.dumps(
|
|
{
|
|
"page": self.page,
|
|
"transaction_datetime": self.transaction_datetime,
|
|
"raw_summary": self.raw_summary,
|
|
"counterparty": self.counterparty,
|
|
"withdrawal": self.withdrawal,
|
|
"deposit": self.deposit,
|
|
"balance_after": self.balance_after,
|
|
"remittance_memo": self.remittance_memo,
|
|
"branch": self.branch,
|
|
},
|
|
ensure_ascii=False,
|
|
separators=(",", ":"),
|
|
)
|
|
|
|
|
|
def _parse_int(value: str) -> int:
|
|
value = value.replace(",", "").replace("원", "").strip()
|
|
return int(value) if value and re.fullmatch(r"-?\d+", value) else 0
|
|
|
|
|
|
def _mask_account_number(account_number: str) -> str:
|
|
digits = re.sub(r"\D", "", account_number)
|
|
if len(digits) <= 4:
|
|
return f"****{digits}" if digits else ""
|
|
return f"****{digits[-4:]}"
|
|
|
|
|
|
def _extract_metadata(first_page_text: str) -> dict[str, Any]:
|
|
lines = [line.strip() for line in first_page_text.splitlines() if line.strip()]
|
|
|
|
def after_label(label: str) -> str:
|
|
try:
|
|
return lines[lines.index(label) + 1]
|
|
except (ValueError, IndexError):
|
|
return ""
|
|
|
|
account_number = after_label("계좌번호")
|
|
query_period = ""
|
|
for line in lines:
|
|
if re.fullmatch(r"\d{4}\.\d{2}\.\d{2}\s+~\s+\d{4}\.\d{2}\.\d{2}", line):
|
|
query_period = line
|
|
break
|
|
query_generated_at = ""
|
|
for line in lines:
|
|
if line.startswith("조회기준일시:"):
|
|
query_generated_at = line.replace("조회기준일시:", "").strip()
|
|
break
|
|
return {
|
|
"account_name": after_label("계좌명"),
|
|
"account_number_masked": _mask_account_number(account_number),
|
|
"holder_name": after_label("예금주"),
|
|
"final_balance": _parse_int(after_label("총 잔액")),
|
|
"withdrawable_amount": _parse_int(after_label("출금가능금액")),
|
|
"header_total_deposit": _parse_int(after_label("총 입금금액")),
|
|
"header_total_withdrawal": _parse_int(after_label("총 출금금액")),
|
|
"query_generated_at": query_generated_at,
|
|
"query_period": query_period,
|
|
}
|
|
|
|
|
|
def _open_pdf(path: Path, password: str) -> Any:
|
|
if fitz is None:
|
|
raise RuntimeError("PyMuPDF is required. Install pymupdf to parse KB Bank PDFs.")
|
|
doc = fitz.open(path)
|
|
if doc.needs_pass and not doc.authenticate(password):
|
|
raise ValueError("PDF password is required or incorrect")
|
|
return doc
|
|
|
|
|
|
def parse_kb_bank_pdf(file_path: str, password: str) -> dict[str, Any]:
|
|
path = Path(file_path).expanduser().resolve()
|
|
if not path.exists():
|
|
raise FileNotFoundError(f"PDF file not found: {path}")
|
|
source_hash = hashlib.sha256(path.read_bytes()).hexdigest()
|
|
doc = _open_pdf(path, password)
|
|
metadata = _extract_metadata(doc[0].get_text("text") if doc.page_count else "")
|
|
columns = [
|
|
("datetime", 0, 115),
|
|
("summary", 115, 165),
|
|
("counterparty", 165, 255),
|
|
("withdrawal", 255, 315),
|
|
("deposit", 315, 375),
|
|
("balance", 375, 450),
|
|
("memo", 450, 520),
|
|
("branch", 520, 999),
|
|
]
|
|
transactions: list[KbPdfTransaction] = []
|
|
for page_index, page in enumerate(doc, start=1):
|
|
lines: dict[int, list[tuple[float, str]]] = {}
|
|
for x0, y0, _x1, _y1, word, _block, _line, _wordno in page.get_text("words"):
|
|
# Header rows live above 110. Footer/help text lives below roughly 735.
|
|
if y0 < 110 or y0 > 735:
|
|
continue
|
|
key = round(y0 / 3) * 3
|
|
lines.setdefault(key, []).append((x0, word))
|
|
for y in sorted(lines):
|
|
cells: dict[str, list[str]] = {name: [] for name, _start, _end in columns}
|
|
for x, word in sorted(lines[y]):
|
|
for name, start, end in columns:
|
|
if start <= x < end:
|
|
cells[name].append(word)
|
|
break
|
|
values = {key: " ".join(parts).strip() for key, parts in cells.items()}
|
|
raw_datetime = values["datetime"]
|
|
if not re.fullmatch(r"20\d{2}\.\d{2}\.\d{2}\s+\d{2}:\d{2}:\d{2}", raw_datetime):
|
|
continue
|
|
dt = datetime.strptime(raw_datetime, "%Y.%m.%d %H:%M:%S")
|
|
transactions.append(
|
|
KbPdfTransaction(
|
|
page=page_index,
|
|
transaction_datetime=dt.isoformat(),
|
|
transaction_date=dt.date().isoformat(),
|
|
raw_summary=values["summary"],
|
|
counterparty=values["counterparty"],
|
|
withdrawal=_parse_int(values["withdrawal"]),
|
|
deposit=_parse_int(values["deposit"]),
|
|
balance_after=_parse_int(values["balance"]),
|
|
remittance_memo=values["memo"],
|
|
branch=values["branch"],
|
|
)
|
|
)
|
|
validation = validate_balance_flow(transactions)
|
|
return {
|
|
"file_path": str(path),
|
|
"source_hash": source_hash,
|
|
"page_count": doc.page_count,
|
|
"metadata": metadata,
|
|
"transactions": transactions,
|
|
"validation": validation,
|
|
}
|
|
|
|
|
|
def validate_balance_flow(transactions: list[KbPdfTransaction]) -> dict[str, Any]:
|
|
if not transactions:
|
|
return {
|
|
"ok": True,
|
|
"row_count": 0,
|
|
"opening_balance": None,
|
|
"final_balance": None,
|
|
"total_deposit": 0,
|
|
"total_withdrawal": 0,
|
|
"net_change": 0,
|
|
"errors": [],
|
|
}
|
|
opening_balance = transactions[0].balance_after + transactions[0].withdrawal - transactions[0].deposit
|
|
previous_balance = opening_balance
|
|
errors = []
|
|
total_deposit = 0
|
|
total_withdrawal = 0
|
|
for index, tx in enumerate(transactions, start=1):
|
|
expected = previous_balance - tx.withdrawal + tx.deposit
|
|
if expected != tx.balance_after:
|
|
errors.append(
|
|
{
|
|
"row_number": index,
|
|
"transaction_datetime": tx.transaction_datetime,
|
|
"expected_balance": expected,
|
|
"actual_balance": tx.balance_after,
|
|
"difference": tx.balance_after - expected,
|
|
}
|
|
)
|
|
previous_balance = tx.balance_after
|
|
total_deposit += tx.deposit
|
|
total_withdrawal += tx.withdrawal
|
|
final_balance = transactions[-1].balance_after
|
|
return {
|
|
"ok": not errors,
|
|
"row_count": len(transactions),
|
|
"opening_balance": opening_balance,
|
|
"final_balance": final_balance,
|
|
"total_deposit": total_deposit,
|
|
"total_withdrawal": total_withdrawal,
|
|
"net_change": total_deposit - total_withdrawal,
|
|
"balance_delta": final_balance - opening_balance,
|
|
"errors": errors,
|
|
}
|
|
|
|
|
|
def transaction_preview(tx: KbPdfTransaction, source_hash: str) -> dict[str, Any]:
|
|
return {
|
|
"transaction_datetime": tx.transaction_datetime,
|
|
"transaction_date": tx.transaction_date,
|
|
"type": tx.type,
|
|
"amount": tx.amount,
|
|
"merchant": tx.counterparty,
|
|
"payment_method": tx.raw_summary,
|
|
"balance_after": tx.balance_after,
|
|
"branch": tx.branch,
|
|
"memo": tx.remittance_memo,
|
|
"external_id": tx.external_id(source_hash),
|
|
}
|