Add torrent acquisition integrations

.env.example

@@ -53,6 +53,10 @@ DEV_PAGES_USE_VITE=true
 DEV_PAGES_VITE_URL=http://localhost:5173
 DEV_PAGES_MANIFEST_PATH=frontend/dev-pages/dist/.vite/manifest.json
 
+# Execute enabled acquisition rules on this interval (seconds). Set a value that
+# respects each indexer's API limits.
+ACQUISITION_AUTOMATION_INTERVAL_SECONDS=300
+
 # Prefer file-based PowerSync keys for local development.
 # Generate them with: ./scripts/generate_dev_powersync_keys.sh
 POWERSYNC_JWT_PRIVATE_KEY_FILE=.local/powersync/private.pem

alembic/versions/d4e5f6a7b8c9_add_acquisition.py

@@ -0,0 +1,79 @@
+"""add acquisition endpoints, rules and jobs
+
+Revision ID: d4e5f6a7b8c9
+Revises: c3f8b2a9d1e4
+"""
+
+from collections.abc import Sequence
+
+import sqlalchemy as sa
+from sqlalchemy.dialects import postgresql
+
+from alembic import op
+
+revision: str = "d4e5f6a7b8c9"
+down_revision: str | Sequence[str] | None = "c3f8b2a9d1e4"
+branch_labels: str | Sequence[str] | None = None
+depends_on: str | Sequence[str] | None = None
+
+
+def upgrade() -> None:
+    """Create private acquisition integration tables."""
+    op.create_table(
+        "acquisition_endpoints",
+        sa.Column("endpoint_id", sa.Uuid(), nullable=False),
+        sa.Column("owner_user_id", sa.Uuid(), nullable=False),
+        sa.Column("name", sa.String(120), nullable=False),
+        sa.Column("kind", sa.String(32), nullable=False),
+        sa.Column("base_url", sa.String(2048), nullable=False),
+        sa.Column("credentials", postgresql.JSONB(), nullable=True),
+        sa.Column("settings", postgresql.JSONB(), nullable=True),
+        sa.Column("enabled", sa.Boolean(), server_default=sa.text("true"), nullable=False),
+        sa.Column("created_at", sa.DateTime(timezone=True), server_default=sa.text("now()"), nullable=False),
+        sa.Column("updated_at", sa.DateTime(timezone=True), server_default=sa.text("now()"), nullable=False),
+        sa.ForeignKeyConstraint(["owner_user_id"], ["users.user_id"], ondelete="CASCADE"),
+        sa.PrimaryKeyConstraint("endpoint_id"),
+    )
+    op.create_index("ix_acquisition_endpoints_owner_user_id", "acquisition_endpoints", ["owner_user_id"])
+    op.create_table(
+        "acquisition_rules",
+        sa.Column("rule_id", sa.Uuid(), nullable=False),
+        sa.Column("owner_user_id", sa.Uuid(), nullable=False),
+        sa.Column("name", sa.String(120), nullable=False),
+        sa.Column("query", sa.String(500), nullable=False),
+        sa.Column("endpoint_ids", postgresql.JSONB(), nullable=True),
+        sa.Column("download_client_id", sa.Uuid(), nullable=True),
+        sa.Column("filters", postgresql.JSONB(), nullable=True),
+        sa.Column("enabled", sa.Boolean(), server_default=sa.text("true"), nullable=False),
+        sa.Column("last_run_at", sa.DateTime(timezone=True), nullable=True),
+        sa.Column("created_at", sa.DateTime(timezone=True), server_default=sa.text("now()"), nullable=False),
+        sa.ForeignKeyConstraint(["download_client_id"], ["acquisition_endpoints.endpoint_id"]),
+        sa.ForeignKeyConstraint(["owner_user_id"], ["users.user_id"], ondelete="CASCADE"),
+        sa.PrimaryKeyConstraint("rule_id"),
+    )
+    op.create_index("ix_acquisition_rules_owner_user_id", "acquisition_rules", ["owner_user_id"])
+    op.create_table(
+        "acquisition_jobs",
+        sa.Column("job_id", sa.Uuid(), nullable=False),
+        sa.Column("owner_user_id", sa.Uuid(), nullable=False),
+        sa.Column("endpoint_id", sa.Uuid(), nullable=False),
+        sa.Column("rule_id", sa.Uuid(), nullable=True),
+        sa.Column("title", sa.String(500), nullable=False),
+        sa.Column("download_url", sa.Text(), nullable=False),
+        sa.Column("status", sa.String(32), server_default=sa.text("'queued'"), nullable=False),
+        sa.Column("client_reference", sa.String(255), nullable=True),
+        sa.Column("error", sa.Text(), nullable=True),
+        sa.Column("created_at", sa.DateTime(timezone=True), server_default=sa.text("now()"), nullable=False),
+        sa.ForeignKeyConstraint(["endpoint_id"], ["acquisition_endpoints.endpoint_id"]),
+        sa.ForeignKeyConstraint(["owner_user_id"], ["users.user_id"], ondelete="CASCADE"),
+        sa.ForeignKeyConstraint(["rule_id"], ["acquisition_rules.rule_id"], ondelete="SET NULL"),
+        sa.PrimaryKeyConstraint("job_id"),
+    )
+    op.create_index("ix_acquisition_jobs_owner_user_id", "acquisition_jobs", ["owner_user_id"])
+
+
+def downgrade() -> None:
+    """Drop acquisition tables."""
+    op.drop_table("acquisition_jobs")
+    op.drop_table("acquisition_rules")
+    op.drop_table("acquisition_endpoints")

papyrus/api/routes/__init__.py

@@ -3,6 +3,7 @@
 from fastapi import APIRouter, FastAPI
 
 from papyrus.api.routes import (
+    acquisition,
     annotations,
     auth,
     bookmarks,
@@ -26,6 +27,7 @@
 
 api_router = APIRouter()
 
+api_router.include_router(acquisition.router, prefix="/acquisition", tags=["Acquisition"])
 api_router.include_router(auth.router, prefix="/auth", tags=["Auth"])
 api_router.include_router(users.router, prefix="/users", tags=["Users"])
 api_router.include_router(books.router, prefix="/books", tags=["Books"])

papyrus/api/routes/acquisition.py

@@ -0,0 +1,209 @@
+"""Private torrent and indexer acquisition endpoints."""
+
+from typing import Annotated
+from uuid import UUID
+
+from fastapi import APIRouter, Depends, HTTPException, status
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from papyrus.api.deps import CurrentUserId
+from papyrus.core.database import get_db
+from papyrus.models.acquisition import AcquisitionEndpoint as AcquisitionEndpointModel
+from papyrus.models.acquisition import AcquisitionJob as AcquisitionJobModel
+from papyrus.models.acquisition import AcquisitionRule as AcquisitionRuleModel
+from papyrus.schemas.acquisition import (
+    AcquisitionEndpoint,
+    AcquisitionEndpointCreate,
+    AcquisitionEndpointUpdate,
+    AcquisitionJob,
+    AcquisitionRule,
+    AcquisitionRuleCreate,
+    ArrCommandRequest,
+    Release,
+    SearchRequest,
+    SubmitRequest,
+)
+from papyrus.services.acquisition import (
+    dispatch_arr_command,
+    owned_endpoint,
+    run_rule,
+    search_endpoint,
+    submit_to_client,
+)
+
+router = APIRouter()
+DbSession = Annotated[AsyncSession, Depends(get_db)]
+
+
+def _credentials(api_key: str | None, username: str | None, password: str | None) -> dict[str, str] | None:
+    values = {
+        key: value for key, value in {"api_key": api_key, "username": username, "password": password}.items() if value
+    }
+    return values or None
+
+
+@router.get("/endpoints", response_model=list[AcquisitionEndpoint])
+async def list_endpoints(user_id: CurrentUserId, db: DbSession) -> list[AcquisitionEndpointModel]:
+    result = await db.execute(
+        select(AcquisitionEndpointModel)
+        .where(AcquisitionEndpointModel.owner_user_id == user_id)
+        .order_by(AcquisitionEndpointModel.name)
+    )
+    return list(result.scalars())
+
+
+@router.post("/endpoints", response_model=AcquisitionEndpoint, status_code=status.HTTP_201_CREATED)
+async def create_endpoint(
+    user_id: CurrentUserId, request: AcquisitionEndpointCreate, db: DbSession
+) -> AcquisitionEndpointModel:
+    endpoint = AcquisitionEndpointModel(
+        owner_user_id=user_id,
+        name=request.name,
+        kind=request.kind.value,
+        base_url=str(request.base_url),
+        credentials=_credentials(
+            request.api_key.get_secret_value() if request.api_key else None,
+            request.username.get_secret_value() if request.username else None,
+            request.password.get_secret_value() if request.password else None,
+        ),
+        settings=request.settings,
+    )
+    db.add(endpoint)
+    await db.commit()
+    await db.refresh(endpoint)
+    return endpoint
+
+
+@router.patch("/endpoints/{endpoint_id}", response_model=AcquisitionEndpoint)
+async def update_endpoint(
+    user_id: CurrentUserId, endpoint_id: UUID, request: AcquisitionEndpointUpdate, db: DbSession
+) -> AcquisitionEndpointModel:
+    endpoint = await owned_endpoint(db, user_id, endpoint_id)
+    updates = request.model_dump(exclude_unset=True, exclude={"api_key", "username", "password"})
+    for field, value in updates.items():
+        setattr(endpoint, field, str(value) if field == "base_url" else value)
+    replacement = _credentials(
+        request.api_key.get_secret_value() if request.api_key else None,
+        request.username.get_secret_value() if request.username else None,
+        request.password.get_secret_value() if request.password else None,
+    )
+    if replacement:
+        endpoint.credentials = {**(endpoint.credentials or {}), **replacement}
+    await db.commit()
+    await db.refresh(endpoint)
+    return endpoint
+
+
+@router.delete("/endpoints/{endpoint_id}", status_code=status.HTTP_204_NO_CONTENT)
+async def delete_endpoint(user_id: CurrentUserId, endpoint_id: UUID, db: DbSession) -> None:
+    endpoint = await owned_endpoint(db, user_id, endpoint_id)
+    await db.delete(endpoint)
+    await db.commit()
+
+
+@router.post("/search", response_model=list[Release])
+async def search_releases(user_id: CurrentUserId, request: SearchRequest, db: DbSession) -> list[Release]:
+    statement = select(AcquisitionEndpointModel).where(
+        AcquisitionEndpointModel.owner_user_id == user_id,
+        AcquisitionEndpointModel.enabled.is_(True),
+        AcquisitionEndpointModel.kind.in_(("prowlarr", "torznab", "newznab")),
+    )
+    if request.endpoint_ids:
+        statement = statement.where(AcquisitionEndpointModel.endpoint_id.in_(request.endpoint_ids))
+    result = await db.execute(statement)
+    releases: list[Release] = []
+    for endpoint in result.scalars():
+        releases.extend(await search_endpoint(endpoint, request.query))
+    return releases
+
+
+@router.post("/submissions", response_model=AcquisitionJob, status_code=status.HTTP_201_CREATED)
+async def submit_release(user_id: CurrentUserId, request: SubmitRequest, db: DbSession) -> AcquisitionJobModel:
+    endpoint = await owned_endpoint(db, user_id, request.endpoint_id)
+    job = AcquisitionJobModel(
+        owner_user_id=user_id, endpoint_id=endpoint.endpoint_id, title=request.title, download_url=request.download_url
+    )
+    db.add(job)
+    try:
+        job.client_reference = await submit_to_client(
+            endpoint, request.download_url, request.category, request.save_path
+        )
+        job.status = "submitted"
+    except HTTPException as exc:
+        job.status = "failed"
+        job.error = str(exc.detail)
+    await db.commit()
+    await db.refresh(job)
+    return job
+
+
+@router.post("/arr/{endpoint_id}/commands", response_model=AcquisitionJob, status_code=status.HTTP_201_CREATED)
+async def run_arr_command(
+    user_id: CurrentUserId, endpoint_id: UUID, request: ArrCommandRequest, db: DbSession
+) -> AcquisitionJobModel:
+    """Delegate a managed search/acquisition to Readarr or another Arr app."""
+    endpoint = await owned_endpoint(db, user_id, endpoint_id)
+    job = AcquisitionJobModel(
+        owner_user_id=user_id,
+        endpoint_id=endpoint.endpoint_id,
+        title=request.command,
+        download_url=f"arr-command:{request.command}",
+    )
+    db.add(job)
+    try:
+        job.client_reference = await dispatch_arr_command(endpoint, request.command, request.ids)
+        job.status = "submitted"
+    except HTTPException as exc:
+        job.status = "failed"
+        job.error = str(exc.detail)
+    await db.commit()
+    await db.refresh(job)
+    return job
+
+
+@router.get("/jobs", response_model=list[AcquisitionJob])
+async def list_jobs(user_id: CurrentUserId, db: DbSession) -> list[AcquisitionJobModel]:
+    result = await db.execute(
+        select(AcquisitionJobModel)
+        .where(AcquisitionJobModel.owner_user_id == user_id)
+        .order_by(AcquisitionJobModel.created_at.desc())
+    )
+    return list(result.scalars())
+
+
+@router.get("/rules", response_model=list[AcquisitionRule])
+async def list_rules(user_id: CurrentUserId, db: DbSession) -> list[AcquisitionRuleModel]:
+    result = await db.execute(select(AcquisitionRuleModel).where(AcquisitionRuleModel.owner_user_id == user_id))
+    return list(result.scalars())
+
+
+@router.post("/rules", response_model=AcquisitionRule, status_code=status.HTTP_201_CREATED)
+async def create_rule(user_id: CurrentUserId, request: AcquisitionRuleCreate, db: DbSession) -> AcquisitionRuleModel:
+    await owned_endpoint(db, user_id, request.download_client_id)
+    rule = AcquisitionRuleModel(
+        owner_user_id=user_id,
+        name=request.name,
+        query=request.query,
+        endpoint_ids=[str(value) for value in request.endpoint_ids] if request.endpoint_ids else None,
+        download_client_id=request.download_client_id,
+        filters=request.filters,
+        enabled=request.enabled,
+    )
+    db.add(rule)
+    await db.commit()
+    await db.refresh(rule)
+    return rule
+
+
+@router.post("/rules/{rule_id}/run", response_model=list[AcquisitionJob])
+async def run_acquisition_rule(user_id: CurrentUserId, rule_id: UUID, db: DbSession) -> list[AcquisitionJobModel]:
+    result = await db.execute(
+        select(AcquisitionRuleModel).where(
+            AcquisitionRuleModel.rule_id == rule_id, AcquisitionRuleModel.owner_user_id == user_id
+        )
+    )
+    rule = result.scalar_one_or_none()
+    if rule is None:
+        raise HTTPException(status_code=404, detail="Acquisition rule not found")
+    return await run_rule(db, rule)

papyrus/config.py

@@ -71,6 +71,7 @@ class Settings(BaseSettings):
     dev_pages_use_vite: bool = False
     dev_pages_vite_url: str = "http://localhost:5173"
     dev_pages_manifest_path: str = "frontend/dev-pages/dist/.vite/manifest.json"
+    acquisition_automation_interval_seconds: int = 300
 
     @field_validator("debug", mode="before")
     @classmethod

papyrus/main.py

@@ -1,8 +1,9 @@
 """FastAPI application factory and configuration."""
 
+import asyncio
 import logging
 from collections.abc import AsyncGenerator
-from contextlib import asynccontextmanager
+from contextlib import asynccontextmanager, suppress
 
 import structlog
 from fastapi import FastAPI, Request, status
@@ -16,9 +17,11 @@
 
 from papyrus.api.routes import api_router, include_debug_routers
 from papyrus.config import get_settings
+from papyrus.core.database import async_session_maker
 from papyrus.core.dev_pages import DEV_PAGES_DIST_DIR, DEV_PAGES_STATIC_URL
 from papyrus.core.exceptions import AppError
 from papyrus.core.rate_limit import limiter
+from papyrus.services.acquisition import run_enabled_rules
 
 settings = get_settings()
 
@@ -40,9 +43,20 @@ def configure_logging() -> None:
 @asynccontextmanager
 async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]:
     """Application lifespan events."""
-    # Startup
-    yield
-    # Shutdown
+
+    async def acquisition_worker() -> None:
+        while True:
+            async with async_session_maker() as session:
+                await run_enabled_rules(session)
+            await asyncio.sleep(settings.acquisition_automation_interval_seconds)
+
+    worker = asyncio.create_task(acquisition_worker())
+    try:
+        yield
+    finally:
+        worker.cancel()
+        with suppress(asyncio.CancelledError):
+            await worker
 
 
 def create_app() -> FastAPI:

papyrus/models/__init__.py

@@ -1,11 +1,15 @@
 from papyrus.core.database import Base
+from papyrus.models.acquisition import AcquisitionEndpoint, AcquisitionJob, AcquisitionRule
 from papyrus.models.auth import AuthExchangeCode, AuthSession, EmailActionToken, PasswordCredential, UserIdentity
 from papyrus.models.media import MediaAsset
 from papyrus.models.powersync_demo import PowerSyncDemoItem
 from papyrus.models.sync import SyncBook
 from papyrus.models.user import User
 
 __all__ = [
+    "AcquisitionEndpoint",
+    "AcquisitionJob",
+    "AcquisitionRule",
     "AuthExchangeCode",
     "AuthSession",
     "Base",