jarca-ledger-mcp/jarca_ledger_mcp/application/shared.py

from __future__ import annotations

import csv
import hashlib
import json
import re
import sqlite3
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,
    parse_seoulpay_xlsx,
    parse_toss_bank_pdf,
    parse_toss_securities_pdf,
    seoulpay_transaction_preview,
    toss_bank_transaction_preview,
    toss_transaction_preview,
    transaction_preview,
)
from ..infrastructure.sqlite import db, init_db, 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()