king
0b04a05b26
- CategoryStore: 분류 항목 JSON 저장소(DATA_DIR/expense_categories.json) - 관리자 설정 페이지(/expense/settings) + 분류 CRUD API - _normalize 분류 고정목록 검증 제거 — 추가 분류 즉시 등록 폼/집계 반영 - index/reports 분류를 동적 목록으로 교체 Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
99 lines
3.5 KiB
Python
99 lines
3.5 KiB
Python
"""개인경비 분류(category) 설정 저장소.
|
|
|
|
- 저장 위치: DATA_DIR/expense_categories.json
|
|
- 관리자만 추가/삭제. 저장 즉시 사용자 등록 폼/집계에 반영.
|
|
- JSON 파일 기반 — expense_db(PostgreSQL) 모드와 무관하게 동작(설정값이라
|
|
트랜잭션 데이터와 분리). DB 스키마 변경(superuser SQL) 불필요.
|
|
- 동시성: ExpenseStore 와 동일 패턴(threading.Lock + temp→rename 원자적 쓰기).
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
import tempfile
|
|
import threading
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
# 최초 1회 시드. 기존 하드코딩 CATEGORIES 와 동일.
|
|
DEFAULT_CATEGORIES: tuple[str, ...] = (
|
|
"식대", "교통", "숙박", "비품", "접대", "통신", "기타",
|
|
)
|
|
|
|
|
|
class CategoryStore:
|
|
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({"categories": list(DEFAULT_CATEGORIES)})
|
|
|
|
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 = {}
|
|
cats = data.get("categories")
|
|
if not isinstance(cats, list) or not cats:
|
|
data["categories"] = list(DEFAULT_CATEGORIES)
|
|
else:
|
|
# 문자열만, 공백 제거, 중복 제거(순서 유지)
|
|
seen: set[str] = set()
|
|
clean: list[str] = []
|
|
for c in cats:
|
|
name = str(c).strip()
|
|
if name and name not in seen:
|
|
seen.add(name)
|
|
clean.append(name)
|
|
data["categories"] = clean or list(DEFAULT_CATEGORIES)
|
|
return data
|
|
|
|
def _write_atomic(self, data: dict[str, Any]) -> None:
|
|
fd, tmp = tempfile.mkstemp(
|
|
prefix=".expense_categories.", 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(self) -> list[str]:
|
|
with self._lock:
|
|
return list(self._read()["categories"])
|
|
|
|
def add(self, name: str) -> list[str]:
|
|
name = str(name or "").strip()
|
|
if not name:
|
|
raise ValueError("분류명을 입력하세요.")
|
|
if len(name) > 30:
|
|
raise ValueError("분류명은 30자 이하로 입력하세요.")
|
|
with self._lock:
|
|
data = self._read()
|
|
if name in data["categories"]:
|
|
raise ValueError(f"이미 존재하는 분류입니다: {name}")
|
|
data["categories"].append(name)
|
|
self._write_atomic(data)
|
|
return list(data["categories"])
|
|
|
|
def delete(self, name: str) -> list[str]:
|
|
name = str(name or "").strip()
|
|
with self._lock:
|
|
data = self._read()
|
|
if name not in data["categories"]:
|
|
raise KeyError(name)
|
|
if len(data["categories"]) <= 1:
|
|
raise ValueError("분류는 최소 1개 이상이어야 합니다.")
|
|
data["categories"] = [c for c in data["categories"] if c != name]
|
|
self._write_atomic(data)
|
|
return list(data["categories"])
|