mrwnslz
e246a32212
Dieselben verschlüsselten Listen wie der Web-Reader (annotations.py). Die Anker des Webs (Block-Index plus Zeichenposition im Rohtext) werden auf den angezeigten Text umgerechnet (marks.py, blocks.collapse_map), alte XPath-Anker über das Zitat gefunden. Im Reader: v markiert (Cursor je Wort, Anfang/Ende, Farbe oder Notiz), n legt eine Notiz an, b ein Lesezeichen, a öffnet die Liste (springen, Notiz ändern, löschen). Der Server ersetzt die Liste als Ganzes; die TUI legt ihre Änderungen vor dem Schreiben dreiwege je Eintrag-ID auf die aktuelle Liste des Servers, offline bleiben sie vorgemerkt. Geprüft: die echten resolveAnchorRange des Webs lösen 81 TUI-Anker exakt auf (tools/parity/web_anchors.cjs), 34 von 34 echten Web-Markierungen werden angezeigt, der Server-Smoke schreibt und mischt zwei Geräte. 75 Tests. Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
84 lines
3.6 KiB
Python
84 lines
3.6 KiB
Python
"""HTTP client for the diora server, authenticated by the account's API token
|
||
(Authorization: Bearer; the server sets request.user for every endpoint).
|
||
|
||
Only what the reader needs: the book list with the reading state (`GET /books/` – the same
|
||
rows the apps pull, without the notes' blobs that `/api/sync/` carries), a book's encrypted
|
||
file, and a position write (`POST /books/<id>/progress/`, docs/progress-sync.md).
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import time
|
||
|
||
import requests
|
||
|
||
USER_AGENT = "diora-tui"
|
||
|
||
|
||
class ApiError(Exception):
|
||
pass
|
||
|
||
|
||
def _headers(token: str) -> dict[str, str]:
|
||
return {"Authorization": f"Bearer {token}", "User-Agent": USER_AGENT}
|
||
|
||
|
||
def _check(resp: requests.Response, what: str) -> None:
|
||
if resp.status_code == 401:
|
||
raise ApiError("Anmeldung abgelehnt — Token ungültig oder neu erzeugt? `diora-tui login` ausführen.")
|
||
if resp.status_code == 404:
|
||
raise ApiError(f"{what} nicht gefunden (gelöscht oder anderes Konto?)")
|
||
resp.raise_for_status()
|
||
|
||
|
||
def fetch_books(server_url: str, token: str) -> tuple[list[dict], int | None, int]:
|
||
"""(books, the server's clock in ms from X-Server-Now or None, this machine's clock in ms when
|
||
the answer came). Each book: id, meta_ct, meta_iv, uploaded_at, is_read, has_cover and the
|
||
reading state (scroll_fraction, position_anchor, read_at, device, furthest_*)."""
|
||
resp = requests.get(f"{server_url}/books/", headers=_headers(token), timeout=60)
|
||
device_now = int(time.time() * 1000)
|
||
_check(resp, "Bücherliste")
|
||
try:
|
||
server_now = int(resp.headers.get("X-Server-Now", ""))
|
||
except ValueError:
|
||
server_now = None
|
||
return resp.json(), server_now, device_now
|
||
|
||
|
||
def fetch_prefs(server_url: str, token: str) -> dict:
|
||
"""The library's encrypted settings: {prefs_ct, prefs_iv}, both '' while there are none."""
|
||
resp = requests.get(f"{server_url}/books/prefs/", headers=_headers(token), timeout=30)
|
||
_check(resp, "Bibliothekseinstellungen")
|
||
return resp.json()
|
||
|
||
|
||
def fetch_book_data(server_url: str, token: str, book_id: int) -> dict:
|
||
resp = requests.get(f"{server_url}/books/{book_id}/data/", headers=_headers(token), timeout=300)
|
||
_check(resp, f"Buch {book_id}")
|
||
return resp.json()
|
||
|
||
|
||
def fetch_annotations(server_url: str, token: str, book_id: int, kind: str) -> dict:
|
||
"""A book's encrypted highlights or bookmarks: {ct, iv} (both null while there are none)."""
|
||
resp = requests.get(f"{server_url}/books/{book_id}/{kind}/", headers=_headers(token), timeout=60)
|
||
_check(resp, f"Buch {book_id}")
|
||
return resp.json()
|
||
|
||
|
||
def post_annotations(server_url: str, token: str, book_id: int, kind: str, blob: dict) -> None:
|
||
"""Replace a book's highlights or bookmarks as a whole (the server doesn't merge)."""
|
||
resp = requests.post(f"{server_url}/books/{book_id}/{kind}/", headers=_headers(token), json=blob, timeout=60)
|
||
_check(resp, f"Buch {book_id}")
|
||
|
||
|
||
def post_progress(server_url: str, token: str, book_id: int, *, anchor: str, fraction: float, read_at: int,
|
||
device: str, force: bool = False) -> dict:
|
||
"""Send a queued position. `sent_at` is the moment it leaves: the server corrects this
|
||
machine's clock error with it. The answer is the state after the write."""
|
||
body = {"scroll_fraction": fraction, "position_anchor": anchor, "read_at": read_at,
|
||
"sent_at": int(time.time() * 1000), "device": device}
|
||
if force:
|
||
body["force"] = True
|
||
resp = requests.post(f"{server_url}/books/{book_id}/progress/", headers=_headers(token), json=body, timeout=30)
|
||
_check(resp, f"Buch {book_id}")
|
||
return resp.json()
|