.epub" (see # remote.py) — reused here to find the server book id for progress push. _SERVER_ID_RE = re.compile(r"^(\d{4,7}) - ") def _server_book_id(path: Path) -> int | None: m = _SERVER_ID_RE.match(path.name) return int(m.group(1)) if m else None class FootnoteScreen(ModalScreen[None]): BINDINGS = [Binding("escape,f,q", "dismiss_self", "Schließen")] def __init__(self, notes: list[str]) -> None: super().__init__() self._notes = notes def compose(self): with VerticalScroll(id="footnote-body"): for note in self._notes: yield Static(note, classes="footnote-note") yield Label("Escape/f zum Schließen", id="footnote-hint") def action_dismiss_self(self) -> None: self.dismiss(None) class ReaderScreen(Screen): BINDINGS = [ Binding("j,down", "scroll_down_line", "Runter", show=False), Binding("k,up", "scroll_up_line", "Hoch", show=False), Binding("n", "next_chapter", "Nächstes Kapitel"), Binding("p", "prev_chapter", "Vorheriges Kapitel"), Binding("f", "peek_footnote", "Fußnote"), Binding("q,escape", "back", "Zurück zur Bibliothek"), ] def __init__(self, path: Path) -> None: super().__init__() self.path = path self.book: blocks.FlatBook | None = None self.book_layout: layout.BookLayout = _EMPTY_LAYOUT self._server_id = _server_book_id(path) self._remote_cfg = config_mod.load() if self._server_id is not None else None self._last_pushed_anchor = "" def compose(self): yield Header() yield LoadingIndicator(id="reader-loading") with Container(id="reader-scroll"): # Stays visible (empty) from the start rather than toggling display # on once loaded — a widget that's just been switched from hidden # to visible hasn't been through a layout pass yet, so its `.size` # is still (0, 0) and an immediate scroll_to() right after has # nothing to clamp against and silently resets to 0. self._book_view = ContinuousBookView(_EMPTY_LAYOUT) yield self._book_view yield Footer() def on_mount(self) -> None: self._load_book(self._content_width()) @work(thread=True) def _load_book(self, width: int) -> None: cached = cache.load(self.path, width) if cached is not None: book, book_layout = cached else: book = blocks.load_flat_book(self.path) console = Console(width=width) book_layout = layout.build_layout([b.text for b in book.blocks], width, console) cache.save(self.path, width, book, book_layout) self.app.call_from_thread(self._on_book_loaded, book, book_layout, width) def _on_book_loaded(self, book: blocks.FlatBook, book_layout: layout.BookLayout, width: int) -> None: self.book = book self.sub_title = book.title self._apply_layout(book_layout, width) self.query_one("#reader-loading", LoadingIndicator).display = False self.call_after_refresh(self._restore_position) self.set_interval(AUTOSAVE_INTERVAL, self._autosave) def _content_width(self) -> int: return max(20, min(MAX_LINE_WIDTH, self.size.width - 4)) def _apply_layout(self, book_layout: layout.BookLayout, width: int) -> None: self.book_layout = book_layout self._book_view.styles.width = width self._book_view.book_layout = book_layout self._book_view.virtual_size = Size(width, book_layout.total_rows) self._book_view.refresh() def _build_layout(self) -> None: assert self.book is not None width = self._content_width() console = Console(width=width) block_texts = [b.text for b in self.book.blocks] new_layout = layout.build_layout(block_texts, width, console) cache.save(self.path, width, self.book, new_layout) self._apply_layout(new_layout, width) def on_resize(self) -> None: if self.book is None: return old_y = self._book_view.scroll_y anchor_block, anchor_frac = layout.get_position_anchor(self.book_layout, old_y) self._build_layout() new_y = layout.scroll_y_for_anchor(self.book_layout, anchor_block, anchor_frac) self._book_view.scroll_to(y=new_y, animate=False, immediate=True) def _restore_position(self, attempt: int = 0) -> None: if self.book is None: return saved = progress.load(self.book.id) if saved is None: return parsed = blocks.parse_anchor(saved.anchor) if parsed is None: return block_index, inner_fraction = parsed y = layout.scroll_y_for_anchor(self.book_layout, block_index, inner_fraction) if y <= 0: return self._book_view.scroll_to(y=y, animate=False, immediate=True) # A widget that's only just become part of the layout doesn't always # honor an immediate scroll on the first attempt (its own size/scroll # bounds can still be mid-update) — verify it actually landed and # retry a bounded number of times rather than guessing a fixed delay. if attempt < 20 and abs(self._book_view.scroll_y - y) > 1: self.call_after_refresh(lambda: self._restore_position(attempt + 1)) def _current_anchor_str(self) -> str: block_index, inner_fraction = layout.get_position_anchor(self.book_layout, self._book_view.scroll_y) return blocks.format_anchor(block_index, inner_fraction) def _autosave(self) -> None: self._save_progress() def _save_progress(self) -> None: if self.book is None or not self.book_layout.heights: return anchor = self._current_anchor_str() advanced = progress.save(self.book.id, progress.Position(anchor=anchor)) if advanced and self._remote_cfg is not None and self._server_id is not None: self._push_remote_progress(anchor) @work(thread=True, exclusive=True, group="progress-push") def _push_remote_progress(self, anchor: str) -> None: if anchor == self._last_pushed_anchor or self._remote_cfg is None or self._server_id is None: return try: api.post_progress( self._remote_cfg.server_url, self._remote_cfg.api_token, self._server_id, scroll_fraction=0.0, position_anchor=anchor, force=False, ) self._last_pushed_anchor = anchor except Exception: pass # best-effort — local progress is already saved regardless def on_unmount(self) -> None: self._save_progress() def action_scroll_down_line(self) -> None: self._book_view.scroll_relative(y=1, animate=False) def action_scroll_up_line(self) -> None: self._book_view.scroll_relative(y=-1, animate=False) def _current_chapter_index(self) -> int: if self.book is None: return 0 block_index, _ = layout.get_position_anchor(self.book_layout, self._book_view.scroll_y) return bisect.bisect_right(self.book.chapter_start_block, block_index) - 1 def _jump_to_block(self, block_index: int) -> None: y = layout.scroll_y_for_anchor(self.book_layout, block_index, 0.0) self._book_view.scroll_to(y=y, animate=False, immediate=True) def action_next_chapter(self) -> None: if self.book is None: return chapter = self._current_chapter_index() if chapter + 1 < len(self.book.chapter_start_block): self._jump_to_block(self.book.chapter_start_block[chapter + 1]) self._save_progress() def action_prev_chapter(self) -> None: if self.book is None: return chapter = self._current_chapter_index() if chapter > 0: self._jump_to_block(self.book.chapter_start_block[chapter - 1]) self._save_progress() def action_peek_footnote(self) -> None: if self.book is None or not self.book_layout.heights: return current_block, _ = layout.get_position_anchor(self.book_layout, self._book_view.scroll_y) forward = [(i, b) for i, b in enumerate(self.book.blocks) if i >= current_block and b.footnotes] backward = [(i, b) for i, b in enumerate(self.book.blocks) if i < current_block and b.footnotes] candidate = forward[0] if forward else (backward[-1] if backward else None) if candidate is None: self.notify("Keine Fußnote in diesem Buch gefunden.", timeout=3) return _, block = candidate notes: list[str] = [] for ref in block.footnotes: target_idx = self.book.footnote_targets.get(ref.target_id) if target_idx is None: continue notes.append(self.book.blocks[target_idx].text) if not notes: self.notify("Fußnote konnte nicht aufgelöst werden.", timeout=3) return self.app.push_screen(FootnoteScreen(notes)) def action_back(self) -> None: self.app.pop

"""Continuous whole-book reader screen — mirrors the web reader's single scrollable view (see blocks.py's module docstring) instead of the old per-chapter pagination, so reading position is expressed in the same "blockIndex:innerFraction" anchor space as the server and the web client. Uses book_view.ContinuousBookView (a Line-API widget) rather than dumping the whole book into one Static: for large books (tens of thousands of blocks) a single giant renderable made Textual's layout/paint pass take upwards of a minute, whereas the Line API only ever renders the rows on screen. """ from __future__ import annotations import bisect import re from pathlib import Path from rich.console import Console from textual import work from textual.binding import Binding from textual.containers import Container, VerticalScroll from textual.geometry import Size from textual.screen import ModalScreen, Screen from textual.widgets import Footer, Header, Label, LoadingIndicator, Static from . import api, blocks, cache, config as config_mod, layout, progress from .book_view import ContinuousBookView MAX_LINE_WIDTH = 120 AUTOSAVE_INTERVAL = 5.0 _EMPTY_LAYOUT = layout.BookLayout(starts=[], heights=[], row_strips=[], total_rows=0, width=1) # Downloaded-via-sync files are named " - .epub" (see # remote.py) — reused here to find the server book id for progress push. _SERVER_ID_RE = re.compile(r"^(\d{4,7}) - ") def _server_book_id(path: Path) -> int | None: m = _SERVER_ID_RE.match(path.name) return int(m.group(1)) if m else None class FootnoteScreen(ModalScreen[None]): BINDINGS = [Binding("escape,f,q", "dismiss_self", "Schließen")] def __init__(self, notes: list[str]) -> None: super().__init__() self._notes = notes def compose(self): with VerticalScroll(id="footnote-body"): for note in self._notes: yield Static(note, classes="footnote-note") yield Label("Escape/f zum Schließen", id="footnote-hint") def action_dismiss_self(self) -> None: self.dismiss(None) class ReaderScreen(Screen): BINDINGS = [ Binding("j,down", "scroll_down_line", "Runter", show=False), Binding("k,up", "scroll_up_line", "Hoch", show=False), Binding("n", "next_chapter", "Nächstes Kapitel"), Binding("p", "prev_chapter", "Vorheriges Kapitel"), Binding("f", "peek_footnote", "Fußnote"), Binding("q,escape", "back", "Zurück zur Bibliothek"), ] def __init__(self, path: Path) -> None: super().__init__() self.path = path self.book: blocks.FlatBook | None = None self.book_layout: layout.BookLayout = _EMPTY_LAYOUT self._server_id = _server_book_id(path) self._remote_cfg = config_mod.load() if self._server_id is not None else None self._last_pushed_anchor = "" def compose(self): yield Header() yield LoadingIndicator(id="reader-loading") with Container(id="reader-scroll"): # Stays visible (empty) from the start rather than toggling display # on once loaded — a widget that's just been switched from hidden # to visible hasn't been through a layout pass yet, so its `.size` # is still (0, 0) and an immediate scroll_to() right after has # nothing to clamp against and silently resets to 0. self._book_view = ContinuousBookView(_EMPTY_LAYOUT) yield self._book_view yield Footer() def on_mount(self) -> None: self._load_book(self._content_width()) @work(thread=True) def _load_book(self, width: int) -> None: cached = cache.load(self.path, width) if cached is not None: book, book_layout = cached else: book = blocks.load_flat_book(self.path) console = Console(width=width) book_layout = layout.build_layout([b.text for b in book.blocks], width, console) cache.save(self.path, width, book, book_layout) self.app.call_from_thread(self._on_book_loaded, book, book_layout, width) def _on_book_loaded(self, book: blocks.FlatBook, book_layout: layout.BookLayout, width: int) -> None: self.book = book self.sub_title = book.title self._apply_layout(book_layout, width) self.query_one("#reader-loading", LoadingIndicator).display = False self.call_after_refresh(self._restore_position) self.set_interval(AUTOSAVE_INTERVAL, self._autosave) def _content_width(self) -> int: return max(20, min(MAX_LINE_WIDTH, self.size.width - 4)) def _apply_layout(self, book_layout: layout.BookLayout, width: int) -> None: self.book_layout = book_layout self._book_view.styles.width = width self._book_view.book_layout = book_layout self._book_view.virtual_size = Size(width, book_layout.total_rows) self._book_view.refresh() def _build_layout(self) -> None: assert self.book is not None width = self._content_width() console = Console(width=width) block_texts = [b.text for b in self.book.blocks] new_layout = layout.build_layout(block_texts, width, console) cache.save(self.path, width, self.book, new_layout) self._apply_layout(new_layout, width) def on_resize(self) -> None: if self.book is None: return old_y = self._book_view.scroll_y anchor_block, anchor_frac = layout.get_position_anchor(self.book_layout, old_y) self._build_layout() new_y = layout.scroll_y_for_anchor(self.book_layout, anchor_block, anchor_frac) self._book_view.scroll_to(y=new_y, animate=False, immediate=True) def _restore_position(self, attempt: int = 0) -> None: if self.book is None: return saved = progress.load(self.book.id) if saved is None: return parsed = blocks.parse_anchor(saved.anchor) if parsed is None: return block_index, inner_fraction = parsed y = layout.scroll_y_for_anchor(self.book_layout, block_index, inner_fraction) if y <= 0: return self._book_view.scroll_to(y=y, animate=False, immediate=True) # A widget that's only just become part of the layout doesn't always # honor an immediate scroll on the first attempt (its own size/scroll # bounds can still be mid-update) — verify it actually landed and # retry a bounded number of times rather than guessing a fixed delay. if attempt < 20 and abs(self._book_view.scroll_y - y) > 1: self.call_after_refresh(lambda: self._restore_position(attempt + 1)) def _current_anchor_str(self) -> str: block_index, inner_fraction = layout.get_position_anchor(self.book_layout, self._book_view.scroll_y) return blocks.format_anchor(block_index, inner_fraction) def _autosave(self) -> None: self._save_progress() def _save_progress(self) -> None: if self.book is None or not self.book_layout.heights: return anchor = self._current_anchor_str() advanced = progress.save(self.book.id, progress.Position(anchor=anchor)) if advanced and self._remote_cfg is not None and self._server_id is not None: self._push_remote_progress(anchor) @work(thread=True, exclusive=True, group="progress-push") def _push_remote_progress(self, anchor: str) -> None: if anchor == self._last_pushed_anchor or self._remote_cfg is None or self._server_id is None: return try: api.post_progress( self._remote_cfg.server_url, self._remote_cfg.api_token, self._server_id, scroll_fraction=0.0, position_anchor=anchor, force=False, ) self._last_pushed_anchor = anchor except Exception: pass # best-effort — local progress is already saved regardless def on_unmount(self) -> None: self._save_progress() def action_scroll_down_line(self) -> None: self._book_view.scroll_relative(y=1, animate=False) def action_scroll_up_line(self) -> None: self._book_view.scroll_relative(y=-1, animate=False) def _current_chapter_index(self) -> int: if self.book is None: return 0 block_index, _ = layout.get_position_anchor(self.book_layout, self._book_view.scroll_y) return bisect.bisect_right(self.book.chapter_start_block, block_index) - 1 def _jump_to_block(self, block_index: int) -> None: y = layout.scroll_y_for_anchor(self.book_layout, block_index, 0.0) self._book_view.scroll_to(y=y, animate=False, immediate=True) def action_next_chapter(self) -> None: if self.book is None: return chapter = self._current_chapter_index() if chapter + 1 < len(self.book.chapter_start_block): self._jump_to_block(self.book.chapter_start_block[chapter + 1]) self._save_progress() def action_prev_chapter(self) -> None: if self.book is None: return chapter = self._current_chapter_index() if chapter > 0: self._jump_to_block(self.book.chapter_start_block[chapter - 1]) self._save_progress() def action_peek_footnote(self) -> None: if self.book is None or not self.book_layout.heights: return current_block, _ = layout.get_position_anchor(self.book_layout, self._book_view.scroll_y) forward = [(i, b) for i, b in enumerate(self.book.blocks) if i >= current_block and b.footnotes] backward = [(i, b) for i, b in enumerate(self.book.blocks) if i < current_block and b.footnotes] candidate = forward[0] if forward else (backward[-1] if backward else None) if candidate is None: self.notify("Keine Fußnote in diesem Buch gefunden.", timeout=3) return _, block = candidate notes: list[str] = [] for ref in block.footnotes: target_idx = self.book.footnote_targets.get(ref.target_id) if target_idx is None: continue notes.append(self.book.blocks[target_idx].text) if not notes: self.notify("Fußnote konnte nicht aufgelöst werden.", timeout=3) return self.app.push_screen(FootnoteScreen(notes)) def action_back(self) -> None: self.app.pop_screen()