from __future__ import annotations import logging from typing import Any from django.db.models import QuerySet, Q from django_filters.rest_framework import DjangoFilterBackend from rest_framework import parsers, permissions, status, viewsets from rest_framework.decorators import action from rest_framework.filters import OrderingFilter, SearchFilter from rest_framework.permissions import AllowAny, IsAuthenticated from rest_framework.request import Request from rest_framework.response import Response from apps.books.models import Book, BookChapter, EBook, ReadingProgress, ReadingSettings from apps.books.serializers import ( BookDetailSerializer, BookListSerializer, BookSerializer, BookChapterSerializer, EBookContentSerializer, EBookDetailSerializer, EBookListSerializer, EBookTocSerializer, EBookUploadSerializer, ReadingProgressSerializer, ReadingSettingsSerializer, ) logger = logging.getLogger(__name__) class BookViewSet(viewsets.ModelViewSet): queryset = Book.objects.all() permission_classes = [IsAuthenticated] filter_backends = [DjangoFilterBackend, SearchFilter, OrderingFilter] filterset_fields = ["author", "genre", "reading_status"] search_fields = ["title", "author", "genre"] ordering_fields = ["title", "author", "genre", "created_at"] ordering = ["title"] def get_serializer_class(self): if self.action == "retrieve": return BookDetailSerializer if self.action == "list": return BookListSerializer return BookSerializer def get_queryset(self) -> QuerySet[Book]: qs = super().get_queryset() query = self.request.query_params.get("q", "").strip() if query: qs = qs.filter(Q(title__icontains=query) | Q(author__icontains=query) | Q(genre__icontains=query)) return qs @action(detail=False, methods=["get"], permission_classes=[AllowAny]) def genres(self, request: Request) -> Response: genre_list = Book.objects.values_list("genre", flat=True).distinct().order_by("genre") return Response([g for g in genre_list if g]) @action(detail=False, methods=["get"], permission_classes=[AllowAny]) def authors(self, request: Request) -> Response: author_list = Book.objects.values_list("author", flat=True).distinct().order_by("author") return Response([a for a in author_list if a]) class IsEBookOwner(permissions.BasePermission): def has_object_permission(self, request: Request, view: object, obj: EBook) -> bool: return obj.user == request.user class EBookViewSet(viewsets.ModelViewSet): parser_classes = [parsers.MultiPartParser, parsers.FormParser, parsers.JSONParser] permission_classes = [IsAuthenticated, IsEBookOwner] def get_serializer_class(self): if self.action == "create": return EBookUploadSerializer if self.action in ("list",): return EBookListSerializer if self.action in ("toc",): return BookChapterSerializer return EBookDetailSerializer def get_queryset(self): return EBook.objects.filter(user=self.request.user).select_related("reading_progress", "user") @action(detail=True, methods=["get", "patch"]) def progress(self, request: Request, pk: int | None = None) -> Response: ebook = self.get_object() progress_obj, _created = ReadingProgress.objects.get_or_create(user=request.user, ebook=ebook) if request.method == "GET": serializer = ReadingProgressSerializer(progress_obj) return Response(serializer.data) serializer = ReadingProgressSerializer(progress_obj, data=request.data, partial=True) serializer.is_valid(raise_exception=True) serializer.save() return Response(serializer.data) @action(detail=True, methods=["post"]) def process(self, request: Request, pk: int | None = None) -> Response: """Trigger e-book processing: metadata extraction, TOC building, page counting.""" ebook = self.get_object() if not ebook.file: return Response({"error": "No file found for this e-book."}, status=status.HTTP_400_BAD_REQUEST) try: from apps.books.services import process_ebook file_path = ebook.file.path result = process_ebook(file_path, original_filename=ebook.filename()) # Update ebook with extracted data ebook.format = result.get("format", ebook.format) ebook.page_count = result.get("page_count", 0) ebook.metadata_json = result.get("metadata", {}) ebook.save(update_fields=["format", "page_count", "metadata_json", "updated_at"]) # Store chapters in DB raw_toc: list[dict[str, Any]] = result.get("toc", []) BookChapter.objects.filter(ebook=ebook).delete() _store_chapters(ebook, raw_toc) return Response({ "format": ebook.format, "page_count": ebook.page_count, "metadata": ebook.metadata_json, "toc_count": len(raw_toc), "status": "processed", }) except Exception as exc: logger.exception("Failed to process ebook %s", ebook.id) return Response({"error": f"Processing failed: {exc}"}, status=status.HTTP_500_INTERNAL_SERVER_ERROR) @action(detail=True, methods=["get"]) def toc(self, request: Request, pk: int | None = None) -> Response: """Return hierarchical table of contents.""" ebook = self.get_object() chapters = BookChapter.objects.filter(ebook=ebook).order_by("index").select_related("ebook") serializer = BookChapterSerializer(chapters, many=True) return Response({ "chapters": serializer.data, "format": ebook.format, "page_count": ebook.page_count, }) @action(detail=True, methods=["get"]) def content(self, request: Request, pk: int | None = None) -> Response: """Return paginated content for a given page number. Query params: page (int): page/chapter index to fetch (1-indexed, default: 1) """ ebook = self.get_object() page = max(1, int(request.query_params.get("page", 1))) chapters = list(BookChapter.objects.filter(ebook=ebook).order_by("index").select_related("ebook")) total_pages = len(chapters) or ebook.page_count or 1 chapter: BookChapter | None = None chapter_title = "" content_html = "" if chapters and 0 <= page - 1 < len(chapters): ch = chapters[page - 1] chapter_title = ch.title content_html = _fetch_chapter_content(ebook, ch) serializer = EBookContentSerializer(data={ "page": page, "total_pages": total_pages, "content": content_html, "chapter_title": chapter_title, "format": ebook.format, }) serializer.is_valid(raise_exception=True) return Response(serializer.data) def _store_chapters(ebook: EBook, toc: list[dict[str, Any]], parent_index: int = 0) -> None: """Recursively store TOC entries as BookChapter records.""" for idx, entry in enumerate(toc): BookChapter.objects.create( ebook=ebook, title=entry.get("title", "Untitled"), index=parent_index + idx, href=entry.get("href", ""), children=entry.get("children", []), ) children = entry.get("children", []) if children: _store_chapters(ebook, children, parent_index + idx + 1) def _fetch_chapter_content(ebook: EBook, chapter: BookChapter) -> str: """Fetch HTML content for a chapter from the e-book file.""" if ebook.format == "epub": return _fetch_epub_chapter_content(ebook.file.path, chapter) return "" def _fetch_epub_chapter_content(file_path: str, chapter: BookChapter) -> str: """Extract HTML content of a specific EPUB chapter by href.""" try: from ebooklib import epub from bs4 import BeautifulSoup except ImportError: return "" try: book = epub.read_epub(file_path) href = chapter.href or "" # Find the item by href for item in book.get_items(): item_name = item.get_name() or "" if href and (item_name.endswith(href) or href.endswith(item_name)): content = item.get_content() soup = BeautifulSoup(content, "html.parser") # Clean up — remove body/html/head wrappers, keep inner content body = soup.find("body") if body: return str(body) return str(soup) return "" except Exception: logger.exception("Failed to fetch EPUB chapter content for %s", chapter.href) return "" class ReadingSettingsViewSet(viewsets.GenericViewSet): permission_classes = [IsAuthenticated] serializer_class = ReadingSettingsSerializer def get_queryset(self): return ReadingSettings.objects.filter(user=self.request.user) def list(self, request: Request) -> Response: settings_obj, _created = ReadingSettings.objects.get_or_create(user=request.user) serializer = self.get_serializer(settings_obj) return Response(serializer.data) def partial_update(self, request: Request) -> Response: settings_obj, _created = ReadingSettings.objects.get_or_create(user=request.user) serializer = self.get_serializer(settings_obj, data=request.data, partial=True) serializer.is_valid(raise_exception=True) serializer.save() return Response(serializer.data)