king/dbx-main
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
163c6c2f00875c496188112eb1391c82a307c4de
dbx-main/app/main.py
T
king c6fb8ed375 feat(cafe24): 상품 상세페이지 관리 모듈 Phase 1 
카페24 관리자에 직접 접속하지 않고 상품 상세페이지(description HTML)를
편집·예약 적용·복원하기 위한 모듈의 기반을 만든다. Phase 1 은 공통
Integration 계층, cafe24_db, OAuth 연결 화면까지다.

카페24 OAuth/API 클라이언트를 상품관리 모듈 안에 두지 않고
app/integrations/cafe24/ 로 분리했다. 향후 추가할 주문관리(주문 조회·송장
일괄등록·취소/반품/교환)가 같은 토큰과 클라이언트를 그대로 재사용해야 하기
때문이다. 라우터에서 httpx 를 직접 부르지 않고 Cafe24Client 만 쓰게 해서
재시도·rate limit·API 로그·토큰 갱신을 한 곳에 모았다.

토큰은 Fernet 으로 암호화해 저장한다(CAFE24_TOKEN_SECRET). DB 덤프가
유출돼도 access/refresh token 이 평문으로 남지 않게 하기 위함이며, API 로그와
연결 상태 화면에는 토큰·시크릿을 일절 기록/표시하지 않는다.

토큰 갱신은 행 잠금(SELECT ... FOR UPDATE) 안에서 한다. 카페24는 refresh
token 을 회전시키므로, 이후 추가될 예약 worker 컨테이너와 web 컨테이너가
동시에 갱신하면 한쪽 토큰이 무효화된다.

기존 파일 변경은 목록에 한 줄씩 추가하는 형태로 44줄뿐이며 기존 라우트·
테이블·인증 로직은 건드리지 않았다. CAFE24_DB_URL 미설정 시 store 가 None
이라 앱은 정상 기동하고 모듈만 "설정 필요" 안내를 표시한다.

가드 헬퍼를 common.py 로 분리한 것은 router.py 가 routes_system.py 를
include 하는 구조에서 순환 import 가 생기기 때문이다.

검증: 신규 테스트 16개 통과(암호화 왕복, 토큰 만료·자동갱신, 상태 노출 시
토큰 미유출, 재시도 예산, 예약 상태 전이). dispatch 기존 테스트 9개 통과.
cafe24_db_init.sql 은 로컬에 Docker 가 없어 미실행 — 서버 적용 시 확인 필요.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
2026-08-14 00:23:02 +09:00

667 lines
25 KiB
Python
Raw Blame History

import inspect
import os
from pathlib import Path
from typing import Any
from authlib.integrations.starlette_client import OAuth, OAuthError
from dotenv import load_dotenv
from fastapi import Depends, FastAPI, HTTPException, Request
from fastapi.responses import HTMLResponse, JSONResponse, RedirectResponse
from fastapi.staticfiles import StaticFiles
from fastapi.templating import Jinja2Templates
from jinja2 import ChoiceLoader, FileSystemLoader
from pydantic import BaseModel
from starlette.middleware.sessions import SessionMiddleware
from .modules.cafe24 import build_cafe24_store
from .modules.cafe24 import router as cafe24_router
from .modules.cupang import build_cupang_store, build_itemcode_reader
from .modules.cupang import router as cupang_router
from .modules.dispatch import build_dispatch_store
from .modules.dispatch import router as dispatch_router
from .modules.expense import CategoryStore, build_expense_store
from .modules.expense import router as expense_router
from .modules.malaysia import build_malaysia_itemcode, build_malaysia_store
from .modules.malaysia import router as malaysia_router
from .modules.project import build_project_store
from .modules.project import router as project_router
from .modules.vacation import build_vacation_store
from .modules.vacation import router as vacation_router
from .store import (
APPROVER_KEYS,
MODULE_KEYS,
SUPER_ADMIN_EMAIL,
UserStore,
allowed_modules,
has_module,
is_admin,
)
# 권한 키 한글 라벨 (admin.html / 사이드바 공용)
MODULE_LABELS: dict[str, str] = {
"corm": "CORM",
"order": "Order",
"expense": "개인경비",
"vacation": "휴가",
"cupang": "쿠팡 밀크런",
"malaysia": "말레이시아 재고관리",
"dispatch": "말레이시아 배송",
"project": "프로젝트 관리",
"cafe24": "카페24 상품관리",
"expense_approver": "개인경비",
"vacation_approver": "휴가",
}
# OMS(orderlist) 와 공유하는 세션 키. SessionMiddleware 의 session_cookie 도 동일 이름.
SESSION_COOKIE_DEFAULT = "session"
load_dotenv()
BASE_DIR = Path(__file__).resolve().parent
ALLOWED_DOMAIN = "dbxcorp.co.kr"
def _data_dir() -> Path:
"""사용자/권한 JSON 저장소 위치. 컨테이너 재배포에도 살아남도록
DATA_DIR 환경변수로 마운트된 볼륨을 가리킬 수 있다."""
override = os.getenv("DATA_DIR", "").strip()
if override:
return Path(override)
return BASE_DIR / "data"
DATA_DIR = _data_dir()
def env(name: str, default: str = "") -> str:
value = os.getenv(name, "").strip()
return value if value else default
def _require_session_secret() -> str:
secret = env("SESSION_SECRET_KEY", "")
if not secret:
raise RuntimeError(
"SESSION_SECRET_KEY 환경변수가 설정되지 않았습니다. "
"openssl rand -hex 32 로 새 값을 만들어 .env 에 넣고 컨테이너를 재기동하세요. "
"OMS(orderlist) 와 SSO 하려면 양쪽 .env 에 같은 값이어야 합니다."
)
return secret
def build_google_oauth() -> OAuth:
oauth = OAuth()
oauth.register(
name="google",
client_id=env("GOOGLE_CLIENT_ID"),
client_secret=env("GOOGLE_CLIENT_SECRET"),
server_metadata_url="https://accounts.google.com/.well-known/openid-configuration",
client_kwargs={"scope": "openid email profile"},
)
return oauth
app = FastAPI(title="DBX 메인 페이지")
app.add_middleware(
SessionMiddleware,
secret_key=_require_session_secret(),
session_cookie=env("SESSION_COOKIE_NAME", SESSION_COOKIE_DEFAULT),
https_only=env("SESSION_COOKIE_SECURE", "true").lower() == "true",
same_site="lax",
max_age=int(env("SESSION_MAX_AGE", "28800")),
path="/",
)
app.mount("/static", StaticFiles(directory=str(BASE_DIR / "static")), name="static")
# 모듈별 templates 디렉토리를 추가로 검색하도록 ChoiceLoader 설정.
# 신규 모듈 추가 시 아래 리스트에 `BASE_DIR / "modules" / "<name>" / "templates"` 만 추가.
_MODULE_TEMPLATE_DIRS = [
BASE_DIR / "modules" / "expense" / "templates",
BASE_DIR / "modules" / "cupang" / "templates",
BASE_DIR / "modules" / "vacation" / "templates",
BASE_DIR / "modules" / "malaysia" / "templates",
BASE_DIR / "modules" / "dispatch" / "templates",
BASE_DIR / "modules" / "project" / "templates",
BASE_DIR / "modules" / "cafe24" / "templates",
]
templates = Jinja2Templates(directory=str(BASE_DIR / "templates"))
templates.env.loader = ChoiceLoader(
[
FileSystemLoader(str(BASE_DIR / "templates")),
*[FileSystemLoader(str(p)) for p in _MODULE_TEMPLATE_DIRS if p.exists()],
]
)
# dispatch 모듈 템플릿 필터: 배송사 로고 URL / 송장 조회 딥링크.
# {{ provider | courier_logo }} · {{ tracking | courier_track(provider) }}
from .modules.dispatch.store import ( # noqa: E402
courier_logo_url as _courier_logo_url,
)
from .modules.dispatch.store import ( # noqa: E402
courier_tracking_url as _courier_tracking_url,
)
templates.env.filters["courier_logo"] = _courier_logo_url
templates.env.filters["courier_track"] = _courier_tracking_url
oauth = build_google_oauth()
user_store = UserStore(DATA_DIR / "users.json")
# 모듈별 데이터 저장소.
# EXPENSE_DB_URL 가 있으면 expense_db(PostgreSQL), 없으면 JSON 파일.
app.state.data_dir = DATA_DIR # 모듈에서 첨부 저장 경로 등으로 참조
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)
app.state.itemcode_reader = build_itemcode_reader()
# 휴가 관리: VACATION_DB_URL 없으면 store=None(라우터가 "설정 필요" 안내).
app.state.vacation_store = build_vacation_store(dsn=env("VACATION_DB_URL") or None)
# 말레이시아 재고관리: MALAYSIA_STOCK_DB_URL 없으면 store=None(라우터가 "설정 필요" 안내).
# 상품명은 위 itemcode_reader(읽기 전용) 재사용.
app.state.malaysia_store = build_malaysia_store(dsn=env("MALAYSIA_STOCK_DB_URL") or None)
# 세트 BOM 은 itemcode_db(set_components)에서 읽는다(ITEMCODE_DB_URL 재사용).
app.state.malaysia_itemcode = build_malaysia_itemcode()
# 말레이시아 배송(TikTok 출고): DISPATCH_DB_URL 없으면 store=None(라우터가 "설정 필요" 안내).
# 업로드 원본은 DATA_DIR/dispatch/ 아래 저장(개인정보는 저장하지 않음).
app.state.dispatch_store = build_dispatch_store(dsn=env("DISPATCH_DB_URL") or None)
# 프로젝트 관리(아사나식): PROJECT_DB_URL 없으면 store=None(라우터가 "설정 필요" 안내).
# 메일 알림은 SMTP_* 환경변수 기반(app/mail.py). 미설정 시 조용히 skip.
app.state.project_store = build_project_store(dsn=env("PROJECT_DB_URL") or None)
# 카페24 상품관리: CAFE24_DB_URL 없으면 store=None(라우터가 "설정 필요" 안내).
# 카페24 API 호출/토큰은 app/integrations/cafe24 공통 계층(CAFE24_* 환경변수).
app.state.cafe24_store = build_cafe24_store(dsn=env("CAFE24_DB_URL") or None)
# 모듈 라우터 등록 — 신규 모듈 추가 시 여기 한 줄.
app.include_router(expense_router)
app.include_router(cupang_router)
app.include_router(vacation_router)
app.include_router(malaysia_router)
app.include_router(dispatch_router)
app.include_router(project_router)
app.include_router(cafe24_router)
def public_url_for(request: Request, route_name: str) -> str:
public_base_url = env("PUBLIC_BASE_URL").rstrip("/")
if public_base_url:
return f"{public_base_url}{request.url_for(route_name).path}"
return str(request.url_for(route_name))
def get_session_user(request: Request) -> dict[str, Any] | None:
"""세션에서 로그인 사용자(이메일/이름/사진) 추출. OMS SSO 호환."""
user = request.session.get("user")
if isinstance(user, dict):
return user
email = request.session.get("user_email")
if email:
return {
"email": str(email),
"name": request.session.get("user_name") or str(email),
"picture": request.session.get("user_picture", "") or "",
}
return None
def get_current_user_record(request: Request) -> dict[str, Any] | None:
"""세션 + 저장소를 합쳐 권한이 포함된 사용자 레코드 반환."""
sess = get_session_user(request)
if not sess:
return None
rec = user_store.get(sess["email"])
if rec is None:
# 세션은 살아있지만 저장소에 없음 — 세션 무효화
return None
return rec
def require_admin(request: Request) -> dict[str, Any]:
rec = get_current_user_record(request)
if rec is None:
raise HTTPException(status_code=401, detail="로그인이 필요합니다.")
if not is_admin(rec):
raise HTTPException(status_code=403, detail="관리자 권한이 필요합니다.")
return rec
def safe_next(raw: str | None) -> str:
"""Open redirect 방지: 같은 호스트의 절대 경로만 허용."""
if not raw:
return "/"
if not raw.startswith("/") or raw.startswith("//") or raw.startswith("/\\"):
return "/"
return raw
def render_template(
request: Request,
name: str,
context: dict[str, Any] | None = None,
status_code: int = 200,
) -> HTMLResponse:
template_context = {"request": request, **(context or {})}
first_param = next(iter(inspect.signature(templates.TemplateResponse).parameters))
if first_param == "request":
return templates.TemplateResponse(
request,
name,
template_context,
status_code=status_code,
)
return templates.TemplateResponse(
name,
template_context,
status_code=status_code,
)
def is_allowed_google_user(userinfo: dict[str, Any]) -> tuple[bool, str]:
email = str(userinfo.get("email", "")).lower().strip()
email_verified = bool(userinfo.get("email_verified"))
domain = email.rsplit("@", 1)[-1] if "@" in email else ""
if not email_verified:
return False, "Google 계정 이메일 인증이 확인되지 않았습니다."
if domain != ALLOWED_DOMAIN:
return False, "회사 Google Workspace 계정만 접속할 수 있습니다."
return True, ""
# ── ERP 메뉴 정의 ──────────────────────────────────────────────
# 각 항목: key(권한키), title, description, url(env override), status(ready|preparing), category
def _menu_items_for(user_rec: dict[str, Any]) -> list[dict[str, Any]]:
items = [
{
"key": "corm",
"title": "CORM",
"subtitle": "CS · 발주 · 반품 · 코드관리",
"description": "고객 응대와 발주/반품, 코드 관리 업무를 한 곳에서 처리합니다.",
"url": env("CS_ORDER_URL", "/corm/"),
"health_url": "/corm/health/db",
"status": "ready",
"category": "운영",
},
{
"key": "order",
"title": "Order",
"subtitle": "고객 주문 데이터베이스",
"description": "고객 주문 내역을 조회·검색하고 관련 데이터를 관리합니다.",
"url": env("CUSTOMER_ORDER_LIST_URL", "/orderlist/"),
"health_url": "/orderlist/health/db",
"status": "ready",
"category": "운영",
},
{
"key": "expense",
"title": "개인경비",
"subtitle": "Personal Expense",
"description": "법인카드/개인경비 사용 내역을 등록·증빙하고 정산을 신청합니다.",
"url": "/expense/",
"health_url": "/expense/health",
"status": "ready",
"category": "관리",
},
{
"key": "cupang",
"title": "쿠팡 밀크런",
"subtitle": "Coupang Milk-run",
"description": "쿠팡 밀크런 출고 일정·박스 계산·입고센터를 달력에서 관리합니다.",
"url": "/cupang/",
"health_url": "/cupang/health",
"status": "ready",
"category": "운영",
},
{
"key": "vacation",
"title": "휴가",
"subtitle": "Vacation",
"description": "연차/반차/특별휴가 신청과 잔여일수, 결재 현황을 달력에서 관리합니다.",
"url": "/vacation/",
"health_url": "/vacation/health",
"status": "ready",
"category": "관리",
},
{
"key": "malaysia",
"title": "말레이시아 재고관리",
"subtitle": "Malaysia Stock",
"description": "말레이시아 창고 입출고·세트 BOM·일일 재고조사를 관리합니다.",
"url": "/malaysia/",
"health_url": "/malaysia/health",
"status": "ready",
"category": "운영",
},
{
"key": "dispatch",
"title": "말레이시아 배송",
"subtitle": "Malaysia Dispatch",
"description": "TikTok 출고 파일을 올리면 박스별 작업 리스트·피킹 요약·Kagayaku 전달표를 자동 생성합니다.",
"url": "/dispatch/",
"health_url": "/dispatch/health",
"status": "ready",
"category": "운영",
},
{
"key": "project",
"title": "프로젝트 관리",
"subtitle": "Projects",
"description": "프로젝트·서브프로젝트·업무를 달력/타임라인/보드로 관리합니다(아사나식).",
"url": "/project/",
"health_url": "/project/health",
"status": "ready",
"category": "관리",
},
{
"key": "cafe24",
"title": "카페24 상품관리",
"subtitle": "Cafe24 Products",
"description": "카페24 관리자에 들어가지 않고 상품 상세페이지를 편집·예약 적용하고 이전 버전으로 되돌립니다.",
"url": "/cafe24/",
"health_url": "/cafe24/health",
"status": "ready",
"category": "운영",
},
]
allowed = allowed_modules(user_rec)
for item in items:
item["allowed"] = item["key"] in allowed
return items
def _icon_svg(name: str) -> str:
"""좌측 사이드바용 인라인 아이콘. 외부 의존 없는 작은 SVG."""
paths = {
"home": '<path d="M3 11 12 3l9 8"/><path d="M5 10v10h14V10"/>',
"expense": '<rect x="3" y="6" width="18" height="13" rx="2"/><path d="M3 10h18"/><path d="M7 15h4"/>',
"vacation": '<path d="M8 2v4"/><path d="M16 2v4"/><rect x="3" y="6" width="18" height="15" rx="2"/><path d="M3 11h18"/>',
"corm": '<path d="M21 11.5a8.4 8.4 0 0 1-.9 3.8 8.5 8.5 0 0 1-7.6 4.7 8.4 8.4 0 0 1-3.8-.9L3 21l1.9-5.7a8.4 8.4 0 0 1-.9-3.8 8.5 8.5 0 0 1 4.7-7.6 8.4 8.4 0 0 1 3.8-.9h.5a8.5 8.5 0 0 1 8 8v.5z"/>',
"order": '<rect x="3" y="3" width="18" height="18" rx="2"/><line x1="3" y1="9" x2="21" y2="9"/><line x1="9" y1="21" x2="9" y2="9"/>',
"cupang": '<rect x="3" y="4" width="18" height="18" rx="2"/><line x1="3" y1="10" x2="21" y2="10"/><line x1="8" y1="2" x2="8" y2="6"/><line x1="16" y1="2" x2="16" y2="6"/>',
"malaysia": '<path d="M21 16V8a2 2 0 0 0-1-1.73l-7-4a2 2 0 0 0-2 0l-7 4A2 2 0 0 0 3 8v8a2 2 0 0 0 1 1.73l7 4a2 2 0 0 0 2 0l7-4A2 2 0 0 0 21 16z"/><polyline points="3.27 6.96 12 12.01 20.73 6.96"/><line x1="12" y1="22.08" x2="12" y2="12"/>',
"dispatch": '<rect x="1" y="3" width="15" height="13"/><path d="M16 8h4l3 3v5h-7V8z"/><circle cx="5.5" cy="18.5" r="2.5"/><circle cx="18.5" cy="18.5" r="2.5"/>',
"project": '<rect x="3" y="4" width="18" height="16" rx="2"/><line x1="3" y1="9" x2="21" y2="9"/><line x1="8" y1="13" x2="13" y2="13"/><line x1="8" y1="16" x2="11" y2="16"/>',
"cafe24": '<rect x="3" y="4" width="18" height="16" rx="2"/><path d="M7 9h10"/><path d="M7 13h7"/><path d="M7 17h4"/>',
"modules": '<rect x="3" y="3" width="7" height="7"/><rect x="14" y="3" width="7" height="7"/><rect x="3" y="14" width="7" height="7"/><rect x="14" y="14" width="7" height="7"/>',
}
body = paths.get(name, paths["modules"])
return (
'<svg width="16" height="16" viewBox="0 0 24 24" fill="none" '
'stroke="currentColor" stroke-width="1.8" stroke-linecap="round" stroke-linejoin="round">'
f"{body}</svg>"
)
def build_erp_nav(
user_rec: dict[str, Any], active: str | None = None
) -> list[dict[str, Any]]:
"""좌측 사이드바 메뉴. 모듈 정의(_menu_items_for)와 동기화한다.
각 항목: key, label, group, url, target, icon, active, disabled, disabled_reason.
"""
items: list[dict[str, Any]] = [
{
"key": "home",
"label": "홈",
"group": "ERP",
"url": "/",
"target": "_self",
"icon": _icon_svg("home"),
},
]
for m in _menu_items_for(user_rec):
if not m["allowed"]:
continue
target = "_blank" if m["url"].startswith("http") else "_self"
items.append(
{
"key": m["key"],
"label": m["title"],
"group": m["category"],
"url": m["url"] if m["status"] == "ready" else "#",
"target": target,
"icon": _icon_svg(m["key"]),
"disabled": m["status"] != "ready",
"disabled_reason": "준비중" if m["status"] != "ready" else None,
}
)
# 같은 그룹끼리 묶이도록 정렬(그룹 헤더 중복 방지). 그룹 내 순서는 유지(stable).
group_order = {"ERP": 0, "운영": 1, "관리": 2}
items.sort(key=lambda it: group_order.get(it["group"], 9))
for it in items:
it["active"] = it["key"] == active
return items
@app.get("/", response_class=HTMLResponse)
async def home(request: Request) -> HTMLResponse:
sess = get_session_user(request)
if not sess:
return render_template(request, "login.html")
user_rec = user_store.get(sess["email"])
if user_rec is None:
# 도메인은 통과했으나 저장소에 없음 — 세션 정리 후 재로그인
request.session.clear()
return render_template(request, "login.html")
# 슈퍼 관리자 → 업무 모듈 선택 화면(main.html).
# 일반 사용자 → ERP 메인(좌측 메뉴 + 우측 콘텐츠).
if user_rec.get("is_super_admin"):
menu_items = _menu_items_for(user_rec)
return render_template(
request,
"main.html",
{
"user": user_rec,
"menu_items": menu_items,
"is_admin": is_admin(user_rec),
},
)
return render_template(
request,
"erp_home.html",
{
"user": user_rec,
"is_admin": is_admin(user_rec),
"nav_items": build_erp_nav(user_rec, active="home"),
"page_title": "ERP 홈",
"page_subtitle": "오늘의 업무를 시작하세요.",
},
)
@app.get("/modules", response_class=HTMLResponse)
async def modules_page(request: Request) -> HTMLResponse:
"""업무 모듈 선택 화면. 슈퍼 관리자가 사이드바에서 다시 진입할 때 사용."""
user_rec = get_current_user_record(request)
if user_rec is None:
return RedirectResponse(url="/login", status_code=303)
return render_template(
request,
"main.html",
{
"user": user_rec,
"menu_items": _menu_items_for(user_rec),
"is_admin": is_admin(user_rec),
},
)
@app.get("/login")
async def login(request: Request):
if not env("GOOGLE_CLIENT_ID") or not env("GOOGLE_CLIENT_SECRET"):
return render_template(
request,
"denied.html",
{"reason": "Google OAuth 환경 변수가 아직 설정되지 않았습니다."},
status_code=500,
)
request.session["_post_login_next"] = safe_next(request.query_params.get("next"))
redirect_uri = public_url_for(request, "auth_google")
return await oauth.google.authorize_redirect(
request,
redirect_uri,
hd=ALLOWED_DOMAIN,
prompt="select_account",
)
@app.get("/auth/google")
async def auth_google(request: Request):
try:
token = await oauth.google.authorize_access_token(request)
userinfo = token.get("userinfo")
if userinfo is None:
userinfo = await oauth.google.userinfo(token=token)
except OAuthError as exc:
return render_template(
request,
"denied.html",
{"reason": f"Google 로그인 실패: {exc.error}"},
status_code=401,
)
allowed, reason = is_allowed_google_user(dict(userinfo))
if not allowed:
request.session.clear()
return render_template(
request,
"denied.html",
{"reason": reason},
status_code=403,
)
email = str(userinfo.get("email", "")).lower().strip()
name = userinfo.get("name") or email
picture = userinfo.get("picture", "") or ""
# 저장소에 사용자 등록/갱신 (신규는 권한 0 — 관리자가 부여)
user_store.upsert_login(email=email, name=name, picture=picture)
# OMS 와 공유하는 top-level 키 (SSO 계약)
request.session["user_email"] = email
request.session["user_name"] = name
request.session["user_picture"] = picture
request.session["user"] = {
"email": email,
"name": name,
"picture": picture,
}
next_url = safe_next(request.session.pop("_post_login_next", "/"))
return RedirectResponse(url=next_url, status_code=303)
@app.get("/logout")
async def logout(request: Request) -> RedirectResponse:
request.session.clear()
return RedirectResponse(url="/", status_code=303)
@app.get("/healthz")
async def healthz() -> dict[str, str]:
return {"status": "ok"}
# ── 관리자 페이지 ──────────────────────────────────────────────
@app.get("/admin", response_class=HTMLResponse)
async def admin_page(request: Request) -> HTMLResponse:
rec = get_current_user_record(request)
if rec is None:
return RedirectResponse(url="/login", status_code=303)
if not is_admin(rec):
return render_template(
request,
"denied.html",
{"reason": "관리자만 접근할 수 있는 페이지입니다."},
status_code=403,
)
users = sorted(user_store.list_all(), key=lambda u: (u["email"] != SUPER_ADMIN_EMAIL, u["email"]))
return render_template(
request,
"admin.html",
{
"user": rec,
"users": users,
"module_keys": list(MODULE_KEYS),
"module_labels": MODULE_LABELS,
"approver_keys": list(APPROVER_KEYS),
"super_admin_email": SUPER_ADMIN_EMAIL,
"is_admin": True,
},
)
# ── 관리자 API ──────────────────────────────────────────────
class UpdatePermissionsBody(BaseModel):
role: str | None = None
modules: dict[str, bool] | None = None
class CreateUserBody(BaseModel):
email: str
name: str = ""
role: str = "user"
modules: dict[str, bool] | None = None
@app.get("/api/users")
async def api_list_users(_: dict[str, Any] = Depends(require_admin)) -> JSONResponse:
return JSONResponse({"users": user_store.list_all()})
@app.post("/api/users")
async def api_create_user(
body: CreateUserBody,
_: dict[str, Any] = Depends(require_admin),
) -> JSONResponse:
try:
rec = user_store.create_user(
email=body.email,
name=body.name,
role=body.role,
modules=body.modules,
)
except ValueError as exc:
raise HTTPException(status_code=400, detail=str(exc))
return JSONResponse({"user": rec}, status_code=201)
@app.put("/api/users/{email}")
async def api_update_user(
email: str,
body: UpdatePermissionsBody,
_: dict[str, Any] = Depends(require_admin),
) -> JSONResponse:
try:
rec = user_store.update_permissions(
email=email, role=body.role, modules=body.modules
)
except PermissionError as exc:
raise HTTPException(status_code=403, detail=str(exc))
except KeyError as exc:
raise HTTPException(status_code=404, detail=str(exc))
except ValueError as exc:
raise HTTPException(status_code=400, detail=str(exc))
return JSONResponse({"user": rec})
@app.delete("/api/users/{email}")
async def api_delete_user(
email: str,
_: dict[str, Any] = Depends(require_admin),
) -> JSONResponse:
try:
user_store.delete(email)
except PermissionError as exc:
raise HTTPException(status_code=403, detail=str(exc))
return JSONResponse({"ok": True})
Reference in New Issue View Git Blame Copy Permalink