diora-web/tui/diora_tui/api.py

"""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()