genichin
d9182f28f2
Merge BAT-017 / REQ-021 implementation after delivery-runner implementation, independent QC, and passing GitHub Actions CI.
296 lines
9.4 KiB
Python
296 lines
9.4 KiB
Python
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import json
|
|
import re
|
|
from dataclasses import dataclass
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
import fitz
|
|
|
|
from .provenance_policy import allowlisted_json_payload
|
|
|
|
DATE_RE = re.compile(r"^\d{4}\.\d{2}\.\d{2}$")
|
|
FOOTER_RE = re.compile(r"^\d+\s*/\s*\d+$")
|
|
NUM_RE = re.compile(r"^-?\d{1,3}(?:,\d{3})*(?:\.\d+)?$|^-?\d+(?:\.\d+)?$")
|
|
DOLLAR_RE = re.compile(r"^\(\$\s*-?\d+(?:,\d{3})*(?:\.\d+)?\)$")
|
|
HEADER_LINES = {
|
|
"거래일자",
|
|
"거래구분",
|
|
"종목명(종목코드)",
|
|
"환율",
|
|
"거래수량",
|
|
"거래대금",
|
|
"정산금액",
|
|
"단가",
|
|
"수수료",
|
|
"거래세",
|
|
"제세금",
|
|
"변제/연체합",
|
|
"잔고",
|
|
"잔액",
|
|
}
|
|
|
|
|
|
def _clean(text: str) -> str:
|
|
return re.sub(r"\s+", " ", text.replace("\u00a0", " ")).strip()
|
|
|
|
|
|
def _parse_number(value: str) -> float:
|
|
cleaned = _clean(value).replace(",", "")
|
|
if cleaned.startswith("($") and cleaned.endswith(")"):
|
|
cleaned = cleaned[2:-1].strip()
|
|
return float(cleaned)
|
|
|
|
|
|
def _parse_int(value: str) -> int:
|
|
return int(round(_parse_number(value)))
|
|
|
|
|
|
def _is_num(value: str) -> bool:
|
|
value = _clean(value)
|
|
return bool(NUM_RE.match(value) or DOLLAR_RE.match(value))
|
|
|
|
|
|
def _split_numeric_line(line: str) -> list[str]:
|
|
line = _clean(line)
|
|
parts = line.split()
|
|
if len(parts) > 1 and all(NUM_RE.match(p) for p in parts):
|
|
return parts
|
|
return [line]
|
|
|
|
|
|
def _normalize_lines(text: str) -> list[str]:
|
|
lines: list[str] = []
|
|
for raw in text.splitlines():
|
|
line = _clean(raw)
|
|
if not line or FOOTER_RE.match(line) or line in HEADER_LINES:
|
|
continue
|
|
if line.startswith("수량단위") or line.startswith("금액단위") or line.startswith("발급기준"):
|
|
continue
|
|
lines.append(line)
|
|
return lines
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class TossSecuritiesTransaction:
|
|
row_index: int
|
|
section: str
|
|
transaction_date: str
|
|
action: str
|
|
name: str
|
|
code: str
|
|
amount: int
|
|
type: str
|
|
quantity: float | None
|
|
unit_price: float | None
|
|
fee: int
|
|
tax: int
|
|
balance_after: int | None
|
|
raw_numbers: list[str]
|
|
raw_block: list[str]
|
|
|
|
def external_id(self, source_hash: str) -> str:
|
|
raw = "|".join(
|
|
[
|
|
source_hash,
|
|
str(self.row_index),
|
|
self.section,
|
|
self.transaction_date,
|
|
self.action,
|
|
self.name,
|
|
str(self.amount),
|
|
json.dumps(self.raw_numbers, ensure_ascii=False),
|
|
]
|
|
)
|
|
return "tosssec:" + hashlib.sha256(raw.encode("utf-8")).hexdigest()
|
|
|
|
def raw_payload(self) -> str:
|
|
return allowlisted_json_payload(
|
|
{
|
|
"row_index": self.row_index,
|
|
"section": self.section,
|
|
"transaction_date": self.transaction_date,
|
|
"action": self.action,
|
|
"name": self.name,
|
|
"code": self.code,
|
|
"amount": self.amount,
|
|
"type": self.type,
|
|
"quantity": self.quantity,
|
|
"unit_price": self.unit_price,
|
|
"fee": self.fee,
|
|
"tax": self.tax,
|
|
"balance_after": self.balance_after,
|
|
"raw_numbers": self.raw_numbers,
|
|
"raw_block": self.raw_block,
|
|
},
|
|
allowed_keys=("row_index", "section", "quantity", "unit_price", "fee", "tax"),
|
|
)
|
|
|
|
|
|
def _extract_metadata(text: str) -> dict[str, Any]:
|
|
compact = _clean(text)
|
|
holder = ""
|
|
account = ""
|
|
date_range = ""
|
|
issue_no = ""
|
|
m = re.search(r"성명\s+([^\s]+)", compact)
|
|
if m:
|
|
holder = m.group(1)
|
|
m = re.search(r"계좌 번호\s+([0-9\-]+)", compact)
|
|
if m:
|
|
account = m.group(1)
|
|
m = re.search(r"조회 기간\s+(\d{4}년\s+\d{1,2}월\s+\d{1,2}일\s+~\s+\d{4}년\s+\d{1,2}월\s+\d{1,2}일)", compact)
|
|
if m:
|
|
date_range = m.group(1)
|
|
m = re.search(r"거래내역서\s+([0-9A-Z\-]+)\s+발급번호", compact)
|
|
if m:
|
|
issue_no = m.group(1)
|
|
return {
|
|
"holder_name": holder,
|
|
"account_number_masked": account,
|
|
"date_range": date_range,
|
|
"issue_no": issue_no,
|
|
"account_name": "토스증권 계좌",
|
|
}
|
|
|
|
|
|
def _blocks_from_lines(lines: list[str]) -> list[tuple[str, list[str]]]:
|
|
blocks: list[tuple[str, list[str]]] = []
|
|
section = "KRW"
|
|
current: list[str] | None = None
|
|
current_section = section
|
|
for line in lines:
|
|
if line == "원화 거래내역":
|
|
section = "KRW"
|
|
continue
|
|
if line == "달러 거래내역":
|
|
section = "USD"
|
|
continue
|
|
if DATE_RE.match(line):
|
|
if current:
|
|
blocks.append((current_section, current))
|
|
current = [line]
|
|
current_section = section
|
|
elif current is not None:
|
|
current.append(line)
|
|
if current:
|
|
blocks.append((current_section, current))
|
|
return blocks
|
|
|
|
|
|
def _parse_block(row_index: int, section: str, block: list[str]) -> TossSecuritiesTransaction | None:
|
|
if len(block) < 3:
|
|
return None
|
|
date_iso = block[0].replace(".", "-")
|
|
action = block[1]
|
|
body_start = 2
|
|
if "(" in action and ")" not in action and len(block) > 2:
|
|
action = f"{action} {block[2]}"
|
|
body_start = 3
|
|
action = action.replace("케 이뱅크", "케이뱅크")
|
|
name_parts: list[str] = []
|
|
numeric_cells: list[str] = []
|
|
numeric_started = False
|
|
for line in block[body_start:]:
|
|
pieces = _split_numeric_line(line)
|
|
if not numeric_started and all(_is_num(p) for p in pieces):
|
|
numeric_started = True
|
|
if numeric_started:
|
|
numeric_cells.extend(p for p in pieces if _is_num(p))
|
|
else:
|
|
name_parts.append(line)
|
|
name = " ".join(name_parts).strip()
|
|
code = ""
|
|
code_match = re.search(r"\(([A-Z0-9]+)\)", name)
|
|
if code_match:
|
|
code = code_match.group(1)
|
|
krw_numbers = [n for n in numeric_cells if not _clean(n).startswith("($")]
|
|
if len(krw_numbers) < 2:
|
|
return None
|
|
has_exchange = section == "USD" or action.startswith("환전") or action.startswith("외화")
|
|
amount_idx = 3 if has_exchange and len(krw_numbers) > 3 else 2 if len(krw_numbers) > 2 else 1
|
|
amount = abs(_parse_int(krw_numbers[amount_idx]))
|
|
quantity_idx = 1 if has_exchange and len(krw_numbers) > 1 else 0
|
|
quantity = _parse_number(krw_numbers[quantity_idx]) if len(krw_numbers) > quantity_idx else None
|
|
price_idx = 4 if has_exchange else 3
|
|
unit_price = _parse_number(krw_numbers[price_idx]) if len(krw_numbers) > price_idx else None
|
|
fee_idx = 5 if has_exchange else 4
|
|
fee = abs(_parse_int(krw_numbers[fee_idx])) if len(krw_numbers) > fee_idx else 0
|
|
tax_start = 6 if has_exchange else 5
|
|
tax = sum(abs(_parse_int(x)) for x in krw_numbers[tax_start : min(tax_start + 2, len(krw_numbers))])
|
|
balance_after = _parse_int(krw_numbers[-1]) if krw_numbers else None
|
|
tx_type = "income" if any(k in action for k in ["입금", "판매", "환급"]) else "expense"
|
|
if any(k in action for k in ["출금", "구매"]):
|
|
tx_type = "expense"
|
|
return TossSecuritiesTransaction(
|
|
row_index=row_index,
|
|
section=section,
|
|
transaction_date=date_iso,
|
|
action=action,
|
|
name=name,
|
|
code=code,
|
|
amount=amount,
|
|
type=tx_type,
|
|
quantity=quantity,
|
|
unit_price=unit_price,
|
|
fee=fee,
|
|
tax=tax,
|
|
balance_after=balance_after,
|
|
raw_numbers=numeric_cells,
|
|
raw_block=block,
|
|
)
|
|
|
|
|
|
def parse_toss_securities_pdf(file_path: str) -> dict[str, Any]:
|
|
path = Path(file_path).expanduser()
|
|
if not path.exists():
|
|
raise FileNotFoundError(f"Toss Securities 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 Securities PDF is not supported yet")
|
|
all_text = "\n".join(page.get_text("text") for page in doc)
|
|
metadata = _extract_metadata(all_text)
|
|
lines: list[str] = []
|
|
for page in doc:
|
|
lines.extend(_normalize_lines(page.get_text("text")))
|
|
transactions: list[TossSecuritiesTransaction] = []
|
|
for idx, (section, block) in enumerate(_blocks_from_lines(lines), start=1):
|
|
tx = _parse_block(idx, section, block)
|
|
if tx and tx.amount > 0:
|
|
transactions.append(tx)
|
|
validation = {
|
|
"ok": bool(transactions),
|
|
"parsed_rows": len(transactions),
|
|
"errors": [] if transactions else ["no transactions 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: TossSecuritiesTransaction, source_hash: str) -> dict[str, Any]:
|
|
return {
|
|
"transaction_date": tx.transaction_date,
|
|
"type": tx.type,
|
|
"amount": tx.amount,
|
|
"action": tx.action,
|
|
"name": tx.name,
|
|
"code": tx.code,
|
|
"section": tx.section,
|
|
"quantity": tx.quantity,
|
|
"unit_price": tx.unit_price,
|
|
"fee": tx.fee,
|
|
"tax": tx.tax,
|
|
"balance_after": tx.balance_after,
|
|
"external_id": tx.external_id(source_hash),
|
|
}
|