8ab5223790
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
828 lines
29 KiB
Python
828 lines
29 KiB
Python
"""휴가 관리 모듈 라우터.
|
|
|
|
- 경로: /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)
|
|
|
|
|
|
@router.post("/{request_id}/delete")
|
|
async def delete(
|
|
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.hard_delete(
|
|
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))
|
|
except ValueError as exc:
|
|
raise HTTPException(status_code=409, detail=str(exc))
|
|
return RedirectResponse(url="/vacation/", status_code=303)
|