"""휴가 관리 모듈 라우터. - 경로: /vacation - 권한: 로그인 + `vacation` 모듈 권한 (관리자는 항상 통과). 서버 측 검사. 승인/반려는 `vacation_approver` 또는 admin. - 데이터: VacationDBStore (vacation_db / PostgreSQL) 전용. VACATION_DB_URL 미설정 시 store 가 None 이며, 각 페이지는 "설정 필요" 안내를 보여준다. """ from __future__ import annotations import calendar as _calendar import io from datetime import date from typing import Any from fastapi import APIRouter, Depends, Form, HTTPException, Request from fastapi.responses import ( HTMLResponse, JSONResponse, RedirectResponse, StreamingResponse, ) from app.timezone import now_kst, today_kst from .store import HALF_LABELS, HALVES, VACATION_TYPES router = APIRouter(prefix="/vacation", tags=["vacation"]) # ──────────────────────────────────────────────────────────── # 공용 헬퍼 # ──────────────────────────────────────────────────────────── def _store(request: Request) -> Any: return getattr(request.app.state, "vacation_store", None) def _require_user(request: Request) -> dict[str, Any]: from app.main import get_current_user_record # noqa: WPS433 from app.store import has_module # noqa: WPS433 user = get_current_user_record(request) if user is None: raise HTTPException(status_code=401, detail="로그인이 필요합니다.") if not has_module(user, "vacation"): raise HTTPException(status_code=403, detail="휴가 관리 모듈 권한이 없습니다.") 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, "vacation_approver")): raise HTTPException(status_code=403, detail="휴가 승인자 권한이 필요합니다.") return user def _is_approver(request: Request, user: dict[str, Any]) -> bool: from app.store import has_module, is_admin # noqa: WPS433 return is_admin(user) or has_module(user, "vacation_approver") def _render_config_needed(request: Request, user: dict[str, Any]) -> HTMLResponse: from app.main import build_erp_nav, render_template # noqa: WPS433 from app.store import is_admin # noqa: WPS433 return render_template( request, "denied.html", { "reason": "휴가 관리 모듈이 아직 설정되지 않았습니다. " "VACATION_DB_URL 환경변수를 설정하고 scripts/sql/vacation_db_init.sql 로 " "vacation_db 를 초기화한 뒤 컨테이너를 재기동하세요.", "user": user, "is_admin": is_admin(user), "nav_items": build_erp_nav(user, active="vacation"), }, status_code=503, ) def _guard( request: Request, ) -> tuple[Any, dict[str, Any]] | HTMLResponse | RedirectResponse: """로그인+권한+store 점검. 페이지 핸들러 진입부에서 사용.""" from app.main import get_current_user_record, render_template # noqa: WPS433 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 has_module(user, "vacation"): return render_template( request, "denied.html", {"reason": "휴가 관리 모듈 접근 권한이 없습니다.", "is_admin": is_admin(user)}, status_code=403, ) store = _store(request) if store is None: return _render_config_needed(request, user) return store, user def _ym(request: Request) -> tuple[int, int]: today = today_kst() try: year = int(request.query_params.get("year") or today.year) month = int(request.query_params.get("month") or today.month) except ValueError: year, month = today.year, today.month if not (1 <= month <= 12): year, month = today.year, today.month return year, month # 상태별 bar 스타일 클래스 (CSS 와 매핑) _STATUS_CLASS = { "작성중": "vac-bar-draft", "제출": "vac-bar-submit", "승인": "vac-bar-approve", "반려": "vac-bar-reject", "취소": "vac-bar-cancel", } def _assign_lanes(bars: list[dict[str, Any]]) -> int: """주(week) 내 bar 들에 겹치지 않는 lane(행) 번호를 그리디 배정. bars 는 같은 주의 segment 들. 각 bar 에 'lane' 키를 추가하고, 사용된 lane 수를 반환. """ bars.sort(key=lambda b: (b["start_col"], -b["span"])) lane_end: list[int] = [] # lane 별 마지막 점유 col(포함) for b in bars: placed = False for li, end_col in enumerate(lane_end): if b["start_col"] > end_col: b["lane"] = li lane_end[li] = b["start_col"] + b["span"] - 1 placed = True break if not placed: b["lane"] = len(lane_end) lane_end.append(b["start_col"] + b["span"] - 1) return len(lane_end) def _build_calendar( store: Any, year: int, month: int, sel: str ) -> dict[str, Any]: """월간 달력 데이터 + bar lane 레이아웃 계산.""" cal = _calendar.Calendar(firstweekday=6) # 일요일 시작 weeks_dates = cal.monthdatescalendar(year, month) range_start = weeks_dates[0][0] range_end = weeks_dates[-1][-1] red_set = store.red_holiday_set( date_from=range_start.isoformat(), date_to=range_end.isoformat() ) holiday_names = { h["holiday_date"]: h["name"] for h in store.list_holidays() if h.get("is_red") } requests = store.list_overlapping( date_from=range_start.isoformat(), date_to=range_end.isoformat() ) today = today_kst() weeks: list[dict[str, Any]] = [] for week in weeks_dates: w_start, w_end = week[0], week[-1] days = [ { "date": d.isoformat(), "day": d.day, "in_month": d.month == month, "is_today": d == today, "is_selected": d.isoformat() == sel, "is_sunday": d.weekday() == 6, "is_saturday": d.weekday() == 5, "is_holiday": d.isoformat() in red_set, "holiday_name": holiday_names.get(d.isoformat(), ""), } for d in week ] # 이 주에 걸치는 bar segment bars: list[dict[str, Any]] = [] for r in requests: rs = date.fromisoformat(r["start_date"]) re_ = date.fromisoformat(r["end_date"]) if re_ < w_start or rs > w_end: continue seg_start = max(rs, w_start) seg_end = min(re_, w_end) start_col = (seg_start - w_start).days # 0..6 span = (seg_end - seg_start).days + 1 label = f"{r.get('owner_name') or r.get('owner')} {r['vacation_type']}" bars.append( { "id": r["id"], "label": label, "status": r["status"], "status_class": _STATUS_CLASS.get(r["status"], "vac-bar-draft"), "start_col": start_col, "span": span, "continues_left": rs < w_start, "continues_right": re_ > w_end, "days": r["days"], } ) lane_count = _assign_lanes(bars) weeks.append({"days": days, "bars": bars, "lane_count": lane_count}) return {"weeks": weeks, "requests": requests, "sel_date": sel} # ════════════════════════════════════════════════════════════ # 메인 — 월간 달력 + 선택일 휴가 리스트 # ════════════════════════════════════════════════════════════ @router.get("/", response_class=HTMLResponse) async def index(request: Request) -> HTMLResponse: from app.main import build_erp_nav, render_template # noqa: WPS433 from app.store import is_admin # noqa: WPS433 guard = _guard(request) if not isinstance(guard, tuple): return guard store, user = guard year, month = _ym(request) sel = request.query_params.get("date") or "" today = today_kst() if not sel: sel = ( today.isoformat() if (today.year == year and today.month == month) else f"{year:04d}-{month:02d}-01" ) caldata = _build_calendar(store, year, month, sel) sel_requests = [ r for r in caldata["requests"] if r["start_date"] <= sel <= r["end_date"] ] prev_y, prev_m = (year - 1, 12) if month == 1 else (year, month - 1) next_y, next_m = (year + 1, 1) if month == 12 else (year, month + 1) summary = store.balance_summary(user_email=user["email"], year=year) is_approver = _is_approver(request, user) pending_count = len(store.list_pending()) if is_approver else 0 return render_template( request, "vacation/index.html", { "user": user, "is_admin": is_admin(user), "is_approver": is_approver, "pending_count": pending_count, "nav_items": build_erp_nav(user, active="vacation"), "page_title": "휴가 관리", "page_subtitle": f"{year}년 {month}월 휴가 달력", "year": year, "month": month, "prev_y": prev_y, "prev_m": prev_m, "next_y": next_y, "next_m": next_m, "today": today.isoformat(), "weekdays": ["일", "월", "화", "수", "목", "금", "토"], "weeks": caldata["weeks"], "selected_date": sel, "sel_requests": sel_requests, "balance": summary, "half_labels": HALF_LABELS, }, ) # ════════════════════════════════════════════════════════════ # 휴가 신청 — 등록 / 수정 / 상세 # ════════════════════════════════════════════════════════════ def _form_context(request: Request, user: dict[str, Any]) -> dict[str, Any]: from app.main import build_erp_nav # noqa: WPS433 from app.store import is_admin # noqa: WPS433 return { "user": user, "is_admin": is_admin(user), "is_approver": _is_approver(request, user), "nav_items": build_erp_nav(user, active="vacation"), "vacation_types": list(VACATION_TYPES), "halves": list(HALVES), "half_labels": HALF_LABELS, } @router.get("/new", response_class=HTMLResponse) async def new_form(request: Request) -> HTMLResponse: from app.main import render_template # noqa: WPS433 guard = _guard(request) if not isinstance(guard, tuple): return guard _store_obj, user = guard ctx = _form_context(request, user) ctx.update( { "page_title": "휴가 신청", "page_subtitle": "휴가 종류 · 기간 · 사유 입력", "mode": "new", "req": None, "default_date": today_kst().isoformat(), } ) return render_template(request, "vacation/form.html", ctx) def _payload_from_form( vacation_type: str, start_date: str, end_date: str, start_half: str, end_half: str, reason: str, ) -> dict[str, Any]: return { "vacation_type": vacation_type, "start_date": start_date, "end_date": end_date or start_date, "start_half": start_half, "end_half": end_half, "reason": reason, } @router.post("/new") async def create( request: Request, vacation_type: str = Form(...), start_date: str = Form(...), end_date: str = Form(""), start_half: str = Form("full"), end_half: str = Form("full"), reason: str = Form(""), action: str = Form("submit"), # "draft" | "submit" user: dict[str, Any] = Depends(_require_user), ) -> RedirectResponse: store = _store(request) if store is None: raise HTTPException(status_code=503, detail="vacation_db 미설정") status = "작성중" if action == "draft" else "제출" payload = _payload_from_form( vacation_type, start_date, end_date, start_half, end_half, reason ) try: req = store.create_request( owner=user["email"], owner_name=user.get("name") or user["email"], payload=payload, status=status, ) except ValueError as exc: raise HTTPException(status_code=400, detail=str(exc)) return RedirectResponse(url=f"/vacation/{req['id']}", status_code=303) @router.get("/pending", response_class=HTMLResponse) async def pending_page(request: Request) -> HTMLResponse: from app.main import build_erp_nav, render_template # noqa: WPS433 from app.store import is_admin # noqa: WPS433 guard = _guard(request) if not isinstance(guard, tuple): return guard store, user = guard if not _is_approver(request, user): return render_template( request, "denied.html", {"reason": "휴가 승인자 권한이 없습니다.", "is_admin": is_admin(user)}, status_code=403, ) pending = store.list_pending() return render_template( request, "vacation/pending.html", { "user": user, "is_admin": is_admin(user), "is_approver": True, "nav_items": build_erp_nav(user, active="vacation"), "page_title": "휴가 — 승인 대기", "page_subtitle": f"제출 상태 {len(pending)}건", "items": pending, "half_labels": HALF_LABELS, }, ) @router.get("/settings", response_class=HTMLResponse) async def settings_page(request: Request) -> HTMLResponse: from app.main import build_erp_nav, render_template # noqa: WPS433 from app.store import is_admin # noqa: WPS433 guard = _guard(request) if not isinstance(guard, tuple): return guard store, user = guard if not is_admin(user): return render_template( request, "denied.html", {"reason": "휴가 설정은 관리자만 접근할 수 있습니다.", "is_admin": False}, status_code=403, ) today = today_kst() try: year = int(request.query_params.get("year") or today.year) except ValueError: year = today.year balances = store.list_balances(year=year) for b in balances: b["used_days"] = store.used_days(user_email=b["user_email"], year=year) return render_template( request, "vacation/settings.html", { "user": user, "is_admin": True, "is_approver": True, "nav_items": build_erp_nav(user, active="vacation"), "page_title": "휴가 — 설정", "page_subtitle": "공휴일 · 연차 잔여 관리", "year": year, "holidays": store.list_holidays(year=year), "balances": balances, }, ) @router.post("/settings/holidays") async def holiday_upsert( request: Request, holiday_date: str = Form(...), name: str = Form(...), kind: str = Form("public"), is_red: str = Form(""), user: dict[str, Any] = Depends(_require_user), ) -> RedirectResponse: from app.store import is_admin # noqa: WPS433 if not is_admin(user): raise HTTPException(status_code=403, detail="관리자만 가능합니다.") store = _store(request) if store is None: raise HTTPException(status_code=503, detail="vacation_db 미설정") try: store.upsert_holiday( holiday_date=holiday_date, name=name, kind=kind, is_red=is_red in ("1", "true", "on", "True"), ) except ValueError as exc: raise HTTPException(status_code=400, detail=str(exc)) year = holiday_date[:4] return RedirectResponse(url=f"/vacation/settings?year={year}", status_code=303) @router.post("/settings/holidays/{holiday_id:int}/delete") async def holiday_delete( request: Request, holiday_id: int, year: str = Form(""), user: dict[str, Any] = Depends(_require_user), ) -> RedirectResponse: from app.store import is_admin # noqa: WPS433 if not is_admin(user): raise HTTPException(status_code=403, detail="관리자만 가능합니다.") store = _store(request) if store is None: raise HTTPException(status_code=503, detail="vacation_db 미설정") try: store.delete_holiday(holiday_id=holiday_id) except KeyError: raise HTTPException(status_code=404, detail="공휴일을 찾을 수 없습니다.") suffix = f"?year={year}" if year else "" return RedirectResponse(url=f"/vacation/settings{suffix}", status_code=303) @router.post("/settings/balances") async def balance_upsert( request: Request, user_email: str = Form(...), year: int = Form(...), total_days: float = Form(...), memo: str = Form(""), user: dict[str, Any] = Depends(_require_user), ) -> RedirectResponse: from app.store import is_admin # noqa: WPS433 if not is_admin(user): raise HTTPException(status_code=403, detail="관리자만 가능합니다.") store = _store(request) if store is None: raise HTTPException(status_code=503, detail="vacation_db 미설정") try: store.upsert_balance( user_email=user_email, year=year, total_days=total_days, memo=memo ) except ValueError as exc: raise HTTPException(status_code=400, detail=str(exc)) return RedirectResponse(url=f"/vacation/settings?year={year}", status_code=303) @router.get("/api/calendar") async def api_calendar( request: Request, user: dict[str, Any] = Depends(_require_user) ) -> JSONResponse: """달력 비동기 데이터(JSON). year/month 쿼리.""" store = _store(request) if store is None: raise HTTPException(status_code=503, detail="vacation_db 미설정") year, month = _ym(request) sel = request.query_params.get("date") or f"{year:04d}-{month:02d}-01" caldata = _build_calendar(store, year, month, sel) return JSONResponse( { "year": year, "month": month, "weeks": caldata["weeks"], "requests": caldata["requests"], } ) @router.get("/export.xlsx") async def export_xlsx( request: Request, user: dict[str, Any] = Depends(_require_user) ) -> StreamingResponse: from openpyxl import Workbook # 지연 import store = _store(request) if store is None: raise HTTPException(status_code=503, detail="vacation_db 미설정") today = today_kst() try: year = int(request.query_params.get("year") or today.year) month = int(request.query_params.get("month") or 0) except ValueError: year, month = today.year, 0 if 1 <= month <= 12: last = _calendar.monthrange(year, month)[1] date_from = f"{year:04d}-{month:02d}-01" date_to = f"{year:04d}-{month:02d}-{last:02d}" else: date_from = f"{year:04d}-01-01" date_to = f"{year:04d}-12-31" rows = store.list_for_export(date_from=date_from, date_to=date_to) wb = Workbook() ws = wb.active ws.title = "vacation" header = [ "신청자", "휴가종류", "시작일", "종료일", "시작구분", "종료구분", "사용일수", "상태", "승인자", "승인/반려일", "사유", "반려사유", ] ws.append(header) for r in rows: ws.append([ r.get("owner_name") or r.get("owner", ""), r.get("vacation_type", ""), r.get("start_date", ""), r.get("end_date", ""), HALF_LABELS.get(r.get("start_half", "full"), ""), HALF_LABELS.get(r.get("end_half", "full"), ""), r.get("days", 0), r.get("status", ""), r.get("approver_email", "") or "", r.get("decided_at", "") or "", r.get("reason", "") or "", r.get("reject_reason", "") or "", ]) widths = [24, 10, 12, 12, 8, 8, 8, 8, 24, 22, 30, 30] 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) scope = f"{year}{('%02d' % month) if (1 <= month <= 12) else ''}" fname = f"vacation_{scope}_{now_kst().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": "vacation"} @router.get("/{request_id}", response_class=HTMLResponse) async def detail(request: Request, request_id: str) -> HTMLResponse: from app.main import build_erp_nav, render_template # noqa: WPS433 from app.store import is_admin # noqa: WPS433 guard = _guard(request) if not isinstance(guard, tuple): return guard store, user = guard req = store.get_request(request_id=request_id) if not req: return render_template( request, "denied.html", {"reason": "휴가 신청을 찾을 수 없습니다.", "is_admin": is_admin(user)}, status_code=404, ) is_owner = req["owner"] == user["email"] is_approver = _is_approver(request, user) if not (is_owner or is_approver): return render_template( request, "denied.html", {"reason": "본인 또는 승인자만 조회할 수 있습니다.", "is_admin": is_admin(user)}, status_code=403, ) return render_template( request, "vacation/detail.html", { "user": user, "is_admin": is_admin(user), "is_approver": is_approver, "is_owner": is_owner, "nav_items": build_erp_nav(user, active="vacation"), "page_title": "휴가 상세", "page_subtitle": f"{req['start_date']} · {req['vacation_type']}", "req": req, "half_labels": HALF_LABELS, "can_edit": is_owner and req["status"] in ("작성중", "반려"), }, ) @router.get("/{request_id}/edit", response_class=HTMLResponse) async def edit_form(request: Request, request_id: str) -> HTMLResponse: from app.main import render_template # noqa: WPS433 from app.store import is_admin # noqa: WPS433 guard = _guard(request) if not isinstance(guard, tuple): return guard store, user = guard req = store.get_request(request_id=request_id) if not req: return render_template( request, "denied.html", {"reason": "휴가 신청을 찾을 수 없습니다.", "is_admin": is_admin(user)}, status_code=404, ) if req["owner"] != user["email"]: return render_template( request, "denied.html", {"reason": "본인 신청만 수정할 수 있습니다.", "is_admin": is_admin(user)}, status_code=403, ) if req["status"] not in ("작성중", "반려"): return render_template( request, "denied.html", {"reason": f"{req['status']} 상태에서는 수정할 수 없습니다.", "is_admin": is_admin(user)}, status_code=409, ) ctx = _form_context(request, user) ctx.update( { "page_title": "휴가 신청 수정", "page_subtitle": "작성중/반려 상태만 수정 가능", "mode": "edit", "req": req, "default_date": req["start_date"], } ) return render_template(request, "vacation/form.html", ctx) @router.post("/{request_id}/edit") async def update( request: Request, request_id: str, vacation_type: str = Form(...), start_date: str = Form(...), end_date: str = Form(""), start_half: str = Form("full"), end_half: str = Form("full"), reason: str = Form(""), action: str = Form("save"), # "save" | "submit" user: dict[str, Any] = Depends(_require_user), ) -> RedirectResponse: store = _store(request) if store is None: raise HTTPException(status_code=503, detail="vacation_db 미설정") payload = _payload_from_form( vacation_type, start_date, end_date, start_half, end_half, reason ) try: store.update_request(request_id=request_id, owner=user["email"], payload=payload) if action == "submit": store.submit(request_id=request_id, owner=user["email"]) except KeyError: raise HTTPException(status_code=404, detail="휴가 신청을 찾을 수 없습니다.") except PermissionError as exc: raise HTTPException(status_code=403, detail=str(exc)) except ValueError as exc: raise HTTPException(status_code=400, detail=str(exc)) return RedirectResponse(url=f"/vacation/{request_id}", status_code=303) @router.post("/{request_id}/submit") async def submit( request: Request, request_id: str, user: dict[str, Any] = Depends(_require_user), ) -> RedirectResponse: store = _store(request) if store is None: raise HTTPException(status_code=503, detail="vacation_db 미설정") try: store.submit(request_id=request_id, owner=user["email"]) except ValueError as exc: raise HTTPException(status_code=409, detail=str(exc)) return RedirectResponse(url=f"/vacation/{request_id}", status_code=303) @router.post("/{request_id}/approve") async def approve( request: Request, request_id: str, user: dict[str, Any] = Depends(_require_approver), ) -> RedirectResponse: store = _store(request) if store is None: raise HTTPException(status_code=503, detail="vacation_db 미설정") try: store.approve(request_id=request_id, approver_email=user["email"]) except ValueError as exc: raise HTTPException(status_code=409, detail=str(exc)) return RedirectResponse(url=f"/vacation/{request_id}", status_code=303) @router.post("/{request_id}/reject") async def reject( request: Request, request_id: str, reject_reason: str = Form(...), user: dict[str, Any] = Depends(_require_approver), ) -> RedirectResponse: store = _store(request) if store is None: raise HTTPException(status_code=503, detail="vacation_db 미설정") try: store.reject( request_id=request_id, approver_email=user["email"], reason=reject_reason ) except ValueError as exc: raise HTTPException(status_code=409, detail=str(exc)) return RedirectResponse(url=f"/vacation/{request_id}", status_code=303) @router.post("/{request_id}/cancel") async def cancel( request: Request, request_id: str, user: dict[str, Any] = Depends(_require_user), ) -> RedirectResponse: from app.store import is_admin # noqa: WPS433 store = _store(request) if store is None: raise HTTPException(status_code=503, detail="vacation_db 미설정") try: store.cancel( request_id=request_id, user_email=user["email"], is_admin=is_admin(user) ) except KeyError: raise HTTPException(status_code=404, detail="휴가 신청을 찾을 수 없습니다.") except PermissionError as exc: raise HTTPException(status_code=403, detail=str(exc)) return RedirectResponse(url=f"/vacation/{request_id}", status_code=303)