ctms/backend/app/core/project_permissions.py

from __future__ import annotations

import uuid

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

from app.models.study_role_permission import StudyRolePermission

PROJECT_PERMISSION_ROLES = ("ADMIN", "PM", "CRA", "PV", "MEDICAL_REVIEW", "IMP", "QA")

PROJECT_PERMISSION_MODULES = (
    {"key": "project_members", "label": "项目成员", "description": "维护项目账号、成员角色与启停状态"},
    {"key": "sites", "label": "中心管理", "description": "维护中心资料与 CRA 绑定"},
    {"key": "audit_export", "label": "审计日志导出", "description": "导出项目审计日志"},
    {"key": "project_overview", "label": "项目总览", "description": "查看项目整体进度与中心概览", "writable": False},
    {"key": "project_milestones", "label": "项目里程碑", "description": "维护项目级里程碑"},
    {"key": "fees", "label": "合同费用管理", "description": "维护合同、费用与付款"},
    {"key": "materials", "label": "物资管理", "description": "维护药品出入库与物资设备"},
    {"key": "file_versions", "label": "文件版本管理", "description": "维护文件版本、分发与确认"},
    {"key": "startup_ethics", "label": "立项与伦理", "description": "维护立项、可行性与伦理资料"},
    {"key": "startup_auth", "label": "启动与授权", "description": "维护启动会、培训与授权"},
    {"key": "subjects", "label": "参与者管理", "description": "维护参与者、访视与 PD"},
    {"key": "risk_issues", "label": "风险问题", "description": "维护 SAE、PD 与监查问题"},
    {"key": "monitoring_audit", "label": "监查稽查", "description": "维护监查稽查记录"},
    {"key": "etmf", "label": "eTMF", "description": "维护 eTMF 文件"},
    {"key": "faq", "label": "FAQ", "description": "维护项目 FAQ 分类、问题与回复"},
    {"key": "shared_library", "label": "共享库", "description": "维护注意事项、支持性文件与说明文件"},
)
MANAGEMENT_BACKEND_PERMISSION_MODULES = {"project_members", "sites", "audit_export"}
READ_ONLY_PERMISSION_MODULES = {
    module["key"]
    for module in PROJECT_PERMISSION_MODULES
    if module.get("writable") is False
}

DEFAULT_PROJECT_ROLE_PERMISSIONS: dict[str, dict[str, dict[str, bool]]] = {
    "ADMIN": {
        module["key"]: {"read": True, "write": module["key"] not in READ_ONLY_PERMISSION_MODULES}
        for module in PROJECT_PERMISSION_MODULES
    },
    "PM": {
        "project_overview": {"read": True, "write": False},
        "project_milestones": {"read": True, "write": True},
        "fees": {"read": True, "write": True},
        "materials": {"read": True, "write": True},
        "file_versions": {"read": True, "write": True},
        "startup_ethics": {"read": True, "write": True},
        "startup_auth": {"read": True, "write": True},
        "subjects": {"read": True, "write": True},
        "risk_issues": {"read": True, "write": True},
        "monitoring_audit": {"read": True, "write": True},
        "etmf": {"read": True, "write": True},
        "faq": {"read": True, "write": True},
        "shared_library": {"read": True, "write": True},
        "project_members": {"read": True, "write": True},
        "sites": {"read": True, "write": True},
        "audit_export": {"read": True, "write": False},
    },
    "CRA": {
        "project_overview": {"read": True, "write": False},
        "project_milestones": {"read": True, "write": True},
        "fees": {"read": True, "write": True},
        "materials": {"read": True, "write": True},
        "file_versions": {"read": True, "write": True},
        "startup_ethics": {"read": True, "write": True},
        "startup_auth": {"read": True, "write": True},
        "subjects": {"read": True, "write": True},
        "risk_issues": {"read": True, "write": True},
        "monitoring_audit": {"read": True, "write": True},
        "etmf": {"read": True, "write": True},
        "faq": {"read": True, "write": True},
        "shared_library": {"read": True, "write": True},
        "project_members": {"read": False, "write": False},
        "sites": {"read": False, "write": False},
        "audit_export": {"read": True, "write": False},
    },
    "PV": {
        "project_overview": {"read": True, "write": False},
        "project_milestones": {"read": True, "write": False},
        "fees": {"read": True, "write": False},
        "materials": {"read": True, "write": False},
        "file_versions": {"read": True, "write": False},
        "startup_ethics": {"read": True, "write": False},
        "startup_auth": {"read": True, "write": False},
        "subjects": {"read": True, "write": False},
        "risk_issues": {"read": True, "write": True},
        "monitoring_audit": {"read": True, "write": True},
        "etmf": {"read": True, "write": False},
        "faq": {"read": True, "write": True},
        "shared_library": {"read": True, "write": True},
        "project_members": {"read": False, "write": False},
        "sites": {"read": False, "write": False},
        "audit_export": {"read": False, "write": False},
    },
    "MEDICAL_REVIEW": {
        "project_overview": {"read": True, "write": False},
        "project_milestones": {"read": False, "write": False},
        "fees": {"read": False, "write": False},
        "materials": {"read": False, "write": False},
        "file_versions": {"read": False, "write": False},
        "startup_ethics": {"read": False, "write": False},
        "startup_auth": {"read": False, "write": False},
        "subjects": {"read": True, "write": False},
        "risk_issues": {"read": True, "write": True},
        "monitoring_audit": {"read": True, "write": False},
        "etmf": {"read": False, "write": False},
        "faq": {"read": True, "write": True},
        "shared_library": {"read": True, "write": True},
        "project_members": {"read": False, "write": False},
        "sites": {"read": False, "write": False},
        "audit_export": {"read": False, "write": False},
    },
    "IMP": {
        "project_overview": {"read": True, "write": False},
        "project_milestones": {"read": True, "write": False},
        "fees": {"read": True, "write": False},
        "materials": {"read": True, "write": True},
        "file_versions": {"read": True, "write": True},
        "startup_ethics": {"read": True, "write": False},
        "startup_auth": {"read": True, "write": True},
        "subjects": {"read": True, "write": False},
        "risk_issues": {"read": True, "write": False},
        "monitoring_audit": {"read": True, "write": False},
        "etmf": {"read": True, "write": False},
        "faq": {"read": True, "write": True},
        "shared_library": {"read": True, "write": True},
        "project_members": {"read": False, "write": False},
        "sites": {"read": False, "write": False},
        "audit_export": {"read": False, "write": False},
    },
    "QA": {
        "project_overview": {"read": True, "write": False},
        "project_milestones": {"read": True, "write": False},
        "fees": {"read": True, "write": False},
        "materials": {"read": True, "write": False},
        "file_versions": {"read": True, "write": False},
        "startup_ethics": {"read": True, "write": False},
        "startup_auth": {"read": True, "write": False},
        "subjects": {"read": True, "write": False},
        "risk_issues": {"read": True, "write": False},
        "monitoring_audit": {"read": True, "write": True},
        "etmf": {"read": True, "write": False},
        "faq": {"read": True, "write": False},
        "shared_library": {"read": True, "write": True},
        "project_members": {"read": False, "write": False},
        "sites": {"read": False, "write": False},
        "audit_export": {"read": False, "write": False},
    },
}


def _empty_action_state() -> dict[str, bool]:
    return {"read": False, "write": False}


def _normalize_action_state(value: dict | None, role: str) -> dict[str, bool]:
    if role == "ADMIN":
        return {"read": True, "write": True}
    read = bool((value or {}).get("read", False))
    write = bool((value or {}).get("write", False))
    if write:
        read = True
    return {"read": read, "write": write}


def _normalize_module_action_state(value: dict | None, role: str, module: str) -> dict[str, bool]:
    if role == "ADMIN":
        return {"read": True, "write": module not in READ_ONLY_PERMISSION_MODULES}
    if module in MANAGEMENT_BACKEND_PERMISSION_MODULES and role != "PM":
        return _empty_action_state()
    if module in READ_ONLY_PERMISSION_MODULES:
        return {"read": bool((value or {}).get("read", False) or (value or {}).get("write", False)), "write": False}
    return _normalize_action_state(value, role)


def normalize_permission_matrix(matrix: dict | None = None) -> dict[str, dict[str, dict[str, bool]]]:
    normalized: dict[str, dict[str, dict[str, bool]]] = {}
    matrix = matrix or {}
    for role in PROJECT_PERMISSION_ROLES:
        normalized[role] = {}
        role_matrix = matrix.get(role) or DEFAULT_PROJECT_ROLE_PERMISSIONS.get(role, {})
        for module in PROJECT_PERMISSION_MODULES:
            module_key = module["key"]
            normalized[role][module_key] = _normalize_module_action_state(
                role_matrix.get(module_key, _empty_action_state()),
                role,
                module_key,
            )
    return normalized


async def get_project_role_permissions(db: AsyncSession, study_id: uuid.UUID) -> dict[str, dict[str, dict[str, bool]]]:
    result = await db.execute(select(StudyRolePermission).where(StudyRolePermission.study_id == study_id))
    rows = result.scalars().all()
    if not rows:
        return normalize_permission_matrix(DEFAULT_PROJECT_ROLE_PERMISSIONS)
    matrix = normalize_permission_matrix(DEFAULT_PROJECT_ROLE_PERMISSIONS)
    for row in rows:
        if row.role not in PROJECT_PERMISSION_ROLES:
            continue
        if row.module not in matrix[row.role]:
            continue
        matrix[row.role][row.module] = _normalize_module_action_state(
            {"read": row.can_read, "write": row.can_write},
            row.role,
            row.module,
        )
    return matrix


async def replace_project_role_permissions(
    db: AsyncSession,
    study_id: uuid.UUID,
    payload: dict[str, dict[str, dict[str, bool]]],
) -> dict[str, dict[str, dict[str, bool]]]:
    matrix = normalize_permission_matrix(payload)
    await db.execute(delete(StudyRolePermission).where(StudyRolePermission.study_id == study_id))
    for role, role_matrix in matrix.items():
        if role == "ADMIN":
            continue
        for module, actions in role_matrix.items():
            db.add(
                StudyRolePermission(
                    study_id=study_id,
                    role=role,
                    module=module,
                    can_read=actions["read"],
                    can_write=actions["write"],
                )
            )
    await db.commit()
    return matrix


async def role_has_project_permission(
    db: AsyncSession,
    study_id: uuid.UUID,
    role: str | None,
    module: str,
    action: str,
) -> bool:
    if role == "ADMIN":
        return True
    matrix = await get_project_role_permissions(db, study_id)
    actions = matrix.get(role or "", {}).get(module)
    if not actions:
        return False
    if action == "write":
        return bool(actions["write"])
    return bool(actions["read"])