from __future__ import annotations import hashlib import io from dataclasses import dataclass from datetime import datetime from pathlib import Path from typing import Any @dataclass(frozen=True) class SeoulPayTransaction: row_index: int channel: str transaction_no: str transaction_datetime: str transaction_date: str merchant: str business_no: str merchant_id: str transaction_kind: str payment_kind: str amount: int supply_amount: int service_fee: int vat: int tax_status: str @property def type(self) -> str: text = f"{self.transaction_kind} {self.payment_kind}" if any(keyword in text for keyword in ("취소", "환불", "승인취소")): return "refund" return "expense" def external_id(self, source_hash: str) -> str: base = self.transaction_no or f"{self.transaction_datetime}:{self.merchant}:{self.amount}:{self.row_index}" return f"seoulpay_xlsx:{source_hash}:{base}" def raw_payload(self) -> str: import json return json.dumps( { "row_index": self.row_index, "channel": self.channel, "transaction_no": self.transaction_no, "transaction_datetime": self.transaction_datetime, "merchant": self.merchant, "business_no": self.business_no, "merchant_id": self.merchant_id, "transaction_kind": self.transaction_kind, "payment_kind": self.payment_kind, "amount": self.amount, "supply_amount": self.supply_amount, "service_fee": self.service_fee, "vat": self.vat, "tax_status": self.tax_status, }, ensure_ascii=False, ) def _cell_text(value: Any) -> str: if value is None: return "" return str(value).strip() def _amount(value: Any) -> int: if value is None or value == "": return 0 if isinstance(value, (int, float)): return int(round(float(value))) text = str(value).replace(",", "").strip() if not text: return 0 return int(round(float(text))) def _datetime_text(value: Any) -> tuple[str, str]: if isinstance(value, datetime): dt = value.replace(microsecond=0) return dt.isoformat(sep=" "), dt.date().isoformat() text = _cell_text(value) if not text: raise ValueError("결제일 is required") for fmt in ("%Y-%m-%d %H:%M:%S", "%Y.%m.%d %H:%M:%S", "%Y/%m/%d %H:%M:%S"): try: dt = datetime.strptime(text, fmt) return dt.isoformat(sep=" "), dt.date().isoformat() except ValueError: pass # Excel sometimes returns an ISO-ish string; keep date if it is already first 10 chars. if len(text) >= 10: return text, text[:10].replace(".", "-").replace("/", "-") raise ValueError(f"unsupported 결제일 format: {text}") def _load_workbook(file_path: str, password: str = ""): try: import openpyxl except ImportError as e: # pragma: no cover - environment guard raise RuntimeError("openpyxl is required to parse SeoulPay XLSX files") from e path = Path(file_path) data = path.read_bytes() try: return openpyxl.load_workbook(io.BytesIO(data), data_only=True) except Exception as plain_error: if not password: raise ValueError("서울페이 XLSX 파일이 암호화되어 있습니다. password가 필요합니다.") from plain_error try: import msoffcrypto except ImportError as e: # pragma: no cover - environment guard raise RuntimeError("msoffcrypto-tool is required to decrypt password-protected SeoulPay files") from e try: office = msoffcrypto.OfficeFile(io.BytesIO(data)) office.load_key(password=password) decrypted = io.BytesIO() office.decrypt(decrypted) decrypted.seek(0) return openpyxl.load_workbook(decrypted, data_only=True) except Exception as e: raise ValueError(f"서울페이 XLSX 복호화/파싱 실패: {e}") from e def parse_seoulpay_xlsx(file_path: str, password: str = "") -> dict[str, Any]: path = Path(file_path) if not path.exists(): raise FileNotFoundError(str(path)) source_hash = hashlib.sha256(path.read_bytes()).hexdigest() wb = _load_workbook(str(path), password=password) ws = wb[wb.sheetnames[0]] metadata: dict[str, Any] = {"source": "seoulpay_xlsx"} header_row = None headers: list[str] = [] for idx, row in enumerate(ws.iter_rows(values_only=True), start=1): first = _cell_text(row[0] if row else None) second = _cell_text(row[1] if row and len(row) > 1 else None) if first == "성명": metadata["holder_name"] = second elif first == "생년월일": metadata["birth_date_masked"] = "********" if second else "" elif first == "조회기간": metadata["period"] = second if "거래번호" in [_cell_text(v) for v in row] and "결제일" in [_cell_text(v) for v in row]: header_row = idx headers = [_cell_text(v) for v in row] break if header_row is None: raise ValueError("서울페이 이용내역 헤더를 찾을 수 없습니다") index = {name: i for i, name in enumerate(headers) if name} required = ["결제채널", "거래번호", "결제일", "가맹점", "거래유형", "결제구분", "거래금액"] missing = [name for name in required if name not in index] if missing: raise ValueError(f"서울페이 필수 컬럼 누락: {', '.join(missing)}") transactions: list[SeoulPayTransaction] = [] errors: list[str] = [] for row_index in range(header_row + 1, ws.max_row + 1): values = [cell.value for cell in ws[row_index]] if not any(v not in (None, "") for v in values): continue try: tx_dt, tx_date = _datetime_text(values[index["결제일"]]) tx = SeoulPayTransaction( row_index=row_index, channel=_cell_text(values[index["결제채널"]]), transaction_no=_cell_text(values[index["거래번호"]]), transaction_datetime=tx_dt, transaction_date=tx_date, merchant=_cell_text(values[index["가맹점"]]), business_no=_cell_text(values[index.get("사업자번호", -1)]) if "사업자번호" in index else "", merchant_id=_cell_text(values[index.get("가맹점ID", -1)]) if "가맹점ID" in index else "", transaction_kind=_cell_text(values[index["거래유형"]]), payment_kind=_cell_text(values[index["결제구분"]]), amount=_amount(values[index["거래금액"]]), supply_amount=_amount(values[index.get("공급가액", -1)]) if "공급가액" in index else 0, service_fee=_amount(values[index.get("봉사료", -1)]) if "봉사료" in index else 0, vat=_amount(values[index.get("부가세", -1)]) if "부가세" in index else 0, tax_status=_cell_text(values[index.get("과세여부", -1)]) if "과세여부" in index else "", ) if tx.amount <= 0: raise ValueError("거래금액이 0 이하입니다") transactions.append(tx) except Exception as e: errors.append(f"row {row_index}: {e}") return { "file_path": str(path), "source_hash": source_hash, "sheet_name": ws.title, "metadata": metadata, "transactions": transactions, "validation": {"ok": bool(transactions) and not errors, "parsed_rows": len(transactions), "errors": errors}, } def transaction_preview(tx: SeoulPayTransaction, source_hash: str) -> dict[str, Any]: return { "transaction_datetime": tx.transaction_datetime, "transaction_date": tx.transaction_date, "type": tx.type, "channel": tx.channel, "transaction_no": tx.transaction_no, "merchant": tx.merchant, "transaction_kind": tx.transaction_kind, "payment_kind": tx.payment_kind, "amount": tx.amount, "tax_status": tx.tax_status, "external_id": tx.external_id(source_hash), }