From 8c11ded7e014cdd2f2aac4a3504a5bc5459c938c Mon Sep 17 00:00:00 2001 From: Karolis Strazdas Date: Sun, 4 Oct 2026 00:12:24 +0300 Subject: [PATCH 1/2] refactor: split acquisition services and refresh development guidance --- AGENTS.md | 6 +- README.md | 95 +- papyrus/models/powersync_demo.py | 4 +- papyrus/services/acquisition.py | 1310 ----------------- papyrus/services/acquisition/__init__.py | 62 + papyrus/services/acquisition/jobs.py | 542 +++++++ papyrus/services/acquisition/providers.py | 588 ++++++++ .../services/acquisition/release_tokens.py | 84 ++ papyrus/services/acquisition/transport.py | 85 ++ papyrus/services/acquisition/types.py | 72 + scripts/export_openapi.py | 61 + tests/api/routes/test_acquisition.py | 5 +- tests/api/routes/test_auth_sandbox.py | 16 +- tests/api/routes/test_powersync_sandbox.py | 6 +- tests/integration/test_auth_smoke.py | 4 +- tests/services/test_acquisition.py | 33 +- tests/services/test_auth.py | 4 +- tests/test_openapi_export.py | 27 + 18 files changed, 1633 insertions(+), 1371 deletions(-) delete mode 100644 papyrus/services/acquisition.py create mode 100644 papyrus/services/acquisition/__init__.py create mode 100644 papyrus/services/acquisition/jobs.py create mode 100644 papyrus/services/acquisition/providers.py create mode 100644 papyrus/services/acquisition/release_tokens.py create mode 100644 papyrus/services/acquisition/transport.py create mode 100644 papyrus/services/acquisition/types.py create mode 100644 scripts/export_openapi.py create mode 100644 tests/test_openapi_export.py diff --git a/AGENTS.md b/AGENTS.md index c85c445..d180067 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -6,7 +6,8 @@ - `papyrus/services`: service-layer business logic. Prefer new domain modules here instead of growing route handlers. - `papyrus/schemas`: Pydantic request and response models. - `papyrus/models`: SQLAlchemy models and metadata exports used by Alembic. -- `papyrus/core`: shared infrastructure such as config, database, exceptions, and security. +- `papyrus/config.py`: validated environment configuration. +- `papyrus/core`: database, exceptions, security and shared infrastructure. - `alembic`: Alembic environment and migration revisions. - `tests/api/routes`: endpoint behavior and contract tests. - `tests/services`: service-layer tests. @@ -77,8 +78,7 @@ Local auth testing supports Mailpit for SMTP capture, a dev auth sandbox at `/__ Use `.env.example`, `tests/api/routes/test_auth.py`, and `tests/integration/test_auth_smoke.py` for current configuration and test entry points. -The auth, Flutter integration, and PowerSync sandbox guides linked by the README -are absent from this checkout; do not assume their contents or invent commands from them. +See `README.md` for current auth/sync routes, local sandbox entry points and test lanes. For client/server contracts, use the workspace's `papyrus-sync-contract` skill and `../.agents/skills/papyrus-sync-contract/references/contract-map.md` when developing diff --git a/README.md b/README.md index 87797be..47db1c4 100644 --- a/README.md +++ b/README.md @@ -1,41 +1,86 @@ # Papyrus Server -FastAPI backend for Papyrus authentication, metadata, file storage, and -PowerSync-backed synchronization. +FastAPI services for authentication, book metadata, media, OPDS relay and +PowerSync-backed synchronization. PostgreSQL stores account-scoped library data; +the pinned PowerSync service replicates it to client SQLite databases. -## Auth And Sync +## Local setup -- Email/password auth uses `POST /v1/auth/register` and - `POST /v1/auth/login`. -- Google auth starts at `GET /v1/auth/oauth/google/start` and finishes through - `POST /v1/auth/exchange-code`. -- PowerSync credentials come from `POST /v1/auth/powersync-token`. -- PowerSync uploads use `POST /v1/sync/powersync-upload`. +Run from this repository using Python 3.12 and Docker Compose: -Read the focused guides: +```bash +uv sync --locked --extra dev +./scripts/bootstrap_local.sh +npm --prefix frontend/dev-pages ci +npm --prefix frontend/dev-pages run dev +``` + +Bootstrap creates missing development keys, starts the local databases, applies +Alembic migrations and configures PowerSync replication. Read `.env.example` and +`scripts/bootstrap_local.sh` before running it against an existing environment. +The Vite command is optional and serves the development sandbox assets. + +## Auth and sync contracts + +The API prefix is configured by `API_PREFIX`; `.env.example` uses `/v1`. + +| Operation | Route under the configured prefix | +| --- | --- | +| Register / sign in | `POST /auth/register`, `POST /auth/login` | +| Start Google sign-in | `GET /auth/oauth/google/start` | +| Exchange the browser auth code | `POST /auth/exchange-code` | +| Refresh the session | `POST /auth/refresh` | +| Get PowerSync credentials | `POST /auth/powersync-token` | +| Upload offline changes | `POST /sync/powersync-upload` | -- [Flutter auth and PowerSync integration](docs/flutter-auth-integration.md) -- [Authentication testing](docs/auth-testing.md) -- [PowerSync sandbox](docs/powersync-sandbox.md) -- [Managed book acquisition](docs/acquisition-downloads.md) +Contracts are implemented in [auth routes](papyrus/api/routes/auth.py), +[sync routes](papyrus/api/routes/sync.py) and +[PowerSync replication rules](powersync/sync-config.yaml). The server validates +ownership and tombstones and commits mixed upload batches atomically. Media +removal happens after commit. Client token refresh and profile-scoped persistence +remain client responsibilities. -## Local Setup +With `DEBUG=true`, `/__dev/auth-sandbox` and `/__dev/powersync-sandbox` provide +local development tools. They are excluded from the public schema. Mailpit in +Compose captures local SMTP. See [.env.example](.env.example) and +[auth smoke tests](tests/integration/test_auth_smoke.py) for provider setup; +provider-backed tests require explicit SMTP/Google configuration. -Run from `server/`: +[Managed acquisition routes](papyrus/api/routes/acquisition.py) use owner-scoped +endpoints, jobs and rules. Provider adapters, job lifecycle and release tokens +live in [the acquisition service](papyrus/services/acquisition/). The monitor +imports selected media and handles retry/cancellation. [OPDS relay setup](docs/opds-relay.md) +describes catalog transport and origin restrictions. + +## Verification + +Inside the Papyrus workspace, run: ```bash -uv sync --extra dev -./scripts/bootstrap_local.sh -npm --prefix frontend/dev-pages install -npm --prefix frontend/dev-pages run dev +../tools/papyrus check server +../tools/papyrus test server ``` -The bootstrap is idempotent: it creates development keys when missing, starts -the databases, applies Alembic migrations, configures logical replication, and -starts the healthy server and pinned PowerSync services. +The CLI excludes provider `auth_smoke` tests and checks for a separate local test +database. Pytest fixtures drop/recreate tables, including most route tests without +an `integration` marker. Direct pytest requires a distinct `*_test` database; +do not run concurrent suites on it. Read [test fixtures](tests/conftest.py) before +setting test database variables. Run provider smoke tests separately once configured. -## Checks +For schema changes, run `uv run --locked alembic current` and `heads`, apply the +reviewed migrations, then verify `current` and `check`. Keep migrations separate +from structural refactors. + +## Public API documentation + +The docs repository renders a generated schema rather than maintaining copied +endpoint definitions. Export with deterministic documentation settings: ```bash -uv run pytest --cov --cov-report html +uv run --locked python scripts/export_openapi.py ../docs/_static/openapi.json +uv run --locked python scripts/export_openapi.py ../docs/_static/openapi.json --check ``` + +Export does not load `.env`, start services or connect to a database. Set the docs +workflow's server revision to the source commit used for the snapshot. The runtime +`/openapi.json` remains the specification for a configured deployment. diff --git a/papyrus/models/powersync_demo.py b/papyrus/models/powersync_demo.py index 83e7308..fb636f9 100644 --- a/papyrus/models/powersync_demo.py +++ b/papyrus/models/powersync_demo.py @@ -17,7 +17,9 @@ class PowerSyncDemoItem(Base): __tablename__ = "powersync_demo_items" item_id: Mapped[UUID] = mapped_column(Uuid, primary_key=True, default=uuid4) - owner_user_id: Mapped[UUID] = mapped_column(ForeignKey("users.user_id", ondelete="CASCADE"), nullable=False, index=True) + owner_user_id: Mapped[UUID] = mapped_column( + ForeignKey("users.user_id", ondelete="CASCADE"), nullable=False, index=True + ) title: Mapped[str] = mapped_column(String(255), nullable=False) notes: Mapped[str | None] = mapped_column(Text, nullable=True) created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False, server_default=func.now()) diff --git a/papyrus/services/acquisition.py b/papyrus/services/acquisition.py deleted file mode 100644 index 7027fbd..0000000 --- a/papyrus/services/acquisition.py +++ /dev/null @@ -1,1310 +0,0 @@ -"""Adapters for indexer protocols and self-hosted download clients.""" - -from __future__ import annotations - -import asyncio -import base64 -import json -import re -from dataclasses import dataclass -from datetime import UTC, datetime, timedelta -from pathlib import PurePosixPath -from typing import Any, Self -from urllib.error import HTTPError, URLError -from urllib.parse import urlencode, urljoin -from urllib.request import Request, urlopen -from uuid import UUID, uuid4 -from xml.etree import ElementTree - -from fastapi import HTTPException, status -from sqlalchemy import func, select -from sqlalchemy.ext.asyncio import AsyncSession - -from papyrus.config import get_settings -from papyrus.core.security import decrypt_secret_payload, encrypt_secret_payload -from papyrus.models.acquisition import AcquisitionEndpoint, AcquisitionJob, AcquisitionRule -from papyrus.models.sync import SyncBook -from papyrus.schemas.acquisition import AcquisitionEndpointTest -from papyrus.services.media import BOOK_EXTENSIONS - -RELEASE_TOKEN_LIFETIME = timedelta(minutes=5) -SUBMISSION_LEASE_LIFETIME = timedelta(minutes=5) - - -@dataclass(frozen=True, slots=True) -class ReleaseCandidate: - title: str - download_url: str - protocol: str - indexer: str - size_bytes: int | None - seeders: int | None - publish_date: datetime | None - format_hints: list[str] - - -@dataclass(frozen=True, slots=True) -class ReleaseTokenPayload: - endpoint_id: UUID - owner_user_id: UUID - title: str - download_url: str - protocol: str - indexer: str - size_bytes: int | None - seeders: int | None - publish_date: datetime | None - format_hints: list[str] - - -@dataclass(frozen=True, slots=True) -class BatchSubmissionResult: - index: int - job: AcquisitionJob | None - error: str | None - - -@dataclass(frozen=True, slots=True) -class QbittorrentTorrent: - hash: str - state: str - progress_basis_points: int - downloaded_bytes: int - total_bytes: int - download_speed_bytes_per_second: int - eta_seconds: int | None - - -@dataclass(frozen=True, slots=True) -class QbittorrentFile: - index: int - name: str - size_bytes: int - progress_basis_points: int - priority: int - - -@dataclass(frozen=True, slots=True) -class JobFileCandidate: - index: int - name: str - size_bytes: int - progress_basis_points: int - priority: int - supported: bool - - -@dataclass(slots=True) -class QbittorrentClient: - endpoint: AcquisitionEndpoint - cookie: str - - @classmethod - async def connect(cls, endpoint: AcquisitionEndpoint) -> Self: - if endpoint.kind != "qbittorrent": - raise HTTPException(status_code=422, detail="Endpoint is not qBittorrent") - - credentials = _credentials(endpoint) - login = urlencode( - { - "username": credentials.get("username", ""), - "password": credentials.get("password", ""), - } - ).encode() - response_status, headers, response_payload = await _request( - _url(endpoint, "api/v2/auth/login"), - method="POST", - headers={"Content-Type": "application/x-www-form-urlencoded"}, - body=login, - ) - - if not _qbittorrent_login_succeeded(response_status, response_payload): - raise HTTPException(status_code=502, detail="qBittorrent authentication failed") - - return cls( - endpoint=endpoint, - cookie=_header_value(headers, "Set-Cookie").split(";", 1)[0], - ) - - async def find_torrent( - self, - *, - tag: str, - torrent_hash: str | None = None, - ) -> QbittorrentTorrent: - payload: list[dict[str, object]] = [] - - if torrent_hash is not None: - payload = await self._get_json_array( - "api/v2/torrents/info", - {"hashes": torrent_hash}, - ) - - if not payload: - payload = await self._get_json_array("api/v2/torrents/info", {"tag": tag}) - - if not payload: - raise HTTPException(status_code=404, detail="qBittorrent torrent not found") - - if len(payload) != 1: - raise HTTPException(status_code=409, detail="qBittorrent tag matched multiple torrents") - - item = payload[0] - progress_basis_points = _progress_basis_points(item.get("progress")) - completed_bytes = _optional_int(item.get("completed")) - - return QbittorrentTorrent( - hash=_required_string(item, "hash", "qBittorrent torrent"), - state=_required_string(item, "state", "qBittorrent torrent"), - progress_basis_points=progress_basis_points, - downloaded_bytes=completed_bytes - if completed_bytes is not None - else _required_int(item, "downloaded", "qBittorrent torrent"), - total_bytes=_required_int(item, "total_size", "qBittorrent torrent"), - download_speed_bytes_per_second=_required_int(item, "dlspeed", "qBittorrent torrent"), - eta_seconds=0 if progress_basis_points == 10_000 else _optional_int(item.get("eta")), - ) - - async def files(self, torrent_hash: str) -> list[QbittorrentFile]: - payload = await self._get_json_array("api/v2/torrents/files", {"hash": torrent_hash}) - return [ - QbittorrentFile( - index=_required_int(item, "index", "qBittorrent file"), - name=_required_string(item, "name", "qBittorrent file"), - size_bytes=_required_int(item, "size", "qBittorrent file"), - progress_basis_points=_progress_basis_points(item.get("progress")), - priority=_required_int(item, "priority", "qBittorrent file"), - ) - for item in payload - ] - - async def select_file(self, torrent_hash: str, *, selected_index: int, file_indices: list[int]) -> None: - if selected_index not in file_indices: - raise HTTPException(status_code=422, detail="Selected qBittorrent file was not found") - - other_indices = [index for index in file_indices if index != selected_index] - - await self._post_form( - "api/v2/torrents/pause", - {"hashes": torrent_hash}, - fallback_path="api/v2/torrents/stop", - ) - - if other_indices: - await self._post_form( - "api/v2/torrents/filePrio", - { - "hash": torrent_hash, - "id": "|".join(str(index) for index in other_indices), - "priority": "0", - }, - ) - - await self._post_form( - "api/v2/torrents/filePrio", - { - "hash": torrent_hash, - "id": str(selected_index), - "priority": "1", - }, - ) - await self._post_form( - "api/v2/torrents/resume", - {"hashes": torrent_hash}, - fallback_path="api/v2/torrents/start", - ) - - async def pause(self, torrent_hash: str) -> None: - await self._post_form( - "api/v2/torrents/pause", - {"hashes": torrent_hash}, - fallback_path="api/v2/torrents/stop", - ) - - async def delete_torrent(self, torrent_hash: str) -> None: - await self._post_form( - "api/v2/torrents/delete", - { - "hashes": torrent_hash, - "deleteFiles": "true", - }, - ) - - async def _get_json_array(self, path: str, params: dict[str, str]) -> list[dict[str, object]]: - response_status, _, payload = await _request( - _url(self.endpoint, f"{path}?{urlencode(params)}"), - headers={"Cookie": self.cookie}, - ) - - if response_status >= 400: - raise HTTPException(status_code=502, detail="qBittorrent request failed") - - return _json_array(payload, "qBittorrent") - - async def _post_form( - self, - path: str, - values: dict[str, str], - *, - fallback_path: str | None = None, - ) -> None: - response_status, _, _ = await _request( - _url(self.endpoint, path), - method="POST", - headers={ - "Content-Type": "application/x-www-form-urlencoded", - "Cookie": self.cookie, - }, - body=urlencode(values).encode(), - ) - - if response_status == 404 and fallback_path is not None: - response_status, _, _ = await _request( - _url(self.endpoint, fallback_path), - method="POST", - headers={ - "Content-Type": "application/x-www-form-urlencoded", - "Cookie": self.cookie, - }, - body=urlencode(values).encode(), - ) - - if response_status >= 400: - raise HTTPException(status_code=502, detail="qBittorrent request failed") - - -def _required_string(item: dict[str, object], key: str, subject: str) -> str: - value = item.get(key) - - if not isinstance(value, str) or not value: - raise HTTPException(status_code=502, detail=f"{subject} returned invalid data") - - return value - - -def _required_int(item: dict[str, object], key: str, subject: str) -> int: - value = item.get(key) - - if not isinstance(value, int) or isinstance(value, bool): - raise HTTPException(status_code=502, detail=f"{subject} returned invalid data") - - return value - - -def _progress_basis_points(value: object) -> int: - if not isinstance(value, int | float) or isinstance(value, bool): - raise HTTPException(status_code=502, detail="qBittorrent returned invalid progress") - - return min(10_000, max(0, round(float(value) * 10_000))) - - -def create_release_token( - release: ReleaseCandidate, - endpoint: AcquisitionEndpoint, - *, - now: datetime | None = None, -) -> str: - issued_at = now or datetime.now(UTC) - - if endpoint.endpoint_id is None: - raise ValueError("Release tokens require a persisted endpoint") - - return encrypt_secret_payload( - { - "endpoint_id": str(endpoint.endpoint_id), - "owner_user_id": str(endpoint.owner_user_id), - "title": release.title, - "download_url": release.download_url, - "protocol": release.protocol, - "indexer": release.indexer, - "size_bytes": "" if release.size_bytes is None else str(release.size_bytes), - "seeders": "" if release.seeders is None else str(release.seeders), - "publish_date": release.publish_date.isoformat() if release.publish_date is not None else "", - "format_hints": json.dumps(release.format_hints), - "expires_at": str(int((issued_at + RELEASE_TOKEN_LIFETIME).timestamp())), - } - ) - - -def decode_release_token( - token: str, - owner_user_id: UUID, - *, - now: datetime | None = None, -) -> ReleaseTokenPayload: - try: - payload = decrypt_secret_payload(token) - expires_at = int(payload["expires_at"]) - token_owner_user_id = UUID(payload["owner_user_id"]) - endpoint_id = UUID(payload["endpoint_id"]) - format_hints = json.loads(payload["format_hints"]) - size_bytes = int(payload["size_bytes"]) if payload["size_bytes"] else None - seeders = int(payload["seeders"]) if payload["seeders"] else None - publish_date = datetime.fromisoformat(payload["publish_date"]) if payload["publish_date"] else None - - if ( - token_owner_user_id != owner_user_id - or expires_at <= int((now or datetime.now(UTC)).timestamp()) - or not isinstance(format_hints, list) - or not all(isinstance(value, str) for value in format_hints) - ): - raise ValueError - - return ReleaseTokenPayload( - endpoint_id=endpoint_id, - owner_user_id=token_owner_user_id, - title=payload["title"], - download_url=payload["download_url"], - protocol=payload["protocol"], - indexer=payload["indexer"], - size_bytes=size_bytes, - seeders=seeders, - publish_date=publish_date, - format_hints=format_hints, - ) - except (KeyError, TypeError, ValueError, json.JSONDecodeError) as exc: - raise HTTPException(status_code=400, detail="Release token is invalid or expired") from exc - - -def _url(endpoint: AcquisitionEndpoint, path: str) -> str: - return urljoin(endpoint.base_url.rstrip("/") + "/", path.lstrip("/")) - - -async def _request( - url: str, *, method: str = "GET", headers: dict[str, str] | None = None, body: bytes | None = None -) -> tuple[int, dict[str, str], bytes]: - """Perform a bounded blocking HTTP request off the event loop.""" - - def send() -> tuple[int, dict[str, str], bytes]: - request = Request(url, data=body, headers=headers or {}, method=method) - try: - with urlopen(request, timeout=15) as response: # noqa: S310 - user-owned self-hosted integrations - return response.status, dict(response.headers.items()), response.read(5_000_000) - except HTTPError as exc: - return exc.code, dict(exc.headers.items()), exc.read(1_000_000) - except URLError as exc: - raise HTTPException(status_code=502, detail=f"Integration request failed: {exc.reason}") from exc - - return await asyncio.to_thread(send) - - -def _credentials(endpoint: AcquisitionEndpoint) -> dict[str, str]: - credentials = endpoint.credentials or {} - encrypted = credentials.get("encrypted") - if encrypted is None: - return credentials - try: - return decrypt_secret_payload(encrypted) - except ValueError as exc: - raise HTTPException(status_code=500, detail="Stored integration credentials are invalid") from exc - - -def _json_value(payload: bytes, integration: str) -> object: - try: - return json.loads(payload) - except (UnicodeDecodeError, json.JSONDecodeError) as exc: - raise HTTPException(status_code=502, detail=f"{integration} returned invalid JSON") from exc - - -def _json_object(payload: bytes, integration: str) -> dict[str, object]: - value = _json_value(payload, integration) - if not isinstance(value, dict): - raise HTTPException(status_code=502, detail=f"{integration} returned an invalid response") - return value - - -def _json_array(payload: bytes, integration: str) -> list[dict[str, object]]: - value = _json_value(payload, integration) - if not isinstance(value, list) or not all(isinstance(item, dict) for item in value): - raise HTTPException(status_code=502, detail=f"{integration} returned an invalid response") - return value - - -def _require_deluge_result(payload: bytes) -> object: - response = _json_object(payload, "Deluge") - result = response.get("result") - if response.get("error") is not None or result is None or result is False: - raise HTTPException(status_code=502, detail="Deluge rejected the request") - return result - - -async def search_endpoint(endpoint: AcquisitionEndpoint, query: str) -> list[ReleaseCandidate]: - """Search Prowlarr or a Torznab-compatible torrent indexer.""" - credentials = _credentials(endpoint) - if endpoint.kind == "prowlarr": - request_url = _url(endpoint, f"api/v1/search?{urlencode({'query': query})}") - response_status, _, payload = await _request(request_url, headers={"X-Api-Key": credentials.get("api_key", "")}) - if response_status >= 400: - raise HTTPException(status_code=502, detail="Prowlarr search failed") - data = _json_array(payload, "Prowlarr") - releases: list[ReleaseCandidate] = [] - - for item in data: - download_url = item.get("downloadUrl") or item.get("magnetUrl") or item.get("guid") - protocol = item.get("protocol", "torrent") - - if not isinstance(download_url, str) or not download_url or protocol != "torrent": - continue - - title_value = item.get("title") - indexer_value = item.get("indexer") - title = title_value if isinstance(title_value, str) and title_value else "Untitled" - indexer = indexer_value if isinstance(indexer_value, str) and indexer_value else "Prowlarr" - - releases.append( - ReleaseCandidate( - title=title, - download_url=download_url, - protocol="torrent", - indexer=indexer, - size_bytes=_optional_int(item.get("size")), - seeders=_optional_int(item.get("seeders")), - publish_date=None, - format_hints=_format_hints(title, download_url), - ) - ) - - return releases - - params = urlencode({"t": "search", "q": query, "apikey": credentials.get("api_key", "")}) - response_status, _, payload = await _request(_url(endpoint, f"api?{params}")) - if response_status >= 400: - raise HTTPException(status_code=502, detail=f"{endpoint.kind.title()} search failed") - return _parse_torznab(payload, endpoint.name) - - -def _parse_torznab(payload: bytes, indexer: str) -> list[ReleaseCandidate]: - try: - root = ElementTree.fromstring(payload) - except ElementTree.ParseError as exc: - raise HTTPException(status_code=502, detail="Indexer returned invalid XML") from exc - releases: list[ReleaseCandidate] = [] - for item in root.findall(".//item"): - enclosure = item.find("enclosure") - link = (enclosure.get("url") if enclosure is not None else None) or item.findtext("link") - if not link: - continue - attrs = {child.attrib.get("name"): child.attrib.get("value") for child in item if child.tag.endswith("attr")} - size = attrs.get("size") - seeders = attrs.get("seeders") - releases.append( - ReleaseCandidate( - title=item.findtext("title") or "Untitled", - download_url=link, - protocol="torrent", - indexer=indexer, - size_bytes=int(size) if size is not None and size.isdigit() else None, - seeders=int(seeders) if seeders is not None and seeders.isdigit() else None, - publish_date=None, - format_hints=_format_hints(item.findtext("title") or "", link), - ) - ) - return releases - - -def _format_hints(title: str, download_url: str) -> list[str]: - searchable = f"{title} {download_url}".lower() - return [ - extension - for extension in ("epub", "pdf", "mobi", "azw3", "txt", "cbr", "cbz") - if re.search(rf"(? int | None: - return value if isinstance(value, int) and not isinstance(value, bool) else None - - -async def submit_to_client( - endpoint: AcquisitionEndpoint, - download_url: str, - category: str | None, - save_path: str | None, - *, - tags: list[str] | None = None, -) -> str | None: - """Submit a magnet or torrent URL to a supported BitTorrent client.""" - if endpoint.kind == "qbittorrent": - return await _submit_qbittorrent(endpoint, download_url, category, save_path, tags=tags) - if endpoint.kind == "transmission": - return await _submit_transmission(endpoint, download_url, save_path) - if endpoint.kind == "deluge": - return await _submit_deluge(endpoint, download_url, save_path) - raise HTTPException(status_code=422, detail="Endpoint is not a download client") - - -async def dispatch_arr_command(endpoint: AcquisitionEndpoint, command: str, ids: list[int]) -> str | None: - """Start an acquisition/search command in a Servarr application. - - The Arr apps own their library and release-grab decisions, so they must be - driven through their command API rather than handed an arbitrary magnet. - """ - if endpoint.kind not in {"readarr", "sonarr", "radarr", "lidarr", "whisparr"}: - raise HTTPException(status_code=422, detail="Endpoint is not a Servarr application") - - allowed_commands = { - "readarr": {"AuthorSearch", "BookSearch"}, - "sonarr": {"SeriesSearch", "EpisodeSearch", "MissingEpisodeSearch"}, - "radarr": {"MoviesSearch", "MissingMoviesSearch"}, - "lidarr": {"ArtistSearch", "AlbumSearch", "MissingAlbumSearch"}, - "whisparr": {"SeriesSearch", "EpisodeSearch", "MissingEpisodeSearch"}, - } - if command not in allowed_commands[endpoint.kind]: - raise HTTPException(status_code=422, detail="Command is not supported by this Servarr application") - - id_field = { - "AuthorSearch": "authorIds", - "BookSearch": "bookIds", - "SeriesSearch": "seriesId", - "EpisodeSearch": "episodeIds", - "MissingEpisodeSearch": "seriesId", - "MoviesSearch": "movieIds", - "MissingMoviesSearch": "movieIds", - "ArtistSearch": "artistIds", - "AlbumSearch": "albumIds", - "MissingAlbumSearch": "artistIds", - }[command] - payload: dict[str, object] = {"name": command} - if ids: - payload[id_field] = ids[0] if id_field in {"seriesId"} else ids - - response_status, _, response_payload = await _request( - _url(endpoint, "api/v3/command"), - method="POST", - headers={"Content-Type": "application/json", "X-Api-Key": _credentials(endpoint).get("api_key", "")}, - body=json.dumps(payload).encode(), - ) - if response_status >= 400: - raise HTTPException(status_code=502, detail=f"{endpoint.kind.title()} command failed") - response = _json_object(response_payload, endpoint.kind.title()) - return str(response.get("id")) if response.get("id") is not None else None - - -async def _submit_qbittorrent( - endpoint: AcquisitionEndpoint, - download_url: str, - category: str | None, - save_path: str | None, - *, - tags: list[str] | None = None, -) -> str | None: - credentials = _credentials(endpoint) - login = urlencode( - {"username": credentials.get("username", ""), "password": credentials.get("password", "")} - ).encode() - response_status, headers, response_payload = await _request( - _url(endpoint, "api/v2/auth/login"), - method="POST", - headers={"Content-Type": "application/x-www-form-urlencoded"}, - body=login, - ) - if not _qbittorrent_login_succeeded(response_status, response_payload): - raise HTTPException(status_code=502, detail="qBittorrent authentication failed") - payload = {"urls": download_url} - if category: - payload["category"] = category - if save_path: - payload["savepath"] = save_path - if tags: - payload["tags"] = ",".join(tags) - response_status, _, _ = await _request( - _url(endpoint, "api/v2/torrents/add"), - method="POST", - headers={ - "Content-Type": "application/x-www-form-urlencoded", - "Cookie": _header_value(headers, "Set-Cookie").split(";", 1)[0], - }, - body=urlencode(payload).encode(), - ) - if response_status >= 400: - raise HTTPException(status_code=502, detail="qBittorrent rejected the release") - return None - - -def _header_value(headers: dict[str, str], name: str) -> str: - normalized_name = name.casefold() - return next((value for key, value in headers.items() if key.casefold() == normalized_name), "") - - -def _qbittorrent_login_succeeded(response_status: int, response_payload: bytes) -> bool: - return response_status < 400 and (response_status == 204 or response_payload.strip() == b"Ok.") - - -async def _submit_transmission(endpoint: AcquisitionEndpoint, download_url: str, save_path: str | None) -> str | None: - credentials = _credentials(endpoint) - arguments: dict[str, str] = {"filename": download_url} - if save_path: - arguments["download-dir"] = save_path - body = json.dumps({"method": "torrent-add", "arguments": arguments}).encode() - headers = {"Content-Type": "application/json"} - if credentials.get("username"): - token = base64.b64encode(f"{credentials['username']}:{credentials.get('password', '')}".encode()).decode() - headers["Authorization"] = f"Basic {token}" - response_status, response_headers, payload = await _request( - _url(endpoint, "transmission/rpc"), method="POST", headers=headers, body=body - ) - if response_status == 409: - headers["X-Transmission-Session-Id"] = response_headers.get("X-Transmission-Session-Id", "") - response_status, _, payload = await _request( - _url(endpoint, "transmission/rpc"), method="POST", headers=headers, body=body - ) - if response_status >= 400: - raise HTTPException(status_code=502, detail="Transmission rejected the release") - - response = _json_object(payload, "Transmission") - if response.get("result") != "success": - raise HTTPException(status_code=502, detail="Transmission rejected the release") - - response_arguments = response.get("arguments") - if not isinstance(response_arguments, dict): - raise HTTPException(status_code=502, detail="Transmission returned an invalid response") - - torrent = response_arguments.get("torrent-added") or response_arguments.get("torrent-duplicate") - if not isinstance(torrent, dict): - return None - - reference = torrent.get("hashString") - return str(reference) if reference is not None else None - - -async def _submit_deluge(endpoint: AcquisitionEndpoint, download_url: str, save_path: str | None) -> str | None: - credentials = _credentials(endpoint) - headers = {"Content-Type": "application/json"} - login = json.dumps({"method": "auth.login", "params": [credentials.get("password", "")], "id": 1}).encode() - response_status, response_headers, login_payload = await _request( - _url(endpoint, "json"), method="POST", headers=headers, body=login - ) - if response_status >= 400: - raise HTTPException(status_code=502, detail="Deluge authentication failed") - - try: - _require_deluge_result(login_payload) - except HTTPException as exc: - raise HTTPException(status_code=502, detail="Deluge authentication failed") from exc - - options = {"download_location": save_path} if save_path else {} - method = "core.add_torrent_magnet" if download_url.startswith("magnet:") else "core.add_torrent_url" - body = json.dumps({"method": method, "params": [download_url, options], "id": 2}).encode() - headers["Cookie"] = response_headers.get("Set-Cookie", "").split(";", 1)[0] - response_status, _, payload = await _request(_url(endpoint, "json"), method="POST", headers=headers, body=body) - if response_status >= 400: - raise HTTPException(status_code=502, detail="Deluge rejected the release") - - try: - result = _require_deluge_result(payload) - except HTTPException as exc: - raise HTTPException(status_code=502, detail="Deluge rejected the release") from exc - - return str(result) - - -async def owned_endpoint(session: AsyncSession, owner_user_id: Any, endpoint_id: Any) -> AcquisitionEndpoint: - result = await session.execute( - select(AcquisitionEndpoint).where( - AcquisitionEndpoint.endpoint_id == endpoint_id, AcquisitionEndpoint.owner_user_id == owner_user_id - ) - ) - endpoint = result.scalar_one_or_none() - if endpoint is None: - raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Acquisition endpoint not found") - return endpoint - - -async def owned_job( - session: AsyncSession, - owner_user_id: UUID, - job_id: UUID, - *, - for_update: bool = False, -) -> AcquisitionJob: - statement = select(AcquisitionJob).where( - AcquisitionJob.job_id == job_id, - AcquisitionJob.owner_user_id == owner_user_id, - ) - - if for_update: - statement = statement.with_for_update() - - result = await session.execute(statement.execution_options(populate_existing=for_update)) - job = result.scalar_one_or_none() - - if job is None: - raise HTTPException(status_code=404, detail="Acquisition job not found") - - return job - - -async def paginated_jobs( - session: AsyncSession, - owner_user_id: UUID, - *, - limit: int, - offset: int, -) -> tuple[list[AcquisitionJob], int]: - total = await session.scalar( - select(func.count()).select_from(AcquisitionJob).where(AcquisitionJob.owner_user_id == owner_user_id) - ) - result = await session.execute( - select(AcquisitionJob) - .where(AcquisitionJob.owner_user_id == owner_user_id) - .order_by(AcquisitionJob.created_at.desc(), AcquisitionJob.job_id.desc()) - .limit(limit) - .offset(offset) - ) - - return list(result.scalars()), total or 0 - - -async def job_file_candidates( - session: AsyncSession, - owner_user_id: UUID, - job_id: UUID, -) -> list[JobFileCandidate]: - job = await owned_job(session, owner_user_id, job_id) - - if job.status != "needs_file_selection": - raise HTTPException(status_code=409, detail="Acquisition job does not need file selection") - - if job.endpoint_id is None or job.client_hash is None: - raise HTTPException(status_code=409, detail="Acquisition job is missing its qBittorrent reference") - - endpoint = await owned_endpoint(session, owner_user_id, job.endpoint_id) - client = await QbittorrentClient.connect(endpoint) - files = await client.files(job.client_hash) - - return [ - JobFileCandidate( - index=file.index, - name=file.name, - size_bytes=file.size_bytes, - progress_basis_points=file.progress_basis_points, - priority=file.priority, - supported=_supported_book_file(file.name), - ) - for file in files - ] - - -def _supported_book_file(filename: str) -> bool: - normalized = filename.replace("\\", "/") - extension = PurePosixPath(normalized).suffix.lower().lstrip(".") - return extension in BOOK_EXTENSIONS - - -async def select_job_file( - session: AsyncSession, - owner_user_id: UUID, - job_id: UUID, - file_index: int, -) -> AcquisitionJob: - job = await owned_job( - session, - owner_user_id, - job_id, - for_update=True, - ) - - if job.status != "needs_file_selection": - raise HTTPException(status_code=409, detail="Acquisition job does not need file selection") - - if job.endpoint_id is None or job.client_hash is None: - raise HTTPException(status_code=409, detail="Acquisition job is missing its qBittorrent reference") - - endpoint = await owned_endpoint(session, owner_user_id, job.endpoint_id) - client = await QbittorrentClient.connect(endpoint) - files = await client.files(job.client_hash) - selected = next((file for file in files if file.index == file_index), None) - - if selected is None: - raise HTTPException(status_code=404, detail="qBittorrent file not found") - - if not _supported_book_file(selected.name): - raise HTTPException(status_code=422, detail="Selected file is not a supported book") - - await client.select_file( - job.client_hash, - selected_index=selected.index, - file_indices=[file.index for file in files], - ) - - job.selected_file_path = selected.name - job.status = "downloading" - job.error = None - job.next_poll_at = datetime.now(UTC) - job.updated_at = datetime.now(UTC) - - await session.commit() - await session.refresh(job) - return job - - -async def cancel_job( - session: AsyncSession, - owner_user_id: UUID, - job_id: UUID, -) -> AcquisitionJob: - job = await owned_job( - session, - owner_user_id, - job_id, - for_update=True, - ) - - if job.status == "cancelled": - return job - - if job.status not in {"queued", "submitted", "downloading", "needs_file_selection"}: - raise HTTPException(status_code=409, detail="Acquisition job cannot be cancelled") - - if job.endpoint_id is None: - raise HTTPException(status_code=409, detail="Acquisition job is missing its qBittorrent endpoint") - - endpoint = await owned_endpoint(session, owner_user_id, job.endpoint_id) - client = await QbittorrentClient.connect(endpoint) - torrent_hash = job.client_hash - - if torrent_hash is None: - try: - torrent_hash = (await client.find_torrent(tag=f"papyrus:{job.job_id}")).hash - except HTTPException as exc: - if exc.status_code != 404: - raise - - if torrent_hash is not None: - await client.delete_torrent(torrent_hash) - - now = datetime.now(UTC) - job.status = "cancelled" - job.cancelled_at = now - job.updated_at = now - job.next_poll_at = None - job.error = None - - await session.commit() - await session.refresh(job) - return job - - -async def delete_terminal_job( - session: AsyncSession, - owner_user_id: UUID, - job_id: UUID, -) -> None: - job = await owned_job(session, owner_user_id, job_id) - - if job.status not in {"failed", "cancelled"}: - raise HTTPException(status_code=409, detail="Only failed or cancelled jobs can be removed") - - book = None - - if job.book_id is not None: - result = await session.execute( - select(SyncBook) - .where( - SyncBook.book_id == job.book_id, - SyncBook.owner_user_id == owner_user_id, - ) - .with_for_update() - ) - book = result.scalar_one_or_none() - - if book is not None and book.file_media_id is not None: - raise HTTPException(status_code=409, detail="Imported books must be removed from the library") - - await session.delete(job) - await session.flush() - - if book is not None: - await session.delete(book) - - await session.commit() - - -async def retry_job_import( - session: AsyncSession, - owner_user_id: UUID, - job_id: UUID, -) -> AcquisitionJob: - job = await owned_job( - session, - owner_user_id, - job_id, - for_update=True, - ) - - if job.status != "failed": - raise HTTPException(status_code=409, detail="Only failed jobs can retry import") - - if job.endpoint_id is None: - raise HTTPException(status_code=409, detail="Acquisition job cannot resume without qBittorrent") - - if job.submitted_at is None: - raise HTTPException(status_code=409, detail="Acquisition job was not submitted to qBittorrent") - - now = datetime.now(UTC) - job.status = "downloading" - job.retry_count += 1 - job.error = None - job.next_poll_at = now - job.updated_at = now - - await session.commit() - await session.refresh(job) - return job - - -async def submit_release_batch( - session: AsyncSession, - owner_user_id: UUID, - endpoint_id: UUID, - release_tokens: list[str], -) -> list[BatchSubmissionResult]: - endpoint = await owned_endpoint(session, owner_user_id, endpoint_id) - - if endpoint.kind != "qbittorrent": - raise HTTPException(status_code=422, detail="Managed downloads require qBittorrent") - - if not endpoint.enabled: - raise HTTPException(status_code=409, detail="Download client is disabled") - - if endpoint.download_root is None: - raise HTTPException(status_code=409, detail="qBittorrent download root is not configured") - - if get_settings().acquisition_import_root is None: - raise HTTPException(status_code=409, detail="Acquisition import root is not configured") - - results: list[BatchSubmissionResult] = [] - - for index, token in enumerate(release_tokens): - try: - release = decode_release_token(token, owner_user_id) - except HTTPException as exc: - results.append(BatchSubmissionResult(index=index, job=None, error=str(exc.detail))) - continue - - if release.protocol != "torrent" or not release.download_url.startswith(("magnet:", "http://", "https://")): - results.append(BatchSubmissionResult(index=index, job=None, error="Release token is invalid or expired")) - continue - - book_id = uuid4() - job_id = uuid4() - book = SyncBook( - book_id=book_id, - owner_user_id=owner_user_id, - title=release.title, - custom_metadata={ - "acquisition": { - "job_id": str(job_id), - "provisional": True, - } - }, - ) - job = AcquisitionJob( - job_id=job_id, - owner_user_id=owner_user_id, - endpoint_id=endpoint.endpoint_id, - book_id=book_id, - title=release.title, - download_url=None, - status="queued", - next_poll_at=datetime.now(UTC), - lease_owner=f"submission:{job_id}", - lease_until=datetime.now(UTC) + SUBMISSION_LEASE_LIFETIME, - ) - - session.add(book) - await session.flush() - - session.add(job) - await session.flush() - - await session.commit() - await session.refresh(job) - - try: - job.client_reference = await submit_to_client( - endpoint, - release.download_url, - "papyrus", - _managed_download_path(endpoint.download_root, owner_user_id, job_id), - tags=[f"papyrus:{job_id}"], - ) - job.status = "submitted" - job.submitted_at = datetime.now(UTC) - job.next_poll_at = datetime.now(UTC) - except HTTPException as exc: - job.status = "failed" - job.error = str(exc.detail) - job.next_poll_at = None - - job.lease_owner = None - job.lease_until = None - - await session.commit() - await session.refresh(job) - results.append(BatchSubmissionResult(index=index, job=job, error=None)) - - return results - - -def _managed_download_path(download_root: str, owner_user_id: UUID, job_id: UUID) -> str: - normalized_root = download_root.rstrip("/\\") - separator = "\\" if "\\" in normalized_root and "/" not in normalized_root else "/" - return separator.join((normalized_root, str(owner_user_id), str(job_id))) - - -async def delete_acquisition_endpoint(session: AsyncSession, owner_user_id: Any, endpoint_id: Any) -> None: - endpoint = await owned_endpoint(session, owner_user_id, endpoint_id) - active_job_id = await session.scalar( - select(AcquisitionJob.job_id) - .where( - AcquisitionJob.owner_user_id == owner_user_id, - AcquisitionJob.endpoint_id == endpoint_id, - AcquisitionJob.status.not_in({"completed", "failed", "cancelled"}), - ) - .limit(1) - ) - - if active_job_id is not None: - raise HTTPException(status_code=409, detail="Endpoint has active acquisition jobs") - - result = await session.execute( - select(AcquisitionRule).where(AcquisitionRule.owner_user_id == owner_user_id).with_for_update() - ) - - for rule in result.scalars(): - endpoint_ids = rule.endpoint_ids or [] - remaining_endpoint_ids = [value for value in endpoint_ids if value != str(endpoint_id)] - indexer_deleted = remaining_endpoint_ids != endpoint_ids - download_client_deleted = rule.download_client_id == endpoint_id - - if indexer_deleted: - rule.endpoint_ids = remaining_endpoint_ids - - if download_client_deleted: - rule.download_client_id = None - - if download_client_deleted or (indexer_deleted and not remaining_endpoint_ids): - rule.enabled = False - - await session.flush() - - await session.delete(endpoint) - await session.commit() - - -async def build_test_endpoint( - session: AsyncSession, - owner_user_id: Any, - request: AcquisitionEndpointTest, -) -> AcquisitionEndpoint: - stored_endpoint = None - stored_credentials: dict[str, str] = {} - - if request.endpoint_id is not None: - stored_endpoint = await owned_endpoint(session, owner_user_id, request.endpoint_id) - stored_credentials = _credentials(stored_endpoint) - - credentials = dict(stored_credentials) - for field in ("api_key", "username", "password"): - value = getattr(request, field) - if value is not None: - credentials[field] = value.get_secret_value() - - if request.kind is not None: - kind = request.kind.value - elif stored_endpoint is not None: - kind = stored_endpoint.kind - else: - raise HTTPException(status_code=422, detail="Endpoint kind is required") - - if request.base_url is not None: - base_url = str(request.base_url) - elif stored_endpoint is not None: - base_url = stored_endpoint.base_url - else: - raise HTTPException(status_code=422, detail="Endpoint URL is required") - - encrypted_credentials = {"encrypted": encrypt_secret_payload(credentials)} if credentials else None - - return AcquisitionEndpoint( - owner_user_id=owner_user_id, - name=stored_endpoint.name if stored_endpoint is not None else "Connection test", - kind=kind, - base_url=base_url, - credentials=encrypted_credentials, - settings=stored_endpoint.settings if stored_endpoint is not None else None, - ) - - -async def test_endpoint_connection(endpoint: AcquisitionEndpoint) -> None: - credentials = _credentials(endpoint) - - if endpoint.kind == "prowlarr": - response_status, _, payload = await _request( - _url(endpoint, "api/v1/system/status"), - headers={"X-Api-Key": credentials.get("api_key", "")}, - ) - if response_status >= 400: - raise HTTPException(status_code=502, detail="Prowlarr connection test failed") - - _json_object(payload, "Prowlarr") - return - - if endpoint.kind == "torznab": - params = urlencode({"t": "caps", "apikey": credentials.get("api_key", "")}) - response_status, _, payload = await _request(_url(endpoint, f"api?{params}")) - if response_status >= 400: - raise HTTPException(status_code=502, detail="Torznab connection test failed") - - try: - ElementTree.fromstring(payload) - except ElementTree.ParseError as exc: - raise HTTPException(status_code=502, detail="Torznab returned invalid XML") from exc - return - - if endpoint.kind in {"readarr", "sonarr", "radarr", "lidarr", "whisparr"}: - response_status, _, payload = await _request( - _url(endpoint, "api/v3/system/status"), - headers={"X-Api-Key": credentials.get("api_key", "")}, - ) - if response_status >= 400: - raise HTTPException(status_code=502, detail=f"{endpoint.kind.title()} connection test failed") - - _json_object(payload, endpoint.kind.title()) - return - - if endpoint.kind == "qbittorrent": - login = urlencode( - {"username": credentials.get("username", ""), "password": credentials.get("password", "")} - ).encode() - response_status, _, payload = await _request( - _url(endpoint, "api/v2/auth/login"), - method="POST", - headers={"Content-Type": "application/x-www-form-urlencoded"}, - body=login, - ) - if not _qbittorrent_login_succeeded(response_status, payload): - raise HTTPException(status_code=502, detail="qBittorrent authentication failed") - return - - if endpoint.kind == "transmission": - body = json.dumps({"method": "session-get", "arguments": {}}).encode() - headers = {"Content-Type": "application/json"} - if credentials.get("username"): - token = base64.b64encode(f"{credentials['username']}:{credentials.get('password', '')}".encode()).decode() - headers["Authorization"] = f"Basic {token}" - - response_status, response_headers, payload = await _request( - _url(endpoint, "transmission/rpc"), - method="POST", - headers=headers, - body=body, - ) - if response_status == 409: - headers["X-Transmission-Session-Id"] = response_headers.get("X-Transmission-Session-Id", "") - response_status, _, payload = await _request( - _url(endpoint, "transmission/rpc"), - method="POST", - headers=headers, - body=body, - ) - if response_status >= 400 or _json_object(payload, "Transmission").get("result") != "success": - raise HTTPException(status_code=502, detail="Transmission connection test failed") - return - - if endpoint.kind == "deluge": - login = json.dumps({"method": "auth.login", "params": [credentials.get("password", "")], "id": 1}).encode() - response_status, _, payload = await _request( - _url(endpoint, "json"), - method="POST", - headers={"Content-Type": "application/json"}, - body=login, - ) - if response_status >= 400: - raise HTTPException(status_code=502, detail="Deluge authentication failed") - - try: - _require_deluge_result(payload) - except HTTPException as exc: - raise HTTPException(status_code=502, detail="Deluge authentication failed") from exc - return - - raise HTTPException(status_code=422, detail="Endpoint kind is not supported") - - -async def run_rule(session: AsyncSession, rule: AcquisitionRule) -> list[AcquisitionJob]: - """Run one rule once; callers may schedule this from their worker/cron service.""" - client = await owned_endpoint(session, rule.owner_user_id, rule.download_client_id) - if client.kind in {"readarr", "sonarr", "radarr", "lidarr", "whisparr"}: - filters = rule.filters or {} - command = filters.get("arr_command") - ids = filters.get("arr_ids", []) - if ( - not isinstance(command, str) - or not isinstance(ids, list) - or not all(isinstance(value, int) for value in ids) - ): - raise HTTPException( - status_code=422, - detail="Arr rules require filters.arr_command and filters.arr_ids", - ) - job = AcquisitionJob( - owner_user_id=rule.owner_user_id, - endpoint_id=client.endpoint_id, - rule_id=rule.rule_id, - title=command, - download_url=f"arr-command:{command}", - ) - session.add(job) - try: - job.client_reference = await dispatch_arr_command(client, command, ids) - job.status = "submitted" - except HTTPException as exc: - job.status = "failed" - job.error = str(exc.detail) - rule.last_run_at = datetime.now(UTC) - await session.commit() - return [job] - - endpoint_ids = rule.endpoint_ids or [] - result = await session.execute( - select(AcquisitionEndpoint).where( - AcquisitionEndpoint.owner_user_id == rule.owner_user_id, - AcquisitionEndpoint.endpoint_id.in_(endpoint_ids), - AcquisitionEndpoint.enabled.is_(True), - ) - ) - releases = [release for endpoint in result.scalars() for release in await search_endpoint(endpoint, rule.query)] - if not releases: - rule.last_run_at = datetime.now(UTC) - await session.commit() - return [] - selected = sorted(releases, key=lambda release: (release.seeders or 0, release.size_bytes or 0), reverse=True)[0] - job = AcquisitionJob( - owner_user_id=rule.owner_user_id, - endpoint_id=client.endpoint_id, - rule_id=rule.rule_id, - title=selected.title, - download_url=selected.download_url, - ) - session.add(job) - try: - job.client_reference = await submit_to_client(client, selected.download_url, None, None) - job.status = "submitted" - except HTTPException as exc: - job.status = "failed" - job.error = str(exc.detail) - rule.last_run_at = datetime.now(UTC) - await session.commit() - return [job] - - -async def run_enabled_rules(session: AsyncSession) -> None: - """Run enabled rules, isolating a failed remote integration from the others.""" - result = await session.execute(select(AcquisitionRule).where(AcquisitionRule.enabled.is_(True))) - for rule in result.scalars(): - try: - await run_rule(session, rule) - except HTTPException: - await session.rollback() diff --git a/papyrus/services/acquisition/__init__.py b/papyrus/services/acquisition/__init__.py new file mode 100644 index 0000000..59be792 --- /dev/null +++ b/papyrus/services/acquisition/__init__.py @@ -0,0 +1,62 @@ +"""Stable acquisition service imports; implementations live in focused modules.""" + +from .jobs import ( + build_test_endpoint, + cancel_job, + delete_acquisition_endpoint, + delete_terminal_job, + job_file_candidates, + owned_endpoint, + owned_job, + paginated_jobs, + retry_job_import, + run_enabled_rules, + run_rule, + select_job_file, + submit_release_batch, +) +from .providers import ( + QbittorrentClient, + dispatch_arr_command, + search_endpoint, + submit_to_client, + test_endpoint_connection, +) +from .release_tokens import create_release_token, decode_release_token +from .types import ( + BatchSubmissionResult, + JobFileCandidate, + QbittorrentFile, + QbittorrentTorrent, + ReleaseCandidate, + ReleaseTokenPayload, +) + +__all__ = [ + "ReleaseCandidate", + "ReleaseTokenPayload", + "BatchSubmissionResult", + "QbittorrentTorrent", + "QbittorrentFile", + "JobFileCandidate", + "create_release_token", + "decode_release_token", + "QbittorrentClient", + "search_endpoint", + "submit_to_client", + "dispatch_arr_command", + "test_endpoint_connection", + "owned_endpoint", + "owned_job", + "paginated_jobs", + "job_file_candidates", + "select_job_file", + "cancel_job", + "delete_terminal_job", + "retry_job_import", + "submit_release_batch", + "delete_acquisition_endpoint", + "build_test_endpoint", + "run_rule", + "run_enabled_rules", +] diff --git a/papyrus/services/acquisition/jobs.py b/papyrus/services/acquisition/jobs.py new file mode 100644 index 0000000..815452d --- /dev/null +++ b/papyrus/services/acquisition/jobs.py @@ -0,0 +1,542 @@ +"""Owner-scoped acquisition job lifecycle and rules.""" + +from __future__ import annotations + +from datetime import UTC, datetime, timedelta +from pathlib import PurePosixPath +from typing import Any +from uuid import UUID, uuid4 + +from fastapi import HTTPException, status +from sqlalchemy import func, select +from sqlalchemy.ext.asyncio import AsyncSession + +from papyrus.config import get_settings +from papyrus.core.security import encrypt_secret_payload +from papyrus.models.acquisition import AcquisitionEndpoint, AcquisitionJob, AcquisitionRule +from papyrus.models.sync import SyncBook +from papyrus.schemas.acquisition import AcquisitionEndpointTest +from papyrus.services.media import BOOK_EXTENSIONS + +from .providers import QbittorrentClient, dispatch_arr_command, search_endpoint, submit_to_client +from .release_tokens import decode_release_token +from .transport import _credentials +from .types import BatchSubmissionResult, JobFileCandidate + +SUBMISSION_LEASE_LIFETIME = timedelta(minutes=5) + + +async def owned_endpoint(session: AsyncSession, owner_user_id: Any, endpoint_id: Any) -> AcquisitionEndpoint: + result = await session.execute( + select(AcquisitionEndpoint).where( + AcquisitionEndpoint.endpoint_id == endpoint_id, AcquisitionEndpoint.owner_user_id == owner_user_id + ) + ) + endpoint = result.scalar_one_or_none() + if endpoint is None: + raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Acquisition endpoint not found") + return endpoint + + +async def owned_job( + session: AsyncSession, + owner_user_id: UUID, + job_id: UUID, + *, + for_update: bool = False, +) -> AcquisitionJob: + statement = select(AcquisitionJob).where( + AcquisitionJob.job_id == job_id, + AcquisitionJob.owner_user_id == owner_user_id, + ) + + if for_update: + statement = statement.with_for_update() + + result = await session.execute(statement.execution_options(populate_existing=for_update)) + job = result.scalar_one_or_none() + + if job is None: + raise HTTPException(status_code=404, detail="Acquisition job not found") + + return job + + +async def paginated_jobs( + session: AsyncSession, + owner_user_id: UUID, + *, + limit: int, + offset: int, +) -> tuple[list[AcquisitionJob], int]: + total = await session.scalar( + select(func.count()).select_from(AcquisitionJob).where(AcquisitionJob.owner_user_id == owner_user_id) + ) + result = await session.execute( + select(AcquisitionJob) + .where(AcquisitionJob.owner_user_id == owner_user_id) + .order_by(AcquisitionJob.created_at.desc(), AcquisitionJob.job_id.desc()) + .limit(limit) + .offset(offset) + ) + + return list(result.scalars()), total or 0 + + +async def job_file_candidates( + session: AsyncSession, + owner_user_id: UUID, + job_id: UUID, +) -> list[JobFileCandidate]: + job = await owned_job(session, owner_user_id, job_id) + + if job.status != "needs_file_selection": + raise HTTPException(status_code=409, detail="Acquisition job does not need file selection") + + if job.endpoint_id is None or job.client_hash is None: + raise HTTPException(status_code=409, detail="Acquisition job is missing its qBittorrent reference") + + endpoint = await owned_endpoint(session, owner_user_id, job.endpoint_id) + client = await QbittorrentClient.connect(endpoint) + files = await client.files(job.client_hash) + + return [ + JobFileCandidate( + index=file.index, + name=file.name, + size_bytes=file.size_bytes, + progress_basis_points=file.progress_basis_points, + priority=file.priority, + supported=_supported_book_file(file.name), + ) + for file in files + ] + + +def _supported_book_file(filename: str) -> bool: + normalized = filename.replace("\\", "/") + extension = PurePosixPath(normalized).suffix.lower().lstrip(".") + return extension in BOOK_EXTENSIONS + + +async def select_job_file( + session: AsyncSession, + owner_user_id: UUID, + job_id: UUID, + file_index: int, +) -> AcquisitionJob: + job = await owned_job( + session, + owner_user_id, + job_id, + for_update=True, + ) + + if job.status != "needs_file_selection": + raise HTTPException(status_code=409, detail="Acquisition job does not need file selection") + + if job.endpoint_id is None or job.client_hash is None: + raise HTTPException(status_code=409, detail="Acquisition job is missing its qBittorrent reference") + + endpoint = await owned_endpoint(session, owner_user_id, job.endpoint_id) + client = await QbittorrentClient.connect(endpoint) + files = await client.files(job.client_hash) + selected = next((file for file in files if file.index == file_index), None) + + if selected is None: + raise HTTPException(status_code=404, detail="qBittorrent file not found") + + if not _supported_book_file(selected.name): + raise HTTPException(status_code=422, detail="Selected file is not a supported book") + + await client.select_file( + job.client_hash, + selected_index=selected.index, + file_indices=[file.index for file in files], + ) + + job.selected_file_path = selected.name + job.status = "downloading" + job.error = None + job.next_poll_at = datetime.now(UTC) + job.updated_at = datetime.now(UTC) + + await session.commit() + await session.refresh(job) + return job + + +async def cancel_job( + session: AsyncSession, + owner_user_id: UUID, + job_id: UUID, +) -> AcquisitionJob: + job = await owned_job( + session, + owner_user_id, + job_id, + for_update=True, + ) + + if job.status == "cancelled": + return job + + if job.status not in {"queued", "submitted", "downloading", "needs_file_selection"}: + raise HTTPException(status_code=409, detail="Acquisition job cannot be cancelled") + + if job.endpoint_id is None: + raise HTTPException(status_code=409, detail="Acquisition job is missing its qBittorrent endpoint") + + endpoint = await owned_endpoint(session, owner_user_id, job.endpoint_id) + client = await QbittorrentClient.connect(endpoint) + torrent_hash = job.client_hash + + if torrent_hash is None: + try: + torrent_hash = (await client.find_torrent(tag=f"papyrus:{job.job_id}")).hash + except HTTPException as exc: + if exc.status_code != 404: + raise + + if torrent_hash is not None: + await client.delete_torrent(torrent_hash) + + now = datetime.now(UTC) + job.status = "cancelled" + job.cancelled_at = now + job.updated_at = now + job.next_poll_at = None + job.error = None + + await session.commit() + await session.refresh(job) + return job + + +async def delete_terminal_job( + session: AsyncSession, + owner_user_id: UUID, + job_id: UUID, +) -> None: + job = await owned_job(session, owner_user_id, job_id) + + if job.status not in {"failed", "cancelled"}: + raise HTTPException(status_code=409, detail="Only failed or cancelled jobs can be removed") + + book = None + + if job.book_id is not None: + result = await session.execute( + select(SyncBook) + .where( + SyncBook.book_id == job.book_id, + SyncBook.owner_user_id == owner_user_id, + ) + .with_for_update() + ) + book = result.scalar_one_or_none() + + if book is not None and book.file_media_id is not None: + raise HTTPException(status_code=409, detail="Imported books must be removed from the library") + + await session.delete(job) + await session.flush() + + if book is not None: + await session.delete(book) + + await session.commit() + + +async def retry_job_import( + session: AsyncSession, + owner_user_id: UUID, + job_id: UUID, +) -> AcquisitionJob: + job = await owned_job( + session, + owner_user_id, + job_id, + for_update=True, + ) + + if job.status != "failed": + raise HTTPException(status_code=409, detail="Only failed jobs can retry import") + + if job.endpoint_id is None: + raise HTTPException(status_code=409, detail="Acquisition job cannot resume without qBittorrent") + + if job.submitted_at is None: + raise HTTPException(status_code=409, detail="Acquisition job was not submitted to qBittorrent") + + now = datetime.now(UTC) + job.status = "downloading" + job.retry_count += 1 + job.error = None + job.next_poll_at = now + job.updated_at = now + + await session.commit() + await session.refresh(job) + return job + + +async def submit_release_batch( + session: AsyncSession, + owner_user_id: UUID, + endpoint_id: UUID, + release_tokens: list[str], +) -> list[BatchSubmissionResult]: + endpoint = await owned_endpoint(session, owner_user_id, endpoint_id) + + if endpoint.kind != "qbittorrent": + raise HTTPException(status_code=422, detail="Managed downloads require qBittorrent") + + if not endpoint.enabled: + raise HTTPException(status_code=409, detail="Download client is disabled") + + if endpoint.download_root is None: + raise HTTPException(status_code=409, detail="qBittorrent download root is not configured") + + if get_settings().acquisition_import_root is None: + raise HTTPException(status_code=409, detail="Acquisition import root is not configured") + + results: list[BatchSubmissionResult] = [] + + for index, token in enumerate(release_tokens): + try: + release = decode_release_token(token, owner_user_id) + except HTTPException as exc: + results.append(BatchSubmissionResult(index=index, job=None, error=str(exc.detail))) + continue + + if release.protocol != "torrent" or not release.download_url.startswith(("magnet:", "http://", "https://")): + results.append(BatchSubmissionResult(index=index, job=None, error="Release token is invalid or expired")) + continue + + book_id = uuid4() + job_id = uuid4() + book = SyncBook( + book_id=book_id, + owner_user_id=owner_user_id, + title=release.title, + custom_metadata={ + "acquisition": { + "job_id": str(job_id), + "provisional": True, + } + }, + ) + job = AcquisitionJob( + job_id=job_id, + owner_user_id=owner_user_id, + endpoint_id=endpoint.endpoint_id, + book_id=book_id, + title=release.title, + download_url=None, + status="queued", + next_poll_at=datetime.now(UTC), + lease_owner=f"submission:{job_id}", + lease_until=datetime.now(UTC) + SUBMISSION_LEASE_LIFETIME, + ) + + session.add(book) + await session.flush() + + session.add(job) + await session.flush() + + await session.commit() + await session.refresh(job) + + try: + job.client_reference = await submit_to_client( + endpoint, + release.download_url, + "papyrus", + _managed_download_path(endpoint.download_root, owner_user_id, job_id), + tags=[f"papyrus:{job_id}"], + ) + job.status = "submitted" + job.submitted_at = datetime.now(UTC) + job.next_poll_at = datetime.now(UTC) + except HTTPException as exc: + job.status = "failed" + job.error = str(exc.detail) + job.next_poll_at = None + + job.lease_owner = None + job.lease_until = None + + await session.commit() + await session.refresh(job) + results.append(BatchSubmissionResult(index=index, job=job, error=None)) + + return results + + +def _managed_download_path(download_root: str, owner_user_id: UUID, job_id: UUID) -> str: + normalized_root = download_root.rstrip("/\\") + separator = "\\" if "\\" in normalized_root and "/" not in normalized_root else "/" + return separator.join((normalized_root, str(owner_user_id), str(job_id))) + + +async def delete_acquisition_endpoint(session: AsyncSession, owner_user_id: Any, endpoint_id: Any) -> None: + endpoint = await owned_endpoint(session, owner_user_id, endpoint_id) + active_job_id = await session.scalar( + select(AcquisitionJob.job_id) + .where( + AcquisitionJob.owner_user_id == owner_user_id, + AcquisitionJob.endpoint_id == endpoint_id, + AcquisitionJob.status.not_in({"completed", "failed", "cancelled"}), + ) + .limit(1) + ) + + if active_job_id is not None: + raise HTTPException(status_code=409, detail="Endpoint has active acquisition jobs") + + result = await session.execute( + select(AcquisitionRule).where(AcquisitionRule.owner_user_id == owner_user_id).with_for_update() + ) + + for rule in result.scalars(): + endpoint_ids = rule.endpoint_ids or [] + remaining_endpoint_ids = [value for value in endpoint_ids if value != str(endpoint_id)] + indexer_deleted = remaining_endpoint_ids != endpoint_ids + download_client_deleted = rule.download_client_id == endpoint_id + + if indexer_deleted: + rule.endpoint_ids = remaining_endpoint_ids + + if download_client_deleted: + rule.download_client_id = None + + if download_client_deleted or (indexer_deleted and not remaining_endpoint_ids): + rule.enabled = False + + await session.flush() + + await session.delete(endpoint) + await session.commit() + + +async def build_test_endpoint( + session: AsyncSession, + owner_user_id: Any, + request: AcquisitionEndpointTest, +) -> AcquisitionEndpoint: + stored_endpoint = None + stored_credentials: dict[str, str] = {} + + if request.endpoint_id is not None: + stored_endpoint = await owned_endpoint(session, owner_user_id, request.endpoint_id) + stored_credentials = _credentials(stored_endpoint) + + credentials = dict(stored_credentials) + for field in ("api_key", "username", "password"): + value = getattr(request, field) + if value is not None: + credentials[field] = value.get_secret_value() + + if request.kind is not None: + kind = request.kind.value + elif stored_endpoint is not None: + kind = stored_endpoint.kind + else: + raise HTTPException(status_code=422, detail="Endpoint kind is required") + + if request.base_url is not None: + base_url = str(request.base_url) + elif stored_endpoint is not None: + base_url = stored_endpoint.base_url + else: + raise HTTPException(status_code=422, detail="Endpoint URL is required") + + encrypted_credentials = {"encrypted": encrypt_secret_payload(credentials)} if credentials else None + + return AcquisitionEndpoint( + owner_user_id=owner_user_id, + name=stored_endpoint.name if stored_endpoint is not None else "Connection test", + kind=kind, + base_url=base_url, + credentials=encrypted_credentials, + settings=stored_endpoint.settings if stored_endpoint is not None else None, + ) + + +async def run_rule(session: AsyncSession, rule: AcquisitionRule) -> list[AcquisitionJob]: + """Run one rule once; callers may schedule this from their worker/cron service.""" + client = await owned_endpoint(session, rule.owner_user_id, rule.download_client_id) + if client.kind in {"readarr", "sonarr", "radarr", "lidarr", "whisparr"}: + filters = rule.filters or {} + command = filters.get("arr_command") + ids = filters.get("arr_ids", []) + if ( + not isinstance(command, str) + or not isinstance(ids, list) + or not all(isinstance(value, int) for value in ids) + ): + raise HTTPException( + status_code=422, + detail="Arr rules require filters.arr_command and filters.arr_ids", + ) + job = AcquisitionJob( + owner_user_id=rule.owner_user_id, + endpoint_id=client.endpoint_id, + rule_id=rule.rule_id, + title=command, + download_url=f"arr-command:{command}", + ) + session.add(job) + try: + job.client_reference = await dispatch_arr_command(client, command, ids) + job.status = "submitted" + except HTTPException as exc: + job.status = "failed" + job.error = str(exc.detail) + rule.last_run_at = datetime.now(UTC) + await session.commit() + return [job] + + endpoint_ids = rule.endpoint_ids or [] + result = await session.execute( + select(AcquisitionEndpoint).where( + AcquisitionEndpoint.owner_user_id == rule.owner_user_id, + AcquisitionEndpoint.endpoint_id.in_(endpoint_ids), + AcquisitionEndpoint.enabled.is_(True), + ) + ) + releases = [release for endpoint in result.scalars() for release in await search_endpoint(endpoint, rule.query)] + if not releases: + rule.last_run_at = datetime.now(UTC) + await session.commit() + return [] + selected = sorted(releases, key=lambda release: (release.seeders or 0, release.size_bytes or 0), reverse=True)[0] + job = AcquisitionJob( + owner_user_id=rule.owner_user_id, + endpoint_id=client.endpoint_id, + rule_id=rule.rule_id, + title=selected.title, + download_url=selected.download_url, + ) + session.add(job) + try: + job.client_reference = await submit_to_client(client, selected.download_url, None, None) + job.status = "submitted" + except HTTPException as exc: + job.status = "failed" + job.error = str(exc.detail) + rule.last_run_at = datetime.now(UTC) + await session.commit() + return [job] + + +async def run_enabled_rules(session: AsyncSession) -> None: + """Run enabled rules, isolating a failed remote integration from the others.""" + result = await session.execute(select(AcquisitionRule).where(AcquisitionRule.enabled.is_(True))) + for rule in result.scalars(): + try: + await run_rule(session, rule) + except HTTPException: + await session.rollback() diff --git a/papyrus/services/acquisition/providers.py b/papyrus/services/acquisition/providers.py new file mode 100644 index 0000000..1aa3dfd --- /dev/null +++ b/papyrus/services/acquisition/providers.py @@ -0,0 +1,588 @@ +"""Indexer and download-client protocol adapters.""" + +from __future__ import annotations + +import base64 +import json +import re +from dataclasses import dataclass +from typing import Self +from urllib.parse import urlencode +from xml.etree import ElementTree + +from fastapi import HTTPException + +from papyrus.models.acquisition import AcquisitionEndpoint + +from .transport import ( + _credentials, + _header_value, + _json_array, + _json_object, + _qbittorrent_login_succeeded, + _request, + _require_deluge_result, + _url, +) +from .types import QbittorrentFile, QbittorrentTorrent, ReleaseCandidate + + +@dataclass(slots=True) +class QbittorrentClient: + endpoint: AcquisitionEndpoint + cookie: str + + @classmethod + async def connect(cls, endpoint: AcquisitionEndpoint) -> Self: + if endpoint.kind != "qbittorrent": + raise HTTPException(status_code=422, detail="Endpoint is not qBittorrent") + + credentials = _credentials(endpoint) + login = urlencode( + { + "username": credentials.get("username", ""), + "password": credentials.get("password", ""), + } + ).encode() + response_status, headers, response_payload = await _request( + _url(endpoint, "api/v2/auth/login"), + method="POST", + headers={"Content-Type": "application/x-www-form-urlencoded"}, + body=login, + ) + + if not _qbittorrent_login_succeeded(response_status, response_payload): + raise HTTPException(status_code=502, detail="qBittorrent authentication failed") + + return cls( + endpoint=endpoint, + cookie=_header_value(headers, "Set-Cookie").split(";", 1)[0], + ) + + async def find_torrent( + self, + *, + tag: str, + torrent_hash: str | None = None, + ) -> QbittorrentTorrent: + payload: list[dict[str, object]] = [] + + if torrent_hash is not None: + payload = await self._get_json_array( + "api/v2/torrents/info", + {"hashes": torrent_hash}, + ) + + if not payload: + payload = await self._get_json_array("api/v2/torrents/info", {"tag": tag}) + + if not payload: + raise HTTPException(status_code=404, detail="qBittorrent torrent not found") + + if len(payload) != 1: + raise HTTPException(status_code=409, detail="qBittorrent tag matched multiple torrents") + + item = payload[0] + progress_basis_points = _progress_basis_points(item.get("progress")) + completed_bytes = _optional_int(item.get("completed")) + + return QbittorrentTorrent( + hash=_required_string(item, "hash", "qBittorrent torrent"), + state=_required_string(item, "state", "qBittorrent torrent"), + progress_basis_points=progress_basis_points, + downloaded_bytes=completed_bytes + if completed_bytes is not None + else _required_int(item, "downloaded", "qBittorrent torrent"), + total_bytes=_required_int(item, "total_size", "qBittorrent torrent"), + download_speed_bytes_per_second=_required_int(item, "dlspeed", "qBittorrent torrent"), + eta_seconds=0 if progress_basis_points == 10_000 else _optional_int(item.get("eta")), + ) + + async def files(self, torrent_hash: str) -> list[QbittorrentFile]: + payload = await self._get_json_array("api/v2/torrents/files", {"hash": torrent_hash}) + return [ + QbittorrentFile( + index=_required_int(item, "index", "qBittorrent file"), + name=_required_string(item, "name", "qBittorrent file"), + size_bytes=_required_int(item, "size", "qBittorrent file"), + progress_basis_points=_progress_basis_points(item.get("progress")), + priority=_required_int(item, "priority", "qBittorrent file"), + ) + for item in payload + ] + + async def select_file(self, torrent_hash: str, *, selected_index: int, file_indices: list[int]) -> None: + if selected_index not in file_indices: + raise HTTPException(status_code=422, detail="Selected qBittorrent file was not found") + + other_indices = [index for index in file_indices if index != selected_index] + + await self._post_form( + "api/v2/torrents/pause", + {"hashes": torrent_hash}, + fallback_path="api/v2/torrents/stop", + ) + + if other_indices: + await self._post_form( + "api/v2/torrents/filePrio", + { + "hash": torrent_hash, + "id": "|".join(str(index) for index in other_indices), + "priority": "0", + }, + ) + + await self._post_form( + "api/v2/torrents/filePrio", + { + "hash": torrent_hash, + "id": str(selected_index), + "priority": "1", + }, + ) + await self._post_form( + "api/v2/torrents/resume", + {"hashes": torrent_hash}, + fallback_path="api/v2/torrents/start", + ) + + async def pause(self, torrent_hash: str) -> None: + await self._post_form( + "api/v2/torrents/pause", + {"hashes": torrent_hash}, + fallback_path="api/v2/torrents/stop", + ) + + async def delete_torrent(self, torrent_hash: str) -> None: + await self._post_form( + "api/v2/torrents/delete", + { + "hashes": torrent_hash, + "deleteFiles": "true", + }, + ) + + async def _get_json_array(self, path: str, params: dict[str, str]) -> list[dict[str, object]]: + response_status, _, payload = await _request( + _url(self.endpoint, f"{path}?{urlencode(params)}"), + headers={"Cookie": self.cookie}, + ) + + if response_status >= 400: + raise HTTPException(status_code=502, detail="qBittorrent request failed") + + return _json_array(payload, "qBittorrent") + + async def _post_form( + self, + path: str, + values: dict[str, str], + *, + fallback_path: str | None = None, + ) -> None: + response_status, _, _ = await _request( + _url(self.endpoint, path), + method="POST", + headers={ + "Content-Type": "application/x-www-form-urlencoded", + "Cookie": self.cookie, + }, + body=urlencode(values).encode(), + ) + + if response_status == 404 and fallback_path is not None: + response_status, _, _ = await _request( + _url(self.endpoint, fallback_path), + method="POST", + headers={ + "Content-Type": "application/x-www-form-urlencoded", + "Cookie": self.cookie, + }, + body=urlencode(values).encode(), + ) + + if response_status >= 400: + raise HTTPException(status_code=502, detail="qBittorrent request failed") + + +def _required_string(item: dict[str, object], key: str, subject: str) -> str: + value = item.get(key) + + if not isinstance(value, str) or not value: + raise HTTPException(status_code=502, detail=f"{subject} returned invalid data") + + return value + + +def _required_int(item: dict[str, object], key: str, subject: str) -> int: + value = item.get(key) + + if not isinstance(value, int) or isinstance(value, bool): + raise HTTPException(status_code=502, detail=f"{subject} returned invalid data") + + return value + + +def _progress_basis_points(value: object) -> int: + if not isinstance(value, int | float) or isinstance(value, bool): + raise HTTPException(status_code=502, detail="qBittorrent returned invalid progress") + + return min(10_000, max(0, round(float(value) * 10_000))) + + +async def search_endpoint(endpoint: AcquisitionEndpoint, query: str) -> list[ReleaseCandidate]: + """Search Prowlarr or a Torznab-compatible torrent indexer.""" + credentials = _credentials(endpoint) + if endpoint.kind == "prowlarr": + request_url = _url(endpoint, f"api/v1/search?{urlencode({'query': query})}") + response_status, _, payload = await _request(request_url, headers={"X-Api-Key": credentials.get("api_key", "")}) + if response_status >= 400: + raise HTTPException(status_code=502, detail="Prowlarr search failed") + data = _json_array(payload, "Prowlarr") + releases: list[ReleaseCandidate] = [] + + for item in data: + download_url = item.get("downloadUrl") or item.get("magnetUrl") or item.get("guid") + protocol = item.get("protocol", "torrent") + + if not isinstance(download_url, str) or not download_url or protocol != "torrent": + continue + + title_value = item.get("title") + indexer_value = item.get("indexer") + title = title_value if isinstance(title_value, str) and title_value else "Untitled" + indexer = indexer_value if isinstance(indexer_value, str) and indexer_value else "Prowlarr" + + releases.append( + ReleaseCandidate( + title=title, + download_url=download_url, + protocol="torrent", + indexer=indexer, + size_bytes=_optional_int(item.get("size")), + seeders=_optional_int(item.get("seeders")), + publish_date=None, + format_hints=_format_hints(title, download_url), + ) + ) + + return releases + + params = urlencode({"t": "search", "q": query, "apikey": credentials.get("api_key", "")}) + response_status, _, payload = await _request(_url(endpoint, f"api?{params}")) + if response_status >= 400: + raise HTTPException(status_code=502, detail=f"{endpoint.kind.title()} search failed") + return _parse_torznab(payload, endpoint.name) + + +def _parse_torznab(payload: bytes, indexer: str) -> list[ReleaseCandidate]: + try: + root = ElementTree.fromstring(payload) + except ElementTree.ParseError as exc: + raise HTTPException(status_code=502, detail="Indexer returned invalid XML") from exc + releases: list[ReleaseCandidate] = [] + for item in root.findall(".//item"): + enclosure = item.find("enclosure") + link = (enclosure.get("url") if enclosure is not None else None) or item.findtext("link") + if not link: + continue + attrs = {child.attrib.get("name"): child.attrib.get("value") for child in item if child.tag.endswith("attr")} + size = attrs.get("size") + seeders = attrs.get("seeders") + releases.append( + ReleaseCandidate( + title=item.findtext("title") or "Untitled", + download_url=link, + protocol="torrent", + indexer=indexer, + size_bytes=int(size) if size is not None and size.isdigit() else None, + seeders=int(seeders) if seeders is not None and seeders.isdigit() else None, + publish_date=None, + format_hints=_format_hints(item.findtext("title") or "", link), + ) + ) + return releases + + +def _format_hints(title: str, download_url: str) -> list[str]: + searchable = f"{title} {download_url}".lower() + return [ + extension + for extension in ("epub", "pdf", "mobi", "azw3", "txt", "cbr", "cbz") + if re.search(rf"(? int | None: + return value if isinstance(value, int) and not isinstance(value, bool) else None + + +async def submit_to_client( + endpoint: AcquisitionEndpoint, + download_url: str, + category: str | None, + save_path: str | None, + *, + tags: list[str] | None = None, +) -> str | None: + """Submit a magnet or torrent URL to a supported BitTorrent client.""" + if endpoint.kind == "qbittorrent": + return await _submit_qbittorrent(endpoint, download_url, category, save_path, tags=tags) + if endpoint.kind == "transmission": + return await _submit_transmission(endpoint, download_url, save_path) + if endpoint.kind == "deluge": + return await _submit_deluge(endpoint, download_url, save_path) + raise HTTPException(status_code=422, detail="Endpoint is not a download client") + + +async def dispatch_arr_command(endpoint: AcquisitionEndpoint, command: str, ids: list[int]) -> str | None: + """Start an acquisition/search command in a Servarr application. + + The Arr apps own their library and release-grab decisions, so they must be + driven through their command API rather than handed an arbitrary magnet. + """ + if endpoint.kind not in {"readarr", "sonarr", "radarr", "lidarr", "whisparr"}: + raise HTTPException(status_code=422, detail="Endpoint is not a Servarr application") + + allowed_commands = { + "readarr": {"AuthorSearch", "BookSearch"}, + "sonarr": {"SeriesSearch", "EpisodeSearch", "MissingEpisodeSearch"}, + "radarr": {"MoviesSearch", "MissingMoviesSearch"}, + "lidarr": {"ArtistSearch", "AlbumSearch", "MissingAlbumSearch"}, + "whisparr": {"SeriesSearch", "EpisodeSearch", "MissingEpisodeSearch"}, + } + if command not in allowed_commands[endpoint.kind]: + raise HTTPException(status_code=422, detail="Command is not supported by this Servarr application") + + id_field = { + "AuthorSearch": "authorIds", + "BookSearch": "bookIds", + "SeriesSearch": "seriesId", + "EpisodeSearch": "episodeIds", + "MissingEpisodeSearch": "seriesId", + "MoviesSearch": "movieIds", + "MissingMoviesSearch": "movieIds", + "ArtistSearch": "artistIds", + "AlbumSearch": "albumIds", + "MissingAlbumSearch": "artistIds", + }[command] + payload: dict[str, object] = {"name": command} + if ids: + payload[id_field] = ids[0] if id_field in {"seriesId"} else ids + + response_status, _, response_payload = await _request( + _url(endpoint, "api/v3/command"), + method="POST", + headers={"Content-Type": "application/json", "X-Api-Key": _credentials(endpoint).get("api_key", "")}, + body=json.dumps(payload).encode(), + ) + if response_status >= 400: + raise HTTPException(status_code=502, detail=f"{endpoint.kind.title()} command failed") + response = _json_object(response_payload, endpoint.kind.title()) + return str(response.get("id")) if response.get("id") is not None else None + + +async def _submit_qbittorrent( + endpoint: AcquisitionEndpoint, + download_url: str, + category: str | None, + save_path: str | None, + *, + tags: list[str] | None = None, +) -> str | None: + credentials = _credentials(endpoint) + login = urlencode( + {"username": credentials.get("username", ""), "password": credentials.get("password", "")} + ).encode() + response_status, headers, response_payload = await _request( + _url(endpoint, "api/v2/auth/login"), + method="POST", + headers={"Content-Type": "application/x-www-form-urlencoded"}, + body=login, + ) + if not _qbittorrent_login_succeeded(response_status, response_payload): + raise HTTPException(status_code=502, detail="qBittorrent authentication failed") + payload = {"urls": download_url} + if category: + payload["category"] = category + if save_path: + payload["savepath"] = save_path + if tags: + payload["tags"] = ",".join(tags) + response_status, _, _ = await _request( + _url(endpoint, "api/v2/torrents/add"), + method="POST", + headers={ + "Content-Type": "application/x-www-form-urlencoded", + "Cookie": _header_value(headers, "Set-Cookie").split(";", 1)[0], + }, + body=urlencode(payload).encode(), + ) + if response_status >= 400: + raise HTTPException(status_code=502, detail="qBittorrent rejected the release") + return None + + +async def _submit_transmission(endpoint: AcquisitionEndpoint, download_url: str, save_path: str | None) -> str | None: + credentials = _credentials(endpoint) + arguments: dict[str, str] = {"filename": download_url} + if save_path: + arguments["download-dir"] = save_path + body = json.dumps({"method": "torrent-add", "arguments": arguments}).encode() + headers = {"Content-Type": "application/json"} + if credentials.get("username"): + token = base64.b64encode(f"{credentials['username']}:{credentials.get('password', '')}".encode()).decode() + headers["Authorization"] = f"Basic {token}" + response_status, response_headers, payload = await _request( + _url(endpoint, "transmission/rpc"), method="POST", headers=headers, body=body + ) + if response_status == 409: + headers["X-Transmission-Session-Id"] = response_headers.get("X-Transmission-Session-Id", "") + response_status, _, payload = await _request( + _url(endpoint, "transmission/rpc"), method="POST", headers=headers, body=body + ) + if response_status >= 400: + raise HTTPException(status_code=502, detail="Transmission rejected the release") + + response = _json_object(payload, "Transmission") + if response.get("result") != "success": + raise HTTPException(status_code=502, detail="Transmission rejected the release") + + response_arguments = response.get("arguments") + if not isinstance(response_arguments, dict): + raise HTTPException(status_code=502, detail="Transmission returned an invalid response") + + torrent = response_arguments.get("torrent-added") or response_arguments.get("torrent-duplicate") + if not isinstance(torrent, dict): + return None + + reference = torrent.get("hashString") + return str(reference) if reference is not None else None + + +async def _submit_deluge(endpoint: AcquisitionEndpoint, download_url: str, save_path: str | None) -> str | None: + credentials = _credentials(endpoint) + headers = {"Content-Type": "application/json"} + login = json.dumps({"method": "auth.login", "params": [credentials.get("password", "")], "id": 1}).encode() + response_status, response_headers, login_payload = await _request( + _url(endpoint, "json"), method="POST", headers=headers, body=login + ) + if response_status >= 400: + raise HTTPException(status_code=502, detail="Deluge authentication failed") + + try: + _require_deluge_result(login_payload) + except HTTPException as exc: + raise HTTPException(status_code=502, detail="Deluge authentication failed") from exc + + options = {"download_location": save_path} if save_path else {} + method = "core.add_torrent_magnet" if download_url.startswith("magnet:") else "core.add_torrent_url" + body = json.dumps({"method": method, "params": [download_url, options], "id": 2}).encode() + headers["Cookie"] = response_headers.get("Set-Cookie", "").split(";", 1)[0] + response_status, _, payload = await _request(_url(endpoint, "json"), method="POST", headers=headers, body=body) + if response_status >= 400: + raise HTTPException(status_code=502, detail="Deluge rejected the release") + + try: + result = _require_deluge_result(payload) + except HTTPException as exc: + raise HTTPException(status_code=502, detail="Deluge rejected the release") from exc + + return str(result) + + +async def test_endpoint_connection(endpoint: AcquisitionEndpoint) -> None: + credentials = _credentials(endpoint) + + if endpoint.kind == "prowlarr": + response_status, _, payload = await _request( + _url(endpoint, "api/v1/system/status"), + headers={"X-Api-Key": credentials.get("api_key", "")}, + ) + if response_status >= 400: + raise HTTPException(status_code=502, detail="Prowlarr connection test failed") + + _json_object(payload, "Prowlarr") + return + + if endpoint.kind == "torznab": + params = urlencode({"t": "caps", "apikey": credentials.get("api_key", "")}) + response_status, _, payload = await _request(_url(endpoint, f"api?{params}")) + if response_status >= 400: + raise HTTPException(status_code=502, detail="Torznab connection test failed") + + try: + ElementTree.fromstring(payload) + except ElementTree.ParseError as exc: + raise HTTPException(status_code=502, detail="Torznab returned invalid XML") from exc + return + + if endpoint.kind in {"readarr", "sonarr", "radarr", "lidarr", "whisparr"}: + response_status, _, payload = await _request( + _url(endpoint, "api/v3/system/status"), + headers={"X-Api-Key": credentials.get("api_key", "")}, + ) + if response_status >= 400: + raise HTTPException(status_code=502, detail=f"{endpoint.kind.title()} connection test failed") + + _json_object(payload, endpoint.kind.title()) + return + + if endpoint.kind == "qbittorrent": + login = urlencode( + {"username": credentials.get("username", ""), "password": credentials.get("password", "")} + ).encode() + response_status, _, payload = await _request( + _url(endpoint, "api/v2/auth/login"), + method="POST", + headers={"Content-Type": "application/x-www-form-urlencoded"}, + body=login, + ) + if not _qbittorrent_login_succeeded(response_status, payload): + raise HTTPException(status_code=502, detail="qBittorrent authentication failed") + return + + if endpoint.kind == "transmission": + body = json.dumps({"method": "session-get", "arguments": {}}).encode() + headers = {"Content-Type": "application/json"} + if credentials.get("username"): + token = base64.b64encode(f"{credentials['username']}:{credentials.get('password', '')}".encode()).decode() + headers["Authorization"] = f"Basic {token}" + + response_status, response_headers, payload = await _request( + _url(endpoint, "transmission/rpc"), + method="POST", + headers=headers, + body=body, + ) + if response_status == 409: + headers["X-Transmission-Session-Id"] = response_headers.get("X-Transmission-Session-Id", "") + response_status, _, payload = await _request( + _url(endpoint, "transmission/rpc"), + method="POST", + headers=headers, + body=body, + ) + if response_status >= 400 or _json_object(payload, "Transmission").get("result") != "success": + raise HTTPException(status_code=502, detail="Transmission connection test failed") + return + + if endpoint.kind == "deluge": + login = json.dumps({"method": "auth.login", "params": [credentials.get("password", "")], "id": 1}).encode() + response_status, _, payload = await _request( + _url(endpoint, "json"), + method="POST", + headers={"Content-Type": "application/json"}, + body=login, + ) + if response_status >= 400: + raise HTTPException(status_code=502, detail="Deluge authentication failed") + + try: + _require_deluge_result(payload) + except HTTPException as exc: + raise HTTPException(status_code=502, detail="Deluge authentication failed") from exc + return + + raise HTTPException(status_code=422, detail="Endpoint kind is not supported") diff --git a/papyrus/services/acquisition/release_tokens.py b/papyrus/services/acquisition/release_tokens.py new file mode 100644 index 0000000..eb52b18 --- /dev/null +++ b/papyrus/services/acquisition/release_tokens.py @@ -0,0 +1,84 @@ +"""Owner-scoped release selection tokens.""" + +from __future__ import annotations + +import json +from datetime import UTC, datetime, timedelta +from uuid import UUID + +from fastapi import HTTPException + +from papyrus.core.security import decrypt_secret_payload, encrypt_secret_payload +from papyrus.models.acquisition import AcquisitionEndpoint + +from .types import ReleaseCandidate, ReleaseTokenPayload + +RELEASE_TOKEN_LIFETIME = timedelta(minutes=5) + + +def create_release_token( + release: ReleaseCandidate, + endpoint: AcquisitionEndpoint, + *, + now: datetime | None = None, +) -> str: + issued_at = now or datetime.now(UTC) + + if endpoint.endpoint_id is None: + raise ValueError("Release tokens require a persisted endpoint") + + return encrypt_secret_payload( + { + "endpoint_id": str(endpoint.endpoint_id), + "owner_user_id": str(endpoint.owner_user_id), + "title": release.title, + "download_url": release.download_url, + "protocol": release.protocol, + "indexer": release.indexer, + "size_bytes": "" if release.size_bytes is None else str(release.size_bytes), + "seeders": "" if release.seeders is None else str(release.seeders), + "publish_date": release.publish_date.isoformat() if release.publish_date is not None else "", + "format_hints": json.dumps(release.format_hints), + "expires_at": str(int((issued_at + RELEASE_TOKEN_LIFETIME).timestamp())), + } + ) + + +def decode_release_token( + token: str, + owner_user_id: UUID, + *, + now: datetime | None = None, +) -> ReleaseTokenPayload: + try: + payload = decrypt_secret_payload(token) + expires_at = int(payload["expires_at"]) + token_owner_user_id = UUID(payload["owner_user_id"]) + endpoint_id = UUID(payload["endpoint_id"]) + format_hints = json.loads(payload["format_hints"]) + size_bytes = int(payload["size_bytes"]) if payload["size_bytes"] else None + seeders = int(payload["seeders"]) if payload["seeders"] else None + publish_date = datetime.fromisoformat(payload["publish_date"]) if payload["publish_date"] else None + + if ( + token_owner_user_id != owner_user_id + or expires_at <= int((now or datetime.now(UTC)).timestamp()) + or not isinstance(format_hints, list) + or not all(isinstance(value, str) for value in format_hints) + ): + raise ValueError + + return ReleaseTokenPayload( + endpoint_id=endpoint_id, + owner_user_id=token_owner_user_id, + title=payload["title"], + download_url=payload["download_url"], + protocol=payload["protocol"], + indexer=payload["indexer"], + size_bytes=size_bytes, + seeders=seeders, + publish_date=publish_date, + format_hints=format_hints, + ) + except (KeyError, TypeError, ValueError, json.JSONDecodeError) as exc: + raise HTTPException(status_code=400, detail="Release token is invalid or expired") from exc diff --git a/papyrus/services/acquisition/transport.py b/papyrus/services/acquisition/transport.py new file mode 100644 index 0000000..4759a7b --- /dev/null +++ b/papyrus/services/acquisition/transport.py @@ -0,0 +1,85 @@ +"""Bounded integration transport and credential decoding.""" + +from __future__ import annotations + +import asyncio +import json +from urllib.error import HTTPError, URLError +from urllib.parse import urljoin +from urllib.request import Request, urlopen + +from fastapi import HTTPException + +from papyrus.core.security import decrypt_secret_payload +from papyrus.models.acquisition import AcquisitionEndpoint + + +def _url(endpoint: AcquisitionEndpoint, path: str) -> str: + return urljoin(endpoint.base_url.rstrip("/") + "/", path.lstrip("/")) + + +async def _request( + url: str, *, method: str = "GET", headers: dict[str, str] | None = None, body: bytes | None = None +) -> tuple[int, dict[str, str], bytes]: + """Perform a bounded blocking HTTP request off the event loop.""" + + def send() -> tuple[int, dict[str, str], bytes]: + request = Request(url, data=body, headers=headers or {}, method=method) + try: + with urlopen(request, timeout=15) as response: # noqa: S310 - user-owned self-hosted integrations + return response.status, dict(response.headers.items()), response.read(5_000_000) + except HTTPError as exc: + return exc.code, dict(exc.headers.items()), exc.read(1_000_000) + except URLError as exc: + raise HTTPException(status_code=502, detail=f"Integration request failed: {exc.reason}") from exc + + return await asyncio.to_thread(send) + + +def _credentials(endpoint: AcquisitionEndpoint) -> dict[str, str]: + credentials = endpoint.credentials or {} + encrypted = credentials.get("encrypted") + if encrypted is None: + return credentials + try: + return decrypt_secret_payload(encrypted) + except ValueError as exc: + raise HTTPException(status_code=500, detail="Stored integration credentials are invalid") from exc + + +def _json_value(payload: bytes, integration: str) -> object: + try: + return json.loads(payload) + except (UnicodeDecodeError, json.JSONDecodeError) as exc: + raise HTTPException(status_code=502, detail=f"{integration} returned invalid JSON") from exc + + +def _json_object(payload: bytes, integration: str) -> dict[str, object]: + value = _json_value(payload, integration) + if not isinstance(value, dict): + raise HTTPException(status_code=502, detail=f"{integration} returned an invalid response") + return value + + +def _json_array(payload: bytes, integration: str) -> list[dict[str, object]]: + value = _json_value(payload, integration) + if not isinstance(value, list) or not all(isinstance(item, dict) for item in value): + raise HTTPException(status_code=502, detail=f"{integration} returned an invalid response") + return value + + +def _require_deluge_result(payload: bytes) -> object: + response = _json_object(payload, "Deluge") + result = response.get("result") + if response.get("error") is not None or result is None or result is False: + raise HTTPException(status_code=502, detail="Deluge rejected the request") + return result + + +def _header_value(headers: dict[str, str], name: str) -> str: + normalized_name = name.casefold() + return next((value for key, value in headers.items() if key.casefold() == normalized_name), "") + + +def _qbittorrent_login_succeeded(response_status: int, response_payload: bytes) -> bool: + return response_status < 400 and (response_status == 204 or response_payload.strip() == b"Ok.") diff --git a/papyrus/services/acquisition/types.py b/papyrus/services/acquisition/types.py new file mode 100644 index 0000000..6ef5362 --- /dev/null +++ b/papyrus/services/acquisition/types.py @@ -0,0 +1,72 @@ +"""Acquisition integration and job value types.""" + +from __future__ import annotations + +from dataclasses import dataclass +from datetime import datetime +from uuid import UUID + +from papyrus.models.acquisition import AcquisitionJob + + +@dataclass(frozen=True, slots=True) +class ReleaseCandidate: + title: str + download_url: str + protocol: str + indexer: str + size_bytes: int | None + seeders: int | None + publish_date: datetime | None + format_hints: list[str] + + +@dataclass(frozen=True, slots=True) +class ReleaseTokenPayload: + endpoint_id: UUID + owner_user_id: UUID + title: str + download_url: str + protocol: str + indexer: str + size_bytes: int | None + seeders: int | None + publish_date: datetime | None + format_hints: list[str] + + +@dataclass(frozen=True, slots=True) +class BatchSubmissionResult: + index: int + job: AcquisitionJob | None + error: str | None + + +@dataclass(frozen=True, slots=True) +class QbittorrentTorrent: + hash: str + state: str + progress_basis_points: int + downloaded_bytes: int + total_bytes: int + download_speed_bytes_per_second: int + eta_seconds: int | None + + +@dataclass(frozen=True, slots=True) +class QbittorrentFile: + index: int + name: str + size_bytes: int + progress_basis_points: int + priority: int + + +@dataclass(frozen=True, slots=True) +class JobFileCandidate: + index: int + name: str + size_bytes: int + progress_basis_points: int + priority: int + supported: bool diff --git a/scripts/export_openapi.py b/scripts/export_openapi.py new file mode 100644 index 0000000..1d3797c --- /dev/null +++ b/scripts/export_openapi.py @@ -0,0 +1,61 @@ +"""Export public API documentation without local settings, services or secrets.""" + +import argparse +import json +import os +from pathlib import Path +from unittest.mock import patch + +from papyrus.config import Settings + + +def export() -> str: + """Build the schema in a fresh process without starting the app lifespan.""" + with patch.dict(os.environ, {}, clear=True): + settings = Settings( + _env_file=None, + debug=False, + host="127.0.0.1", + port=8080, + api_prefix="/v1", + cors_origins=[], + postgres_user="docs", + postgres_password="docs", + postgres_host="127.0.0.1", + postgres_port=5432, + postgres_db="docs", + secret_key="documentation-placeholder", + algorithm="HS256", + access_token_expire_minutes=60, + refresh_token_expire_days=30, + rate_limit_auth=5, + rate_limit_general=100, + rate_limit_upload=10, + rate_limit_batch=20, + ) + + with patch("papyrus.config.get_settings", return_value=settings): + from papyrus.main import create_app + + schema = create_app().openapi() + + return json.dumps(schema, indent=2, sort_keys=True) + "\n" + + +def main() -> None: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("output", type=Path) + parser.add_argument("--check", action="store_true") + args = parser.parse_args() + content = export() + + if args.check: + if args.output.read_text() != content: + raise SystemExit("API snapshot is stale; regenerate with scripts/export_openapi.py") + + else: + args.output.write_text(content) + + +if __name__ == "__main__": + main() diff --git a/tests/api/routes/test_acquisition.py b/tests/api/routes/test_acquisition.py index 2cd3a1f..66af845 100644 --- a/tests/api/routes/test_acquisition.py +++ b/tests/api/routes/test_acquisition.py @@ -18,6 +18,7 @@ from papyrus.models.sync import SyncBook from papyrus.models.user import User from papyrus.services import acquisition as acquisition_service +from papyrus.services.acquisition import jobs as acquisition_jobs @pytest.fixture(autouse=True) @@ -309,7 +310,7 @@ async def submit_to_client( submissions.append((download_url, category, save_path, tags)) return None - monkeypatch.setattr(acquisition_service, "submit_to_client", submit_to_client) + monkeypatch.setattr(acquisition_jobs, "submit_to_client", submit_to_client) response = await client.post( "/v1/acquisition/submissions/batch", @@ -437,7 +438,7 @@ async def test_batch_submission_isolates_invalid_release_tokens( async def submit_to_client(*args: object, **kwargs: object) -> str | None: return None - monkeypatch.setattr(acquisition_service, "submit_to_client", submit_to_client) + monkeypatch.setattr(acquisition_jobs, "submit_to_client", submit_to_client) response = await client.post( "/v1/acquisition/submissions/batch", diff --git a/tests/api/routes/test_auth_sandbox.py b/tests/api/routes/test_auth_sandbox.py index fee33cb..6168f4d 100644 --- a/tests/api/routes/test_auth_sandbox.py +++ b/tests/api/routes/test_auth_sandbox.py @@ -47,14 +47,16 @@ async def test_auth_sandbox_renders_built_assets_when_manifest_exists( """Test the auth sandbox uses built assets when Vite mode is disabled.""" manifest_path = tmp_path / "manifest.json" manifest_path.write_text( - json.dumps({ - "src/pages/auth-sandbox/main.ts": { - "file": "assets/auth-sandbox.js", - "css": ["assets/auth-sandbox.css"], - "imports": [], - "src": "src/pages/auth-sandbox/main.ts", + json.dumps( + { + "src/pages/auth-sandbox/main.ts": { + "file": "assets/auth-sandbox.js", + "css": ["assets/auth-sandbox.css"], + "imports": [], + "src": "src/pages/auth-sandbox/main.ts", + } } - }), + ), encoding="utf-8", ) diff --git a/tests/api/routes/test_powersync_sandbox.py b/tests/api/routes/test_powersync_sandbox.py index f36d3da..4536d97 100644 --- a/tests/api/routes/test_powersync_sandbox.py +++ b/tests/api/routes/test_powersync_sandbox.py @@ -58,7 +58,7 @@ async def test_powersync_sandbox_registered_in_debug_vite_mode( assert 'data-shell-marker="sticky-status-rail"' in response.text assert 'href="/__dev/auth-sandbox"' in response.text assert 'href="/__dev/powersync-sandbox"' in response.text - assert 'dev-page-nav__link--active' in response.text + assert "dev-page-nav__link--active" in response.text assert 'src="http://vite.test:5173/@vite/client"' in response.text assert 'src="http://vite.test:5173/src/pages/powersync-sandbox/main.ts"' in response.text assert '"powersync_endpoint": "http://localhost:8081"' in response.text @@ -133,9 +133,7 @@ async def test_powersync_sandbox_worker_asset_is_served_from_backend_origin( assert response.headers["x-papyrus-vendor-path"] == "WASQLiteDB.umd.js" assert response.text == "// worker/WASQLiteDB.umd.js" - nested_response = await debug_client.get( - "/__dev/powersync-sandbox/worker/node_modules_pnpm_example_chunk.umd.js" - ) + nested_response = await debug_client.get("/__dev/powersync-sandbox/worker/node_modules_pnpm_example_chunk.umd.js") assert nested_response.status_code == 200 assert nested_response.headers["x-papyrus-vendor-path"] == "node_modules_pnpm_example_chunk.umd.js" assert nested_response.text == "// worker/node_modules_pnpm_example_chunk.umd.js" diff --git a/tests/integration/test_auth_smoke.py b/tests/integration/test_auth_smoke.py index 44e54b5..1a021a3 100644 --- a/tests/integration/test_auth_smoke.py +++ b/tests/integration/test_auth_smoke.py @@ -35,14 +35,14 @@ async def test_smtp_password_reset_smoke( async with test_session_maker() as session: from papyrus.models import User + session.add(User(display_name="SMTP Smoke", primary_email=recipient, primary_email_verified=True)) await session.commit() message = await auth_service.begin_password_reset(session, recipient) assert message == "If the email is registered, a reset link has been sent" -async def test_google_oauth_smoke( -): +async def test_google_oauth_smoke(): """Exercise a live Google-authenticated Papyrus session against the running server.""" if not _env_flag("RUN_GOOGLE_SMOKE_TEST"): pytest.skip("RUN_GOOGLE_SMOKE_TEST is not enabled") diff --git a/tests/services/test_acquisition.py b/tests/services/test_acquisition.py index dc81b55..6b1059c 100644 --- a/tests/services/test_acquisition.py +++ b/tests/services/test_acquisition.py @@ -11,6 +11,7 @@ from papyrus.models.acquisition import AcquisitionEndpoint from papyrus.services import acquisition +from papyrus.services.acquisition import providers def _endpoint(kind: str) -> AcquisitionEndpoint: @@ -82,7 +83,7 @@ async def test_transmission_rejects_rpc_failure(monkeypatch: pytest.MonkeyPatch) async def request(*args: object, **kwargs: object) -> tuple[int, dict[str, str], bytes]: return 200, {}, b'{"result":"invalid or corrupt torrent file","arguments":{}}' - monkeypatch.setattr(acquisition, "_request", request) + monkeypatch.setattr(providers, "_request", request) with pytest.raises(HTTPException) as exc_info: await acquisition.submit_to_client( @@ -106,7 +107,7 @@ async def request(*args: object, **kwargs: object) -> tuple[int, dict[str, str], return 200, {}, b'{"result":"torrent-id","error":null,"id":2}' - monkeypatch.setattr(acquisition, "_request", request) + monkeypatch.setattr(providers, "_request", request) await acquisition.submit_to_client( _endpoint("deluge"), @@ -129,7 +130,7 @@ async def test_deluge_rejects_json_rpc_error(monkeypatch: pytest.MonkeyPatch) -> async def request(*args: object, **kwargs: object) -> tuple[int, dict[str, str], bytes]: return next(responses) - monkeypatch.setattr(acquisition, "_request", request) + monkeypatch.setattr(providers, "_request", request) with pytest.raises(HTTPException) as exc_info: await acquisition.submit_to_client( @@ -146,7 +147,7 @@ async def test_prowlarr_rejects_invalid_json(monkeypatch: pytest.MonkeyPatch) -> async def request(*args: object, **kwargs: object) -> tuple[int, dict[str, str], bytes]: return 200, {}, b"not-json" - monkeypatch.setattr(acquisition, "_request", request) + monkeypatch.setattr(providers, "_request", request) with pytest.raises(HTTPException) as exc_info: await acquisition.search_endpoint(_endpoint("prowlarr"), "book") @@ -159,7 +160,7 @@ async def test_qbittorrent_connection_test_accepts_empty_204_login(monkeypatch: async def request(*args: object, **kwargs: object) -> tuple[int, dict[str, str], bytes]: return 204, {"Set-Cookie": "QBT_SID_8082=test; path=/"}, b"" - monkeypatch.setattr(acquisition, "_request", request) + monkeypatch.setattr(providers, "_request", request) await acquisition.test_endpoint_connection(_endpoint("qbittorrent")) @@ -175,7 +176,7 @@ async def test_qbittorrent_submission_accepts_empty_204_login(monkeypatch: pytes async def request(*args: object, **kwargs: object) -> tuple[int, dict[str, str], bytes]: return next(responses) - monkeypatch.setattr(acquisition, "_request", request) + monkeypatch.setattr(providers, "_request", request) await acquisition.submit_to_client( _endpoint("qbittorrent"), @@ -200,7 +201,7 @@ async def request(*args: object, **kwargs: object) -> tuple[int, dict[str, str], bodies.append(cast(bytes, kwargs["body"])) return next(responses) - monkeypatch.setattr(acquisition, "_request", request) + monkeypatch.setattr(providers, "_request", request) await acquisition.submit_to_client( _endpoint("qbittorrent"), @@ -241,7 +242,7 @@ async def request(url: str, **kwargs: object) -> tuple[int, dict[str, str], byte requests.append((url, cast(dict[str, str], kwargs.get("headers", {})))) return next(responses) - monkeypatch.setattr(acquisition, "_request", request) + monkeypatch.setattr(providers, "_request", request) client = await acquisition.QbittorrentClient.connect(_endpoint("qbittorrent")) torrent = await client.find_torrent(tag="papyrus:job") @@ -287,7 +288,7 @@ async def test_qbittorrent_client_reports_completed_content_bytes_and_zero_eta( async def request(*args: object, **kwargs: object) -> tuple[int, dict[str, str], bytes]: return next(responses) - monkeypatch.setattr(acquisition, "_request", request) + monkeypatch.setattr(providers, "_request", request) client = await acquisition.QbittorrentClient.connect(_endpoint("qbittorrent")) torrent = await client.find_torrent(tag="papyrus:job") @@ -318,7 +319,7 @@ async def request(url: str, **kwargs: object) -> tuple[int, dict[str, str], byte request_urls.append(url) return next(responses) - monkeypatch.setattr(acquisition, "_request", request) + monkeypatch.setattr(providers, "_request", request) client = await acquisition.QbittorrentClient.connect(_endpoint("qbittorrent")) torrent = await client.find_torrent( @@ -350,7 +351,7 @@ async def request(url: str, **kwargs: object) -> tuple[int, dict[str, str], byte requests.append((url, parse_qs(cast(bytes, body).decode()))) return next(responses) - monkeypatch.setattr(acquisition, "_request", request) + monkeypatch.setattr(providers, "_request", request) client = await acquisition.QbittorrentClient.connect(_endpoint("qbittorrent")) await client.select_file("abc123", selected_index=1, file_indices=[0, 1, 2]) @@ -395,7 +396,7 @@ async def request(url: str, **kwargs: object) -> tuple[int, dict[str, str], byte request_urls.append(url) return next(responses) - monkeypatch.setattr(acquisition, "_request", request) + monkeypatch.setattr(providers, "_request", request) client = await acquisition.QbittorrentClient.connect(_endpoint("qbittorrent")) await client.select_file("abc123", selected_index=1, file_indices=[0, 1, 2]) @@ -426,7 +427,7 @@ async def request(url: str, **kwargs: object) -> tuple[int, dict[str, str], byte requests.append((url, parse_qs(cast(bytes, body).decode()))) return next(responses) - monkeypatch.setattr(acquisition, "_request", request) + monkeypatch.setattr(providers, "_request", request) client = await acquisition.QbittorrentClient.connect(_endpoint("qbittorrent")) await client.delete_torrent("abc123") @@ -453,7 +454,7 @@ async def request(*args: object, **kwargs: object) -> tuple[int, dict[str, str], request_headers.append(kwargs.get("headers", {})) return next(responses) - monkeypatch.setattr(acquisition, "_request", request) + monkeypatch.setattr(providers, "_request", request) await acquisition.submit_to_client( _endpoint("qbittorrent"), @@ -469,7 +470,7 @@ async def test_qbittorrent_rejects_failed_login_body(monkeypatch: pytest.MonkeyP async def request(*args: object, **kwargs: object) -> tuple[int, dict[str, str], bytes]: return 200, {}, b"Fails." - monkeypatch.setattr(acquisition, "_request", request) + monkeypatch.setattr(providers, "_request", request) with pytest.raises(HTTPException) as exc_info: await acquisition.submit_to_client( @@ -490,7 +491,7 @@ async def request(url: str, **kwargs: object) -> tuple[int, dict[str, str], byte urls.append(url) return 201, {}, b'{"id":1}' - monkeypatch.setattr(acquisition, "_request", request) + monkeypatch.setattr(providers, "_request", request) await acquisition.dispatch_arr_command(_endpoint("readarr"), "BookSearch", [1]) diff --git a/tests/services/test_auth.py b/tests/services/test_auth.py index c6988d6..6967d7a 100644 --- a/tests/services/test_auth.py +++ b/tests/services/test_auth.py @@ -250,7 +250,9 @@ async def test_reset_password_revokes_all_sessions( ) await session.commit() await auth_service.reset_password(session, plain_token, "NewSecureP@ss123") - session_result = await session.execute(select(AuthSession).where(AuthSession.user_id == register_result.user.user_id)) + session_result = await session.execute( + select(AuthSession).where(AuthSession.user_id == register_result.user.user_id) + ) assert all(auth_session.revoked_at is not None for auth_session in session_result.scalars()) diff --git a/tests/test_openapi_export.py b/tests/test_openapi_export.py new file mode 100644 index 0000000..c9f27be --- /dev/null +++ b/tests/test_openapi_export.py @@ -0,0 +1,27 @@ +"""Public documentation export must be independent of local configuration.""" + +import json +import os +import subprocess +import sys +from pathlib import Path + + +def test_export_ignores_environment_and_dev_routes(tmp_path: Path) -> None: + output = tmp_path / "api.json" + command = [sys.executable, "scripts/export_openapi.py", str(output)] + subprocess.run(command, check=True) + baseline = output.read_text() + environment = { + **os.environ, + "DEBUG": "true", + "API_PREFIX": "/private-prefix", + "PUBLIC_BASE_URL": "https://private-environment.example", + "SECRET_KEY": "private-environment-secret", + } + subprocess.run(command, env=environment, check=True) + assert output.read_text() == baseline + assert "private-environment" not in baseline + assert all(not path.startswith("/__dev") for path in json.loads(baseline)["paths"]) + + subprocess.run([*command, "--check"], check=True) From d01f2196c006824dd4be494b1ed43ff3223cd700 Mon Sep 17 00:00:00 2001 From: Karolis Strazdas Date: Sun, 4 Oct 2026 00:22:21 +0300 Subject: [PATCH 2/2] docs: correct API project link and verify exported route coverage --- papyrus/main.py | 2 +- tests/test_openapi_export.py | 5 ++++- 2 files changed, 5 insertions(+), 2 deletions(-) diff --git a/papyrus/main.py b/papyrus/main.py index 2901145..7041c81 100644 --- a/papyrus/main.py +++ b/papyrus/main.py @@ -117,7 +117,7 @@ def create_app() -> FastAPI: """, contact={ "name": "Papyrus Support", - "url": "https://github.com/Eoic/Papyrus", + "url": "https://github.com/PapyrusReader/papyrus", }, license_info={ "name": "AGPL-3.0", diff --git a/tests/test_openapi_export.py b/tests/test_openapi_export.py index c9f27be..fdf41e7 100644 --- a/tests/test_openapi_export.py +++ b/tests/test_openapi_export.py @@ -22,6 +22,9 @@ def test_export_ignores_environment_and_dev_routes(tmp_path: Path) -> None: subprocess.run(command, env=environment, check=True) assert output.read_text() == baseline assert "private-environment" not in baseline - assert all(not path.startswith("/__dev") for path in json.loads(baseline)["paths"]) + schema = json.loads(baseline) + assert all(not path.startswith("/__dev") for path in schema["paths"]) + assert {"/v1/books", "/v1/auth/login", "/v1/opds/relay", "/v1/sync/powersync-upload"} <= schema["paths"].keys() + assert schema["info"]["contact"]["url"] == "https://github.com/PapyrusReader/papyrus" subprocess.run([*command, "--check"], check=True)