Library-Screen synct jetzt selbstständig statt den expliziten `sync`-Befehl vorauszusetzen: einmal im Hintergrund direkt nach dem Start (neue Bücher/ Fortschritt erscheinen, sobald fertig), einmal abgewartet beim Beenden über q (DioraTuiApp.action_quit). Beides best-effort — ohne gespeicherte Zugangsdaten oder bei Netzwerkfehlern bleibt die lokale Bibliothek unangetastet nutzbar. Bibliotheksliste sortiert jetzt nach zuletzt geöffnetem Buch (progress.json updated_at, neueste zuerst) und blendet gelesene Bücher standardmäßig aus (EBook.is_read aus dem Sync-Snapshot, lokal in library.json gespiegelt) — r-Taste zeigt sie für die Sitzung wieder an. Getestet: vollständig isoliert (Pfad-Konstanten gemonkeypatcht statt echter ~/.config-/~/.local/share-Dateien) gegen einen echten Dev-Server mit zwei Büchern unterschiedlichen Gelesen-Status und Fortschritts-Zeitstempeln — Auto-Sync, Filter-Default, Toggle und Sortierreihenfolge korrekt bestätigt; Sync-vor-dem-Beenden separat verifiziert (kein echter Hänger, nur ein Test-Timing-Artefakt ohne pilot.pause()).
229 lines
8.6 KiB
Python
229 lines
8.6 KiB
Python
"""diora-tui: terminal EPUB reader for diora, syncing books + reading progress
|
|
against a diora server (see project README)."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import asyncio
|
|
import getpass
|
|
from pathlib import Path
|
|
|
|
from textual import work
|
|
from textual.app import App, ComposeResult
|
|
from textual.binding import Binding
|
|
from textual.screen import Screen
|
|
from textual.widgets import Footer, Header, Label, ListItem, ListView
|
|
|
|
from . import config as config_mod
|
|
from . import crypto, epub, library_meta, progress, remote
|
|
from .reader_screen import ReaderScreen
|
|
|
|
DEFAULT_LIBRARY = Path.home() / "Books"
|
|
|
|
|
|
class LibraryScreen(Screen):
|
|
BINDINGS = [
|
|
Binding("enter", "open_selected", "Öffnen"),
|
|
Binding("r", "toggle_read_filter", "Gelesene ein-/ausblenden"),
|
|
Binding("q", "quit", "Beenden"),
|
|
]
|
|
|
|
def __init__(self, library_dir: Path) -> None:
|
|
super().__init__()
|
|
self.library_dir = library_dir
|
|
self.paths: list[Path] = []
|
|
self.visible_paths: list[Path] = []
|
|
self.show_read = False
|
|
|
|
def compose(self) -> ComposeResult:
|
|
yield Header()
|
|
yield ListView(id="library-list")
|
|
yield Footer()
|
|
|
|
def on_mount(self) -> None:
|
|
self.sub_title = str(self.library_dir)
|
|
self._refresh_list()
|
|
cfg = config_mod.load()
|
|
if cfg is not None:
|
|
self._auto_sync(cfg)
|
|
|
|
@work(thread=True)
|
|
def _auto_sync(self, cfg: config_mod.RemoteConfig) -> None:
|
|
try:
|
|
result = remote.sync_library(self.library_dir, cfg)
|
|
except remote.SyncError:
|
|
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:
|
|
if result.downloaded or result.progress_pulled:
|
|
self._refresh_list()
|
|
if result.downloaded:
|
|
self.notify(f"{len(result.downloaded)} neue(s) Buch/Bücher synchronisiert.", timeout=3)
|
|
|
|
def _refresh_list(self) -> None:
|
|
self.paths = epub.scan_library(self.library_dir)
|
|
|
|
entries = []
|
|
for path in self.paths:
|
|
book_id = epub._book_id(path)
|
|
read = library_meta.is_read(book_id)
|
|
if read and not self.show_read:
|
|
continue
|
|
saved = progress.load(book_id)
|
|
last_opened = saved.updated_at if saved else 0.0
|
|
entries.append((last_opened, path, read))
|
|
entries.sort(key=lambda e: e[0], reverse=True)
|
|
self.visible_paths = [path for _, path, _ in entries]
|
|
|
|
list_view = self.query_one("#library-list", ListView)
|
|
list_view.clear()
|
|
if not self.paths:
|
|
list_view.append(ListItem(Label(f"Keine EPUBs gefunden in {self.library_dir}")))
|
|
return
|
|
if not entries:
|
|
list_view.append(ListItem(Label("Alle Bücher als gelesen markiert — 'r' zum Anzeigen")))
|
|
return
|
|
for _, path, read in entries:
|
|
label = f"✓ {path.stem}" if read else path.stem
|
|
list_view.append(ListItem(Label(label)))
|
|
list_view.index = 0
|
|
list_view.focus()
|
|
|
|
def action_toggle_read_filter(self) -> None:
|
|
self.show_read = not self.show_read
|
|
self._refresh_list()
|
|
state = "eingeblendet" if self.show_read else "ausgeblendet"
|
|
self.notify(f"Gelesene Bücher {state}.", timeout=2)
|
|
|
|
def action_open_selected(self) -> None:
|
|
list_view = self.query_one("#library-list", ListView)
|
|
if not self.visible_paths or list_view.index is None:
|
|
return
|
|
self.app.push_screen(ReaderScreen(self.visible_paths[list_view.index]))
|
|
|
|
def on_list_view_selected(self, event: ListView.Selected) -> None:
|
|
self.action_open_selected()
|
|
|
|
|
|
class DioraTuiApp(App):
|
|
CSS_PATH = "app.tcss"
|
|
TITLE = "diora-tui"
|
|
|
|
def __init__(self, library_dir: Path) -> None:
|
|
super().__init__()
|
|
self.library_dir = library_dir
|
|
|
|
def on_mount(self) -> None:
|
|
self.push_screen(LibraryScreen(self.library_dir))
|
|
|
|
async def action_quit(self) -> None:
|
|
cfg = config_mod.load()
|
|
if cfg is not None:
|
|
try:
|
|
await asyncio.to_thread(remote.sync_library, self.library_dir, cfg)
|
|
except remote.SyncError:
|
|
pass # best-effort — don't block quitting on a sync failure
|
|
self.exit()
|
|
|
|
|
|
def _prompt(label: str, *, secret: bool = False) -> str:
|
|
value = (getpass.getpass(f"{label}: ") if secret else input(f"{label}: ")).strip()
|
|
if not value:
|
|
raise SystemExit(f"Abgebrochen: {label} darf nicht leer sein.")
|
|
return value
|
|
|
|
|
|
def _ask_yes_no(question: str) -> bool:
|
|
return input(f"{question} [y/N]: ").strip().lower() in ("y", "yes", "j", "ja")
|
|
|
|
|
|
def _prompt_enc_key() -> str:
|
|
print()
|
|
print("Verschlüsselungs-Key — zwei Wege:")
|
|
print(" [1] Aus Benutzername + Passwort ableiten (wie diora's 'Unlock'-Formular im Browser)")
|
|
print(" [2] Base64-Key direkt einfügen (z.B. per Browser-Konsole exportiert)")
|
|
choice = input("Wahl [1/2, Standard 1]: ").strip() or "1"
|
|
if choice == "2":
|
|
return _prompt("Verschlüsselungs-Key (Base64)", secret=True)
|
|
|
|
print(
|
|
"Hinweis: das liefert nur dann den richtigen Key, wenn dieser Account je über "
|
|
"diora's Passwort-Ableitung ('Unlock with password' im Browser) entsperrt wurde. "
|
|
"War der Key dort nur automatisch zufällig erzeugt, kommt hier ein anderer "
|
|
"(falscher) Key raus — 'sync' meldet das dann als Entschlüsselungsfehler, ohne "
|
|
"etwas kaputtzumachen; in dem Fall stattdessen [2] mit dem exportierten Key nutzen."
|
|
)
|
|
username = _prompt("diora-Benutzername")
|
|
password = _prompt("diora-Passwort", secret=True)
|
|
return crypto.derive_key_b64(username, password)
|
|
|
|
|
|
def _resolve_remote_config(server_override: str | None, *, save: bool) -> config_mod.RemoteConfig:
|
|
cfg = None if server_override else config_mod.load()
|
|
if cfg is not None:
|
|
return cfg
|
|
|
|
print("Keine gespeicherten Zugangsdaten gefunden — bitte einmalig eingeben.")
|
|
server_url = (server_override or _prompt("Server-URL (z.B. https://diora.creamfresh.xyz)")).rstrip("/")
|
|
token = _prompt("API-Token (diora → Einstellungen → /accounts/settings/)", secret=True)
|
|
enc_key = _prompt_enc_key()
|
|
cfg = config_mod.RemoteConfig(server_url=server_url, api_token=token, enc_key_b64=enc_key)
|
|
if save or _ask_yes_no("Zugangsdaten lokal speichern, damit du sie nicht erneut eingeben musst?"):
|
|
config_mod.save(cfg)
|
|
print("Gespeichert.")
|
|
return cfg
|
|
|
|
|
|
def run_sync(library_dir: Path, server_override: str | None, *, save: bool) -> None:
|
|
cfg = _resolve_remote_config(server_override, save=save)
|
|
print(f"Verbinde zu {cfg.server_url} …")
|
|
try:
|
|
result = remote.sync_library(library_dir, cfg)
|
|
except remote.SyncError as e:
|
|
raise SystemExit(f"Fehler: {e}")
|
|
|
|
print(f"{len(result.downloaded)} Buch/Bücher neu heruntergeladen nach {library_dir}.")
|
|
for name in result.downloaded:
|
|
print(f" + {name}")
|
|
if result.unchanged:
|
|
print(f"{result.unchanged} Buch/Bücher bereits lokal vorhanden, übersprungen.")
|
|
if result.skipped_pdf:
|
|
titles = ", ".join(result.skipped_pdf)
|
|
print(f"{len(result.skipped_pdf)} PDF(s) übersprungen (TUI liest aktuell nur EPUB): {titles}")
|
|
if result.failed:
|
|
titles = ", ".join(result.failed)
|
|
print(f"{len(result.failed)} Buch/Bücher konnten nicht entschlüsselt werden (falscher Key?): {titles}")
|
|
if result.progress_pulled:
|
|
print(f"Lesefortschritt für {result.progress_pulled} Buch/Bücher vom Server übernommen.")
|
|
|
|
|
|
def main() -> None:
|
|
parser = argparse.ArgumentParser(
|
|
description="diora-tui — lokaler EPUB-Reader (Vorstufe des diora-Sync-Clients)"
|
|
)
|
|
parser.add_argument(
|
|
"--library",
|
|
type=Path,
|
|
default=DEFAULT_LIBRARY,
|
|
help=f"Verzeichnis mit EPUB-Dateien (Standard: {DEFAULT_LIBRARY})",
|
|
)
|
|
subparsers = parser.add_subparsers(dest="command")
|
|
sync_parser = subparsers.add_parser("sync", help="EPUBs vom diora-Server holen und entschlüsseln")
|
|
sync_parser.add_argument("--server", help="Server-URL, überschreibt gespeicherte Zugangsdaten für diesen Lauf")
|
|
sync_parser.add_argument(
|
|
"--save", action="store_true", help="Eingegebene Zugangsdaten lokal speichern, ohne zu fragen"
|
|
)
|
|
|
|
args = parser.parse_args()
|
|
library_dir = args.library.expanduser()
|
|
|
|
if args.command == "sync":
|
|
run_sync(library_dir, args.server, save=args.save)
|
|
return
|
|
|
|
DioraTuiApp(library_dir).run()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|