dbx-main/app/modules/expense/store.py

"""개인경비 항목 JSON 저장소.

- 저장 위치: DATA_DIR/expense.json
- 사용자(email) 단위 소유. 본인 항목만 조회/수정/삭제.
- 향후 expense_db(PostgreSQL)로 마이그레이션 예정. 현재는 DB 생성 승인 전이라 JSON 사용.
- 동시성: 프로세스 내 threading.Lock + 원자적 쓰기(temp → rename). UserStore 와 동일 패턴.
"""

from __future__ import annotations

import json
import os
import tempfile
import threading
import uuid
from pathlib import Path
from typing import Any

from app.timezone import now_kst_iso

CATEGORIES: tuple[str, ...] = ("식대", "교통", "숙박", "비품", "접대", "통신", "기타")
METHODS: tuple[str, ...] = ("법인카드", "개인지출", "현금")
STATUSES: tuple[str, ...] = ("작성중", "제출", "승인", "반려", "정산완료")
# 승인 완료(결재 승인 이후) 상태 — "승인완료" 집계/내보내기 대상.
APPROVED_STATUSES: tuple[str, ...] = ("승인", "정산완료")


def _now_iso() -> str:
    return now_kst_iso()


class ExpenseStore:
    def __init__(self, path: Path):
        self._path = path
        self._lock = threading.Lock()
        self._path.parent.mkdir(parents=True, exist_ok=True)
        if not self._path.exists():
            self._write_atomic({"items": []})

    def _read(self) -> dict[str, Any]:
        try:
            with self._path.open("r", encoding="utf-8") as f:
                data = json.load(f)
        except (FileNotFoundError, json.JSONDecodeError):
            data = {"items": []}
        if not isinstance(data.get("items"), list):
            data["items"] = []
        return data

    def _write_atomic(self, data: dict[str, Any]) -> None:
        fd, tmp = tempfile.mkstemp(
            prefix=".expense.", suffix=".json.tmp", dir=str(self._path.parent)
        )
        try:
            with os.fdopen(fd, "w", encoding="utf-8") as f:
                json.dump(data, f, ensure_ascii=False, indent=2)
            os.replace(tmp, self._path)
        except Exception:
            try:
                os.unlink(tmp)
            except OSError:
                pass
            raise

    def list_for(self, email: str) -> list[dict[str, Any]]:
        email = email.lower().strip()
        with self._lock:
            data = self._read()
            return [it for it in data["items"] if it.get("owner") == email]

    def list_all(self) -> list[dict[str, Any]]:
        with self._lock:
            return list(self._read()["items"])

    def get(self, *, item_id: str, owner: str) -> dict[str, Any] | None:
        owner = owner.lower().strip()
        with self._lock:
            data = self._read()
            for it in data["items"]:
                if it["id"] == item_id and it.get("owner") == owner:
                    return dict(it)
        return None

    def create(self, *, owner: str, payload: dict[str, Any]) -> dict[str, Any]:
        owner = owner.lower().strip()
        item = self._normalize(payload)
        item["id"] = uuid.uuid4().hex[:12]
        item["owner"] = owner
        item["status"] = payload.get("status") or "작성중"
        item["created_at"] = _now_iso()
        item["updated_at"] = item["created_at"]
        with self._lock:
            data = self._read()
            data["items"].append(item)
            self._write_atomic(data)
        return item

    def update(
        self, *, item_id: str, owner: str, payload: dict[str, Any]
    ) -> dict[str, Any]:
        owner = owner.lower().strip()
        with self._lock:
            data = self._read()
            for idx, it in enumerate(data["items"]):
                if it["id"] == item_id and it.get("owner") == owner:
                    new = self._normalize(payload, base=it)
                    new["id"] = it["id"]
                    new["owner"] = it["owner"]
                    new["created_at"] = it.get("created_at", _now_iso())
                    new["status"] = payload.get("status") or it.get("status", "작성중")
                    new["updated_at"] = _now_iso()
                    data["items"][idx] = new
                    self._write_atomic(data)
                    return new
        raise KeyError(item_id)

    def delete(self, *, item_id: str, owner: str) -> None:
        owner = owner.lower().strip()
        with self._lock:
            data = self._read()
            before = len(data["items"])
            data["items"] = [
                it
                for it in data["items"]
                if not (it["id"] == item_id and it.get("owner") == owner)
            ]
            if len(data["items"]) == before:
                raise KeyError(item_id)
            self._write_atomic(data)

    def summary_for(self, email: str) -> dict[str, Any]:
        items = self.list_for(email)
        total = sum(int(i.get("amount", 0)) for i in items)
        by_status = {s: sum(1 for i in items if i.get("status") == s) for s in STATUSES}
        by_category = {
            c: sum(int(i.get("amount", 0)) for i in items if i.get("category") == c)
            for c in CATEGORIES
        }
        return {
            "count": len(items),
            "total": total,
            "by_status": by_status,
            "by_category": by_category,
        }

    @staticmethod
    def _normalize(
        payload: dict[str, Any], base: dict[str, Any] | None = None
    ) -> dict[str, Any]:
        b = dict(base or {})
        b["spent_at"] = str(payload.get("spent_at") or b.get("spent_at") or "").strip()
        category = str(payload.get("category") or b.get("category") or "기타").strip()
        method = str(payload.get("method") or b.get("method") or "법인카드").strip()
        # 분류는 관리자 설정으로 동적 추가되므로 고정 목록 검증 없이 그대로 저장.
        b["category"] = category or "기타"
        b["method"] = method if method in METHODS else "법인카드"
        b["merchant"] = str(payload.get("merchant") or b.get("merchant") or "").strip()
        try:
            b["amount"] = max(0, int(payload.get("amount") or b.get("amount") or 0))
        except (TypeError, ValueError):
            b["amount"] = 0
        b["memo"] = str(payload.get("memo") or b.get("memo") or "").strip()
        return b