feat(expense): 승인완료 메뉴 — 전 직원 월별 + 엑셀/첨부 zip

- /expense/approved: 승인자/관리자 전용, 월 선택 + 전 직원 승인완료 항목 - 직원별 승인 금액 합계표 + 총합 - 엑셀 다운(리스트 + 직원별합계 시트) - 첨부 일괄 zip: 파일명 yy년 mm월 dd일(요일)_이름.확장자 (업로드일 기준) - store: APPROVED_STATUSES(승인,정산완료) / db: list_approved, approved_attachments Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-06-02 17:46:41 +09:00
parent 506db5ff07
commit a107b06826
5 changed files with 396 additions and 2 deletions
@@ -13,10 +13,12 @@ import mimetypes
 import os
 import re
 import uuid
+import zipfile
+from datetime import datetime
 from pathlib import Path
 from typing import Any

-from app.timezone import now_kst
+from app.timezone import KST, now_kst

 from fastapi import (
    APIRouter,
@@ -69,6 +71,52 @@ def _categories(request: Request) -> list[str]:
    return _category_store(request).list()


+_WEEKDAYS_KR = ("월", "화", "수", "목", "금", "토", "일")
+
+
+def _parse_month(request: Request) -> tuple[int, int, str]:
+    """?month=YYYY-MM 파싱. 누락/오류 시 이번 달(KST)."""
+    raw = request.query_params.get("month") or ""
+    try:
+        y_str, m_str = raw.split("-")
+        year, mon = int(y_str), int(m_str)
+        if not (1 <= mon <= 12):
+            raise ValueError
+    except (ValueError, AttributeError):
+        n = now_kst()
+        year, mon = n.year, n.month
+    return year, mon, f"{year:04d}-{mon:02d}"
+
+
+def _name_map(request: Request) -> dict[str, str]:
+    """email(소문자) → 표시 이름. 미등록 사용자는 이메일 로컬파트."""
+    from app.main import user_store  # noqa: WPS433
+
+    out: dict[str, str] = {}
+    for u in user_store.list_all():
+        email = str(u.get("email", "")).lower().strip()
+        if email:
+            out[email] = u.get("name") or email.split("@")[0]
+    return out
+
+
+def _display_name(email: str, names: dict[str, str]) -> str:
+    email = (email or "").lower().strip()
+    return names.get(email) or (email.split("@")[0] if email else "")
+
+
+def _att_download_name(uploaded_at: Any, name: str, original: str) -> str:
+    """첨부 다운로드 파일명: `yy년 mm월 dd일(요일)_이름{ext}`. 날짜=업로드일(KST)."""
+    if isinstance(uploaded_at, datetime):
+        d = uploaded_at.astimezone(KST)
+    else:
+        d = now_kst()
+    wd = _WEEKDAYS_KR[d.weekday()]
+    base = f"{d.strftime('%y')}년 {d.strftime('%m')}월 {d.strftime('%d')}일({wd})_{name}"
+    ext = os.path.splitext(original or "")[1].lower()
+    return f"{base}{ext}"
+
+
 def _upload_dir(request: Request) -> Path:
    """첨부 저장 루트. DATA_DIR/uploads/expense — DATA_DIR 와 같은 볼륨에 보관."""
    base = getattr(request.app.state, "data_dir", None)
@@ -288,6 +336,170 @@ async def reports_page(request: Request) -> HTMLResponse:
    )


+# ────────────────────────────────────────────────────────────
+# 승인완료 — 전 직원 월별 (승인자/관리자)
+# ────────────────────────────────────────────────────────────
+def _approved_aggregate(
+    items: list[dict[str, Any]], names: dict[str, str]
+) -> list[dict[str, Any]]:
+    """직원별 승인 금액 합계."""
+    agg: dict[str, dict[str, Any]] = {}
+    for it in items:
+        owner = it.get("owner", "")
+        row = agg.setdefault(
+            owner,
+            {"owner": owner, "name": _display_name(owner, names), "count": 0, "total": 0},
+        )
+        row["count"] += 1
+        row["total"] += int(it.get("amount", 0))
+    return sorted(agg.values(), key=lambda x: -x["total"])
+
+
+@router.get("/approved", response_class=HTMLResponse)
+async def approved_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)
+    year, mon, raw = _parse_month(request)
+    items = store.list_approved(year=year, month=mon)
+    names = _name_map(request)
+    for it in items:
+        it["owner_name"] = _display_name(it.get("owner", ""), names)
+    owner_summary = _approved_aggregate(items, names)
+    grand_total = sum(r["total"] for r in owner_summary)
+    return render_template(
+        request,
+        "expense/approved.html",
+        {
+            "user": user,
+            "is_admin": is_admin(user),
+            "items": items,
+            "owner_summary": owner_summary,
+            "grand_total": grand_total,
+            "month": raw,
+            "count": len(items),
+            "nav_items": build_erp_nav(user, active="expense"),
+            "page_title": "개인경비 — 승인완료",
+            "page_subtitle": f"{raw} · 전 직원 승인완료 {len(items)}건",
+        },
+    )
+
+
+@router.get("/api/approved/export.xlsx")
+async def approved_export(
+    request: Request,
+    user: dict[str, Any] = Depends(_require_approver),
+) -> StreamingResponse:
+    from openpyxl import Workbook  # 지연 import
+
+    store = _store(request)
+    _require_db_store(store)
+    year, mon, raw = _parse_month(request)
+    items = store.list_approved(year=year, month=mon)
+    names = _name_map(request)
+
+    wb = Workbook()
+    ws = wb.active
+    ws.title = "승인완료"
+    header = [
+        "사용일", "이름", "이메일", "분류", "결제수단", "가맹점",
+        "금액", "메모", "상태", "승인자", "결재일시",
+    ]
+    ws.append(header)
+    for r in items:
+        owner = r.get("owner", "")
+        ws.append([
+            r.get("spent_at", ""),
+            _display_name(owner, names),
+            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 "",
+        ])
+    widths = [12, 14, 28, 10, 12, 24, 12, 30, 10, 24, 22]
+    for col, w in enumerate(widths, start=1):
+        ws.column_dimensions[ws.cell(row=1, column=col).column_letter].width = w
+
+    # 직원별 합계 시트
+    ws2 = wb.create_sheet("직원별합계")
+    ws2.append(["이름", "이메일", "건수", "합계금액"])
+    for row in _approved_aggregate(items, names):
+        ws2.append([row["name"], row["owner"], row["count"], row["total"]])
+    for col, w in enumerate([14, 28, 8, 16], start=1):
+        ws2.column_dimensions[ws2.cell(row=1, column=col).column_letter].width = w
+
+    buf = io.BytesIO()
+    wb.save(buf)
+    buf.seek(0)
+    fname = f"expense_approved_{raw}.xlsx"
+    return StreamingResponse(
+        buf,
+        media_type="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
+        headers={"Content-Disposition": f'attachment; filename="{fname}"'},
+    )
+
+
+@router.get("/api/approved/attachments.zip")
+async def approved_attachments_zip(
+    request: Request,
+    user: dict[str, Any] = Depends(_require_approver),
+) -> StreamingResponse:
+    """해당 월 승인완료 항목의 첨부 전체를 zip 한 개로. 파일명 규칙 적용."""
+    store = _store(request)
+    _require_db_store(store)
+    year, mon, raw = _parse_month(request)
+    atts = store.approved_attachments(year=year, month=mon)
+    names = _name_map(request)
+
+    buf = io.BytesIO()
+    used: dict[str, int] = {}
+    with zipfile.ZipFile(buf, "w", zipfile.ZIP_DEFLATED) as zf:
+        for a in atts:
+            path = Path(a.get("stored_path", ""))
+            if not path.exists():
+                continue
+            nm = _display_name(a.get("owner", ""), names)
+            fname = _att_download_name(a.get("uploaded_at"), nm, a.get("filename", ""))
+            # 동일명 충돌 → 접미사
+            if fname in used:
+                used[fname] += 1
+                stem, ext = os.path.splitext(fname)
+                arcname = f"{stem}_{used[fname]}{ext}"
+            else:
+                used[fname] = 1
+                arcname = fname
+            zf.write(str(path), arcname=arcname)
+    buf.seek(0)
+    zipname = f"expense_approved_{raw}.zip"
+    return StreamingResponse(
+        buf,
+        media_type="application/zip",
+        headers={"Content-Disposition": f'attachment; filename="{zipname}"'},
+    )
+
+
 # ────────────────────────────────────────────────────────────
 # 승인자 설정 — 분류 관리
 # ────────────────────────────────────────────────────────────