216 lines
8.3 KiB
Python
216 lines
8.3 KiB
Python
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),
|
|
}
|