"""admin_role 表 CRUD + 「角色 → 有效可见页」解析。""" from __future__ import annotations from sqlalchemy import select from sqlalchemy.orm import Session from app.admin.permissions import ( ALL_PAGE_KEYS, BUILTIN_ROLES, SUPER_ADMIN_ROLE, sanitize_pages, ) from app.models.admin_role import AdminRole def ensure_builtin_roles(db: Session) -> None: """空表时播种内建角色(管理员/运营/财务/技术)。幂等:表非空即 no-op。 给 create_all 环境(测试 / 未跑迁移的全新库)兜底,迁移已播种则不重复。 仅 super_admin 为 is_builtin(锁定不可改删);运营/财务/技术可编辑页/删除。""" if db.execute(select(AdminRole.id).limit(1)).first() is not None: return db.add_all([ AdminRole( name=r["name"], label=r["label"], pages=list(r["pages"]), is_builtin=(r["name"] == SUPER_ADMIN_ROLE), ) for r in BUILTIN_ROLES ]) db.commit() def list_roles(db: Session) -> list[AdminRole]: ensure_builtin_roles(db) # super_admin(内建)恒排最前,其余按创建序 return list( db.execute( select(AdminRole).order_by(AdminRole.is_builtin.desc(), AdminRole.id) ).scalars().all() ) def get_role(db: Session, name: str) -> AdminRole | None: return db.execute( select(AdminRole).where(AdminRole.name == name) ).scalar_one_or_none() def create_role(db: Session, *, name: str, label: str, pages: list[str]) -> AdminRole: # 自定义角色:name(key)= label = 用户输入的名称 role = AdminRole(name=name, label=label, pages=sanitize_pages(pages), is_builtin=False) db.add(role) db.commit() db.refresh(role) return role def update_role( db: Session, role: AdminRole, *, label: str | None = None, pages: list[str] | None = None ) -> AdminRole: # 只改展示名 label + 可见页;name(key)不可变(admin_user.role / require_role 承重) if label is not None: role.label = label if pages is not None: role.pages = sanitize_pages(pages) db.commit() db.refresh(role) return role def delete_role(db: Session, role: AdminRole) -> None: db.delete(role) db.commit() def effective_pages_of(db: Session, role_name: str) -> list[str]: """某角色名的有效可见页:super_admin → 全部;其余 → 查表 pages 与目录取交。""" if role_name == SUPER_ADMIN_ROLE: return list(ALL_PAGE_KEYS) ensure_builtin_roles(db) role = get_role(db, role_name) return sanitize_pages(role.pages if role else None)