king/dbx-main
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(expense+admin): 결재 워크플로 + 첨부 + 집계 + 엑셀 + 사용자 직접 등록

Browse Source 
[admin]
- POST /api/users 사용자 직접 등록 (이메일만으로). UserStore.create_user 추가
- admin.html: 등록 폼 + 삭제 버튼 + 한글 라벨 + 신규 권한 표시
- 신규 권한 키: expense_approver, vacation_approver (APPROVER_KEYS)
- is_approver(user_rec, kind) 헬퍼

[expense 워크플로]
- 상태 전이: 작성중 → 제출 → 승인/반려 → 정산완료
- API: /api/items/{id}/{submit,revert,approve,reject,settle}
- 권한: submit/revert=owner, approve/reject/settle=expense_approver 또는 admin
- expense_items 컬럼: approver_email, decided_at, reject_reason
- 작성중/반려 상태에서만 수정/삭제/첨부 가능

[expense 첨부]
- 영수증/기타 (kind: receipt|other). 최대 20MB, 확장자 화이트리스트
- 저장: DATA_DIR/uploads/expense/{item_id}/{att_id}_{filename}
- expense_attachments 테이블 (CASCADE on item delete)
- API: 업로드/목록/다운로드/삭제

[expense 집계]
- /expense/reports: 연도별 월×카테고리 피벗 표

[expense 엑셀]
- GET /expense/api/export.xlsx?from&to&scope=mine|all
- scope=all 은 승인자/관리자만

[DDL]
- expense_db_init.sql: 신규 컬럼/테이블 포함 (멱등)
- expense_db_002_workflow_attachments.sql: 운영 DB 마이그레이션용

[deps]
- openpyxl>=3.1, python-multipart>=0.0.20

[docs]
- DATABASES.md: 워크플로 다이어그램, attachments 테이블, 마이그레이션
- PROJECT_OVERVIEW.md: 권한 키 표
- DEPLOYMENT.md: DATA_DIR/uploads 안내
This commit is contained in:
king
2026-05-29 02:52:44 +09:00
parent afacc9a7db
commit 550867ba0c
14 changed files with 1492 additions and 142 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/modules/expense/router.py
+504 -8
View File
@@ -2,31 +2,77 @@
- 경로: /expense
- 권한: 로그인 + `expense` 모듈 권한 필요 (관리자는 항상 통과).
- 데이터: ExpenseStore (JSON). app.state.expense_store 에 연결.
- 데이터: ExpenseStore (JSON) 또는 ExpenseDBStore (PostgreSQL).
- 워크플로/첨부/집계/엑셀 내보내기는 DB 모드에서만 완전 지원.
"""
from __future__ import annotations
import io
import mimetypes
import os
import re
import uuid
from datetime import datetime
from pathlib import Path
from typing import Any
from fastapi import APIRouter, Depends, HTTPException, Request
from fastapi.responses import HTMLResponse, JSONResponse, RedirectResponse
from fastapi import (
APIRouter,
Depends,
File,
Form,
HTTPException,
Request,
UploadFile,
)
from fastapi.responses import (
FileResponse,
HTMLResponse,
JSONResponse,
RedirectResponse,
StreamingResponse,
)
from pydantic import BaseModel, Field
from .store import CATEGORIES, METHODS, STATUSES, ExpenseStore
router = APIRouter(prefix="/expense", tags=["expense"])
# 첨부 파일 정책
ALLOWED_EXTS = {
".pdf", ".png", ".jpg", ".jpeg", ".webp", ".gif",
".heic", ".tif", ".tiff",
".doc", ".docx", ".xls", ".xlsx", ".ppt", ".pptx",
".hwp", ".hwpx", ".txt", ".csv", ".zip",
}
MAX_UPLOAD_BYTES = 20 * 1024 * 1024 # 20MB
def _store(request: Request) -> ExpenseStore:
def _store(request: Request) -> Any:
store = getattr(request.app.state, "expense_store", None)
if store is None:
raise RuntimeError("ExpenseStore 가 app.state 에 등록되지 않았습니다.")
return store
def _upload_dir(request: Request) -> Path:
"""첨부 저장 루트. DATA_DIR/uploads/expense — DATA_DIR 와 같은 볼륨에 보관."""
base = getattr(request.app.state, "data_dir", None)
if base is None:
base = Path(os.getenv("DATA_DIR", "/data"))
root = Path(base) / "uploads" / "expense"
root.mkdir(parents=True, exist_ok=True)
return root
def _safe_filename(name: str) -> str:
name = os.path.basename(name or "").strip() or "file"
name = re.sub(r"[\\/:*?\"<>|\r\n\t]", "_", name)
return name[:200]
def _require_user(request: Request) -> dict[str, Any]:
# 순환 import 회피: 핸들러 호출 시점에 main 모듈에서 helper 를 가져온다.
from app.main import get_current_user_record # noqa: WPS433
from app.store import has_module # noqa: WPS433
@@ -38,6 +84,31 @@ def _require_user(request: Request) -> dict[str, Any]:
return user
def _require_approver(request: Request) -> dict[str, Any]:
from app.main import get_current_user_record # noqa: WPS433
from app.store import has_module, 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) or has_module(user, "expense_approver")):
raise HTTPException(status_code=403, detail="개인경비 승인자 권한이 필요합니다.")
return user
def _require_db_store(store: Any) -> None:
"""워크플로/첨부/집계는 DB 모드 전용. JSON 폴백에서는 501."""
if not hasattr(store, "submit"):
raise HTTPException(
status_code=501,
detail="이 기능은 expense_db (PostgreSQL) 모드에서만 지원됩니다. "
"EXPENSE_DB_URL 환경변수를 설정하세요.",
)
# ────────────────────────────────────────────────────────────
# Pydantic 모델
# ────────────────────────────────────────────────────────────
class ExpensePayload(BaseModel):
spent_at: str = Field(..., description="YYYY-MM-DD")
category: str
@@ -48,6 +119,13 @@ class ExpensePayload(BaseModel):
status: str | None = None
class RejectBody(BaseModel):
reason: str
# ────────────────────────────────────────────────────────────
# 페이지
# ────────────────────────────────────────────────────────────
@router.get("/", response_class=HTMLResponse)
async def expense_index(request: Request) -> HTMLResponse:
from app.main import ( # noqa: WPS433
@@ -75,12 +153,19 @@ async def expense_index(request: Request) -> HTMLResponse:
reverse=True,
)
summary = store.summary_for(user["email"])
pending_count = 0
is_approver = is_admin(user) or has_module(user, "expense_approver")
if is_approver and hasattr(store, "list_pending_approval"):
pending_count = len(store.list_pending_approval())
return render_template(
request,
"expense/index.html",
{
"user": user,
"is_admin": is_admin(user),
"is_approver": is_approver,
"pending_count": pending_count,
"items": items,
"summary": summary,
"categories": list(CATEGORIES),
@@ -89,19 +174,110 @@ async def expense_index(request: Request) -> HTMLResponse:
"nav_items": build_erp_nav(user, active="expense"),
"page_title": "개인경비",
"page_subtitle": "법인카드 · 개인지출 · 정산 신청",
"supports_workflow": hasattr(store, "submit"),
},
)
@router.get("/pending", response_class=HTMLResponse)
async def pending_page(request: Request) -> HTMLResponse:
"""승인자 대기열 — 제출 상태 항목 전체."""
from app.main import ( # noqa: WPS433
build_erp_nav,
get_current_user_record,
render_template,
)
from app.store import has_module, 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) or has_module(user, "expense_approver")):
return render_template(
request,
"denied.html",
{"reason": "개인경비 승인자 권한이 없습니다."},
status_code=403,
)
store = _store(request)
_require_db_store(store)
pending = store.list_pending_approval()
return render_template(
request,
"expense/pending.html",
{
"user": user,
"is_admin": is_admin(user),
"items": pending,
"nav_items": build_erp_nav(user, active="expense"),
"page_title": "개인경비 — 승인 대기",
"page_subtitle": f"제출 상태 {len(pending)}건",
},
)
@router.get("/reports", response_class=HTMLResponse)
async def reports_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)
store = _store(request)
_require_db_store(store)
year_str = request.query_params.get("year") or str(datetime.now().year)
try:
year = int(year_str)
except ValueError:
year = datetime.now().year
rows = store.monthly_summary(email=user["email"], year=year)
# 피벗: month → {category → total}
months: dict[str, dict[str, int]] = {}
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}
grand_total = sum(cat_totals.values())
return render_template(
request,
"expense/reports.html",
{
"user": user,
"is_admin": is_admin(user),
"year": year,
"months": month_keys,
"pivot": months,
"categories": list(CATEGORIES),
"cat_totals": cat_totals,
"grand_total": grand_total,
"nav_items": build_erp_nav(user, active="expense"),
"page_title": "개인경비 — 집계",
"page_subtitle": f"{year}년 월별/카테고리별",
},
)
# ────────────────────────────────────────────────────────────
# 항목 CRUD API
# ────────────────────────────────────────────────────────────
@router.get("/api/items")
async def list_items(
request: Request,
user: dict[str, Any] = Depends(_require_user),
) -> JSONResponse:
store = _store(request)
return JSONResponse(
{
"items": _store(request).list_for(user["email"]),
"summary": _store(request).summary_for(user["email"]),
"items": store.list_for(user["email"]),
"summary": store.summary_for(user["email"]),
}
)
@@ -123,8 +299,17 @@ async def update_item(
body: ExpensePayload,
user: dict[str, Any] = Depends(_require_user),
) -> JSONResponse:
store = _store(request)
# 워크플로 모드: 작성중/반려 상태에서만 수정 가능
if hasattr(store, "get"):
current = store.get(item_id=item_id, owner=user["email"])
if current and current.get("status") not in ("작성중", "반려"):
raise HTTPException(
status_code=409,
detail=f"{current['status']} 상태에서는 수정할 수 없습니다.",
)
try:
item = _store(request).update(
item = store.update(
item_id=item_id, owner=user["email"], payload=body.model_dump()
)
except KeyError:
@@ -138,6 +323,14 @@ async def delete_item(
item_id: str,
user: dict[str, Any] = Depends(_require_user),
) -> JSONResponse:
store = _store(request)
if hasattr(store, "get"):
current = store.get(item_id=item_id, owner=user["email"])
if current and current.get("status") not in ("작성중", "반려"):
raise HTTPException(
status_code=409,
detail=f"{current['status']} 상태에서는 삭제할 수 없습니다.",
)
try:
_store(request).delete(item_id=item_id, owner=user["email"])
except KeyError:
@@ -145,6 +338,309 @@ async def delete_item(
return JSONResponse({"ok": True})
# ────────────────────────────────────────────────────────────
# 워크플로 API
# ────────────────────────────────────────────────────────────
@router.post("/api/items/{item_id}/submit")
async def submit_item(
request: Request,
item_id: str,
user: dict[str, Any] = Depends(_require_user),
) -> JSONResponse:
store = _store(request)
_require_db_store(store)
try:
item = store.submit(item_id=item_id, owner=user["email"])
except ValueError as exc:
raise HTTPException(status_code=409, detail=str(exc))
return JSONResponse({"item": item})
@router.post("/api/items/{item_id}/revert")
async def revert_item(
request: Request,
item_id: str,
user: dict[str, Any] = Depends(_require_user),
) -> JSONResponse:
store = _store(request)
_require_db_store(store)
try:
item = store.revert_to_draft(item_id=item_id, owner=user["email"])
except ValueError as exc:
raise HTTPException(status_code=409, detail=str(exc))
return JSONResponse({"item": item})
@router.post("/api/items/{item_id}/approve")
async def approve_item(
request: Request,
item_id: str,
user: dict[str, Any] = Depends(_require_approver),
) -> JSONResponse:
store = _store(request)
_require_db_store(store)
try:
item = store.approve(item_id=item_id, approver_email=user["email"])
except ValueError as exc:
raise HTTPException(status_code=409, detail=str(exc))
return JSONResponse({"item": item})
@router.post("/api/items/{item_id}/reject")
async def reject_item(
request: Request,
item_id: str,
body: RejectBody,
user: dict[str, Any] = Depends(_require_approver),
) -> JSONResponse:
store = _store(request)
_require_db_store(store)
try:
item = store.reject(
item_id=item_id, approver_email=user["email"], reason=body.reason
)
except ValueError as exc:
raise HTTPException(status_code=409, detail=str(exc))
return JSONResponse({"item": item})
@router.post("/api/items/{item_id}/settle")
async def settle_item(
request: Request,
item_id: str,
user: dict[str, Any] = Depends(_require_approver),
) -> JSONResponse:
store = _store(request)
_require_db_store(store)
try:
item = store.settle(item_id=item_id, approver_email=user["email"])
except ValueError as exc:
raise HTTPException(status_code=409, detail=str(exc))
return JSONResponse({"item": item})
# ────────────────────────────────────────────────────────────
# 첨부 API
# ────────────────────────────────────────────────────────────
@router.get("/api/items/{item_id}/attachments")
async def list_attachments(
request: Request,
item_id: str,
user: dict[str, Any] = Depends(_require_user),
) -> JSONResponse:
from app.store import has_module, is_admin # noqa: WPS433
store = _store(request)
_require_db_store(store)
item = store.get_any(item_id=item_id)
if not item:
raise HTTPException(status_code=404, detail="항목 없음")
# 본인 또는 승인자/관리자만 조회
if item["owner"] != user["email"] and not (
is_admin(user) or has_module(user, "expense_approver")
):
raise HTTPException(status_code=403, detail="조회 권한 없음")
return JSONResponse({"attachments": store.list_attachments(item_id=item_id)})
@router.post("/api/items/{item_id}/attachments")
async def upload_attachment(
request: Request,
item_id: str,
kind: str = Form("receipt"),
file: UploadFile = File(...),
user: dict[str, Any] = Depends(_require_user),
) -> JSONResponse:
store = _store(request)
_require_db_store(store)
if kind not in ("receipt", "other"):
raise HTTPException(status_code=400, detail="kind 는 receipt|other")
item = store.get(item_id=item_id, owner=user["email"])
if not item:
raise HTTPException(status_code=404, detail="항목 없음")
if item["status"] not in ("작성중", "반려"):
raise HTTPException(
status_code=409,
detail=f"{item['status']} 상태에서는 첨부할 수 없습니다.",
)
raw_name = _safe_filename(file.filename or "")
ext = os.path.splitext(raw_name)[1].lower()
if ext and ext not in ALLOWED_EXTS:
raise HTTPException(
status_code=415,
detail=f"허용되지 않는 확장자: {ext}",
)
# 디스크에 저장 (스트리밍, 사이즈 제한)
att_id = uuid.uuid4().hex[:12]
sub_dir = _upload_dir(request) / item_id
sub_dir.mkdir(parents=True, exist_ok=True)
stored_name = f"{att_id}_{raw_name}"
stored_path = sub_dir / stored_name
size_total = 0
chunk_size = 1024 * 256
try:
with stored_path.open("wb") as out:
while True:
chunk = await file.read(chunk_size)
if not chunk:
break
size_total += len(chunk)
if size_total > MAX_UPLOAD_BYTES:
out.close()
stored_path.unlink(missing_ok=True)
raise HTTPException(
status_code=413,
detail=f"파일 크기 초과 (최대 {MAX_UPLOAD_BYTES // (1024*1024)}MB)",
)
out.write(chunk)
finally:
await file.close()
content_type = (
file.content_type
or mimetypes.guess_type(raw_name)[0]
or "application/octet-stream"
)
rec = store.add_attachment(
item_id=item_id,
owner=user["email"],
kind=kind,
filename=raw_name,
stored_path=str(stored_path),
content_type=content_type,
size_bytes=size_total,
)
return JSONResponse({"attachment": rec}, status_code=201)
@router.get("/api/attachments/{att_id}")
async def download_attachment(
request: Request,
att_id: str,
user: dict[str, Any] = Depends(_require_user),
) -> FileResponse:
from app.store import has_module, is_admin # noqa: WPS433
store = _store(request)
_require_db_store(store)
rec = store.get_attachment(att_id=att_id)
if not rec:
raise HTTPException(status_code=404, detail="첨부 없음")
item = store.get_any(item_id=rec["item_id"])
if not item:
raise HTTPException(status_code=404, detail="항목 없음")
if item["owner"] != user["email"] and not (
is_admin(user) or has_module(user, "expense_approver")
):
raise HTTPException(status_code=403, detail="조회 권한 없음")
path = Path(rec["stored_path"])
if not path.exists():
raise HTTPException(status_code=410, detail="파일이 사라졌습니다.")
return FileResponse(
path=str(path),
filename=rec["filename"],
media_type=rec["content_type"],
)
@router.delete("/api/attachments/{att_id}")
async def delete_attachment(
request: Request,
att_id: str,
user: dict[str, Any] = Depends(_require_user),
) -> JSONResponse:
store = _store(request)
_require_db_store(store)
try:
rec = store.delete_attachment(att_id=att_id, owner=user["email"])
except KeyError:
raise HTTPException(status_code=404, detail="첨부 없음")
# 파일 정리 (실패해도 DB 삭제는 유효 — 고아 파일 남을 수 있으니 로그만)
try:
Path(rec["stored_path"]).unlink(missing_ok=True)
except OSError:
pass
return JSONResponse({"ok": True})
# ────────────────────────────────────────────────────────────
# 엑셀 내보내기
# ────────────────────────────────────────────────────────────
@router.get("/api/export.xlsx")
async def export_xlsx(
request: Request,
user: dict[str, Any] = Depends(_require_user),
) -> StreamingResponse:
from app.store import has_module, is_admin # noqa: WPS433
from openpyxl import Workbook # 지연 import
store = _store(request)
_require_db_store(store)
date_from = request.query_params.get("from")
date_to = request.query_params.get("to")
scope = request.query_params.get("scope", "mine")
target_email: str | None = user["email"]
if scope == "all":
if not (is_admin(user) or has_module(user, "expense_approver")):
raise HTTPException(status_code=403, detail="전체 내보내기 권한 없음")
target_email = None
rows = store.list_for_export(
email=target_email, date_from=date_from, date_to=date_to
)
wb = Workbook()
ws = wb.active
ws.title = "expense"
header = [
"사용일", "소유자", "분류", "결제수단", "가맹점",
"금액", "메모", "상태", "승인자", "결재일시", "반려사유",
"생성일시", "수정일시",
]
ws.append(header)
for r in rows:
ws.append([
r.get("spent_at", ""),
r.get("owner", ""),
r.get("category", ""),
r.get("method", ""),
r.get("merchant", ""),
int(r.get("amount", 0)),
r.get("memo", ""),
r.get("status", ""),
r.get("approver_email", "") or "",
r.get("decided_at", "") or "",
r.get("reject_reason", "") or "",
r.get("created_at", ""),
r.get("updated_at", ""),
])
# 컬럼 너비 대충
widths = [12, 28, 10, 12, 24, 12, 30, 10, 24, 22, 24, 22, 22]
for col, w in enumerate(widths, start=1):
ws.column_dimensions[ws.cell(row=1, column=col).column_letter].width = w
buf = io.BytesIO()
wb.save(buf)
buf.seek(0)
fname_scope = "all" if target_email is None else "mine"
fname = f"expense_{fname_scope}_{datetime.now().strftime('%Y%m%d_%H%M%S')}.xlsx"
return StreamingResponse(
buf,
media_type="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
headers={"Content-Disposition": f'attachment; filename="{fname}"'},
)
@router.get("/health")
async def health() -> dict[str, str]:
return {"status": "ok", "module": "expense"}