dovodi
/
Dovodi_Backend
3
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity
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.
929 Commits
2 Branches
0 Tags
315 MiB
 
 
 
 
Tree: f90c47fd04
Branches Tags
${ item.name }
Create tag ${ searchTerm }
Create branch ${ searchTerm }
from 'f90c47fd04'
${ noResults }
Dovodi_Backend/apps/hadis/views_admin.py

3179 lines
73 KiB
Raw Blame History

from django.db.models import Q, Exists, OuterRef, Count
from rest_framework.authentication import TokenAuthentication
from rest_framework.decorators import action
from rest_framework.parsers import FormParser, JSONParser, MultiPartParser
from rest_framework.permissions import IsAuthenticated
from rest_framework.response import Response
from rest_framework.viewsets import ModelViewSet, ReadOnlyModelViewSet
from apps.account.permissions import IsSuperAdmin
from utils.pagination import StandardResultsSetPagination, OptionalPagination
from .models import (
Hadis, HadisCategory, HadisCollection, HadisInCollection, HadisCorrection, HadisInterpretation, HadisStatus, HadisTag, HadisSect, HadisTransmitter, HadisReference, ReferenceImage,
BookReference, BookReferenceImage, BookReferenceDocument, BookAuthor, BookAuthorRutba, BookAttribute, BookSubjectArea, BookType, BookTag, BookEdition, BookVolume,
Transmitters, TransmitterReliability, NarratorLayer, OpinionStatus, ScholarlyOpinion, TransmitterOpinion, TransmitterOriginalText,
YearValue, NarratorDate,
CorrectionReference, InterpretationReference, OriginalTextReference,
ContentRelease,
CategoryQuranVerse,
City,
Country,
BookPublisher,
Kunya,
TransmitterFeature,
TransmitterProfession,
SahabiStatus,
Madhhab,
RelationType,
)
from .serializers.serializers_admin import (
AdminHadisCategorySerializer,
AdminHadisSectArchiveSerializer,
AdminHadisCategoryArchiveSerializer,
AdminHadisCategoryArchiveDetailSerializer,
AdminHadisCorrectionSerializer,
AdminCorrectionReferenceSerializer,
AdminInterpretationReferenceSerializer,
AdminOriginalTextReferenceSerializer,
AdminHadisReferenceSerializer,
AdminHadisDetailSerializer,
AdminHadisInterpretationSerializer,
AdminHadisCollectionSerializer,
AdminHadisListSerializer,
AdminHadisBriefSerializer,
AdminHadisStatusSerializer,
AdminHadisTagSerializer,
AdminBookSubjectAreaSerializer,
AdminBookTypeSerializer,
AdminBookTagSerializer,
AdminBookAuthorSerializer,
AdminBookAuthorRutbaSerializer,
AdminBookAttributeSerializer,
AdminBookReferenceImageSerializer,
AdminHadisReferenceImageSerializer,
AdminBookReferenceDocumentSerializer,
AdminBookReferenceListSerializer,
AdminBookReferenceDetailSerializer,
AdminTransmitterListSerializer,
AdminTransmitterDetailSerializer,
AdminTransmitterReliabilitySerializer,
AdminYearValueSerializer,
AdminNarratorLayerSerializer,
AdminOpinionStatusSerializer,
AdminScholarlyOpinionSerializer,
AdminHadisTransmitterSerializer,
AdminTransmitterOpinionSerializer,
AdminTransmitterOriginalTextSerializer,
AdminBookEditionSerializer,
AdminBookVolumeSerializer,
AdminContentReleaseSerializer,
AdminCitySerializer,
AdminCountrySerializer,
AdminBookPublisherSerializer,
AdminKunyaSerializer,
AdminTransmitterFeatureSerializer,
AdminTransmitterProfessionSerializer,
AdminSahabiStatusSerializer,
AdminMadhhabSerializer,
AdminRelationTypeSerializer,
)
class AdminHadisCategoryViewSet(ReadOnlyModelViewSet):
serializer_class = AdminHadisCategorySerializer
permission_classes = [IsAuthenticated, IsSuperAdmin]
authentication_classes = [TokenAuthentication]
pagination_class = OptionalPagination
def get_queryset(self):
hadis_subquery = Hadis.objects.filter(Q(category=OuterRef("pk")) | Q(categories=OuterRef("pk")))
children_subquery = HadisCategory.objects.filter(parent=OuterRef("pk"))
quran_verse_subquery = CategoryQuranVerse.objects.filter(category=OuterRef("pk"))
queryset = HadisCategory.objects.select_related("parent", "sect").annotate(
annotated_has_hadith=Exists(hadis_subquery),
annotated_has_children=Exists(children_subquery),
annotated_has_quran_verse=Exists(quran_verse_subquery),
)
has_quran_verse = self.request.query_params.get("has_quran_verse")
exclude_verse_id = self.request.query_params.get("exclude_verse_id")
if has_quran_verse in ["false", "False", "0"]:
if exclude_verse_id:
queryset = queryset.filter(
Q(source_type="quran", quran_verse__isnull=True)
| Q(quran_verse__id=exclude_verse_id)
)
else:
queryset = queryset.filter(source_type="quran", quran_verse__isnull=True)
elif has_quran_verse in ["true", "True", "1"]:
queryset = queryset.filter(quran_verse__isnull=False)
source_type = self.request.query_params.get("source_type")
if source_type:
queryset = queryset.filter(source_type=source_type)
else:
source_type = self.request.query_params.get("source_type")
if source_type:
queryset = queryset.filter(source_type=source_type)
search = self.request.query_params.get("search")
if search:
queryset = queryset.filter(
Q(title__icontains=search) | Q(slug__icontains=search)
)
has_children = self.request.query_params.get("has_children")
if has_children is not None and has_children != "all":
if str(has_children).lower() in ["false", "0"]:
queryset = queryset.filter(annotated_has_children=False)
elif str(has_children).lower() in ["true", "1"]:
queryset = queryset.filter(annotated_has_children=True)
is_leaf = self.request.query_params.get("is_leaf")
if is_leaf is not None and is_leaf != "all":
if str(is_leaf).lower() in ["true", "1"]:
queryset = queryset.filter(annotated_has_children=False)
elif str(is_leaf).lower() in ["false", "0"]:
queryset = queryset.filter(annotated_has_children=True)
is_topic = self.request.query_params.get("is_topic")
if is_topic is not None and is_topic != "all":
if str(is_topic).lower() in ["true", "1"]:
queryset = queryset.filter(is_topic=True)
elif str(is_topic).lower() in ["false", "0"]:
queryset = queryset.filter(is_topic=False)
return queryset.order_by("order", "id")
class DovodiHadisSectViewSet(ModelViewSet):
serializer_class = AdminHadisSectArchiveSerializer
permission_classes = [IsAuthenticated, IsSuperAdmin]
authentication_classes = [TokenAuthentication]
pagination_class = OptionalPagination
parser_classes = (JSONParser, MultiPartParser, FormParser)
def get_queryset(self):
queryset = HadisSect.objects.all()
search_query = self.request.query_params.get("search")
if search_query:
queryset = queryset.filter(
Q(title__icontains=search_query)
| Q(description__icontains=search_query)
| Q(sect_type__icontains=search_query)
)
return queryset.order_by("order", "id")
class DovodiHadisCategoryViewSet(ModelViewSet):
serializer_class = AdminHadisCategoryArchiveSerializer
permission_classes = [IsAuthenticated, IsSuperAdmin]
authentication_classes = [TokenAuthentication]
pagination_class = OptionalPagination
parser_classes = (JSONParser, MultiPartParser, FormParser)
def get_serializer_class(self):
if self.action in ["retrieve"]:
return AdminHadisCategoryArchiveDetailSerializer
return AdminHadisCategoryArchiveSerializer
def get_queryset(self):
hadis_subquery = Hadis.objects.filter(Q(category=OuterRef("pk")) | Q(categories=OuterRef("pk")))
children_subquery = HadisCategory.objects.filter(parent=OuterRef("pk"))
queryset = HadisCategory.objects.select_related("parent", "sect").annotate(
annotated_has_hadith=Exists(hadis_subquery),
annotated_has_children=Exists(children_subquery),
)
search_query = self.request.query_params.get("search")
if search_query:
queryset = queryset.filter(
Q(title__icontains=search_query)
| Q(description__icontains=search_query)
| Q(slug__icontains=search_query)
| Q(source_type__icontains=search_query)
)
source_type = self.request.query_params.get("source_type")
if source_type and source_type != "all":
queryset = queryset.filter(source_type=source_type)
sect_type = self.request.query_params.get("sect_type")
if sect_type and sect_type != "all":
queryset = queryset.filter(sect__sect_type=sect_type)
sect_id = self.request.query_params.get("sect")
if sect_id and sect_id != "all":
queryset = queryset.filter(sect_id=sect_id)
has_hadith = self.request.query_params.get("has_hadith")
if has_hadith is not None and has_hadith != "all":
if str(has_hadith).lower() in ["false", "0"]:
queryset = queryset.filter(annotated_has_hadith=False)
elif str(has_hadith).lower() in ["true", "1"]:
queryset = queryset.filter(annotated_has_hadith=True)
has_children = self.request.query_params.get("has_children")
if has_children is not None and has_children != "all":
if str(has_children).lower() in ["false", "0"]:
queryset = queryset.filter(annotated_has_children=False)
elif str(has_children).lower() in ["true", "1"]:
queryset = queryset.filter(annotated_has_children=True)
is_leaf = self.request.query_params.get("is_leaf")
if is_leaf is not None and is_leaf != "all":
if str(is_leaf).lower() in ["true", "1"]:
queryset = queryset.filter(annotated_has_children=False)
elif str(is_leaf).lower() in ["false", "0"]:
queryset = queryset.filter(annotated_has_children=True)
is_topic = self.request.query_params.get("is_topic")
if is_topic is not None and is_topic != "all":
if str(is_topic).lower() in ["true", "1"]:
queryset = queryset.filter(is_topic=True)
elif str(is_topic).lower() in ["false", "0"]:
queryset = queryset.filter(is_topic=False)
return queryset.order_by("-id")
class AdminHadisStatusViewSet(ModelViewSet):
serializer_class = AdminHadisStatusSerializer
permission_classes = [IsAuthenticated, IsSuperAdmin]
authentication_classes = [TokenAuthentication]
pagination_class = OptionalPagination
parser_classes = (JSONParser, MultiPartParser, FormParser)
def get_queryset(self):
queryset = HadisStatus.objects.all()
search_query = self.request.query_params.get("search")
if search_query:
queryset = queryset.filter(
Q(title__icontains=search_query)
| Q(slug__icontains=search_query)
| Q(description__icontains=search_query)
)
return queryset.order_by("order", "id")
class AdminHadisTagViewSet(ModelViewSet):
serializer_class = AdminHadisTagSerializer
permission_classes = [IsAuthenticated, IsSuperAdmin]
authentication_classes = [TokenAuthentication]
pagination_class = OptionalPagination
parser_classes = (JSONParser, MultiPartParser, FormParser)
def get_queryset(self):
queryset = HadisTag.objects.all()
search_query = self.request.query_params.get("search")
if search_query:
queryset = queryset.filter(title__icontains=search_query)
return queryset.order_by("id")
class AdminHadisCollectionViewSet(ModelViewSet):
serializer_class = AdminHadisCollectionSerializer
permission_classes = [IsAuthenticated, IsSuperAdmin]
authentication_classes = [TokenAuthentication]
pagination_class = OptionalPagination
parser_classes = (MultiPartParser, FormParser, JSONParser)
def get_queryset(self):
queryset = HadisCollection.objects.all()
search_query = self.request.query_params.get("search")
if search_query:
queryset = queryset.filter(
Q(title__icontains=search_query)
| Q(slug__icontains=search_query)
| Q(summary__icontains=search_query)
)
status_filter = self.request.query_params.get("status")
if status_filter in {"true", "false"}:
queryset = queryset.filter(status=status_filter == "true")
return queryset.order_by("-created_at", "-id")
@action(detail=True, methods=["post"], url_path="assign-items")
def assign_items(self, request, pk=None):
collection = self.get_object()
add_ids = request.data.get("add_ids", []) or request.data.get("item_ids", [])
remove_ids = request.data.get("remove_ids", [])
if add_ids:
new_items = [
HadisInCollection(hadis_id=hid, collection_id=collection.id)
for hid in add_ids
]
HadisInCollection.objects.bulk_create(new_items, ignore_conflicts=True)
if remove_ids:
HadisInCollection.objects.filter(collection_id=collection.id, hadis_id__in=remove_ids).delete()
return Response({"status": "success", "attached_count": collection.hadis_items.count()})
class AdminTransmitterReliabilityViewSet(ReadOnlyModelViewSet):
serializer_class = AdminTransmitterReliabilitySerializer
permission_classes = [IsAuthenticated, IsSuperAdmin]
authentication_classes = [TokenAuthentication]
pagination_class = OptionalPagination
def get_queryset(self):
queryset = TransmitterReliability.objects.all()
search_query = self.request.query_params.get("search")
if search_query:
queryset = queryset.filter(
Q(title__icontains=search_query)
| Q(slug__icontains=search_query)
| Q(description__icontains=search_query)
)
return queryset.order_by("-id")
class DovodiNarratorLayerViewSet(ModelViewSet):
serializer_class = AdminNarratorLayerSerializer
permission_classes = [IsAuthenticated, IsSuperAdmin]
authentication_classes = [TokenAuthentication]
pagination_class = OptionalPagination
parser_classes = (JSONParser, MultiPartParser, FormParser)
def get_queryset(self):
queryset = NarratorLayer.objects.select_related("hadis").all()
hadis_id = self.request.query_params.get("hadis") or self.request.query_params.get("hadis_id")
if hadis_id:
queryset = queryset.filter(hadis_id=hadis_id)
search_query = self.request.query_params.get("search")
if search_query:
queryset = queryset.filter(
Q(name__icontains=search_query)
| Q(description__icontains=search_query)
| Q(slug__icontains=search_query)
)
return queryset.order_by("number", "id")
class DovodiTransmitterReliabilityViewSet(ModelViewSet):
serializer_class = AdminTransmitterReliabilitySerializer
permission_classes = [IsAuthenticated, IsSuperAdmin]
authentication_classes = [TokenAuthentication]
pagination_class = OptionalPagination
parser_classes = (JSONParser, MultiPartParser, FormParser)
def get_queryset(self):
queryset = TransmitterReliability.objects.all()
search_query = self.request.query_params.get("search")
if search_query:
queryset = queryset.filter(
Q(title__icontains=search_query)
| Q(slug__icontains=search_query)
| Q(description__icontains=search_query)
)
return queryset.order_by("-id")
class DovodiOpinionStatusViewSet(ModelViewSet):
serializer_class = AdminOpinionStatusSerializer
permission_classes = [IsAuthenticated, IsSuperAdmin]
authentication_classes = [TokenAuthentication]
pagination_class = OptionalPagination
parser_classes = (JSONParser, MultiPartParser, FormParser)
def get_queryset(self):
queryset = OpinionStatus.objects.all()
search_query = self.request.query_params.get("search")
if search_query:
queryset = queryset.filter(
Q(title__icontains=search_query)
| Q(slug__icontains=search_query)
| Q(description__icontains=search_query)
)
return queryset.order_by("-id")
class DovodiScholarlyOpinionViewSet(ModelViewSet):
serializer_class = AdminScholarlyOpinionSerializer
permission_classes = [IsAuthenticated, IsSuperAdmin]
authentication_classes = [TokenAuthentication]
pagination_class = OptionalPagination
parser_classes = (JSONParser, MultiPartParser, FormParser)
def get_queryset(self):
queryset = ScholarlyOpinion.objects.select_related("status").annotate(
transmitters_count_val=Count("transmitter_opinions__transmitter", distinct=True)
)
search_query = self.request.query_params.get("search")
if search_query:
q_filter = (
Q(opinion_text__icontains=search_query)
| Q(status__title__icontains=search_query)
)
if search_query.isdigit():
q_filter |= Q(id=int(search_query))
queryset = queryset.filter(q_filter)
status_id = self.request.query_params.get("status")
if status_id and status_id != "all":
queryset = queryset.filter(status_id=status_id)
return queryset.order_by("-id")
class DovodiYearValueViewSet(ModelViewSet):
serializer_class = AdminYearValueSerializer
permission_classes = [IsAuthenticated, IsSuperAdmin]
authentication_classes = [TokenAuthentication]
pagination_class = OptionalPagination
parser_classes = (JSONParser, MultiPartParser, FormParser)
def get_queryset(self):
try:
YearValue.ensure_default_context_tags()
except Exception:
pass
queryset = YearValue.objects.all()
category = self.request.query_params.get("category") or self.request.query_params.get("date_type")
if category and category != "all":
queryset = queryset.filter(category=category)
parse_type = self.request.query_params.get("parse_type") or self.request.query_params.get("date_mode")
if parse_type and parse_type != "all":
queryset = queryset.filter(parse_type=parse_type)
# If context_only=true or parse_type/date_mode=context, deduplicate #tags by returning one representative record per raw_value
context_only = self.request.query_params.get("context_only")
if context_only in ("true", "1", "True") or parse_type == "context":
queryset = queryset.filter(parse_type="context")
if not category:
# Prefer 'death' record so each #tag appears once in the context manager list
death_tags = set(
YearValue.objects.filter(parse_type="context", category="death").values_list("raw_value", flat=True)
)
queryset = queryset.filter(
Q(category="death") | (Q(category="birth") & ~Q(raw_value__in=death_tags))
)
search_query = self.request.query_params.get("search")
if search_query:
queryset = queryset.filter(
Q(raw_value__icontains=search_query)
| Q(translations__icontains=search_query)
)
return queryset.order_by("parse_type", "raw_value", "id")
@action(detail=False, methods=["post"], url_path="parse-preview")
def parse_preview(self, request):
from .utils.date_engine import parse_canonical_date, build_translations_for_parsed
raw_value = str(request.data.get("raw_value") or "").strip()
category = str(request.data.get("category") or "death").strip()
parsed = parse_canonical_date(raw_value, category=category)
translations = []
if parsed.is_valid:
if parsed.parse_type == "context":
yv = YearValue.objects.filter(raw_value=parsed.raw_value).first()
if yv and yv.translations:
translations = yv.translations
if not translations:
translations = build_translations_for_parsed(parsed, category=category)
return Response({
"raw_value": parsed.raw_value,
"is_valid": parsed.is_valid,
"parse_type": parsed.parse_type,
"value": parsed.value,
"value_min": parsed.value_min,
"value_max": parsed.value_max,
"values_list": parsed.values_list,
"error": parsed.error,
"suggestion": parsed.suggestion,
"translations": translations,
})
class AdminTransmitterViewSet(ModelViewSet):
permission_classes = [IsAuthenticated, IsSuperAdmin]
authentication_classes = [TokenAuthentication]
pagination_class = StandardResultsSetPagination
parser_classes = (MultiPartParser, FormParser, JSONParser)
def get_serializer_class(self):
if self.action == "list":
return AdminTransmitterListSerializer
return AdminTransmitterDetailSerializer
def get_queryset(self):
queryset = Transmitters.objects.select_related("reliability").prefetch_related("narrator_dates__year_value")
if getattr(self, "action", None) != "list":
queryset = queryset.prefetch_related("relatives__relation_type_ref", "relatives__narrator")
search_query = self.request.query_params.get("search")
if search_query:
q_filter = (
Q(full_name__icontains=search_query)
| Q(kunya__icontains=search_query)
| Q(known_as__icontains=search_query)
| Q(nickname__icontains=search_query)
| Q(origin__icontains=search_query)
| Q(lived_in__icontains=search_query)
| Q(died_in__icontains=search_query)
| Q(slug__icontains=search_query)
| Q(legacy_id__icontains=search_query)
| Q(birth_year_hijri__icontains=search_query)
| Q(death_year_hijri__icontains=search_query)
| Q(age_at_death__icontains=search_query)
)
if search_query.isdigit():
q_filter |= Q(legacy_number=int(search_query)) | Q(generation=int(search_query))
queryset = queryset.filter(q_filter)
status_filter = self.request.query_params.get("status")
if status_filter and status_filter != "all":
queryset = queryset.filter(reliability__slug=status_filter)
madhhab_filter = self.request.query_params.get("madhhab")
if madhhab_filter and madhhab_filter != "all":
queryset = queryset.filter(madhhab=madhhab_filter)
generation_filter = self.request.query_params.get("generation")
if generation_filter and generation_filter != "all":
queryset = queryset.filter(generation=generation_filter)
return queryset.order_by("-created_at", "-id")
class AdminHadisViewSet(ModelViewSet):
permission_classes = [IsAuthenticated, IsSuperAdmin]
authentication_classes = [TokenAuthentication]
pagination_class = StandardResultsSetPagination
parser_classes = (MultiPartParser, FormParser, JSONParser)
def get_serializer_class(self):
if self.action == "list":
if self.request.query_params.get("brief") == "true":
return AdminHadisBriefSerializer
return AdminHadisListSerializer
return AdminHadisDetailSerializer
def get_queryset(self):
queryset = (
Hadis.objects.all()
.select_related("category", "hadis_status")
.prefetch_related("categories", "collection_items__collection")
)
search_query = self.request.query_params.get("search")
if search_query:
q_filter = (
Q(title__icontains=search_query)
| Q(title_narrator__icontains=search_query)
| Q(text__icontains=search_query)
| Q(translation__icontains=search_query)
| Q(slug__icontains=search_query)
)
if search_query.isdigit():
q_filter |= Q(number=int(search_query))
queryset = queryset.filter(q_filter)
category_id = self.request.query_params.get("category")
if category_id and category_id != "all":
queryset = queryset.filter(Q(category_id=category_id) | Q(categories__id=category_id)).distinct()
collection_id = self.request.query_params.get("collection") or self.request.query_params.get("collection_id")
if collection_id and collection_id != "all":
queryset = queryset.filter(collection_items__collection_id=collection_id).distinct()
source_type = self.request.query_params.get("source_type")
if source_type:
queryset = queryset.filter(Q(category__source_type=source_type) | Q(categories__source_type=source_type)).distinct()
status_filter = self.request.query_params.get("status")
if status_filter in {"true", "false"}:
queryset = queryset.filter(status=status_filter == "true")
return queryset.order_by("-id")
@action(detail=False, methods=["post"], url_path="bulk-assign-topics")
def bulk_assign_topics(self, request):
from django.db import transaction
from rest_framework import status as drf_status
from .signals import invalidate_hadis_cache
hadis_ids = request.data.get("hadis_ids") or []
category_ids = request.data.get("category_ids") or []
mode = request.data.get("mode", "add")
if not isinstance(hadis_ids, list) or not hadis_ids:
return Response(
{"detail": "hadis_ids must be a non-empty list."},
status=drf_status.HTTP_400_BAD_REQUEST,
)
if not isinstance(category_ids, list) or not category_ids:
return Response(
{"detail": "category_ids must be a non-empty list."},
status=drf_status.HTTP_400_BAD_REQUEST,
)
try:
hadis_ids = [int(hid) for hid in hadis_ids]
category_ids = list(dict.fromkeys(int(cid) for cid in category_ids))
except (ValueError, TypeError):
return Response(
{"detail": "Invalid ID format."},
status=drf_status.HTTP_400_BAD_REQUEST,
)
categories = list(HadisCategory.objects.filter(id__in=category_ids))
if len(categories) != len(category_ids):
return Response(
{"detail": "One or more selected topics do not exist."},
status=drf_status.HTTP_400_BAD_REQUEST,
)
for cat in categories:
if cat.children.exists():
return Response(
{"detail": "Cannot assign items to a parent category that has subcategories. Please select leaf topics only."},
status=drf_status.HTTP_400_BAD_REQUEST,
)
hadises_qs = Hadis.objects.filter(id__in=hadis_ids)
existing_hadis_ids = list(hadises_qs.values_list("id", flat=True))
if not existing_hadis_ids:
return Response(
{"detail": "No matching items found."},
status=drf_status.HTTP_404_NOT_FOUND,
)
through_model = Hadis.categories.through
first_category_id = category_ids[0]
with transaction.atomic():
if mode == "replace":
through_model.objects.filter(hadis_id__in=existing_hadis_ids).delete()
relations = [
through_model(hadis_id=hid, hadiscategory_id=cid)
for hid in existing_hadis_ids
for cid in category_ids
]
through_model.objects.bulk_create(relations, ignore_conflicts=True)
Hadis.objects.filter(id__in=existing_hadis_ids).update(category_id=first_category_id)
else:
relations = [
through_model(hadis_id=hid, hadiscategory_id=cid)
for hid in existing_hadis_ids
for cid in category_ids
]
through_model.objects.bulk_create(relations, ignore_conflicts=True)
Hadis.objects.filter(id__in=existing_hadis_ids, category__isnull=True).update(
category_id=first_category_id
)
invalidate_hadis_cache()
cat_map = {c.id: c for c in categories}
ordered_cats = [cat_map[cid] for cid in category_ids if cid in cat_map]
categories_detail = [
{
"id": c.id,
"title": c.title,
"slug": c.slug,
"source_type": c.source_type,
"is_topic": True,
}
for c in ordered_cats
]
return Response(
{
"updated_count": len(existing_hadis_ids),
"mode": mode,
"categories_detail": categories_detail,
},
status=drf_status.HTTP_200_OK,
)
class AdminHadisCorrectionViewSet(ModelViewSet):
serializer_class = AdminHadisCorrectionSerializer
permission_classes = [IsAuthenticated, IsSuperAdmin]
authentication_classes = [TokenAuthentication]
parser_classes = (MultiPartParser, FormParser, JSONParser)
pagination_class = OptionalPagination
def get_queryset(self):
from django.db.models import Count, OuterRef, Subquery, IntegerField, Prefetch
from django.db.models.functions import Coalesce
from apps.bookmark.models.bookmark import Bookmark
bookmarks_subquery = Bookmark.objects.filter(
service=Bookmark.ServiceChoices.HADITH_CORRECTION,
content_id=OuterRef("id"),
status=True
).values("content_id").annotate(count=Count("id")).values("count")
references_prefetch = Prefetch(
"references",
queryset=CorrectionReference.objects.select_related(
"book_reference__author", "edition", "book_volume"
).defer(
"book_reference__description",
"book_reference__normalized_text",
"book_reference__author__scholarity",
"book_reference__author__features",
).prefetch_related("images", "book_reference__editions").order_by("id"),
)
queryset = (
HadisCorrection.objects.all()
.defer(
"normalized_text",
"embedded_in",
"hadis__text",
"hadis__translation",
"hadis__explanation",
"hadis__description",
"hadis__normalized_text",
"hadis__embedded_in",
)
.select_related("hadis")
.prefetch_related(references_prefetch)
.annotate(
annotated_bookmark_count=Coalesce(Subquery(bookmarks_subquery, output_field=IntegerField()), 0)
)
)
hadis_id = self.request.query_params.get("hadis_id") or self.request.query_params.get("hadis")
if hadis_id:
queryset = queryset.filter(hadis_id=hadis_id)
search = self.request.query_params.get("search")
if search:
from django.db.models import Q
queryset = queryset.filter(
Q(title__icontains=search) | Q(text__icontains=search)
)
return queryset.order_by("-created_at", "-id")
def perform_create(self, serializer):
priority = self.request.data.get("priority", 1)
try:
priority = int(priority)
except (ValueError, TypeError):
priority = 1
hadis_id = self.request.data.get("hadis")
if hadis_id:
from django.db.models import F
HadisCorrection.objects.filter(hadis_id=hadis_id, priority__gte=priority).update(priority=F("priority") + 1)
serializer.save()
@action(detail=False, methods=["post"])
def reorder(self, request):
correction_ids = request.data.get("correction_ids", [])
if not correction_ids:
return Response({"error": "correction_ids list is required"}, status=400)
for index, correction_id in enumerate(correction_ids, start=1):
HadisCorrection.objects.filter(id=correction_id).update(priority=index)
return Response({"status": "success"})
class AdminCorrectionReferenceViewSet(ModelViewSet):
serializer_class = AdminCorrectionReferenceSerializer
permission_classes = [IsAuthenticated, IsSuperAdmin]
authentication_classes = [TokenAuthentication]
parser_classes = (MultiPartParser, FormParser, JSONParser)
pagination_class = OptionalPagination
def get_queryset(self):
qs = CorrectionReference.objects.all().select_related(
"book_reference__author", "edition", "book_volume"
).prefetch_related("images", "book_reference__editions")
correction_id = self.request.query_params.get("correction_id")
if correction_id:
qs = qs.filter(correction_id=correction_id)
return qs.order_by("id")
class AdminHadisReferenceViewSet(ModelViewSet):
serializer_class = AdminHadisReferenceSerializer
permission_classes = [IsAuthenticated, IsSuperAdmin]
authentication_classes = [TokenAuthentication]
parser_classes = (MultiPartParser, FormParser, JSONParser)
pagination_class = OptionalPagination
def get_queryset(self):
qs = HadisReference.objects.all().select_related(
"book_reference__author", "edition", "book_volume"
).prefetch_related("images", "book_reference__editions")
hadis_id = self.request.query_params.get("hadis_id")
if hadis_id:
qs = qs.filter(hadis_id=hadis_id)
return qs.order_by("id")
def _sync_hadis_title(self, instance):
# If the argument has no title, fill it from the first source's book title
hadis = getattr(instance, "hadis", None)
if hadis and not Hadis._has_localized_value(hadis.title):
if hadis.auto_populate_title():
hadis.save()
def perform_create(self, serializer):
self._sync_hadis_title(serializer.save())
def perform_update(self, serializer):
self._sync_hadis_title(serializer.save())
class AdminInterpretationReferenceViewSet(ModelViewSet):
serializer_class = AdminInterpretationReferenceSerializer
permission_classes = [IsAuthenticated, IsSuperAdmin]
authentication_classes = [TokenAuthentication]
parser_classes = (MultiPartParser, FormParser, JSONParser)
pagination_class = OptionalPagination
def get_queryset(self):
qs = InterpretationReference.objects.all().select_related(
"book_reference__author", "edition", "book_volume"
).prefetch_related("images", "book_reference__editions")
interpretation_id = self.request.query_params.get("interpretation_id") or self.request.query_params.get("interpretation")
if interpretation_id:
qs = qs.filter(interpretation_id=interpretation_id)
return qs.order_by("id")
class AdminOriginalTextReferenceViewSet(ModelViewSet):
def perform_create(self, serializer):
instance = serializer.save()
ot = instance.original_text
if ot and (not ot.title or (isinstance(ot.title, list) and not any(isinstance(x, dict) and (x.get('text', '') or x.get('title', '')).strip() for x in ot.title))):
if ot.auto_populate_title():
ot.save(update_fields=['title', 'slug', 'share_link'])
def perform_update(self, serializer):
instance = serializer.save()
ot = instance.original_text
if ot and (not ot.title or (isinstance(ot.title, list) and not any(isinstance(x, dict) and (x.get('text', '') or x.get('title', '')).strip() for x in ot.title))):
if ot.auto_populate_title():
ot.save(update_fields=['title', 'slug', 'share_link'])
serializer_class = AdminOriginalTextReferenceSerializer
permission_classes = [IsAuthenticated, IsSuperAdmin]
authentication_classes = [TokenAuthentication]
parser_classes = (MultiPartParser, FormParser, JSONParser)
pagination_class = OptionalPagination
def get_queryset(self):
qs = OriginalTextReference.objects.all().select_related(
"book_reference__author", "edition", "book_volume"
).prefetch_related("images", "book_reference__editions")
original_text_id = self.request.query_params.get("original_text_id") or self.request.query_params.get("original_text")
if original_text_id:
qs = qs.filter(original_text_id=original_text_id)
return qs.order_by("id")
class AdminHadisInterpretationViewSet(ModelViewSet):
serializer_class = AdminHadisInterpretationSerializer
permission_classes = [IsAuthenticated, IsSuperAdmin]
authentication_classes = [TokenAuthentication]
parser_classes = (MultiPartParser, FormParser, JSONParser)
pagination_class = OptionalPagination
def get_queryset(self):
from django.db.models import Prefetch
references_prefetch = Prefetch(
"references",
queryset=InterpretationReference.objects.select_related(
"book_reference__author", "edition", "book_volume"
).defer(
"book_reference__description",
"book_reference__normalized_text",
"book_reference__author__scholarity",
"book_reference__author__features",
).prefetch_related("images", "book_reference__editions").order_by("id"),
)
queryset = (
HadisInterpretation.objects.all()
.defer("normalized_text", "category__description")
.select_related("category")
.prefetch_related(references_prefetch)
)
category_id = self.request.query_params.get("category_id") or self.request.query_params.get("category")
if category_id:
queryset = queryset.filter(category_id=category_id)
search = self.request.query_params.get("search")
if search:
from django.db.models import Q
queryset = queryset.filter(
Q(title__icontains=search) | Q(text__icontains=search)
)
return queryset.order_by("-created_at", "-id")
def perform_create(self, serializer):
priority = self.request.data.get("priority", 1)
try:
priority = int(priority)
except (ValueError, TypeError):
priority = 1
category_id = self.request.data.get("category")
if category_id:
from django.db.models import F
HadisInterpretation.objects.filter(category_id=category_id, priority__gte=priority).update(priority=F("priority") + 1)
serializer.save()
@action(detail=False, methods=["post"])
def reorder(self, request):
interpretation_ids = request.data.get("interpretation_ids", [])
if not interpretation_ids:
return Response({"error": "interpretation_ids list is required"}, status=400)
for index, interpretation_id in enumerate(interpretation_ids, start=1):
HadisInterpretation.objects.filter(id=interpretation_id).update(priority=index)
return Response({"status": "success"})
class AdminBookSubjectAreaViewSet(ReadOnlyModelViewSet):
serializer_class = AdminBookSubjectAreaSerializer
permission_classes = [IsAuthenticated, IsSuperAdmin]
authentication_classes = [TokenAuthentication]
pagination_class = None
def get_queryset(self):
return BookSubjectArea.objects.all().order_by("-created_at", "-id")
class AdminBookTypeViewSet(ReadOnlyModelViewSet):
serializer_class = AdminBookTypeSerializer
permission_classes = [IsAuthenticated, IsSuperAdmin]
authentication_classes = [TokenAuthentication]
pagination_class = None
def get_queryset(self):
return BookType.objects.all().order_by("-created_at", "-id")
class DovodiBookSubjectAreaViewSet(ModelViewSet):
serializer_class = AdminBookSubjectAreaSerializer
permission_classes = [IsAuthenticated, IsSuperAdmin]
authentication_classes = [TokenAuthentication]
pagination_class = OptionalPagination
parser_classes = (JSONParser, MultiPartParser, FormParser)
def get_queryset(self):
queryset = BookSubjectArea.objects.all()
search_query = self.request.query_params.get("search")
if search_query:
queryset = queryset.filter(title__icontains=search_query)
return queryset.order_by("-created_at", "-id")
class DovodiBookTypeViewSet(ModelViewSet):
serializer_class = AdminBookTypeSerializer
permission_classes = [IsAuthenticated, IsSuperAdmin]
authentication_classes = [TokenAuthentication]
pagination_class = OptionalPagination
parser_classes = (JSONParser, MultiPartParser, FormParser)
def get_queryset(self):
queryset = BookType.objects.all()
search_query = self.request.query_params.get("search")
if search_query:
queryset = queryset.filter(title__icontains=search_query)
return queryset.order_by("-created_at", "-id")
class DovodiBookAuthorRutbaViewSet(ModelViewSet):
serializer_class = AdminBookAuthorRutbaSerializer
permission_classes = [IsAuthenticated, IsSuperAdmin]
authentication_classes = [TokenAuthentication]
pagination_class = OptionalPagination
parser_classes = (JSONParser, MultiPartParser, FormParser)
def get_queryset(self):
queryset = BookAuthorRutba.objects.all()
search_query = self.request.query_params.get("search")
if search_query:
queryset = queryset.filter(
Q(title__icontains=search_query)
| Q(slug__icontains=search_query)
| Q(description__icontains=search_query)
)
return queryset.order_by("-id")
class AdminBookAuthorViewSet(ModelViewSet):
serializer_class = AdminBookAuthorSerializer
permission_classes = [IsAuthenticated, IsSuperAdmin]
authentication_classes = [TokenAuthentication]
pagination_class = OptionalPagination
parser_classes = (MultiPartParser, FormParser, JSONParser)
def get_queryset(self):
queryset = (
BookAuthor.objects
.select_related("rutba_status")
.prefetch_related("author_dates__year_value")
.all()
)
search_query = self.request.query_params.get("search")
if search_query:
queryset = queryset.filter(name__icontains=search_query)
return queryset.order_by("-created_at", "-id")
@action(detail=True, methods=["post"], url_path="assign-references")
def assign_references(self, request, pk=None):
author = self.get_object()
add_ids = request.data.get("add_ids", []) or request.data.get("book_ids", []) or request.data.get("item_ids", [])
remove_ids = request.data.get("remove_ids", [])
if add_ids:
BookReference.objects.filter(id__in=add_ids).update(author=author)
if remove_ids:
BookReference.objects.filter(id__in=remove_ids, author=author).update(author=None)
return Response({"status": "success", "references_count": author.book_references.count()})
class AdminBookAttributeViewSet(ModelViewSet):
serializer_class = AdminBookAttributeSerializer
permission_classes = [IsAuthenticated, IsSuperAdmin]
authentication_classes = [TokenAuthentication]
pagination_class = None
def get_queryset(self):
queryset = BookAttribute.objects.all()
book_reference_id = self.request.query_params.get("book_reference_id")
if book_reference_id:
queryset = queryset.filter(book_reference_id=book_reference_id)
return queryset.order_by("id")
class AdminBookReferenceImageViewSet(ModelViewSet):
serializer_class = AdminBookReferenceImageSerializer
permission_classes = [IsAuthenticated, IsSuperAdmin]
authentication_classes = [TokenAuthentication]
parser_classes = (MultiPartParser, FormParser, JSONParser)
pagination_class = None
def get_queryset(self):
queryset = BookReferenceImage.objects.all()
book_reference_id = self.request.query_params.get("book_reference_id")
if book_reference_id:
queryset = queryset.filter(book_reference_id=book_reference_id)
return queryset.order_by("order", "id")
@action(detail=False, methods=["post"])
def reorder(self, request):
image_ids = request.data.get("image_ids", []) or request.data.get("ids", [])
if not image_ids:
return Response({"error": "image_ids list is required"}, status=400)
for index, img_id in enumerate(image_ids, start=1):
BookReferenceImage.objects.filter(id=img_id).update(order=index)
return Response({"status": "success"})
class AdminHadisReferenceImageViewSet(ModelViewSet):
serializer_class = AdminHadisReferenceImageSerializer
permission_classes = [IsAuthenticated, IsSuperAdmin]
authentication_classes = [TokenAuthentication]
parser_classes = (MultiPartParser, FormParser, JSONParser)
pagination_class = None
def get_queryset(self):
queryset = ReferenceImage.objects.all().select_related("reference")
hadis_id = self.request.query_params.get("hadis_id")
if hadis_id:
queryset = queryset.filter(reference__hadis_id=hadis_id)
reference_id = self.request.query_params.get("reference_id")
if reference_id:
queryset = queryset.filter(reference_id=reference_id)
return queryset.order_by("priority", "id")
def create(self, request, *args, **kwargs):
hadis_id = request.data.get("hadis_id") or request.data.get("hadis")
reference_id = request.data.get("reference_id") or request.data.get("reference")
thumbnail = request.FILES.get("thumbnail") or request.FILES.get("image")
priority = request.data.get("priority", 0)
if not thumbnail:
return Response({"error": "thumbnail image is required"}, status=400)
if hadis_id:
try:
hadis = Hadis.objects.get(id=hadis_id)
except Hadis.DoesNotExist:
return Response({"error": "Hadis not found"}, status=404)
reference = hadis.references.first()
if not reference:
reference = HadisReference.objects.create(hadis=hadis)
elif reference_id:
try:
reference = HadisReference.objects.get(id=reference_id)
except HadisReference.DoesNotExist:
return Response({"error": "HadisReference not found"}, status=404)
else:
return Response({"error": "hadis_id or reference_id is required"}, status=400)
if not priority:
last_img = reference.images.order_by("-priority").first()
priority = (last_img.priority + 1) if last_img else 1
else:
try:
priority = int(priority)
except (ValueError, TypeError):
priority = 1
ref_image = ReferenceImage.objects.create(
reference=reference,
thumbnail=thumbnail,
priority=priority
)
serializer = self.get_serializer(ref_image)
return Response(serializer.data, status=201)
@action(detail=False, methods=["post"])
def reorder(self, request):
image_ids = request.data.get("image_ids", []) or request.data.get("ids", [])
if not image_ids:
return Response({"error": "image_ids list is required"}, status=400)
for index, img_id in enumerate(image_ids, start=1):
ReferenceImage.objects.filter(id=img_id).update(priority=index)
return Response({"status": "success"})
@action(detail=False, methods=["post"])
def bulk_delete(self, request):
image_ids = request.data.get("ids", []) or request.data.get("image_ids", [])
if not image_ids:
return Response({"error": "ids list is required"}, status=400)
deleted_count, _ = ReferenceImage.objects.filter(id__in=image_ids).delete()
return Response({"status": "success", "deleted_count": deleted_count})
class AdminBookReferenceDocumentViewSet(ModelViewSet):
serializer_class = AdminBookReferenceDocumentSerializer
permission_classes = [IsAuthenticated, IsSuperAdmin]
authentication_classes = [TokenAuthentication]
parser_classes = (MultiPartParser, FormParser, JSONParser)
pagination_class = None
def get_queryset(self):
queryset = BookReferenceDocument.objects.all()
book_reference_id = self.request.query_params.get("book_reference_id")
if book_reference_id:
queryset = queryset.filter(book_reference_id=book_reference_id)
return queryset.order_by("order", "id")
class AdminBookReferenceViewSet(ModelViewSet):
permission_classes = [IsAuthenticated, IsSuperAdmin]
authentication_classes = [TokenAuthentication]
pagination_class = StandardResultsSetPagination
parser_classes = (MultiPartParser, FormParser, JSONParser)
def get_serializer_class(self):
if self.action == "list":
return AdminBookReferenceListSerializer
return AdminBookReferenceDetailSerializer
def get_queryset(self):
queryset = BookReference.objects.all().annotate(annotated_hadis_count=Count("hadis_references__hadis", distinct=True)).prefetch_related("subject_area", "volumes", "editions").select_related("type", "author")
search_query = self.request.query_params.get("search")
if search_query:
queryset = queryset.filter(
Q(title__icontains=search_query)
| Q(slug__icontains=search_query)
| Q(editions__isbn__icontains=search_query)
| Q(editions__publisher__icontains=search_query)
| Q(description__icontains=search_query)
| Q(author__name__icontains=search_query)
| Q(tag__title__icontains=search_query)
| Q(editions__tag__title__icontains=search_query)
)
type_id = self.request.query_params.get("type")
if type_id and type_id != "all":
queryset = queryset.filter(type_id=type_id)
madhab = self.request.query_params.get("madhab")
if madhab and madhab != "all":
queryset = queryset.filter(type__sect=madhab)
subject_area_id = self.request.query_params.get("subject_area")
if subject_area_id and subject_area_id != "all":
queryset = queryset.filter(subject_area__id=subject_area_id)
author_id = self.request.query_params.get("author")
if author_id and author_id != "all":
queryset = queryset.filter(author_id=author_id)
ordering = self.request.query_params.get("ordering")
if ordering:
return queryset.distinct().order_by(ordering)
return queryset.distinct().order_by("priority", "-created_at")
class AdminCategoryQuranVerseViewSet(ModelViewSet):
permission_classes = [IsAuthenticated, IsSuperAdmin]
authentication_classes = [TokenAuthentication]
parser_classes = (JSONParser, MultiPartParser, FormParser)
pagination_class = OptionalPagination
def get_serializer_class(self):
from .serializers.serializers_admin import AdminCategoryQuranVerseSerializer
return AdminCategoryQuranVerseSerializer
def get_queryset(self):
from .models.category import CategoryQuranVerse
queryset = (
CategoryQuranVerse.objects.all()
.select_related("category", "category__sect")
.prefetch_related("secondary_translations")
)
search_query = self.request.query_params.get("search")
if search_query:
q_filter = (
Q(surah_name__icontains=search_query)
| Q(ayah_number__icontains=search_query)
| Q(arabic_text__icontains=search_query)
| Q(translation__icontains=search_query)
| Q(translator__icontains=search_query)
| Q(category__title__icontains=search_query)
| Q(category__slug__icontains=search_query)
)
queryset = queryset.filter(q_filter)
category_id = self.request.query_params.get("category")
if category_id and category_id != "all":
queryset = queryset.filter(category_id=category_id)
source_type = self.request.query_params.get("source_type")
if source_type:
queryset = queryset.filter(category__source_type=source_type)
return queryset.order_by("-id")
class AdminHadisTransmitterViewSet(ModelViewSet):
def perform_create(self, serializer):
instance = serializer.save()
hadis = instance.hadis
if hadis and (not hadis.title_narrator or (isinstance(hadis.title_narrator, list) and not any(isinstance(item, dict) and item.get('text', '').strip() for item in hadis.title_narrator))):
if hadis.auto_populate_title_narrator():
hadis.save(update_fields=['title_narrator', 'normalized_text'])
def perform_update(self, serializer):
instance = serializer.save()
hadis = instance.hadis
if hadis and (not hadis.title_narrator or (isinstance(hadis.title_narrator, list) and not any(isinstance(item, dict) and item.get('text', '').strip() for item in hadis.title_narrator))):
if hadis.auto_populate_title_narrator():
hadis.save(update_fields=['title_narrator', 'normalized_text'])
serializer_class = AdminHadisTransmitterSerializer
permission_classes = [IsAuthenticated, IsSuperAdmin]
authentication_classes = [TokenAuthentication]
parser_classes = (JSONParser, MultiPartParser, FormParser)
def get_queryset(self):
queryset = HadisTransmitter.objects.all().select_related("transmitter", "narrator_layer", "status").prefetch_related("uncertain_transmitters")
hadis_id = self.request.query_params.get("hadis_id")
if hadis_id:
queryset = queryset.filter(hadis_id=hadis_id)
return queryset.order_by("order")
@action(detail=False, methods=["post"])
def reorder(self, request):
from rest_framework.decorators import action
from rest_framework.response import Response
from django.db import transaction
hadis_transmitter_ids = request.data.get("hadis_transmitter_ids", [])
if not hadis_transmitter_ids:
return Response({"error": "hadis_transmitter_ids list is required"}, status=400)
with transaction.atomic():
# Pass 1: Set temporary positive offset order indices to avoid duplicate key violations (>= 0)
for index, ht_id in enumerate(hadis_transmitter_ids):
HadisTransmitter.objects.filter(id=ht_id).update(order=10000 + index)
# Pass 2: Set final sequential order indices starting from 0
for index, ht_id in enumerate(hadis_transmitter_ids):
HadisTransmitter.objects.filter(id=ht_id).update(order=index)
return Response({"status": "success"})
class AdminTransmitterOpinionViewSet(ModelViewSet):
permission_classes = [IsAuthenticated, IsSuperAdmin]
authentication_classes = [TokenAuthentication]
serializer_class = AdminTransmitterOpinionSerializer
pagination_class = None
def get_queryset(self):
queryset = TransmitterOpinion.objects.all().select_related("status", "scholar", "opinion", "opinion__status")
transmitter_id = self.request.query_params.get("transmitter_id")
if transmitter_id:
queryset = queryset.filter(transmitter_id=transmitter_id)
return queryset.order_by("order", "id")
@action(detail=False, methods=["post"])
def reorder(self, request):
opinion_ids = request.data.get("opinion_ids", [])
if not opinion_ids:
return Response({"error": "opinion_ids list is required"}, status=400)
from django.db import transaction
with transaction.atomic():
for index, opinion_id in enumerate(opinion_ids):
TransmitterOpinion.objects.filter(id=opinion_id).update(order=10000 + index)
for index, opinion_id in enumerate(opinion_ids):
TransmitterOpinion.objects.filter(id=opinion_id).update(order=index)
return Response({"status": "success"})
class AdminTransmitterOriginalTextViewSet(ModelViewSet):
permission_classes = [IsAuthenticated, IsSuperAdmin]
authentication_classes = [TokenAuthentication]
serializer_class = AdminTransmitterOriginalTextSerializer
parser_classes = (JSONParser, MultiPartParser, FormParser)
pagination_class = None
def get_queryset(self):
queryset = TransmitterOriginalText.objects.all().prefetch_related("references", "references__images")
transmitter_id = self.request.query_params.get("transmitter_id")
if transmitter_id:
queryset = queryset.filter(transmitter_id=transmitter_id)
return queryset.order_by("order", "id")
@action(detail=False, methods=["post"])
def reorder(self, request):
original_text_ids = request.data.get("original_text_ids", [])
if not original_text_ids:
return Response({"error": "original_text_ids list is required"}, status=400)
from django.db import transaction
with transaction.atomic():
for index, ot_id in enumerate(original_text_ids):
TransmitterOriginalText.objects.filter(id=ot_id).update(order=10000 + index)
for index, ot_id in enumerate(original_text_ids):
TransmitterOriginalText.objects.filter(id=ot_id).update(order=index)
return Response({"status": "success"})
class AdminBookEditionViewSet(ModelViewSet):
permission_classes = [IsAuthenticated, IsSuperAdmin]
authentication_classes = [TokenAuthentication]
serializer_class = AdminBookEditionSerializer
pagination_class = OptionalPagination
def get_queryset(self):
queryset = BookEdition.objects.all().select_related(
"tag",
"publisher_item__city__country",
"publisher_item__country",
"city_of_publication_city__country",
"country_of_publication_country",
).prefetch_related("books", "volumes", "researchers", "editors")
book_reference_id = self.request.query_params.get("book_reference_id")
if book_reference_id:
from django.db.models import Q
queryset = queryset.filter(Q(book_reference_id=book_reference_id) | Q(books__id=book_reference_id)).distinct()
search_query = self.request.query_params.get("search")
if search_query:
from django.db.models import Q
queryset = queryset.filter(
Q(title__icontains=search_query) |
Q(edition_number__icontains=search_query) |
Q(publisher__icontains=search_query) |
Q(year_of_publication__icontains=search_query) |
Q(isbn__icontains=search_query) |
Q(books__title__icontains=search_query)
).distinct()
return queryset.order_by("-created_at")
class AdminBookTagViewSet(ModelViewSet):
permission_classes = [IsAuthenticated, IsSuperAdmin]
authentication_classes = [TokenAuthentication]
serializer_class = AdminBookTagSerializer
pagination_class = OptionalPagination
parser_classes = (JSONParser, MultiPartParser, FormParser)
def get_queryset(self):
queryset = BookTag.objects.all()
search_query = self.request.query_params.get("search")
if search_query:
queryset = queryset.filter(title__icontains=search_query)
return queryset.order_by("-id")
class AdminBookVolumeViewSet(ModelViewSet):
permission_classes = [IsAuthenticated, IsSuperAdmin]
authentication_classes = [TokenAuthentication]
serializer_class = AdminBookVolumeSerializer
pagination_class = None
parser_classes = (MultiPartParser, FormParser, JSONParser)
def get_queryset(self):
queryset = BookVolume.objects.all()
book_reference_id = self.request.query_params.get("book_reference_id")
edition_id = self.request.query_params.get("edition_id")
if book_reference_id:
queryset = queryset.filter(book_reference_id=book_reference_id)
if edition_id:
if edition_id in ["none", "null"]:
queryset = queryset.filter(edition__isnull=True)
else:
queryset = queryset.filter(edition_id=edition_id)
return queryset.order_by("title", "id")
import re
def compute_next_version(current_version_str: str) -> str:
"""
Increments the last integer part of a version string.
Examples:
'V 0.51' -> 'V 0.52'
'V0.5' -> 'V0.6'
'1.0.0' -> '1.0.1'
'1' -> '2'
'v1.9' -> 'v1.10'
"""
if not current_version_str or not str(current_version_str).strip():
return "V 1.0"
matches = list(re.finditer(r'\d+', str(current_version_str)))
if not matches:
return f"{str(current_version_str).strip()} 2"
last_match = matches[-1]
start, end = last_match.span()
num_str = last_match.group()
next_num = int(num_str) + 1
if len(num_str) > 1 and num_str.startswith('0'):
formatted_num = str(next_num).zfill(len(num_str))
else:
formatted_num = str(next_num)
return str(current_version_str)[:start] + formatted_num + str(current_version_str)[end:]
class AdminContentReleaseViewSet(ModelViewSet):
serializer_class = AdminContentReleaseSerializer
permission_classes = [IsAuthenticated, IsSuperAdmin]
authentication_classes = [TokenAuthentication]
pagination_class = None
def get_queryset(self):
return ContentRelease.objects.all().order_by("-published_at", "-id")
def list(self, request, *args, **kwargs):
queryset = self.get_queryset()
current_release = queryset.filter(is_active=True).first()
current_data = self.get_serializer(current_release).data if current_release else None
current_ver_name = current_release.version_name if current_release else "V 0.5"
next_suggested = compute_next_version(current_ver_name)
history_data = self.get_serializer(queryset[:50], many=True).data
return Response({
"current_version": current_data,
"next_suggested_version": next_suggested,
"history": history_data
})
@action(detail=False, methods=["post"], url_path="increment")
def increment(self, request):
current_release = self.get_queryset().filter(is_active=True).first()
current_ver_name = current_release.version_name if current_release else "V 0.5"
version_name = request.data.get("version_name")
if not version_name or not str(version_name).strip():
version_name = compute_next_version(current_ver_name)
else:
version_name = str(version_name).strip()
description = request.data.get("description", "") or ""
# Create new content release
new_release = ContentRelease.objects.create(
version_name=version_name,
description=description,
is_active=True
)
from .signals import invalidate_hadis_cache
invalidate_hadis_cache()
# تریگر خودکار ساخت و فشرده‌سازی دیتابیس SQLite در پس‌زمینه
self._trigger_sqlite_export(notes=f"Auto-generated for version {version_name}")
return Response({
"status": "success",
"message": f"Version incremented to {version_name}",
"release": self.get_serializer(new_release).data,
"next_suggested_version": compute_next_version(version_name)
})
def _trigger_sqlite_export(self, notes=""):
"""اجرای پس‌زمینه دستور تولید دیتابیس SQLite بدون متوقف کردن پاسخ HTTP"""
import threading
import logging
from django.core.management import call_command
logger = logging.getLogger(__name__)
def _task():
try:
logger.info("Starting background SQLite export...")
call_command('export_hadis_sqlite', notes=notes)
logger.info("Background SQLite export completed successfully.")
except Exception as e:
logger.error(f"Failed to export SQLite in background: {e}", exc_info=True)
thread = threading.Thread(target=_task, daemon=True)
thread.start()
def perform_create(self, serializer):
instance = serializer.save()
from .signals import invalidate_hadis_cache
invalidate_hadis_cache()
self._trigger_sqlite_export(notes=f"Auto-generated for created version {instance.version_name}")
def perform_update(self, serializer):
super().perform_update(serializer)
from .signals import invalidate_hadis_cache
invalidate_hadis_cache()
def perform_destroy(self, instance):
super().perform_destroy(instance)
from .signals import invalidate_hadis_cache
invalidate_hadis_cache()
class DovodiCountryViewSet(ModelViewSet):
serializer_class = AdminCountrySerializer
permission_classes = [IsAuthenticated, IsSuperAdmin]
authentication_classes = [TokenAuthentication]
pagination_class = OptionalPagination
parser_classes = (JSONParser, MultiPartParser, FormParser)
def get_queryset(self):
queryset = Country.objects.all().order_by("slug")
search_query = self.request.query_params.get("search")
if search_query:
queryset = queryset.filter(
Q(title__icontains=search_query)
| Q(slug__icontains=search_query)
)
return queryset
class DovodiBookPublisherViewSet(ModelViewSet):
serializer_class = AdminBookPublisherSerializer
permission_classes = [IsAuthenticated, IsSuperAdmin]
authentication_classes = [TokenAuthentication]
pagination_class = OptionalPagination
parser_classes = (JSONParser, MultiPartParser, FormParser)
def get_queryset(self):
queryset = BookPublisher.objects.select_related("city", "country").all().order_by("slug")
search_query = self.request.query_params.get("search")
if search_query:
queryset = queryset.filter(
Q(title__icontains=search_query)
| Q(slug__icontains=search_query)
| Q(city__title__icontains=search_query)
| Q(country__title__icontains=search_query)
)
city_id = self.request.query_params.get("city")
if city_id and city_id != "all":
queryset = queryset.filter(city_id=city_id)
country_id = self.request.query_params.get("country")
if country_id and country_id != "all":
queryset = queryset.filter(country_id=country_id)
return queryset
class DovodiCityViewSet(ModelViewSet):
serializer_class = AdminCitySerializer
permission_classes = [IsAuthenticated, IsSuperAdmin]
authentication_classes = [TokenAuthentication]
pagination_class = OptionalPagination
parser_classes = (JSONParser, MultiPartParser, FormParser)
def get_queryset(self):
queryset = City.objects.all().select_related("country").order_by("slug")
search_query = self.request.query_params.get("search")
if search_query:
queryset = queryset.filter(
Q(title__icontains=search_query)
| Q(slug__icontains=search_query)
)
country_id = self.request.query_params.get("country") or self.request.query_params.get("country_id")
if country_id:
queryset = queryset.filter(country_id=country_id)
return queryset
class DovodiKunyaViewSet(ModelViewSet):
serializer_class = AdminKunyaSerializer
permission_classes = [IsAuthenticated, IsSuperAdmin]
authentication_classes = [TokenAuthentication]
pagination_class = OptionalPagination
parser_classes = (JSONParser, MultiPartParser, FormParser)
def get_queryset(self):
queryset = Kunya.objects.all().order_by("slug")
search_query = self.request.query_params.get("search")
if search_query:
queryset = queryset.filter(
Q(title__icontains=search_query)
| Q(slug__icontains=search_query)
)
return queryset
class DovodiTransmitterFeatureViewSet(ModelViewSet):
serializer_class = AdminTransmitterFeatureSerializer
permission_classes = [IsAuthenticated, IsSuperAdmin]
authentication_classes = [TokenAuthentication]
pagination_class = OptionalPagination
parser_classes = (JSONParser, MultiPartParser, FormParser)
def get_queryset(self):
queryset = TransmitterFeature.objects.all().order_by("slug")
search_query = self.request.query_params.get("search")
if search_query:
queryset = queryset.filter(
Q(title__icontains=search_query)
| Q(slug__icontains=search_query)
)
return queryset
class DovodiTransmitterProfessionViewSet(ModelViewSet):
serializer_class = AdminTransmitterProfessionSerializer
permission_classes = [IsAuthenticated, IsSuperAdmin]
authentication_classes = [TokenAuthentication]
pagination_class = OptionalPagination
parser_classes = (JSONParser, MultiPartParser, FormParser)
def get_queryset(self):
queryset = TransmitterProfession.objects.all().order_by("slug")
search_query = self.request.query_params.get("search")
if search_query:
queryset = queryset.filter(
Q(title__icontains=search_query)
| Q(slug__icontains=search_query)
)
return queryset
class DovodiSahabiStatusViewSet(ModelViewSet):
serializer_class = AdminSahabiStatusSerializer
permission_classes = [IsAuthenticated, IsSuperAdmin]
authentication_classes = [TokenAuthentication]
pagination_class = OptionalPagination
parser_classes = (JSONParser, MultiPartParser, FormParser)
def get_queryset(self):
queryset = SahabiStatus.objects.all().order_by("id")
search_query = self.request.query_params.get("search")
if search_query:
queryset = queryset.filter(
Q(title__icontains=search_query)
| Q(slug__icontains=search_query)
)
return queryset
class DovodiMadhhabViewSet(ModelViewSet):
serializer_class = AdminMadhhabSerializer
permission_classes = [IsAuthenticated, IsSuperAdmin]
authentication_classes = [TokenAuthentication]
pagination_class = OptionalPagination
parser_classes = (JSONParser, MultiPartParser, FormParser)
def get_queryset(self):
queryset = Madhhab.objects.all().order_by("slug", "id")
search_query = self.request.query_params.get("search")
if search_query:
queryset = queryset.filter(
Q(title__icontains=search_query)
| Q(slug__icontains=search_query)
)
return queryset
class DovodiRelationTypeViewSet(ModelViewSet):
serializer_class = AdminRelationTypeSerializer
permission_classes = [IsAuthenticated, IsSuperAdmin]
authentication_classes = [TokenAuthentication]
pagination_class = OptionalPagination
parser_classes = (JSONParser, MultiPartParser, FormParser)
def get_queryset(self):
queryset = RelationType.objects.all().order_by("slug", "id")
search_query = self.request.query_params.get("search")
if search_query:
queryset = queryset.filter(
Q(title__icontains=search_query)
| Q(slug__icontains=search_query)
| Q(code__icontains=search_query)
)
return queryset