You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
128 lines
4.9 KiB
128 lines
4.9 KiB
from django.db.models import Q
|
|
from rest_framework.viewsets import ModelViewSet
|
|
from rest_framework.permissions import IsAuthenticated
|
|
from apps.account.permissions import IsSuperAdmin
|
|
from rest_framework.authentication import TokenAuthentication
|
|
from rest_framework.parsers import MultiPartParser, FormParser, JSONParser
|
|
from rest_framework.decorators import action
|
|
from rest_framework.response import Response
|
|
from utils.pagination import StandardResultsSetPagination
|
|
|
|
from .models import Blog, BlogContent, BlogSeo
|
|
from .serializers_admin import (
|
|
AdminBlogListSerializer,
|
|
AdminBlogDetailSerializer,
|
|
AdminBlogContentSerializer,
|
|
AdminBlogSeoSerializer
|
|
)
|
|
|
|
class AdminBlogViewSet(ModelViewSet):
|
|
"""
|
|
Admin CRUD ViewSet for Blog management.
|
|
"""
|
|
permission_classes = [IsAuthenticated, IsSuperAdmin]
|
|
authentication_classes = [TokenAuthentication]
|
|
pagination_class = StandardResultsSetPagination
|
|
# Allow multipart so admins can upload thumbnail images directly
|
|
parser_classes = (MultiPartParser, FormParser, JSONParser)
|
|
|
|
def get_serializer_class(self):
|
|
if self.action == 'list':
|
|
return AdminBlogListSerializer
|
|
return AdminBlogDetailSerializer
|
|
|
|
def get_queryset(self):
|
|
queryset = Blog.objects.all().prefetch_related('contents', 'seos')
|
|
|
|
# Custom Search handling for JSON fields
|
|
search_query = self.request.query_params.get('search', None)
|
|
has_search = False
|
|
if search_query:
|
|
has_search = True
|
|
from django.db.models import Case, When, Value, IntegerField
|
|
# Note: Searching inside a JSON array of dicts in Django can be tricky
|
|
# depending on your DB (PostgreSQL vs SQLite).
|
|
# This is a safe fallback using text casting for SQLite/PG compatibility.
|
|
queryset = queryset.filter(
|
|
Q(title__icontains=search_query) |
|
|
Q(slogan__icontains=search_query)
|
|
).annotate(
|
|
search_relevance=Case(
|
|
When(title__icontains=search_query, then=Value(2)),
|
|
default=Value(1),
|
|
output_field=IntegerField()
|
|
)
|
|
)
|
|
|
|
created_date = self.request.query_params.get('created', None)
|
|
if created_date:
|
|
queryset = queryset.filter(created_at__date=created_date)
|
|
|
|
created_after = self.request.query_params.get('created_after', None)
|
|
if created_after:
|
|
queryset = queryset.filter(created_at__date__gte=created_after)
|
|
|
|
created_before = self.request.query_params.get('created_before', None)
|
|
if created_before:
|
|
queryset = queryset.filter(created_at__date__lte=created_before)
|
|
|
|
if has_search:
|
|
return queryset.order_by('-search_relevance', '-created_at')
|
|
return queryset.order_by('-created_at')
|
|
|
|
class AdminBlogContentViewSet(ModelViewSet):
|
|
"""
|
|
Admin CRUD ViewSet for managing the content blocks inside a specific blog.
|
|
"""
|
|
serializer_class = AdminBlogContentSerializer
|
|
permission_classes = [IsAuthenticated, IsSuperAdmin]
|
|
authentication_classes = [TokenAuthentication]
|
|
parser_classes = (MultiPartParser, FormParser, JSONParser)
|
|
|
|
def get_queryset(self):
|
|
queryset = BlogContent.objects.all()
|
|
|
|
# Filter contents by blog_id if provided in the URL query params
|
|
blog_id = self.request.query_params.get('blog_id', None)
|
|
if blog_id:
|
|
queryset = queryset.filter(blog_id=blog_id)
|
|
|
|
return queryset.order_by('order', 'created_at')
|
|
|
|
@action(detail=False, methods=['post'])
|
|
def reorder(self, request):
|
|
"""
|
|
Reorders content blocks in bulk.
|
|
Format: {"content_ids": [id1, id2, ...]}
|
|
"""
|
|
content_ids = request.data.get('content_ids', [])
|
|
if not content_ids:
|
|
return Response({'error': 'content_ids list is required'}, status=400)
|
|
|
|
from django.db import transaction
|
|
with transaction.atomic():
|
|
# First pass: temporary shift to prevent unique constraint conflicts
|
|
for idx, c_id in enumerate(content_ids):
|
|
BlogContent.objects.filter(id=c_id).update(order=idx + 10000)
|
|
# Second pass: set the correct final order values
|
|
for idx, c_id in enumerate(content_ids):
|
|
BlogContent.objects.filter(id=c_id).update(order=idx + 1)
|
|
|
|
return Response({'status': 'success'})
|
|
|
|
class AdminBlogSeoViewSet(ModelViewSet):
|
|
"""
|
|
Admin CRUD ViewSet for managing SEO meta tags for a specific blog.
|
|
"""
|
|
serializer_class = AdminBlogSeoSerializer
|
|
permission_classes = [IsAuthenticated, IsSuperAdmin]
|
|
authentication_classes = [TokenAuthentication]
|
|
|
|
def get_queryset(self):
|
|
queryset = BlogSeo.objects.all()
|
|
|
|
blog_id = self.request.query_params.get('blog_id', None)
|
|
if blog_id:
|
|
queryset = queryset.filter(blog_id=blog_id)
|
|
|
|
return queryset
|