74 lines
2.5 KiB
Python
74 lines
2.5 KiB
Python
from ninja import Router, Schema, Query
|
||
from pydantic import Field
|
||
from typing import List
|
||
from django.shortcuts import get_object_or_404
|
||
|
||
from accounts.models import User
|
||
from websites.models import Website
|
||
from utils.auth import jwt_auth
|
||
from utils.permissions import manager_required
|
||
|
||
router = Router(tags=["授权管理"])
|
||
|
||
|
||
class AuthorizeIn(Schema):
|
||
user_id: int = Field(..., description="被授权的用户ID")
|
||
website_ids: List[int] = Field(..., description="要授权的网站ID列表")
|
||
|
||
|
||
@router.post("/authorize", auth=jwt_auth)
|
||
@manager_required
|
||
def authorize_user(request, data: AuthorizeIn):
|
||
manager = request.user
|
||
target_user = get_object_or_404(User, id=data.user_id)
|
||
|
||
if target_user.role != "user":
|
||
return {"success": False, "message": "只能授权给普通用户"}
|
||
managed_ids = set(manager.managed_websites.values_list("id", flat=True))
|
||
for wid in data.website_ids:
|
||
if wid not in managed_ids:
|
||
return {"success": False, "message": f"无权授权网站ID：{wid}"}
|
||
|
||
target_user.authorized_websites.add(*data.website_ids)
|
||
|
||
return {
|
||
"success": True,
|
||
"message": f"已授权 {target_user.username} 访问 {len(data.website_ids)} 个网站",
|
||
}
|
||
|
||
@router.get("/authorized-sites", auth=jwt_auth)
|
||
@manager_required
|
||
def get_user_authorized_sites(request, user_id: int = Query(...)):
|
||
target_user = get_object_or_404(User, id=user_id)
|
||
|
||
if target_user.role != "user":
|
||
return {"success": False, "message": "只能查看普通用户的授权信息"}
|
||
|
||
sites = target_user.authorized_websites.all().values("id", "name", "db_alias")
|
||
|
||
return {
|
||
"success": True,
|
||
"user": target_user.username,
|
||
"authorized_websites": list(sites)
|
||
}
|
||
|
||
@router.post("/revoke", auth=jwt_auth)
|
||
@manager_required
|
||
def revoke_authorization(request, data: AuthorizeIn):
|
||
manager = request.user
|
||
target_user = get_object_or_404(User, id=data.user_id)
|
||
|
||
if target_user.role != "user":
|
||
return {"success": False, "message": "只能撤销普通用户的授权"}
|
||
|
||
managed_ids = set(manager.managed_websites.values_list("id", flat=True))
|
||
for wid in data.website_ids:
|
||
if wid not in managed_ids:
|
||
return {"success": False, "message": f"无权撤销网站ID：{wid}"}
|
||
|
||
target_user.authorized_websites.remove(*data.website_ids)
|
||
|
||
return {
|
||
"success": True,
|
||
"message": f"已撤销 {target_user.username} 的 {len(data.website_ids)} 个授权网站"
|
||
} |