dbx-main/app/modules/cafe24/common.py

"""카페24 모듈 공용 가드/컨텍스트 헬퍼.

router.py 와 routes_*.py 가 함께 쓴다(순환 import 방지를 위해 분리).
다른 모듈과 동일한 규칙:
  - JSON API   → require_user() : 401/403 HTTPException
  - HTML 페이지 → guard()       : 리다이렉트 / denied.html 응답 반환
app.main 은 함수 안에서 지연 import 한다(순환 import 방지).
"""

from __future__ import annotations

from typing import Any

from fastapi import HTTPException, Request
from fastapi.responses import HTMLResponse, RedirectResponse

MODULE_KEY = "cafe24"
MODULE_NAME = "카페24 상품관리"

CONFIG_HELP = (
    "카페24 모듈이 아직 설정되지 않았습니다. "
    "CAFE24_DB_URL 환경변수를 설정하고 "
    "scripts/sql/cafe24_db_init.sql 로 cafe24_db 를 초기화한 뒤 "
    "컨테이너를 재기동하세요."
)


def get_store(request: Request) -> Any:
    return getattr(request.app.state, "cafe24_store", None)


def read_lag_grace_minutes() -> int:
    """카페24 읽기 지연 유예시간(분). `CAFE24_READ_LAG_GRACE_MIN`, 기본 360.

    이 시간 안에 우리가 쓴 값이 있으면, 카페24 GET 이 예전 값을 돌려줘도 우리
    마지막 쓰기를 화면·적용 기준으로 삼는다(store.resolve_description 참고).
    """
    import os  # noqa: WPS433

    from . import store  # noqa: WPS433

    return store.parse_grace_minutes(os.getenv("CAFE24_READ_LAG_GRACE_MIN"))


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, MODULE_KEY):
        raise HTTPException(status_code=403, detail=f"{MODULE_NAME} 모듈 권한이 없습니다.")
    return user


def require_admin(request: Request) -> dict[str, Any]:
    """카페24 연결(OAuth)·연결 해제는 관리자만."""
    from app.store import is_admin  # noqa: WPS433

    user = require_user(request)
    if not is_admin(user):
        raise HTTPException(status_code=403, detail="관리자만 카페24 연결을 변경할 수 있습니다.")
    return user


def require_store(request: Request) -> tuple[Any, dict[str, Any]]:
    """JSON API 용 — store 미설정이면 503."""
    user = require_user(request)
    st = get_store(request)
    if st is None:
        raise HTTPException(status_code=503, detail=CONFIG_HELP)
    return st, user


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": CONFIG_HELP,
            "user": user,
            "is_admin": is_admin(user),
            "nav_items": build_erp_nav(user, active=MODULE_KEY),
        },
        status_code=503,
    )


def guard(request: Request):
    """로그인+권한+store 점검. 페이지 핸들러 진입부에서 사용.

    반환이 tuple 이면 (store, user), 아니면 그대로 응답으로 돌려준다.
    """
    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, MODULE_KEY):
        return render_template(
            request,
            "denied.html",
            {
                "reason": f"{MODULE_NAME} 접근 권한이 없습니다.",
                "user": user,
                "is_admin": is_admin(user),
            },
            status_code=403,
        )
    st = get_store(request)
    if st is None:
        return render_config_needed(request, user)
    return st, user


def base_ctx(request: Request, user: dict[str, Any], *, active_tab: str = "") -> 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_super": bool(user.get("is_super_admin")),
        "nav_items": build_erp_nav(user, active=MODULE_KEY),
        "active_tab": active_tab,
    }