Archived
Marko
1b3b709ba5
Backend: - Create hermes Django app with models: ReadingGroup, GroupBook, Section, ReadingSchedule, MemberProgress - EPUB section splitting service with automatic detection and reading time estimation - Section recommendation engine for 4-week meeting schedule - REST API endpoints for groups, books, sections, schedule, and member progress - Manual section adjustment (merge/split) support Frontend: - GroupsPage: list/create reading groups - GroupDetailPage: manage members, upload EPUB to group, view group books - GroupBookPage: section breakdown with merge/split controls, reading schedule, member progress - API client and TypeScript types for all group operations - i18n keys for English and Spanish Shared: - Group-related types and API endpoint constants in packages/shared
242 lines
8.6 KiB
Python
242 lines
8.6 KiB
Python
"""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
|