from django.contrib.auth import authenticate, get_user_model, login, logout from django.contrib.auth.models import Group, Permission from django.contrib.contenttypes.models import ContentType from django.middleware.csrf import get_token from ninja import Router from ninja.errors import HttpError from apps.accounts.models import MenuItem, UserGroup from apps.accounts.permissions import PERMISSIONS from apps.accounts.schemas import ( BasicUserOut, CurrentUserOut, IdsIn, LoginIn, MenuOut, MenuUpdateIn, PasswordResetIn, PermissionCodesIn, PermissionOut, RoleCreateIn, RoleDetailOut, RoleUpdateIn, UserCreateIn, UserGroupCreateIn, UserGroupOut, UserGroupUpdateIn, UserUpdateIn, ) router = Router(tags=["accounts"]) def permission_codes_for_user(user) -> list[str]: if user.is_superuser: return sorted(permission.codename for permission in PERMISSIONS) codes = set() for permission in user.get_all_permissions(): if permission.startswith("system."): codes.add(permission.removeprefix("system.")) return sorted(codes) def permission_objects_for_codes(codes: list[str]): content_type = ContentType.objects.get(app_label="system", model="platformpermission") return list(Permission.objects.filter(content_type=content_type, codename__in=codes)) def serialize_menu(menu: MenuItem, children: list[dict]) -> dict: return { "id": menu.id, "code": menu.code, "title": menu.title, "path": menu.path, "icon": menu.icon, "permission_code": menu.permission_code, "is_visible": menu.is_visible, "is_enabled": menu.is_enabled, "children": children, } def menu_tree_for_permissions(permission_codes: list[str]) -> list[dict]: allowed = set(permission_codes) menus = list( MenuItem.objects.filter(is_visible=True, is_enabled=True) .select_related("parent") .order_by("sort_order", "id") ) children_by_parent: dict[int | None, list[MenuItem]] = {} for menu in menus: if menu.permission_code and menu.permission_code not in allowed: continue children_by_parent.setdefault(menu.parent_id, []).append(menu) def build(parent_id: int | None) -> list[dict]: result = [] for menu in children_by_parent.get(parent_id, []): result.append(serialize_menu(menu, build(menu.id))) return result return build(None) def serialize_current_user(user) -> dict: permission_codes = permission_codes_for_user(user) return { "id": user.id, "username": user.username, "display_name": user.get_full_name() or user.username, "email": user.email, "is_superuser": user.is_superuser, "roles": [{"id": group.id, "name": group.name} for group in user.groups.all()], "user_groups": [ { "id": group.id, "name": group.name, "code": group.code, "description": group.description, "is_active": group.is_active, "created_at": group.created_at, "updated_at": group.updated_at, } for group in user.business_groups.all() ], "permissions": permission_codes, "menus": menu_tree_for_permissions(permission_codes), } def serialize_basic_user(user) -> dict: return { "id": user.id, "username": user.username, "display_name": user.get_full_name() or user.username, "email": user.email, "is_active": user.is_active, "is_staff": user.is_staff, "is_superuser": user.is_superuser, "roles": [{"id": group.id, "name": group.name} for group in user.groups.all()], "user_groups": [ { "id": group.id, "name": group.name, "code": group.code, "description": group.description, "is_active": group.is_active, "created_at": group.created_at, "updated_at": group.updated_at, } for group in user.business_groups.all() ], } def serialize_role(group: Group) -> dict: codes = sorted(group.permissions.values_list("codename", flat=True)) return {"id": group.id, "name": group.name, "permission_codes": codes} def serialize_group(group) -> dict: return { "id": group.id, "name": group.name, "code": group.code, "description": group.description, "is_active": group.is_active, "created_at": group.created_at, "updated_at": group.updated_at, } def serialize_menu_flat(menu: MenuItem) -> dict: return serialize_menu(menu, []) def require_login(request): if not request.user.is_authenticated: raise HttpError(401, "Authentication required") return request.user def require_permission(request, codename: str): user = require_login(request) if user.is_superuser or user.has_perm(f"system.{codename}"): return user raise HttpError(403, "Permission denied") @router.get("/auth/csrf") def csrf(request): return {"csrfToken": get_token(request)} @router.post("/auth/login", response=CurrentUserOut) def login_user(request, payload: LoginIn): user = authenticate(request, username=payload.username, password=payload.password) if user is None or not user.is_active: raise HttpError(401, "Invalid username or password") login(request, user) return serialize_current_user(user) @router.post("/auth/logout") def logout_user(request): logout(request) return {"status": "ok"} @router.get("/auth/me", response=CurrentUserOut) def current_user(request): user = require_login(request) return serialize_current_user(user) @router.get("/users", response=list[BasicUserOut]) def list_users(request): require_permission(request, "system.user.view") return [serialize_basic_user(user) for user in get_user_model().objects.all().order_by("id")] @router.post("/users", response=BasicUserOut) def create_user(request, payload: UserCreateIn): require_permission(request, "system.user.manage") user = get_user_model().objects.create_user( username=payload.username, email=payload.email, password=payload.password, is_active=payload.is_active, ) if payload.display_name: user.first_name = payload.display_name user.save(update_fields=["first_name"]) return serialize_basic_user(user) @router.get("/users/{user_id}", response=BasicUserOut) def get_user(request, user_id: int): require_permission(request, "system.user.view") return serialize_basic_user(get_user_model().objects.get(id=user_id)) @router.patch("/users/{user_id}", response=BasicUserOut) def update_user(request, user_id: int, payload: UserUpdateIn): require_permission(request, "system.user.manage") user = get_user_model().objects.get(id=user_id) update_fields = [] if payload.display_name is not None: user.first_name = payload.display_name update_fields.append("first_name") if payload.email is not None: user.email = payload.email update_fields.append("email") if payload.is_active is not None: user.is_active = payload.is_active update_fields.append("is_active") if update_fields: user.save(update_fields=update_fields) return serialize_basic_user(user) @router.post("/users/{user_id}/reset-password", response=BasicUserOut) def reset_user_password(request, user_id: int, payload: PasswordResetIn): require_permission(request, "system.user.manage") user = get_user_model().objects.get(id=user_id) user.set_password(payload.password) user.save(update_fields=["password"]) return serialize_basic_user(user) @router.post("/users/{user_id}/roles", response=BasicUserOut) def set_user_roles(request, user_id: int, payload: IdsIn): require_permission(request, "system.user.manage") user = get_user_model().objects.get(id=user_id) user.groups.set(Group.objects.filter(id__in=payload.role_ids)) return serialize_basic_user(user) @router.post("/users/{user_id}/user-groups", response=BasicUserOut) def set_user_groups(request, user_id: int, payload: IdsIn): require_permission(request, "system.user.manage") user = get_user_model().objects.get(id=user_id) user.business_groups.set(UserGroup.objects.filter(id__in=payload.user_ids)) return serialize_basic_user(user) @router.get("/user-groups", response=list[UserGroupOut]) def list_user_groups(request): require_permission(request, "system.user_group.view") return [serialize_group(group) for group in UserGroup.objects.all()] @router.post("/user-groups", response=UserGroupOut) def create_user_group(request, payload: UserGroupCreateIn): require_permission(request, "system.user_group.manage") group = UserGroup.objects.create(**payload.dict()) return serialize_group(group) @router.patch("/user-groups/{group_id}", response=UserGroupOut) def update_user_group(request, group_id: int, payload: UserGroupUpdateIn): require_permission(request, "system.user_group.manage") group = UserGroup.objects.get(id=group_id) for field, value in payload.dict(exclude_unset=True).items(): setattr(group, field, value) group.save() return serialize_group(group) @router.post("/user-groups/{group_id}/members", response=UserGroupOut) def set_user_group_members(request, group_id: int, payload: IdsIn): require_permission(request, "system.user_group.manage") group = UserGroup.objects.get(id=group_id) group.users.set(get_user_model().objects.filter(id__in=payload.user_ids)) return serialize_group(group) @router.get("/roles", response=list[RoleDetailOut]) def list_roles(request): require_permission(request, "system.role.view") return [serialize_role(group) for group in Group.objects.all().order_by("id")] @router.post("/roles", response=RoleDetailOut) def create_role(request, payload: RoleCreateIn): require_permission(request, "system.role.manage") group = Group.objects.create(name=payload.name) group.permissions.set(permission_objects_for_codes(payload.permission_codes)) return serialize_role(group) @router.patch("/roles/{role_id}", response=RoleDetailOut) def update_role(request, role_id: int, payload: RoleUpdateIn): require_permission(request, "system.role.manage") group = Group.objects.get(id=role_id) if payload.name is not None: group.name = payload.name group.save(update_fields=["name"]) return serialize_role(group) @router.post("/roles/{role_id}/permissions", response=RoleDetailOut) def set_role_permissions(request, role_id: int, payload: PermissionCodesIn): require_permission(request, "system.role.manage") group = Group.objects.get(id=role_id) group.permissions.set(permission_objects_for_codes(payload.permission_codes)) return serialize_role(group) @router.get("/permissions", response=list[PermissionOut]) def list_permissions(request): require_permission(request, "system.role.view") return [ {"codename": permission.codename, "name": permission.name, "module": permission.module} for permission in PERMISSIONS ] @router.get("/menus", response=list[MenuOut]) def list_menus(request): require_permission(request, "system.menu.view") return [serialize_menu_flat(menu) for menu in MenuItem.objects.all()] @router.patch("/menus/{menu_id}", response=MenuOut) def update_menu(request, menu_id: int, payload: MenuUpdateIn): require_permission(request, "system.menu.manage") menu = MenuItem.objects.get(id=menu_id) for field, value in payload.dict(exclude_unset=True).items(): setattr(menu, field, value) menu.save() return serialize_menu_flat(menu)