feat: 添加所有 Django apps 的 ViewSets

- User ViewSet(个人中心、统计、收藏、评分、搜索)
- Region ViewSet(层级查询、树形结构、文章、服务、统计、评分、收藏)
- Article ViewSet(创建、提交、审核、评论、点赞、统计)
- FeaturedService ViewSet(创建、提交、审核、评论、点赞、评分、统计)
- Moderation ViewSets(版主申请、权限、支持、限制)
- Interaction ViewSets(评论、评分、点赞、收藏、AI审核)

完整实现权限控制、审核流程和交互功能
This commit is contained in:
mashen
2026-04-09 13:44:13 +00:00
parent edec596516
commit d9c6c8ff59
6 changed files with 1129 additions and 4 deletions

View File

@@ -1,8 +1,14 @@
from rest_framework import viewsets, permissions
from rest_framework import viewsets, permissions, status
from rest_framework.decorators import action
from rest_framework.response import Response
from django.db.models import Count, Q
from .models import User
from .serializers import UserSerializer, UserDetailSerializer
from .serializers import (
UserSerializer,
UserDetailSerializer,
UserUpdateSerializer,
UserStatsSerializer
)
class UserViewSet(viewsets.ModelViewSet):
@@ -12,12 +18,136 @@ class UserViewSet(viewsets.ModelViewSet):
permission_classes = [permissions.IsAuthenticated]
def get_serializer_class(self):
if self.action in ['retrieve', 'update', 'partial_update']:
if self.action == 'retrieve' and self.kwargs.get('pk') == 'me':
return UserDetailSerializer
elif self.action in ['update', 'partial_update'] and self.kwargs.get('pk') == 'me':
return UserUpdateSerializer
return UserSerializer
def get_queryset(self):
# Only admins can see all users
if self.request.user.is_admin():
return User.objects.all()
# Regular users can only see themselves
return User.objects.filter(id=self.request.user.id)
def list(self, request, *args, **kwargs):
"""Only admins can list all users."""
if not request.user.is_admin():
return Response(
{'detail': 'You do not have permission to perform this action.'},
status=status.HTTP_403_FORBIDDEN
)
return super().list(request, *args, **kwargs)
def retrieve(self, request, *args, **kwargs):
"""Get user details (me for current user)."""
if kwargs.get('pk') == 'me':
self.kwargs['pk'] = request.user.id
return super().retrieve(request, *args, **kwargs)
def update(self, request, *args, **kwargs):
"""Update user details (only me for regular users)."""
if kwargs.get('pk') == 'me':
self.kwargs['pk'] = request.user.id
elif not request.user.is_admin() and str(kwargs.get('pk')) != str(request.user.id):
return Response(
{'detail': 'You can only update your own profile.'},
status=status.HTTP_403_FORBIDDEN
)
return super().update(request, *args, **kwargs)
@action(detail=False, methods=['get'])
def me(self, request):
"""Get current user."""
"""Get current user details."""
serializer = self.get_serializer(request.user)
return Response(serializer.data)
@action(detail=True, methods=['get'])
def stats(self, request, pk=None):
"""Get user statistics."""
if pk == 'me':
pk = request.user.id
user = self.get_object()
if str(user.id) != str(request.user.id) and not request.user.is_admin():
return Response(
{'detail': 'You do not have permission to view this user\'s stats.'},
status=status.HTTP_403_FORBIDDEN
)
from apps.articles.models import Article
from apps.featured_services.models import FeaturedService
from apps.interactions.models import Comment, Like, Favorite, Rating
return Response({
'articles_count': Article.objects.filter(author=user).count(),
'services_count': FeaturedService.objects.filter(submitter=user).count(),
'comments_count': Comment.objects.filter(author=user).count(),
'likes_count': Like.objects.filter(user=user).count(),
'favorites_count': Favorite.objects.filter(user=user).count(),
'ratings_count': Rating.objects.filter(user=user).count(),
})
@action(detail=True, methods=['get'])
def favorites(self, request, pk=None):
"""Get user's favorites."""
if pk == 'me':
pk = request.user.id
user = self.get_object()
if str(user.id) != str(request.user.id):
return Response(
{'detail': 'You can only view your own favorites.'},
status=status.HTTP_403_FORBIDDEN
)
from apps.interactions.serializers import FavoriteSerializer
favorites = Favorite.objects.filter(user=user).select_related()
serializer = FavoriteSerializer(favorites, many=True)
return Response(serializer.data)
@action(detail=True, methods=['get'])
def ratings(self, request, pk=None):
"""Get user's ratings."""
if pk == 'me':
pk = request.user.id
user = self.get_object()
if str(user.id) != str(request.user.id):
return Response(
{'detail': 'You can only view your own ratings.'},
status=status.HTTP_403_FORBIDDEN
)
from apps.interactions.serializers import RatingSerializer
ratings = Rating.objects.filter(user=user).select_related()
serializer = RatingSerializer(ratings, many=True)
return Response(serializer.data)
@action(detail=False, methods=['get'])
def search(self, request):
"""Search users (admin only)."""
if not request.user.is_admin():
return Response(
{'detail': 'Only admins can search users.'},
status=status.HTTP_403_FORBIDDEN
)
query = request.query_params.get('q', '')
if query:
users = User.objects.filter(
Q(username__icontains=query) |
Q(email__icontains=query) |
Q(first_name__icontains=query)
)
else:
users = User.objects.all()
page = self.paginate_queryset(users)
if page is not None:
serializer = self.get_serializer(page, many=True)
return self.get_paginated_response(serializer.data)
serializer = self.get_serializer(users, many=True)
return Response(serializer.data)