"""Automatic section detection and meeting recommendation for group books.""" from __future__ import annotations import logging from datetime import date, timedelta from typing import Any from django.db.models import QuerySet from apps.books.models import BookChapter, EBook from apps.groups.models import GroupBook, ReadingSchedule, Section logger = logging.getLogger(__name__) WORDS_PER_MINUTE = 250 # Average reading speed def _fetch_chapter_text(ebook: EBook, chapter: BookChapter) -> str: """Extract plain text from a chapter for word counting.""" try: from ebooklib import epub from bs4 import BeautifulSoup except ImportError: return "" try: book = epub.read_epub(ebook.file.path) href = chapter.href or "" 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") body = soup.find("body") if body: return body.get_text(separator=" ", strip=True) return soup.get_text(separator=" ", strip=True) return "" except Exception: logger.exception("Failed to extract text for chapter %s", chapter.id) return "" def _estimate_reading_minutes(text: str) -> int: """Estimate reading time based on word count at WORDS_PER_MINUTE.""" word_count = len(text.split()) if word_count == 0: return 1 return max(1, round(word_count / WORDS_PER_MINUTE)) def detect_sections(group_book: GroupBook) -> list[dict[str, Any]]: """Auto-detect sections from a GroupBook's chapters. Groups consecutive chapters into logical sections based on TOC structure. Top-level TOC entries become sections; if there are very few (< 3), groups of ~5 chapters become sections instead. """ ebook = group_book.ebook chapters = list( BookChapter.objects.filter(ebook=ebook).order_by("index").select_related("ebook") ) if not chapters: return [] top_level = [ch for ch in chapters if not ch.children or len(ch.children) == 0] has_children = [ch for ch in chapters if ch.children and len(ch.children) > 0] sections: list[dict[str, Any]] = [] if len(has_children) >= 3: # Use TOC structure: each top-level chapter (with children) is a section for idx, ch in enumerate(has_children): # Find all child chapters belonging to this parent child_indices = _collect_child_indices(chapters, ch, idx) start_idx = ch.index end_idx = child_indices[-1] if child_indices else start_idx section_chapters = [c for c in chapters if start_idx <= c.index <= end_idx] total_text = "" for sc in section_chapters: total_text += " " + _fetch_chapter_text(ebook, sc) sections.append({ "title": ch.title, "order": idx + 1, "start_chapter_index": start_idx, "end_chapter_index": end_idx + 1, "estimated_reading_minutes": _estimate_reading_minutes(total_text), }) else: # Group chapters into chunks of ~5 chunk_size = max(1, len(top_level) // 6 if len(top_level) > 6 else 5) chunk_size = max(3, min(chunk_size, 10)) group_start = 0 section_order = 1 total = len(top_level) or len(chapters) source = top_level or chapters while group_start < total: group_end = min(group_start + chunk_size, total) chunk = source[group_start:group_end] total_text = "" for ch in chunk: total_text += " " + _fetch_chapter_text(ebook, ch) first_title = chunk[0].title if chunk else "Section" last_title = chunk[-1].title if len(chunk) > 1 else "" title = f"{first_title}" if not last_title or first_title == last_title else f"{first_title} — {last_title}" sections.append({ "title": title, "order": section_order, "start_chapter_index": chunk[0].index, "end_chapter_index": chunk[-1].index + 1, "estimated_reading_minutes": _estimate_reading_minutes(total_text), }) group_start = group_end section_order += 1 return sections def _collect_child_indices(chapters: list[BookChapter], parent: BookChapter, parent_idx: int) -> list[int]: """Collect indices of all chapters that are children of the given parent TOC entry.""" indices: list[int] = [parent.index] child_hrefs: set[str] = set() for child in parent.children: if isinstance(child, dict): child_hrefs.add(child.get("href", "")) elif hasattr(child, "href"): child_hrefs.add(getattr(child, "href", "")) for ch in chapters: if ch.index == parent.index: continue if ch.href in child_hrefs or any( ch.href.endswith(h) or h.endswith(ch.href) for h in child_hrefs ): indices.append(ch.index) # Also include chapters between this parent and the next parent if parent_idx + 1 < len(chapters): next_parent = chapters[parent_idx + 1] for ch in chapters: if parent.index < ch.index < next_parent.index: indices.append(ch.index) return sorted(set(indices)) def apply_sections(group_book: GroupBook) -> list[Section]: """Detect sections and persist them to the database, replacing existing ones.""" Section.objects.filter(group_book=group_book).delete() sections_data = detect_sections(group_book) created: list[Section] = [] for data in sections_data: section = Section.objects.create( group_book=group_book, title=data["title"], order=data["order"], start_chapter_index=data["start_chapter_index"], end_chapter_index=data["end_chapter_index"], estimated_reading_minutes=data["estimated_reading_minutes"], ) created.append(section) return created def recommend_meetings(group_book: GroupBook, num_meetings: int = 4) -> list[dict[str, Any]]: """Recommend which sections to assign to each weekly meeting. Distributes sections across meetings, trying to balance total reading time. Returns a list of meeting assignments ready for schedule creation. """ sections = list( Section.objects.filter(group_book=group_book).order_by("order") ) if not sections: return [] # Calculate total minutes to distribute total_minutes = sum(s.estimated_reading_minutes for s in sections) target_per_meeting = total_minutes / num_meetings meetings: list[dict[str, Any]] = [] current_meeting: list[int] = [] current_minutes = 0 for section in sections: if current_meeting and current_minutes + section.estimated_reading_minutes > target_per_meeting * 1.4: # Start new meeting if adding this section would overshoot too much if len(meetings) < num_meetings - 1: meetings.append({ "meeting_number": len(meetings) + 1, "section_ids": current_meeting, "total_minutes": current_minutes, }) current_meeting = [] current_minutes = 0 current_meeting.append(section.id) current_minutes += section.estimated_reading_minutes # Add the last meeting if current_meeting: meetings.append({ "meeting_number": len(meetings) + 1, "section_ids": current_meeting, "total_minutes": current_minutes, }) # If we have fewer than num_meetings, we could split the largest one # For now, just return what we have return meetings def apply_schedule(group_book: GroupBook, num_meetings: int = 4) -> list[ReadingSchedule]: """Generate and persist a reading schedule.""" ReadingSchedule.objects.filter(group_book=group_book).delete() recommendations = recommend_meetings(group_book, num_meetings) today = date.today() created: list[ReadingSchedule] = [] for rec in recommendations: week_date = today + timedelta(weeks=rec["meeting_number"] - 1) schedule = ReadingSchedule.objects.create( group_book=group_book, meeting_number=rec["meeting_number"], week_date=week_date, section_ids=rec["section_ids"], ) created.append(schedule) return created