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.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.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": "말레이시아 배송", "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" / "" / "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", ] 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()], ] ) # 배송사명 → 로고 URL(dispatch 모듈). 템플릿에서 {{ provider | courier_logo }}. from .modules.dispatch.store import courier_logo_url as _courier_logo_url # noqa: E402 templates.env.filters["courier_logo"] = _courier_logo_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) # 모듈 라우터 등록 — 신규 모듈 추가 시 여기 한 줄. 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) 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": "운영", }, ] 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": '', "expense": '', "vacation": '', "corm": '', "order": '', "cupang": '', "malaysia": '', "dispatch": '', "modules": '', } body = paths.get(name, paths["modules"]) return ( '' f"{body}" ) 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})