ctms/backend/app/core/project_permissions.py

from __future__ import annotations

import uuid

from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

from app.models.api_endpoint_permission import ApiEndpointPermission
from app.models.study import Study
from app.core.api_permissions import (
    API_ENDPOINT_PERMISSIONS,
    OPERATION_PREREQUISITES,
    PROJECT_PERMISSION_ROLES,
    canonical_permission_key,
)
from app.core.permission_cache import get_permission_cache


class PrerequisitePermissionConfirmationRequired(ValueError):
    """权限调整需要授权人确认。"""

    def __init__(self, adjustments: dict):
        super().__init__("需要确认前置权限调整")
        self.adjustments = adjustments


async def _get_project_permission_overrides(
    db: AsyncSession,
    study_id: uuid.UUID,
) -> dict[str, dict[str, bool]]:
    cache = get_permission_cache()
    cached_permissions = cache.get_project_permissions(study_id)
    from app.core.permission_monitor import get_permission_monitor

    monitor = get_permission_monitor()
    if cached_permissions is not None:
        monitor.record_cache_hit()
        return cached_permissions

    monitor.record_cache_miss()
    result = await db.execute(
        select(ApiEndpointPermission).where(
            ApiEndpointPermission.study_id == study_id,
        )
    )
    rows = result.scalars().all()

    permissions: dict[str, dict[str, bool]] = {}
    for row in rows:
        endpoint_key = canonical_permission_key(row.endpoint_key)
        role_permissions = permissions.setdefault(row.role, {})
        if row.endpoint_key == endpoint_key or endpoint_key not in role_permissions:
            role_permissions[endpoint_key] = row.allowed

    cache.set_project_permissions(study_id, permissions)
    return permissions


def _empty_role_adjustment() -> dict[str, dict[str, list[str]]]:
    return {"add": {}, "remove": {}}


def _flatten_role_permissions(
    matrix: dict[str, dict[str, dict[str, bool]]],
    role: str,
) -> dict[str, bool]:
    permissions: dict[str, bool] = {
        endpoint_key: False
        for endpoint_key in API_ENDPOINT_PERMISSIONS.keys()
    }
    if role in matrix:
        for endpoint_key, value in matrix[role].items():
            endpoint_key = canonical_permission_key(endpoint_key)
            if endpoint_key not in API_ENDPOINT_PERMISSIONS:
                continue
            permissions[endpoint_key] = bool(value["allowed"] if isinstance(value, dict) else value)
    return permissions


def _canonicalize_payload(payload: dict[str, dict[str, bool]]) -> dict[str, dict[str, bool]]:
    normalized: dict[str, dict[str, bool]] = {}
    for role, endpoints in payload.items():
        normalized[role] = {}
        for endpoint_key, allowed in endpoints.items():
            endpoint_key = canonical_permission_key(endpoint_key)
            if endpoint_key not in API_ENDPOINT_PERMISSIONS:
                continue
            normalized[role][endpoint_key] = bool(allowed)
    return normalized


def _add_adjustment(target: dict[str, list[str]], key: str, value: str) -> None:
    values = target.setdefault(key, [])
    if value not in values:
        values.append(value)


async def preview_api_permission_prerequisite_adjustments(
    db: AsyncSession,
    study_id: uuid.UUID,
    payload: dict[str, dict[str, bool]],
) -> dict:
    """预检权限草稿需要授权人确认的前置补齐和下游消除项。"""
    normalized_payload = _canonicalize_payload(payload)
    matrix = await get_api_endpoint_permissions(db, study_id)
    roles: dict[str, dict[str, dict[str, list[str]]]] = {}

    for role, endpoints in normalized_payload.items():
        if role in ("ADMIN", "PM"):
            continue

        candidate = _flatten_role_permissions(matrix, role)
        explicitly_denied = {
            endpoint_key
            for endpoint_key, allowed in endpoints.items()
            if not allowed
        }
        for endpoint_key, allowed in endpoints.items():
            candidate[endpoint_key] = allowed

        role_adjustment = _empty_role_adjustment()
        denied_sources = set(explicitly_denied)

        changed = True
        while changed:
            changed = False
            for endpoint_key, allowed in list(candidate.items()):
                if not allowed or endpoint_key in denied_sources:
                    continue
                for prereq in OPERATION_PREREQUISITES.get(endpoint_key, []):
                    prereq = canonical_permission_key(prereq)
                    if prereq in denied_sources:
                        _add_adjustment(role_adjustment["remove"], prereq, endpoint_key)
                        candidate[endpoint_key] = False
                        denied_sources.add(endpoint_key)
                        changed = True
                        break

        changed = True
        while changed:
            changed = False
            for endpoint_key, allowed in list(candidate.items()):
                if not allowed:
                    continue
                for prereq in OPERATION_PREREQUISITES.get(endpoint_key, []):
                    prereq = canonical_permission_key(prereq)
                    if candidate.get(prereq, False):
                        continue
                    _add_adjustment(role_adjustment["add"], endpoint_key, prereq)
                    candidate[prereq] = True
                    changed = True

        if role_adjustment["add"] or role_adjustment["remove"]:
            roles[role] = role_adjustment

    return {
        "requires_confirmation": bool(roles),
        "roles": roles,
    }


def apply_prerequisite_adjustments_to_payload(
    payload: dict[str, dict[str, bool]],
    adjustments: dict,
) -> dict[str, dict[str, bool]]:
    """把已确认的补齐和消除项合并到待保存 payload。"""
    normalized = _canonicalize_payload(payload)
    for role, role_adjustment in adjustments.get("roles", {}).items():
        role_payload = normalized.setdefault(role, {})
        for prerequisites in role_adjustment.get("add", {}).values():
            for prereq in prerequisites:
                role_payload[canonical_permission_key(prereq)] = True
        for dependents in role_adjustment.get("remove", {}).values():
            for dependent in dependents:
                role_payload[canonical_permission_key(dependent)] = False
    return normalized


async def role_has_api_permission(
    db: AsyncSession,
    study_id: uuid.UUID,
    role: str | None,
    endpoint_key: str,
    check_prerequisites: bool = True,
) -> bool:
    """检查角色是否有权访问特定接口"""
    endpoint_key = canonical_permission_key(endpoint_key)
    if endpoint_key not in API_ENDPOINT_PERMISSIONS:
        return False

    if role == "ADMIN":
        return True

    permissions = await _get_project_permission_overrides(db, study_id)

    if role == "PM":
        override = permissions.get("PM", {}).get(endpoint_key)
        if override is not None:
            return override
        return True

    allowed = permissions.get(role or "", {}).get(endpoint_key)
    if allowed is None:
        config = API_ENDPOINT_PERMISSIONS.get(endpoint_key)
        allowed = bool(config and role in config.get("default_roles", []))

    if not allowed:
        return False

    if check_prerequisites:
        prerequisites = OPERATION_PREREQUISITES.get(endpoint_key, [])
        for prereq_endpoint in prerequisites:
            has_prereq = await role_has_api_permission(
                db, study_id, role, prereq_endpoint, check_prerequisites=False
            )
            if not has_prereq:
                return False

    return True


async def get_missing_prerequisites(
    db: AsyncSession,
    study_id: uuid.UUID,
    role: str | None,
    endpoint_key: str,
) -> list[str]:
    """获取缺失的前置权限列表"""
    if role == "ADMIN":
        return []

    missing = []
    endpoint_key = canonical_permission_key(endpoint_key)
    prerequisites = OPERATION_PREREQUISITES.get(endpoint_key, [])
    for prereq_endpoint in prerequisites:
        has_prereq = await role_has_api_permission(
            db, study_id, role, prereq_endpoint, check_prerequisites=False
        )
        if not has_prereq:
            missing.append(prereq_endpoint)

    return missing


async def get_api_endpoint_permissions(
    db: AsyncSession,
    study_id: uuid.UUID,
) -> dict[str, dict[str, dict[str, bool]]]:
    """获取项目的接口级权限矩阵

    返回格式: {role: {endpoint_key: {allowed: bool}}}
    """
    overrides = await _get_project_permission_overrides(db, study_id)

    study_result = await db.execute(select(Study).where(Study.id == study_id))
    study = study_result.scalar_one_or_none()
    active_roles = [role for role in (study.active_roles if study else []) if isinstance(role, str) and role.strip()]
    roles = list(dict.fromkeys([*PROJECT_PERMISSION_ROLES, *active_roles, *overrides.keys()]))
    matrix: dict[str, dict[str, dict[str, bool]]] = {}
    for role in roles:
        matrix[role] = {}
        for endpoint_key, config in API_ENDPOINT_PERMISSIONS.items():
            if role == "PM":
                matrix[role][endpoint_key] = {"allowed": True}
            else:
                default_allowed = role in config.get("default_roles", [])
                matrix[role][endpoint_key] = {"allowed": default_allowed}

    for role, endpoints in overrides.items():
        if role not in matrix:
            matrix[role] = {}
        for endpoint_key, allowed in endpoints.items():
            endpoint_key = canonical_permission_key(endpoint_key)
            if endpoint_key not in API_ENDPOINT_PERMISSIONS:
                continue
            matrix[role][endpoint_key] = {"allowed": allowed}

    return matrix


async def get_effective_api_endpoint_permissions(
    db: AsyncSession,
    study_id: uuid.UUID,
) -> dict[str, dict[str, dict[str, bool]]]:
    """获取考虑前置权限后的有效权限矩阵。"""
    matrix = await get_api_endpoint_permissions(db, study_id)
    effective: dict[str, dict[str, dict[str, bool]]] = {}
    for role, endpoints in matrix.items():
        effective[role] = {}
        for endpoint_key in endpoints.keys():
            effective[role][endpoint_key] = {
                "allowed": await role_has_api_permission(
                    db,
                    study_id,
                    role,
                    endpoint_key,
                    check_prerequisites=True,
                )
            }
    return effective


async def replace_api_endpoint_permissions(
    db: AsyncSession,
    study_id: uuid.UUID,
    payload: dict[str, dict[str, bool]],
    *,
    confirm_prerequisite_adjustments: bool = False,
) -> dict[str, dict[str, dict[str, bool]]]:
    """更新 payload 中指定的角色权限项。未提交的权限项保持不变。

    ADMIN 与 PM 权限不会被持久化：ADMIN 始终拥有全部权限，PM 默认拥有
    全部项目权限，应当通过专门的渠道而不是项目权限矩阵调整。
    """
    normalized_payload = _canonicalize_payload(payload)
    adjustments = await preview_api_permission_prerequisite_adjustments(
        db,
        study_id,
        normalized_payload,
    )
    if adjustments["requires_confirmation"]:
        if not confirm_prerequisite_adjustments:
            raise PrerequisitePermissionConfirmationRequired(adjustments)
        normalized_payload = apply_prerequisite_adjustments_to_payload(
            normalized_payload,
            adjustments,
        )

    for role, endpoints in normalized_payload.items():
        if role in ("ADMIN", "PM"):
            continue
        for endpoint_key, allowed in endpoints.items():
            if endpoint_key not in API_ENDPOINT_PERMISSIONS:
                continue
            result = await db.execute(
                select(ApiEndpointPermission).where(
                    ApiEndpointPermission.study_id == study_id,
                    ApiEndpointPermission.role == role,
                    ApiEndpointPermission.endpoint_key == endpoint_key,
                )
            )
            existing = result.scalar_one_or_none()
            if existing:
                existing.allowed = allowed
            else:
                db.add(
                    ApiEndpointPermission(
                        study_id=study_id,
                        role=role,
                        endpoint_key=endpoint_key,
                        allowed=allowed,
                    )
                )

    await db.commit()

    cache = get_permission_cache()
    cache.invalidate_project_permissions(study_id)
    cache.invalidate_all_member_roles(study_id)
    from app.core.permission_monitor import get_permission_monitor

    get_permission_monitor().record_cache_invalidation()

    return await get_api_endpoint_permissions(db, study_id)