dbx-main/app/store.py

"""JSON 기반 사용자/권한 저장소.

- app/data/users.json 에 사용자 레코드 저장
- 동시성: 파일락 대신 process-내 threading.Lock + 원자적 쓰기(temp → rename)
- 운영시 컨테이너 한 대 가정. 다중 인스턴스 필요해지면 DB 로 교체.
"""
from __future__ import annotations

import json
import os
import tempfile
import threading
from pathlib import Path
from typing import Any, Iterable

from .timezone import now_kst_iso

# 슈퍼 관리자 — 강등/삭제 불가
SUPER_ADMIN_EMAIL = "king@dbxcorp.co.kr"

# 시스템에서 지원하는 권한 키.
# - 접근권한: 모듈 페이지 진입 허용
# - 승인자: 결재 워크플로에서 승인/반려 가능
# 신규 추가 시 여기 + admin.html MODULE_LABELS + 라우터 검사 동시 갱신.
MODULE_KEYS: tuple[str, ...] = (
    "corm",
    "order",
    "expense",
    "vacation",
    "cupang",
    "malaysia",
    "expense_approver",
    "vacation_approver",
)

# 승인자 권한 키 — 결재 워크플로에서 별도 검사할 때 사용
APPROVER_KEYS: tuple[str, ...] = ("expense_approver", "vacation_approver")


def _now_iso() -> str:
    return now_kst_iso()


class UserStore:
    def __init__(self, path: Path):
        self._path = path
        self._lock = threading.Lock()
        self._path.parent.mkdir(parents=True, exist_ok=True)
        if not self._path.exists():
            self._write_atomic({"users": {}})

    def _read(self) -> dict[str, Any]:
        try:
            with self._path.open("r", encoding="utf-8") as f:
                data = json.load(f)
        except (FileNotFoundError, json.JSONDecodeError):
            data = {"users": {}}
        if "users" not in data or not isinstance(data["users"], dict):
            data["users"] = {}
        return data

    def _write_atomic(self, data: dict[str, Any]) -> None:
        fd, tmp = tempfile.mkstemp(
            prefix=".users.", suffix=".json.tmp", dir=str(self._path.parent)
        )
        try:
            with os.fdopen(fd, "w", encoding="utf-8") as f:
                json.dump(data, f, ensure_ascii=False, indent=2)
            os.replace(tmp, self._path)
        except Exception:
            try:
                os.unlink(tmp)
            except OSError:
                pass
            raise

    @staticmethod
    def _default_modules(is_admin: bool) -> dict[str, bool]:
        return {key: is_admin for key in MODULE_KEYS}

    @staticmethod
    def _normalize_modules(modules: dict[str, Any] | None, *, is_admin: bool) -> dict[str, bool]:
        base = UserStore._default_modules(is_admin)
        if isinstance(modules, dict):
            for key in MODULE_KEYS:
                if key in modules:
                    base[key] = bool(modules[key])
        if is_admin:
            for key in MODULE_KEYS:
                base[key] = True
        return base

    def get(self, email: str) -> dict[str, Any] | None:
        email = email.lower().strip()
        with self._lock:
            data = self._read()
            user = data["users"].get(email)
            if user is None:
                return None
            return self._enrich(email, user)

    def list_all(self) -> list[dict[str, Any]]:
        with self._lock:
            data = self._read()
            return [self._enrich(email, rec) for email, rec in data["users"].items()]

    def create_user(
        self,
        *,
        email: str,
        name: str = "",
        role: str = "user",
        modules: dict[str, Any] | None = None,
    ) -> dict[str, Any]:
        """관리자가 이메일만으로 신규 사용자 등록. 로그인 전이라도 권한 부여 가능."""
        email = email.lower().strip()
        if not email or "@" not in email:
            raise ValueError("올바른 이메일 형식이 아닙니다.")
        if role not in ("admin", "user"):
            raise ValueError(f"role 값이 잘못되었습니다: {role}")
        is_super = email == SUPER_ADMIN_EMAIL
        with self._lock:
            data = self._read()
            if email in data["users"]:
                raise ValueError(f"이미 등록된 사용자입니다: {email}")
            now = _now_iso()
            final_role = "admin" if is_super else role
            rec = {
                "email": email,
                "name": name or email.split("@")[0],
                "picture": "",
                "role": final_role,
                "modules": self._normalize_modules(
                    modules, is_admin=(final_role == "admin")
                ),
                "created_at": now,
                "last_login": "",
            }
            data["users"][email] = rec
            self._write_atomic(data)
            return self._enrich(email, rec)

    def upsert_login(self, *, email: str, name: str, picture: str) -> dict[str, Any]:
        """로그인 시 호출. 신규면 생성, 기존이면 last_login/name/picture 갱신."""
        email = email.lower().strip()
        is_super = email == SUPER_ADMIN_EMAIL
        with self._lock:
            data = self._read()
            now = _now_iso()
            existing = data["users"].get(email)
            if existing is None:
                rec = {
                    "email": email,
                    "name": name,
                    "picture": picture,
                    "role": "admin" if is_super else "user",
                    "modules": self._default_modules(is_admin=is_super),
                    "created_at": now,
                    "last_login": now,
                }
            else:
                rec = dict(existing)
                rec["name"] = name or rec.get("name", email)
                rec["picture"] = picture or rec.get("picture", "")
                rec["last_login"] = now
                # 슈퍼 관리자는 항상 admin + 모든 모듈
                if is_super:
                    rec["role"] = "admin"
                    rec["modules"] = self._default_modules(is_admin=True)
                else:
                    rec["modules"] = self._normalize_modules(
                        rec.get("modules"), is_admin=(rec.get("role") == "admin")
                    )
            data["users"][email] = rec
            self._write_atomic(data)
            return self._enrich(email, rec)

    def update_permissions(
        self, *, email: str, role: str | None, modules: dict[str, Any] | None
    ) -> dict[str, Any]:
        """관리자가 다른 사용자의 권한을 수정."""
        email = email.lower().strip()
        if email == SUPER_ADMIN_EMAIL:
            raise PermissionError("슈퍼 관리자 계정은 수정할 수 없습니다.")
        if role is not None and role not in ("admin", "user"):
            raise ValueError(f"role 값이 잘못되었습니다: {role}")
        with self._lock:
            data = self._read()
            rec = data["users"].get(email)
            if rec is None:
                raise KeyError(f"사용자를 찾을 수 없습니다: {email}")
            if role is not None:
                rec["role"] = role
            new_is_admin = rec.get("role") == "admin"
            rec["modules"] = self._normalize_modules(
                modules if modules is not None else rec.get("modules"),
                is_admin=new_is_admin,
            )
            data["users"][email] = rec
            self._write_atomic(data)
            return self._enrich(email, rec)

    def delete(self, email: str) -> None:
        email = email.lower().strip()
        if email == SUPER_ADMIN_EMAIL:
            raise PermissionError("슈퍼 관리자 계정은 삭제할 수 없습니다.")
        with self._lock:
            data = self._read()
            if email in data["users"]:
                del data["users"][email]
                self._write_atomic(data)

    @staticmethod
    def _enrich(email: str, rec: dict[str, Any]) -> dict[str, Any]:
        out = dict(rec)
        out["email"] = email
        out["is_super_admin"] = email == SUPER_ADMIN_EMAIL
        out["modules"] = UserStore._normalize_modules(
            out.get("modules"), is_admin=(out.get("role") == "admin")
        )
        return out


def is_admin(user_rec: dict[str, Any] | None) -> bool:
    return bool(user_rec and user_rec.get("role") == "admin")


def has_module(user_rec: dict[str, Any] | None, module: str) -> bool:
    if not user_rec:
        return False
    if is_admin(user_rec):
        return True
    mods = user_rec.get("modules") or {}
    if mods.get(module):
        return True
    # 승인자(<module>_approver) 권한이 있으면 해당 모듈 접근도 허용 —
    # 본인 경비 등록/제출도 가능해야 하므로.
    if module in APPROVER_KEYS:
        return False
    if mods.get(f"{module}_approver"):
        return True
    return False


def is_approver(user_rec: dict[str, Any] | None, kind: str) -> bool:
    """결재 승인자 권한 검사. kind 예: 'expense', 'vacation'."""
    return has_module(user_rec, f"{kind}_approver")


def allowed_modules(user_rec: dict[str, Any] | None) -> set[str]:
    return {m for m in MODULE_KEYS if has_module(user_rec, m)}