208 lines
7.5 KiB
Python
208 lines
7.5 KiB
Python
from __future__ import annotations
|
|
|
|
import csv as csv
|
|
import sqlite3 as sqlite3
|
|
import hashlib
|
|
import json
|
|
import re
|
|
from datetime import date, datetime, timedelta
|
|
from pathlib import Path
|
|
from typing import Any
|
|
from uuid import uuid4
|
|
from zoneinfo import ZoneInfo
|
|
|
|
from ..infrastructure.parsers import (
|
|
parse_kb_bank_pdf as parse_kb_bank_pdf,
|
|
parse_seoulpay_xlsx as parse_seoulpay_xlsx,
|
|
parse_toss_bank_pdf as parse_toss_bank_pdf,
|
|
parse_toss_securities_pdf as parse_toss_securities_pdf,
|
|
seoulpay_transaction_preview as seoulpay_transaction_preview,
|
|
toss_bank_transaction_preview as toss_bank_transaction_preview,
|
|
toss_transaction_preview as toss_transaction_preview,
|
|
transaction_preview as transaction_preview,
|
|
)
|
|
from ..infrastructure.sqlite import db as db
|
|
from ..infrastructure.sqlite import init_db as init_db
|
|
from ..infrastructure.sqlite import resolve_db_path as resolve_db_path
|
|
|
|
VALID_TRANSACTION_TYPES = {"income", "expense", "transfer", "refund", "adjustment"}
|
|
VALID_ACCOUNT_TYPES = {"bank", "card", "cash", "investment", "loan", "other"}
|
|
VALID_STATEMENT_STATUSES = {"unknown", "import_supported", "manual_only", "partial_import"}
|
|
VALID_LIQUIDITY_BUCKETS = {"cash", "investment", "deposit", "real_estate", "loan", "other"}
|
|
ASSET_BUCKETS = ["cash", "investment", "deposit", "real_estate", "other"]
|
|
LIABILITY_BUCKETS = ["loan"]
|
|
ASSET_BUCKET_ORDER = [*ASSET_BUCKETS, *LIABILITY_BUCKETS]
|
|
RECONCILIATION_TARGET_STATUSES = {"manual_only", "partial_import"}
|
|
BALANCE_RECONCILIATION_TOLERANCE_KRW = 1_000
|
|
BALANCE_RECONCILIATION_REFERENCE_RATIO = 0.001
|
|
VALID_CASHFLOW_KINDS = {"income", "expense", "transfer", "investment", "debt_payment", "exclude", "unknown"}
|
|
CASHFLOW_KIND_SQL = """COALESCE(lc.cashflow_kind, CASE
|
|
WHEN t.category='수입' THEN 'income'
|
|
WHEN t.category='이체' AND t.subcategory='투자이동' THEN 'investment'
|
|
WHEN t.category='이체' THEN 'transfer'
|
|
WHEN t.category='금융' AND t.subcategory='대출상환' THEN 'debt_payment'
|
|
WHEN t.category='금융' THEN 'exclude'
|
|
WHEN t.category='미분류' THEN 'unknown'
|
|
WHEN COALESCE(t.category, '')='' THEN 'unknown'
|
|
ELSE 'expense'
|
|
END)"""
|
|
|
|
CROSS_TRANSFER_MATCH_TAG = "cross_transfer_match:"
|
|
CROSS_TRANSFER_REVIEW_TAG = "cross_transfer_review:"
|
|
DEFAULT_LEDGER_BACKUP_DIR = Path.home() / ".local" / "share" / "jarca" / "backups" / "ledger"
|
|
KST = ZoneInfo("Asia/Seoul")
|
|
BACKUP_RESTORE_POLICY = {
|
|
"mcp_restore_exposed": False,
|
|
"restore_requires_explicit_user_approval": True,
|
|
"restore_entrypoint": "scripts/ledger_restore_smoke.py 또는 docs/runbooks/ledger-backup-restore.md",
|
|
}
|
|
|
|
|
|
def _normalize_counterparty_for_match(value: str) -> str:
|
|
if not value:
|
|
return ""
|
|
# normalize party names for fuzzy matching across KB/토스뱅크 records
|
|
normalized = re.sub(r"\\([^)]*\\)", "", value)
|
|
normalized = normalized.lower().strip()
|
|
normalized = normalized.replace("토스뱅크", "").replace("토스", "")
|
|
normalized = normalized.replace("kb", "").replace("국민은행", "")
|
|
normalized = re.sub(r"\\s+", "", normalized)
|
|
normalized = re.sub(r"[^0-9가-힣a-zA-Z]", "", normalized)
|
|
return normalized
|
|
|
|
|
|
def _party_match_score(left: str, right: str) -> int:
|
|
left_norm = _normalize_counterparty_for_match(left)
|
|
right_norm = _normalize_counterparty_for_match(right)
|
|
if not left_norm or not right_norm:
|
|
return 0
|
|
if left_norm == right_norm:
|
|
return 2
|
|
if left_norm in right_norm or right_norm in left_norm:
|
|
return 1
|
|
return 0
|
|
|
|
|
|
def _append_memo_tag(raw_memo: str, tag: str) -> str:
|
|
memo = (raw_memo or "").strip()
|
|
if not memo:
|
|
return tag
|
|
if tag in memo:
|
|
return memo
|
|
return f"{memo}; {tag}"
|
|
|
|
|
|
def _memo_has_tag(raw_memo: str, tag_prefix: str) -> bool:
|
|
memo = (raw_memo or "").strip()
|
|
return tag_prefix in memo
|
|
|
|
|
|
def _current_kst_datetime() -> datetime:
|
|
return datetime.now(KST).replace(microsecond=0)
|
|
|
|
|
|
def current_kst_date() -> date:
|
|
return _current_kst_datetime().date()
|
|
|
|
|
|
def current_kst_month() -> str:
|
|
return current_kst_date().strftime("%Y-%m")
|
|
|
|
|
|
def now_iso() -> str:
|
|
return _current_kst_datetime().isoformat()
|
|
|
|
def new_id(prefix: str) -> str:
|
|
return f"{prefix}_{uuid4().hex[:12]}"
|
|
|
|
def ok(**data: Any) -> dict[str, Any]:
|
|
return {"success": True, **data}
|
|
|
|
def fail(error_code: str, message: str, **data: Any) -> dict[str, Any]:
|
|
return {"success": False, "error_code": error_code, "error": message, **data}
|
|
|
|
def row_to_dict(row: Any) -> dict[str, Any]:
|
|
d = dict(row)
|
|
if "tags" in d:
|
|
try:
|
|
d["tags"] = json.loads(d["tags"] or "[]")
|
|
except json.JSONDecodeError:
|
|
d["tags"] = []
|
|
for k in ("active", "is_fixed", "is_correction"):
|
|
if k in d:
|
|
d[k] = bool(d[k])
|
|
return d
|
|
|
|
def normalize_date(value: str | None, default_today: bool = False) -> str:
|
|
if not value:
|
|
if default_today:
|
|
return current_kst_date().isoformat()
|
|
raise ValueError("date is required")
|
|
return date.fromisoformat(value[:10]).isoformat()
|
|
|
|
def normalize_month(value: str | None) -> str:
|
|
if not value:
|
|
return current_kst_month()
|
|
if re.fullmatch(r"\d{4}-\d{2}", value):
|
|
return value
|
|
return date.fromisoformat(value[:10]).strftime("%Y-%m")
|
|
|
|
|
|
def normalize_statement_status(value: str | None) -> str:
|
|
status = (value or "unknown").strip() or "unknown"
|
|
if status not in VALID_STATEMENT_STATUSES:
|
|
raise ValueError(f"statement_status must be one of {sorted(VALID_STATEMENT_STATUSES)}")
|
|
return status
|
|
|
|
|
|
def normalize_liquidity_bucket(value: str | None) -> str:
|
|
bucket = (value or "other").strip() or "other"
|
|
if bucket not in VALID_LIQUIDITY_BUCKETS:
|
|
raise ValueError(f"liquidity_bucket must be one of {sorted(VALID_LIQUIDITY_BUCKETS)}")
|
|
return bucket
|
|
|
|
|
|
def normalize_snapshot_at(value: str | None, default_now: bool = True) -> str:
|
|
if not value:
|
|
if default_now:
|
|
return now_iso()
|
|
raise ValueError("baseline_at is required")
|
|
raw = value.strip()
|
|
if re.fullmatch(r"\d{4}-\d{2}", raw):
|
|
month_start = date.fromisoformat(raw + "-01")
|
|
if month_start.month == 12:
|
|
next_month = date(month_start.year + 1, 1, 1)
|
|
else:
|
|
next_month = date(month_start.year, month_start.month + 1, 1)
|
|
last_day = next_month - timedelta(days=1)
|
|
return f"{last_day.isoformat()}T23:59:00+09:00"
|
|
if "T" not in raw and len(raw) == 10:
|
|
return f"{normalize_date(raw)}T23:59:00+09:00"
|
|
datetime.fromisoformat(raw.replace("Z", "+00:00"))
|
|
return raw
|
|
|
|
def normalize_tags(tags: Any) -> str:
|
|
if tags is None or tags == "":
|
|
return "[]"
|
|
if isinstance(tags, str):
|
|
parts = [p.strip() for p in re.split(r"[,#]", tags) if p.strip()]
|
|
return json.dumps(parts, ensure_ascii=False)
|
|
if isinstance(tags, list):
|
|
return json.dumps([str(x).strip() for x in tags if str(x).strip()], ensure_ascii=False)
|
|
raise ValueError("tags must be a list or comma-separated string")
|
|
|
|
|
|
def _month_range(month: str) -> tuple[str, str]:
|
|
m = normalize_month(month)
|
|
start = date.fromisoformat(f"{m}-01")
|
|
next_month = date(start.year + (start.month // 12), ((start.month % 12) + 1), 1)
|
|
return start.isoformat(), next_month.isoformat()
|
|
|
|
|
|
def make_dedupe_key(transaction_date: str, amount: int, merchant: str = "", account_id: str = "", external_id: str = "") -> str:
|
|
if external_id:
|
|
raw = f"external:{external_id}"
|
|
else:
|
|
raw = f"{transaction_date}|{amount}|{merchant.strip().lower()}|{account_id.strip().lower()}"
|
|
return hashlib.sha256(raw.encode("utf-8")).hexdigest()
|