"""Recommendation engine for splitting book chapters into balanced weekly meetings. Algorithm: 1. Gather all BookChapters for an ebook (flat list, respecting hierarchy). 2. For each chapter, estimate reading time based on content length or default weight. 3. Partition chapters into N meetings (default 4) minimizing per-meeting time variance while respecting chapter boundaries (never split a chapter). 4. Optionally prefer natural boundaries: if a chapter's children exist, keep siblings together where possible. """ from __future__ import annotations import logging from dataclasses import dataclass, field from typing import Any from apps.books.models import BookChapter, EBook, MeetingSection, ReadingSchedule logger = logging.getLogger(__name__) DEFAULT_MEETING_COUNT = 4 @dataclass class ChapterWithWeight: chapter_id: int title: str href: str index: int weight: float # estimated reading time weight children: list[int] = field(default_factory=list) def _collect_chapters(ebook: EBook) -> list[ChapterWithWeight]: """Collect all chapters for an ebook as a flat, ordered list with weights.""" chapters_qs = ( BookChapter.objects.filter(ebook=ebook) .order_by("index") .prefetch_related("ebook") ) chapters = list(chapters_qs) # Build a map of chapter id -> ChapterWithWeight result: list[ChapterWithWeight] = [] for ch in chapters: weight = _estimate_chapter_weight(ch) result.append( ChapterWithWeight( chapter_id=ch.id, title=ch.title, href=ch.href or "", index=ch.index, weight=weight, ) ) return result def _estimate_chapter_weight(chapter: BookChapter) -> float: """Estimate reading time weight for a chapter. Defaults to 1.0 per chapter. Sub-chapters (those with non-empty children) get a weight proportional to their child count + 1. """ children = chapter.children or [] child_count = len(children) if child_count > 0: return float(child_count + 1) return 1.0 def _partition_chapters( chapters: list[ChapterWithWeight], num_meetings: int = DEFAULT_MEETING_COUNT, ) -> list[list[ChapterWithWeight]]: """Partition chapters into `num_meetings` groups minimizing weight variance. Uses a greedy approach: compute target weight per meeting, then place chapters sequentially, starting a new meeting when cumulative weight exceeds the target and adding the overflow to the next meeting. Never splits a chapter — boundaries are always at chapter edges. """ if not chapters: return [[] for _ in range(num_meetings)] total_weight = sum(ch.weight for ch in chapters) target_weight = total_weight / num_meetings meetings: list[list[ChapterWithWeight]] = [] current_meeting: list[ChapterWithWeight] = [] current_weight = 0.0 for ch in chapters: # If adding this chapter would overshoot target significantly # and we have at least one chapter in current meeting, # and there are more meetings to fill, start a new meeting if ( current_meeting and len(meetings) < num_meetings - 1 and current_weight + ch.weight > target_weight * 1.4 and current_weight >= target_weight * 0.5 ): meetings.append(current_meeting) current_meeting = [] current_weight = 0.0 current_meeting.append(ch) current_weight += ch.weight # Add the last meeting if current_meeting: meetings.append(current_meeting) # Pad or balance: if we have fewer meetings than requested, split the largest while len(meetings) < num_meetings and len(meetings) >= 1: # Find the largest meeting to split largest_idx = max( range(len(meetings)), key=lambda i: sum(c.weight for c in meetings[i]), ) largest = meetings[largest_idx] if len(largest) <= 1: break mid = len(largest) // 2 meetings[largest_idx] = largest[:mid] meetings.insert(largest_idx + 1, largest[mid:]) # If we have more meetings than requested (shouldn't normally happen), # merge the smallest adjacent pair while len(meetings) > num_meetings: best_merge = min( range(len(meetings) - 1), key=lambda i: sum(c.weight for c in meetings[i]) + sum(c.weight for c in meetings[i + 1]), ) merged = meetings[best_merge] + meetings[best_merge + 1] meetings[best_merge] = merged del meetings[best_merge + 1] return meetings def generate_recommendations( ebook: EBook, user, meeting_count: int = DEFAULT_MEETING_COUNT, ) -> ReadingSchedule: """Generate section-based reading recommendations for an ebook. Creates (or replaces) a ReadingSchedule with MeetingSections. Args: ebook: The EBook to generate recommendations for. user: The requesting user. meeting_count: Number of weekly meetings (default 4). Returns: The created ReadingSchedule instance. """ chapters = _collect_chapters(ebook) # Delete any existing schedule for this ebook/user ReadingSchedule.objects.filter(ebook=ebook, user=user).delete() schedule = ReadingSchedule.objects.create( ebook=ebook, user=user, meeting_count=meeting_count, ) if not chapters: # Create empty meetings for i in range(meeting_count): MeetingSection.objects.create( schedule=schedule, meeting_index=i, title=f"Week {i + 1}", ) return schedule partition = _partition_chapters(chapters, meeting_count) for meeting_idx, meeting_chapters in enumerate(partition): chapter_ids = [ch.chapter_id for ch in meeting_chapters] total_weight = sum(ch.weight for ch in meeting_chapters) # Build a descriptive title from first and last chapter first = meeting_chapters[0].title if meeting_chapters else "No chapters" last = meeting_chapters[-1].title if len(meeting_chapters) > 1 else "" if last and last != first: title = f"{first} → {last}" else: title = first # Truncate title if too long if len(title) > 200: title = title[:197] + "..." # Estimate time: ~5 minutes per weight point (adjustable heuristic) estimated_time = max(5, int(total_weight * 5)) MeetingSection.objects.create( schedule=schedule, meeting_index=meeting_idx, title=title, estimated_time_minutes=estimated_time, chapter_ids=chapter_ids, ) return schedule def regenerate_recommendations(schedule: ReadingSchedule) -> ReadingSchedule: """Regenerate recommendations for an existing schedule.""" return generate_recommendations( ebook=schedule.ebook, user=schedule.user, meeting_count=schedule.meeting_count, ) def confirm_schedule(schedule: ReadingSchedule) -> ReadingSchedule: """Confirm/finalize a reading schedule.""" schedule.confirmed = True schedule.save(update_fields=["confirmed", "updated_at"]) return schedule