diff --git a/app/main.py b/app/main.py index 2462a67..f869e93 100644 --- a/app/main.py +++ b/app/main.py @@ -15,7 +15,7 @@ from starlette.middleware.sessions import SessionMiddleware from .modules.cupang import build_cupang_store, build_itemcode_reader from .modules.cupang import router as cupang_router -from .modules.expense import build_expense_store +from .modules.expense import CategoryStore, build_expense_store from .modules.expense import router as expense_router from .modules.vacation import build_vacation_store from .modules.vacation import router as vacation_router @@ -127,6 +127,8 @@ app.state.expense_store = build_expense_store( dsn=env("EXPENSE_DB_URL") or None, json_path=DATA_DIR / "expense.json", ) +# 분류(category) 설정 — 관리자가 추가/삭제, 저장 즉시 반영. 항상 JSON 파일. +app.state.expense_category_store = CategoryStore(DATA_DIR / "expense_categories.json") # 쿠팡 밀크런: CUPANG_DB_URL 없으면 store=None(라우터가 "설정 필요" 안내). # 상품 검색은 itemcode_db 읽기 전용(미설정 시 수동 입력 폴백). app.state.cupang_store = build_cupang_store(dsn=env("CUPANG_DB_URL") or None) diff --git a/app/modules/expense/__init__.py b/app/modules/expense/__init__.py index 5d502a8..0fe20ed 100644 --- a/app/modules/expense/__init__.py +++ b/app/modules/expense/__init__.py @@ -9,12 +9,15 @@ from pathlib import Path from typing import Any +from .categories import DEFAULT_CATEGORIES, CategoryStore from .router import router from .store import CATEGORIES, METHODS, STATUSES, ExpenseStore __all__ = [ "router", "ExpenseStore", + "CategoryStore", + "DEFAULT_CATEGORIES", "CATEGORIES", "METHODS", "STATUSES", diff --git a/app/modules/expense/categories.py b/app/modules/expense/categories.py new file mode 100644 index 0000000..c3b6f10 --- /dev/null +++ b/app/modules/expense/categories.py @@ -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"]) diff --git a/app/modules/expense/db.py b/app/modules/expense/db.py index 1cace89..6fed8d8 100644 --- a/app/modules/expense/db.py +++ b/app/modules/expense/db.py @@ -482,7 +482,8 @@ class ExpenseDBStore: 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 if category in CATEGORIES else "기타" + # 분류는 관리자 설정으로 동적 추가되므로 고정 목록 검증 없이 그대로 저장. + 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: diff --git a/app/modules/expense/router.py b/app/modules/expense/router.py index 2f879ce..8c118b1 100644 --- a/app/modules/expense/router.py +++ b/app/modules/expense/router.py @@ -36,7 +36,7 @@ from fastapi.responses import ( ) from pydantic import BaseModel, Field -from .store import CATEGORIES, METHODS, STATUSES, ExpenseStore +from .store import METHODS, STATUSES, ExpenseStore router = APIRouter(prefix="/expense", tags=["expense"]) @@ -57,6 +57,18 @@ def _store(request: Request) -> Any: return store +def _category_store(request: Request) -> Any: + store = getattr(request.app.state, "expense_category_store", None) + if store is None: + raise RuntimeError("CategoryStore 가 app.state 에 등록되지 않았습니다.") + return store + + +def _categories(request: Request) -> list[str]: + """현재 활성 분류 목록 — 등록 폼/집계에서 사용.""" + return _category_store(request).list() + + def _upload_dir(request: Request) -> Path: """첨부 저장 루트. DATA_DIR/uploads/expense — DATA_DIR 와 같은 볼륨에 보관.""" base = getattr(request.app.state, "data_dir", None) @@ -97,6 +109,18 @@ def _require_approver(request: Request) -> dict[str, Any]: return user +def _require_admin(request: Request) -> dict[str, Any]: + from app.main import get_current_user_record # noqa: WPS433 + from app.store import is_admin # noqa: WPS433 + + user = get_current_user_record(request) + if user is None: + raise HTTPException(status_code=401, detail="로그인이 필요합니다.") + if not is_admin(user): + raise HTTPException(status_code=403, detail="관리자 권한이 필요합니다.") + return user + + def _require_db_store(store: Any) -> None: """워크플로/첨부/집계는 DB 모드 전용. JSON 폴백에서는 501.""" if not hasattr(store, "submit"): @@ -124,6 +148,10 @@ class RejectBody(BaseModel): reason: str +class CategoryBody(BaseModel): + name: str + + # ──────────────────────────────────────────────────────────── # 페이지 # ──────────────────────────────────────────────────────────── @@ -169,7 +197,7 @@ async def expense_index(request: Request) -> HTMLResponse: "pending_count": pending_count, "items": items, "summary": summary, - "categories": list(CATEGORIES), + "categories": _categories(request), "methods": list(METHODS), "statuses": list(STATUSES), "nav_items": build_erp_nav(user, active="expense"), @@ -244,7 +272,13 @@ async def reports_page(request: Request) -> HTMLResponse: for r in rows: months.setdefault(r["month"], {})[r["category"]] = r["total"] month_keys = sorted(months.keys()) - cat_totals = {c: sum(months[m].get(c, 0) for m in month_keys) for c in CATEGORIES} + # 활성 분류 + 데이터에만 있는(삭제된) 분류 모두 포함 + categories = list(_categories(request)) + for m in month_keys: + for c in months[m]: + if c not in categories: + categories.append(c) + cat_totals = {c: sum(months[m].get(c, 0) for m in month_keys) for c in categories} grand_total = sum(cat_totals.values()) return render_template( @@ -256,7 +290,7 @@ async def reports_page(request: Request) -> HTMLResponse: "year": year, "months": month_keys, "pivot": months, - "categories": list(CATEGORIES), + "categories": categories, "cat_totals": cat_totals, "grand_total": grand_total, "nav_items": build_erp_nav(user, active="expense"), @@ -266,6 +300,79 @@ async def reports_page(request: Request) -> HTMLResponse: ) +# ──────────────────────────────────────────────────────────── +# 관리자 설정 — 분류 관리 +# ──────────────────────────────────────────────────────────── +@router.get("/settings", response_class=HTMLResponse) +async def settings_page(request: Request) -> HTMLResponse: + """관리자 전용 설정 — 분류 추가/삭제.""" + from app.main import ( # noqa: WPS433 + build_erp_nav, + get_current_user_record, + render_template, + ) + from app.store import is_admin # noqa: WPS433 + + user = get_current_user_record(request) + if user is None: + return RedirectResponse(url="/login", status_code=303) + if not is_admin(user): + return render_template( + request, + "denied.html", + {"reason": "개인경비 설정은 관리자만 접근할 수 있습니다."}, + status_code=403, + ) + return render_template( + request, + "expense/settings.html", + { + "user": user, + "is_admin": True, + "categories": _categories(request), + "nav_items": build_erp_nav(user, active="expense"), + "page_title": "개인경비 — 설정", + "page_subtitle": "분류 항목 관리", + }, + ) + + +@router.get("/api/categories") +async def list_categories( + request: Request, + user: dict[str, Any] = Depends(_require_user), +) -> JSONResponse: + return JSONResponse({"categories": _categories(request)}) + + +@router.post("/api/categories") +async def add_category( + request: Request, + body: CategoryBody, + user: dict[str, Any] = Depends(_require_admin), +) -> JSONResponse: + try: + categories = _category_store(request).add(body.name) + except ValueError as exc: + raise HTTPException(status_code=400, detail=str(exc)) + return JSONResponse({"categories": categories}, status_code=201) + + +@router.delete("/api/categories/{name}") +async def delete_category( + request: Request, + name: str, + user: dict[str, Any] = Depends(_require_admin), +) -> JSONResponse: + try: + categories = _category_store(request).delete(name) + except KeyError: + raise HTTPException(status_code=404, detail="분류를 찾을 수 없습니다.") + except ValueError as exc: + raise HTTPException(status_code=400, detail=str(exc)) + return JSONResponse({"categories": categories}) + + # ──────────────────────────────────────────────────────────── # 항목 CRUD API # ──────────────────────────────────────────────────────────── diff --git a/app/modules/expense/store.py b/app/modules/expense/store.py index 3317c43..b2af956 100644 --- a/app/modules/expense/store.py +++ b/app/modules/expense/store.py @@ -149,7 +149,8 @@ class ExpenseStore: 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 if category in CATEGORIES else "기타" + # 분류는 관리자 설정으로 동적 추가되므로 고정 목록 검증 없이 그대로 저장. + 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: diff --git a/app/modules/expense/templates/expense/index.html b/app/modules/expense/templates/expense/index.html index 54ddcef..2cbb1d3 100644 --- a/app/modules/expense/templates/expense/index.html +++ b/app/modules/expense/templates/expense/index.html @@ -15,6 +15,9 @@ {% if is_approver %} 엑셀(전체) {% endif %} + {% if is_admin %} + ⚙ 설정 + {% endif %} diff --git a/app/modules/expense/templates/expense/settings.html b/app/modules/expense/templates/expense/settings.html new file mode 100644 index 0000000..1e04146 --- /dev/null +++ b/app/modules/expense/templates/expense/settings.html @@ -0,0 +1,119 @@ +{% extends "erp_base.html" %} + +{% block content %} +
+ +
+ ← 개인경비 +
+ +
+
+

분류 항목 관리

+ 추가/삭제 즉시 사용자 등록 폼에 반영됩니다. +
+ +
+ +
+ +
+
+ +
+ + + + + + + + + {% for c in categories %} + + + + + {% endfor %} + +
분류명동작
{{ c }} + +
+
+

+ 삭제해도 이미 등록된 경비 항목의 분류는 그대로 유지됩니다. 분류는 최소 1개 이상이어야 합니다. +

+
+ +
+ + +{% endblock %} + +{% block scripts %} + +{% endblock %}