feat(expense): 관리자 분류 설정 메뉴 추가 (동적 분류)
- CategoryStore: 분류 항목 JSON 저장소(DATA_DIR/expense_categories.json) - 관리자 설정 페이지(/expense/settings) + 분류 CRUD API - _normalize 분류 고정목록 검증 제거 — 추가 분류 즉시 등록 폼/집계 반영 - index/reports 분류를 동적 목록으로 교체 Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
@@ -0,0 +1,98 @@
|
||||
"""개인경비 분류(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"])
|
||||
Reference in New Issue
Block a user