diff --git a/accounts/api/authorize.py b/accounts/api/authorize.py new file mode 100644 index 0000000..b278812 --- /dev/null +++ b/accounts/api/authorize.py @@ -0,0 +1,74 @@ +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)} 个授权网站" + } \ No newline at end of file diff --git a/accounts/models.py b/accounts/models.py index 8a63bb5..011838c 100644 --- a/accounts/models.py +++ b/accounts/models.py @@ -1,6 +1,6 @@ from django.contrib.auth.models import AbstractUser from django.db import models - +from websites.models import Website class User(AbstractUser): ROLE_CHOICES = [ @@ -9,7 +9,18 @@ class User(AbstractUser): ('user', '普通用户'), ] role = models.CharField(max_length=20, choices=ROLE_CHOICES, default='user', help_text="用户角色") - + managed_websites = models.ManyToManyField( + Website, + blank=True, + related_name="managers", + help_text="分管理可管理的网站" + ) + authorized_websites = models.ManyToManyField( + Website, + blank=True, + related_name="authorized_users", + help_text="普通用户被授权可访问的网站" + ) def is_admin(self): return self.role == 'admin' diff --git a/api.py b/api.py index 6e2fff0..e56768a 100644 --- a/api.py +++ b/api.py @@ -2,8 +2,10 @@ from ninja import NinjaAPI from resumes.api.views import router as resume_router from accounts.api.auth import auth_router from accounts.api.user import user_router +from accounts.api.authorize import router api = NinjaAPI(title="简历管理 API") api.add_router("/resumes/", resume_router) api.add_router("/auth", auth_router) api.add_router("/users", user_router) +api.add_router("/authorize", router) \ No newline at end of file diff --git a/resumes/api/views.py b/resumes/api/views.py index 77a10b5..941bcd9 100644 --- a/resumes/api/views.py +++ b/resumes/api/views.py @@ -1,11 +1,16 @@ from ninja import Router, Query + +from accounts.models import User from resumes.models import ResumeBasic from resumes.api.schemas import ResumeBasicOut, PaginatedResumes from typing import Optional +from utils.auth import jwt_auth +from utils.permissions import login_required router = Router(tags=["简历"]) -@router.get("/", response=PaginatedResumes) +@router.get("/", response=PaginatedResumes, auth=jwt_auth) +@login_required def list_resumes( request, job_status: Optional[str] = Query(None), @@ -16,8 +21,16 @@ def list_resumes( limit: int = 10, offset: int = 0 ): + user = request.user qs = ResumeBasic.objects.all() + if user.is_admin(): + pass # 管理员访问全部 + elif user.is_manager(): + qs = qs.filter(source_id__in=user.managed_websites.values_list("id", flat=True)) + elif user.is_user(): + qs = qs.filter(source_id__in=user.authorized_websites.values_list("id", flat=True)) + if job_status: qs = qs.filter(job_status=job_status) if age: @@ -28,6 +41,7 @@ def list_resumes( qs = qs.filter(source_id=source_id) if keyword: qs = qs.filter(crawl_keywords__icontains=keyword) + total = qs.count() results = qs[offset:offset + limit]