""" 角色权限管理 API — 管理会计OS 支持在页面上配置角色可访问的模块和操作权限 """ from fastapi import APIRouter, Depends, HTTPException from sqlalchemy.orm import Session from app.database import get_db from app.models import RolePermission from app.auth_middleware import require_auth, require_role router = APIRouter(prefix="/api/cma/permissions", tags=["权限管理"]) # 模块定义(所有可配置的模块) MODULES = [ {"key": "dashboard", "name": "驾驶舱"}, {"key": "kpis", "name": "KPI字典"}, {"key": "kpi_detail", "name": "KPI详情"}, {"key": "maps", "name": "战略地图"}, {"key": "alerts", "name": "预警中心"}, {"key": "ai_analysis", "name": "AI分析"}, {"key": "data_source", "name": "数据管理"}, {"key": "budget", "name": "预算管理"}, {"key": "deviation", "name": "差异分析"}, {"key": "cost", "name": "成本分析"}, {"key": "predict", "name": "预测模拟"}, {"key": "org", "name": "组织管理"}, {"key": "user_manage", "name": "用户管理"}, {"key": "system_config", "name": "通知配置"}, {"key": "role_permissions", "name": "角色权限"}, {"key": "action_plans", "name": "改善行动"}, {"key": "alignment", "name": "KPI目标对齐"}, ] ACTIONS = [ {"key": "read", "name": "读取"}, {"key": "write", "name": "写入"}, {"key": "import", "name": "导入"}, {"key": "export", "name": "导出"}, {"key": "delete", "name": "删除"}, {"key": "approve", "name": "审批"}, {"key": "admin", "name": "管理"}, ] ROLES = [ {"code": "ceo", "name": "CEO"}, {"code": "finance", "name": "财务"}, {"code": "business", "name": "业务"}, {"code": "it", "name": "IT运维"}, ] # 默认权限 DEFAULT_ROUTE_PERMISSIONS = { "ceo": ["dashboard", "kpis", "kpi_detail", "maps", "alerts", "ai_analysis", "data_source", "budget", "deviation", "cost", "predict", "org", "user_manage", "system_config", "role_permissions", "action_plans", "alignment"], "finance": ["dashboard", "kpis", "kpi_detail", "maps", "alerts", "ai_analysis", "data_source", "budget", "deviation", "cost", "predict"], "business": ["dashboard", "kpis", "kpi_detail", "alerts", "budget", "deviation"], "it": ["dashboard", "kpis", "kpi_detail", "alerts", "data_source", "budget", "deviation", "cost", "predict", "org", "user_manage", "system_config"], } DEFAULT_ACTION_PERMISSIONS = { "ceo": ["read", "approve"], "finance": ["read", "write", "import", "export"], "business": ["read", "write"], "it": ["read", "write", "delete", "admin"], } def _get_or_create_defaults(db: Session): """获取配置,不存在则创建默认值""" route_perm = db.query(RolePermission).filter(RolePermission.key == "route_permissions").first() if not route_perm: route_perm = RolePermission(key="route_permissions", value=DEFAULT_ROUTE_PERMISSIONS) db.add(route_perm) action_perm = db.query(RolePermission).filter(RolePermission.key == "action_permissions").first() if not action_perm: action_perm = RolePermission(key="action_permissions", value=DEFAULT_ACTION_PERMISSIONS) db.add(action_perm) db.commit() db.refresh(route_perm) db.refresh(action_perm) return route_perm, action_perm @router.get("/modules") def list_modules(): """返回模块和动作定义""" return { "modules": MODULES, "actions": ACTIONS, "roles": ROLES, } @router.get("/config") def get_permissions(db: Session = Depends(get_db)): """获取当前权限配置""" route_perm, action_perm = _get_or_create_defaults(db) return { "route_permissions": route_perm.value, "action_permissions": action_perm.value, } @router.put("/config") def update_permissions( data: dict, db: Session = Depends(get_db), _=Depends(require_role("ceo", "it")), ): """更新权限配置""" route_perm, action_perm = _get_or_create_defaults(db) if "route_permissions" in data: route_perm.value = data["route_permissions"] if "action_permissions" in data: action_perm.value = data["action_permissions"] db.commit() return {"message": "权限配置已更新"}