king
9c58b022e2
- ERP 셸: 좌측 사이드바 + 상단 헤더 + 콘텐츠 영역(erp_base.html)
- 분기: 슈퍼관리자 → 업무모듈 카드, 일반사용자 → ERP 메인(erp_home.html)
- 신규 라우트: /modules (슈퍼관리자 모듈 카드 재진입)
- 개인경비 모듈: app/modules/expense/ (라우터/저장소/템플릿)
- JSON 저장소(ExpenseStore) + PostgreSQL 저장소(ExpenseDBStore)
- 팩토리: EXPENSE_DB_URL 있으면 DB, 없으면 JSON 폴백
- DB: scripts/sql/expense_db_init.sql (멱등 DDL, postgres-db 컨테이너)
- expense_db, 역할 expense_app, 테이블 expense_items, 인덱스/트리거
- 마이그레이션: scripts/migrate_expense_json_to_db.py (멱등, --dry-run)
- 의존성: psycopg[binary,pool]>=3.2
- 환경변수: EXPENSE_DB_URL 추가
- 문서: CLAUDE.md 작업 기준 + docs/{PROJECT_OVERVIEW,SERVER_ARCHITECTURE,DATABASES,DEPLOYMENT}.md
- main-app 배포 경로 /opt/www/main 고정 명시
- 위험 명령(DROP/TRUNCATE/rm -rf/volume 삭제) 사용자 승인 정책
508 lines
17 KiB
Python
508 lines
17 KiB
Python
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.expense import build_expense_store
|
|
from .modules.expense import router as expense_router
|
|
from .store import (
|
|
MODULE_KEYS,
|
|
SUPER_ADMIN_EMAIL,
|
|
UserStore,
|
|
allowed_modules,
|
|
has_module,
|
|
is_admin,
|
|
)
|
|
|
|
# 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",
|
|
]
|
|
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()],
|
|
]
|
|
)
|
|
|
|
oauth = build_google_oauth()
|
|
user_store = UserStore(DATA_DIR / "users.json")
|
|
|
|
# 모듈별 데이터 저장소.
|
|
# EXPENSE_DB_URL 가 있으면 expense_db(PostgreSQL), 없으면 JSON 파일.
|
|
app.state.expense_store = build_expense_store(
|
|
dsn=env("EXPENSE_DB_URL") or None,
|
|
json_path=DATA_DIR / "expense.json",
|
|
)
|
|
|
|
# 모듈 라우터 등록 — 신규 모듈 추가 시 여기 한 줄.
|
|
app.include_router(expense_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": "vacation",
|
|
"title": "휴가",
|
|
"subtitle": "Vacation",
|
|
"description": "연차/반차/특별휴가 신청과 잔여일수, 결재 현황을 확인합니다.",
|
|
"url": "#",
|
|
"health_url": None,
|
|
"status": "preparing",
|
|
"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"/>',
|
|
"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,
|
|
}
|
|
)
|
|
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),
|
|
"super_admin_email": SUPER_ADMIN_EMAIL,
|
|
"is_admin": True,
|
|
},
|
|
)
|
|
|
|
|
|
# ── 관리자 API ──────────────────────────────────────────────
|
|
class UpdatePermissionsBody(BaseModel):
|
|
role: str | None = None
|
|
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.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})
|