mrwnslz
695037fb9c
library.py hält die Regeln der Web-Bibliothek (53-library.js) als reine Funktionen: Status Angefangen/Ungelesen/Gelesen, Ordner (mehrere je Buch, Reihenfolge aus den verschlüsselten Bibliothekseinstellungen, "Ohne Ordner"), Suche ohne Akzente, sechs Sortierungen mit natürlicher Zahlenordnung, Weiterlesen-Karte aus den zuletzt gelesenen, Bücher/Zeitschriften. library_screen.py zeigt sie an (Tasten 1-4, o, /, s, m, c; Chips und Karte klickbar). Metadaten (Ordner, Seitenzahl, Zeitschrift) und die Ordnerreihenfolge kommen beim Abgleich; fehlt die Länge, rechnet die TUI sie für geladene Bücher aus dem Text (1800 Zeichen je Seite). crypto.encrypt für Stufe 3. 54 Tests, Smoke prüft Ordner/Länge/Prefs gegen den lokalen Server. Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
388 lines
16 KiB
Python
388 lines
16 KiB
Python
"""The library screen: the web library (docs/library.md) in a terminal — "Weiterlesen" card,
|
||
status and folder chips, search, sort, Bücher/Zeitschriften. The rules are in library.py.
|
||
|
||
Keys: 1–4 status (Alle/Angefangen/Ungelesen/Gelesen), o folder, / search, s sort, m Bücher or
|
||
Zeitschriften, c continue the book read last, Enter open. The chips and the card also answer
|
||
the mouse.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
from pathlib import Path
|
||
|
||
from rich.text import Text
|
||
from textual import work
|
||
from textual.binding import Binding
|
||
from textual.screen import ModalScreen, Screen
|
||
from textual.markup import escape
|
||
from textual.widgets import Footer, Header, Input, Label, ListItem, ListView, OptionList, Static
|
||
|
||
from . import blocks, config as config_mod, epub, library, library_meta, progress, remote
|
||
from .reader_screen import ReaderScreen
|
||
from .statusbar import StatusBar
|
||
|
||
PAGES_NOTE_EVERY = 4 # seconds between redraws while lengths are worked out
|
||
|
||
|
||
class ChoiceScreen(ModalScreen[int]):
|
||
"""Pick one of a few options; dismisses with its index, or None on Escape."""
|
||
|
||
BINDINGS = [Binding("escape,q", "cancel", "Abbrechen")]
|
||
|
||
def __init__(self, title: str, options: list[str], current: int = 0) -> None:
|
||
super().__init__()
|
||
self._title = title
|
||
self._options = options
|
||
self._current = current
|
||
|
||
def compose(self):
|
||
yield Label(self._title, id="choice-title")
|
||
yield OptionList(*[Text(o) for o in self._options], id="choice-list")
|
||
|
||
def on_mount(self) -> None:
|
||
ol = self.query_one(OptionList)
|
||
ol.highlighted = max(0, min(self._current, len(self._options) - 1))
|
||
ol.focus()
|
||
|
||
def on_option_list_option_selected(self, event: OptionList.OptionSelected) -> None:
|
||
self.dismiss(event.option_index)
|
||
|
||
def action_cancel(self) -> None:
|
||
self.dismiss(None)
|
||
|
||
|
||
class LibraryScreen(Screen):
|
||
BINDINGS = [
|
||
Binding("enter", "open_selected", "Öffnen"),
|
||
Binding("c", "continue_reading", "Weiterlesen"),
|
||
Binding("/", "search", "Suchen"),
|
||
Binding("s", "pick_sort", "Sortierung"),
|
||
Binding("o", "pick_folder", "Ordner"),
|
||
Binding("m", "toggle_section", "Bücher/Zeitschriften"),
|
||
Binding("1", "set_status('all')", "Alle", show=False),
|
||
Binding("2", "set_status('started')", "Angefangen", show=False),
|
||
Binding("3", "set_status('unread')", "Ungelesen", show=False),
|
||
Binding("4", "set_status('read')", "Gelesen", show=False),
|
||
Binding("escape", "reset", "Zurücksetzen", show=False),
|
||
Binding("q", "quit", "Beenden"),
|
||
]
|
||
|
||
def __init__(self, library_dir: Path) -> None:
|
||
super().__init__()
|
||
self.library_dir = library_dir
|
||
self.rows: list[library.Row] = [] # every book
|
||
self.section_rows: list[library.Row] = [] # of the section shown (Bücher or Zeitschriften)
|
||
self.shown: list[library.Row] = [] # the list on screen
|
||
self.going: library.Row | None = None # the card's book
|
||
self.folder_names: list[str] = []
|
||
self.q = ""
|
||
self.status = "all"
|
||
self.folder: str | None = None # None: all, '': without folder, else a name
|
||
self.sort = library_meta.get_extra("sort", "recent")
|
||
if self.sort not in dict(library.SORTS):
|
||
self.sort = "recent"
|
||
self.magazines = False
|
||
self._lengths_running = False
|
||
self._lengths_tried: set[str] = set()
|
||
self._lengths_note = ""
|
||
|
||
# ---- layout
|
||
|
||
def compose(self):
|
||
yield Header()
|
||
yield Static("", id="lib-chips")
|
||
yield Input(placeholder="Titel, Autor …", id="lib-search")
|
||
yield Static("", id="lib-card")
|
||
yield Static("", id="lib-note")
|
||
yield ListView(id="library-list")
|
||
yield StatusBar()
|
||
yield Footer()
|
||
|
||
def on_mount(self) -> None:
|
||
self.query_one("#lib-search").display = False
|
||
self.sub_title = str(self.library_dir)
|
||
self._refresh_list()
|
||
cfg = config_mod.load()
|
||
if cfg is not None:
|
||
self._auto_sync(cfg)
|
||
|
||
def on_screen_resume(self) -> None:
|
||
self._refresh_list() # a book was read: the order and the percentages changed
|
||
|
||
# ---- sync
|
||
|
||
@work(thread=True)
|
||
def _auto_sync(self, cfg: config_mod.RemoteConfig) -> None:
|
||
try:
|
||
result = remote.sync_library(self.library_dir, cfg)
|
||
except remote.SyncError as e:
|
||
self.app.call_from_thread(self.notify, str(e), severity="warning", timeout=5)
|
||
return # best-effort — e.g. offline; the local library still works
|
||
self.app.call_from_thread(self._on_auto_sync_done, result)
|
||
|
||
def _on_auto_sync_done(self, result: remote.SyncResult) -> None:
|
||
self._refresh_list()
|
||
if result.new_titles:
|
||
self.notify(f"{len(result.new_titles)} neue(s) Buch/Bücher auf dem Server.", timeout=3)
|
||
|
||
# ---- the list
|
||
|
||
def _gather(self) -> list[library.Row]:
|
||
store = progress.default()
|
||
remote_books = library_meta.remote_entries()
|
||
files = epub.scan_library(self.library_dir)
|
||
keys = {epub._book_id(p) for p in files} | set(remote_books)
|
||
states = {k: st for k in keys if (st := store.get(k)) is not None}
|
||
return library.build_rows(files, remote_books, states, library_meta.local_pages(), epub._book_id)
|
||
|
||
def _refresh_list(self) -> None:
|
||
self.rows = self._gather()
|
||
self.section_rows = library.section(self.rows, self.magazines)
|
||
order = library_meta.get_extra("folder_order", [])
|
||
self.folder_names = [n for n, _ in library.folder_list(self.section_rows, order)]
|
||
if self.folder not in (None, "") and self.folder not in self.folder_names:
|
||
self.folder = None
|
||
self._draw()
|
||
self._fill_lengths()
|
||
|
||
def _draw(self) -> None:
|
||
self._draw_chips()
|
||
list_view = self.query_one("#library-list", ListView)
|
||
selected = self.shown[list_view.index].key if self.shown and list_view.index is not None and list_view.index < len(self.shown) else None
|
||
|
||
shown = library.shown(self.section_rows, q=self.q, status=self.status, folder=self.folder, sort=self.sort)
|
||
# The card: only on the plain list, and its book is not repeated below (once, on top).
|
||
plain = self.status == "all" and self.folder is None and not self.q
|
||
self.going, recent = library.continue_rows(self.section_rows) if plain else (None, [])
|
||
self._draw_card(recent)
|
||
if self.going is not None and self.going in shown:
|
||
shown.remove(self.going)
|
||
self.shown = shown
|
||
|
||
list_view.clear()
|
||
if not self.rows:
|
||
list_view.append(ListItem(Label(f"Keine EPUBs gefunden in {self.library_dir}")))
|
||
elif not self.section_rows:
|
||
list_view.append(ListItem(Label("Noch keine Zeitschriften — m wechselt zu den Büchern." if self.magazines
|
||
else "Alle Bücher stehen bei den Zeitschriften — m wechselt dorthin.")))
|
||
elif not self.shown and self.going is None:
|
||
list_view.append(ListItem(Label("Nichts gefunden.")))
|
||
by_length = self.sort in ("short", "long")
|
||
for r in self.shown:
|
||
list_view.append(ListItem(Label(Text(self._label(r, by_length)))))
|
||
if self.shown:
|
||
idx = next((i for i, r in enumerate(self.shown) if r.key == selected), 0)
|
||
list_view.index = idx
|
||
if not self.query_one("#lib-search", Input).display: # not while a search is open
|
||
list_view.focus()
|
||
self._draw_note()
|
||
|
||
@staticmethod
|
||
def _label(r: library.Row, by_length: bool) -> str:
|
||
mark = "✓ " if r.is_read else ""
|
||
cloud = "☁ " if r.path is None else ""
|
||
author = f" — {r.author}" if r.author else ""
|
||
pct = f" {library.percent(r)} %" if (r.fraction > 0 and not r.is_read) else ""
|
||
length = f" · {r.pages} S." if by_length and r.pages else ""
|
||
return f"{mark}{cloud}{r.title}{author}{pct}{length}"
|
||
|
||
def _draw_chips(self) -> None:
|
||
counts = library.status_counts(self.section_rows)
|
||
parts = []
|
||
for i, (sid, label) in enumerate(library.STATUSES):
|
||
text = label if sid == "all" else f"{label} {counts[sid]}"
|
||
parts.append(self._chip(f"{i + 1} {text}", f"screen.set_status('{sid}')", self.status == sid))
|
||
folders = library.folder_list(self.section_rows, library_meta.get_extra("folder_order", []))
|
||
if folders:
|
||
parts.append("[dim]│[/]")
|
||
parts.append(self._chip("Alle Ordner", "screen.set_folder('all')", self.folder is None))
|
||
for i, (name, count) in enumerate(folders):
|
||
parts.append(self._chip(f"{name} {count}", f"screen.set_folder('{i}')", self.folder == name))
|
||
none = sum(1 for r in self.section_rows if not r.folders)
|
||
if none:
|
||
parts.append(self._chip(f"Ohne Ordner {none}", "screen.set_folder('none')", self.folder == ""))
|
||
section = "Zeitschriften" if self.magazines else "Bücher"
|
||
sort = dict(library.SORTS)[self.sort]
|
||
head = f"[b]{section}[/] [dim]Sortierung:[/] {escape(sort)} "
|
||
self.query_one("#lib-chips", Static).update(head + " ".join(parts))
|
||
|
||
@staticmethod
|
||
def _chip(label: str, action: str, on: bool) -> str:
|
||
label = escape(label)
|
||
return f"[@click={action}]" + (f"[b reverse] {label} [/]" if on else f" [underline]{label}[/] ") + "[/]"
|
||
|
||
def _draw_card(self, recent: list[library.Row]) -> None:
|
||
card = self.query_one("#lib-card", Static)
|
||
b = self.going
|
||
if b is None:
|
||
card.display = False
|
||
return
|
||
card.display = True
|
||
ago = library.ago(b.read_at)
|
||
pct = library.percent(b)
|
||
bar = "█" * (pct // 5) + "░" * (20 - pct // 5)
|
||
pages = f" · Seite {max(1, int(b.pages * b.fraction + 0.5))} von {b.pages}" if b.pages else ""
|
||
where = f" · zuletzt auf {escape(b.device)}" if b.device and b.device != progress.DEVICE_NAME else ""
|
||
lines = [
|
||
f"[@click=screen.continue_reading][dim]Weiterlesen{' · ' + ago if ago else ''}{where}[/]",
|
||
f"[b]{escape(b.title)}[/]",
|
||
escape(b.author) if b.author else "",
|
||
f"{bar} {pct} %{pages} [dim](c)[/][/]",
|
||
]
|
||
if recent:
|
||
lines.append("[dim]Zuletzt gelesen:[/] " + " · ".join(
|
||
f"{escape(r.title)}{' ✓' if r.is_read else ''}" for r in recent))
|
||
card.update("\n".join(l for l in lines if l))
|
||
|
||
def _draw_note(self) -> None:
|
||
note = self.query_one("#lib-note", Static)
|
||
text = self._lengths_note
|
||
if not text and self.sort in ("short", "long"):
|
||
unknown = sum(1 for r in self.section_rows if not r.pages)
|
||
if unknown:
|
||
text = f"{unknown} Bücher ohne bekannte Länge (noch nicht geladen) stehen am Ende."
|
||
note.update(escape(text))
|
||
note.display = bool(text)
|
||
|
||
# ---- lengths (the web works them out once a length sort is chosen: the file is needed)
|
||
|
||
def _fill_lengths(self) -> None:
|
||
if self._lengths_running or self.sort not in ("short", "long"):
|
||
return
|
||
todo = [r for r in self.rows if not r.pages and r.path is not None and r.key not in self._lengths_tried]
|
||
if not todo:
|
||
return
|
||
self._lengths_running = True
|
||
self._work_lengths([(r.key, r.path) for r in todo])
|
||
|
||
@work(thread=True, exclusive=True, group="lengths")
|
||
def _work_lengths(self, todo: list[tuple[str, Path]]) -> None:
|
||
import time
|
||
|
||
shown_at = time.monotonic()
|
||
try:
|
||
for done, (key, path) in enumerate(todo, 1):
|
||
self._lengths_tried.add(key)
|
||
try:
|
||
pages = blocks.book_pages(path)
|
||
except Exception:
|
||
pages = 0 # stays at the end
|
||
if pages:
|
||
library_meta.set_local_pages(key, pages)
|
||
self._lengths_note = f"Länge wird ermittelt … {done} von {len(todo)} Büchern"
|
||
if time.monotonic() - shown_at > PAGES_NOTE_EVERY:
|
||
shown_at = time.monotonic()
|
||
self.app.call_from_thread(self._refresh_list_keep_note)
|
||
finally:
|
||
self._lengths_note = ""
|
||
self._lengths_running = False
|
||
self.app.call_from_thread(self._refresh_list)
|
||
|
||
def _refresh_list_keep_note(self) -> None:
|
||
self.rows = self._gather()
|
||
self.section_rows = library.section(self.rows, self.magazines)
|
||
self._draw()
|
||
|
||
# ---- actions
|
||
|
||
def action_set_status(self, status: str) -> None:
|
||
self.status = "all" if self.status == status else status
|
||
self._draw()
|
||
|
||
def action_set_folder(self, which: str) -> None:
|
||
"""`which`: 'all', 'none' (without a folder) or the index of a folder chip."""
|
||
if which == "all":
|
||
self.folder = None
|
||
elif which == "none":
|
||
self.folder = ""
|
||
elif which.isdigit() and int(which) < len(self.folder_names):
|
||
name = self.folder_names[int(which)]
|
||
self.folder = None if self.folder == name else name
|
||
self._draw()
|
||
|
||
def action_pick_folder(self) -> None:
|
||
values: list[str | None] = [None] + self.folder_names + ([""] if any(not r.folders for r in self.section_rows) else [])
|
||
if len(values) == 1:
|
||
self.notify("Keine Ordner angelegt.", timeout=2)
|
||
return
|
||
names = ["Alle Ordner" if v is None else "Ohne Ordner" if v == "" else v for v in values]
|
||
|
||
def picked(i: int | None) -> None:
|
||
if i is not None:
|
||
self.folder = values[i]
|
||
self._draw()
|
||
|
||
self.app.push_screen(ChoiceScreen("Ordner", names, values.index(self.folder)), picked)
|
||
|
||
def action_pick_sort(self) -> None:
|
||
ids = [s for s, _ in library.SORTS]
|
||
|
||
def picked(i: int | None) -> None:
|
||
if i is None:
|
||
return
|
||
self.sort = ids[i]
|
||
library_meta.set_extra("sort", self.sort)
|
||
self._draw()
|
||
self._fill_lengths()
|
||
|
||
self.app.push_screen(ChoiceScreen("Sortieren nach", [n for _, n in library.SORTS], ids.index(self.sort)), picked)
|
||
|
||
def action_toggle_section(self) -> None:
|
||
self.magazines = not self.magazines
|
||
self.folder = None
|
||
self._refresh_list()
|
||
|
||
def action_search(self) -> None:
|
||
box = self.query_one("#lib-search", Input)
|
||
box.display = True
|
||
box.focus()
|
||
|
||
def on_input_changed(self, event: Input.Changed) -> None:
|
||
self.q = event.value
|
||
self._draw()
|
||
|
||
def on_input_submitted(self, event: Input.Submitted) -> None:
|
||
self.query_one("#library-list", ListView).focus()
|
||
|
||
def action_reset(self) -> None:
|
||
box = self.query_one("#lib-search", Input)
|
||
if box.display or self.q or self.status != "all" or self.folder is not None:
|
||
box.value = ""
|
||
box.display = False
|
||
self.q, self.status, self.folder = "", "all", None
|
||
self._draw()
|
||
|
||
def action_continue_reading(self) -> None:
|
||
if self.going is not None:
|
||
self._open(self.going)
|
||
else:
|
||
self.notify("Nichts zum Weiterlesen — ein Buch öffnen.", timeout=3)
|
||
|
||
def action_open_selected(self) -> None:
|
||
list_view = self.query_one("#library-list", ListView)
|
||
if not self.shown or list_view.index is None or list_view.index >= len(self.shown):
|
||
return
|
||
self._open(self.shown[list_view.index])
|
||
|
||
def _open(self, row: library.Row) -> None:
|
||
if row.path is not None:
|
||
self.app.push_screen(ReaderScreen(row.path))
|
||
return
|
||
cfg = config_mod.load()
|
||
if cfg is None or row.server_id is None:
|
||
self.notify("Nicht angemeldet — `diora-tui login`.", severity="error")
|
||
return
|
||
self.notify(f"Lade „{row.title}“ …", timeout=4)
|
||
self._download_and_open(cfg, row)
|
||
|
||
@work(thread=True, exclusive=True, group="download")
|
||
def _download_and_open(self, cfg: config_mod.RemoteConfig, row: library.Row) -> None:
|
||
try:
|
||
path = remote.download_book(self.library_dir, cfg, row.server_id, row.title)
|
||
except remote.SyncError as e:
|
||
self.app.call_from_thread(self.notify, str(e), severity="error", timeout=6)
|
||
return
|
||
self.app.call_from_thread(self.app.push_screen, ReaderScreen(path))
|
||
|
||
def on_list_view_selected(self, event: ListView.Selected) -> None:
|
||
self.action_open_selected()
|