shanxia2.0/backend/app/core/delete_guard.py

"""删除权限：仅创建人可删（super_admin 例外）。"""
from __future__ import annotations

from typing import Any, Dict, Iterable, Optional, Union

from fastapi import HTTPException

RowLike = Union[Any, Dict[str, Any]]


def normalize_user_info(
    user_info: Optional[Dict[str, Any]] = None,
    *,
    user_id: Optional[str] = None,
    actor_name: Optional[str] = None,
) -> Dict[str, Any]:
    info: Dict[str, Any] = dict(user_info or {})
    uid = (user_id or info.get("user_id") or info.get("id") or "").strip()
    if uid:
        info["user_id"] = uid
    name = (actor_name or info.get("real_name") or info.get("username") or info.get("userName") or info.get("name") or "").strip()
    if name:
        info["real_name"] = name
    return info


def user_info_from_context(ctx_or_dict: Any) -> Dict[str, Any]:
    if isinstance(ctx_or_dict, dict):
        return normalize_user_info(ctx_or_dict)
    info: Dict[str, Any] = {}
    if hasattr(ctx_or_dict, "to_user_info"):
        info.update(ctx_or_dict.to_user_info())
    name = (getattr(ctx_or_dict, "name", None) or "").strip()
    if name:
        info["real_name"] = name
    return normalize_user_info(info)


def is_super_admin(user_info: Optional[Dict[str, Any]]) -> bool:
    if not user_info:
        return False
    return (user_info.get("role") or "").strip() == "super_admin"


def row_creator_id(row: RowLike) -> str:
    if row is None:
        return ""
    if isinstance(row, dict):
        return (row.get("create_by") or row.get("createBy") or "").strip()
    return (getattr(row, "create_by", None) or getattr(row, "createBy", None) or "").strip()


def _actor_ids(user_info: Optional[Dict[str, Any]]) -> set[str]:
    if not user_info:
        return set()
    ids: set[str] = set()
    for key in ("user_id", "id"):
        val = (user_info.get(key) or "").strip()
        if val:
            ids.add(val)
    return ids


def _actor_names(user_info: Optional[Dict[str, Any]]) -> set[str]:
    if not user_info:
        return set()
    names: set[str] = set()
    for key in ("real_name", "username", "userName", "name"):
        val = (user_info.get(key) or "").strip()
        if val:
            names.add(val)
    return names


def can_delete_by_creator(
    row: RowLike,
    user_info: Optional[Dict[str, Any]] = None,
    *,
    user_id: Optional[str] = None,
    actor_name: Optional[str] = None,
) -> bool:
    info = normalize_user_info(user_info, user_id=user_id, actor_name=actor_name)
    if is_super_admin(info):
        return True
    creator = row_creator_id(row)
    if not creator:
        return False
    if creator in _actor_ids(info):
        return True
    if creator in _actor_names(info):
        return True
    return False


def assert_row_deletable_by_creator(
    row: RowLike,
    user_info: Optional[Dict[str, Any]] = None,
    *,
    user_id: Optional[str] = None,
    actor_name: Optional[str] = None,
    message: str = "仅创建人可删除",
) -> None:
    if can_delete_by_creator(row, user_info, user_id=user_id, actor_name=actor_name):
        return
    raise HTTPException(status_code=403, detail=message)


def assert_rows_deletable_by_creator(
    rows: Iterable[RowLike],
    user_info: Optional[Dict[str, Any]] = None,
    *,
    user_id: Optional[str] = None,
    actor_name: Optional[str] = None,
    message: str = "仅创建人可删除",
) -> None:
    info = normalize_user_info(user_info, user_id=user_id, actor_name=actor_name)
    for row in rows:
        assert_row_deletable_by_creator(row, info, message=message)