jellyfin-scene-search/scene_search/api.py

from __future__ import annotations

import hmac
from pathlib import Path

from fastapi import Depends, FastAPI, Header, HTTPException, Query
from fastapi.responses import FileResponse
from pydantic import BaseModel, Field

from .config import settings
from .database import connect, init_db
from .index_jobs import IndexJobManager
from .search_service import SearchService

WEB_INDEX_PATH = Path(__file__).parent / "web" / "index.html"

app = FastAPI(
    title="Jellyfin Scene Search API",
    version="1.0.0",
)

_service: SearchService | None = None
_index_jobs: IndexJobManager | None = None


class IndexRequest(BaseModel):
    library_ids: list[str] = Field(min_length=1)
    force: bool = False


def get_search_service() -> SearchService:
    global _service
    if _service is None:
        _service = SearchService(
            db_path=settings.db_path,
            meili_url=settings.meili_url,
            meili_master_key=settings.meili_master_key,
            meili_index=settings.meili_index,
            model_id=settings.model_id,
            device=settings.device,
            semantic_ratio=settings.hybrid_semantic_ratio,
        )
    return _service


def get_index_jobs() -> IndexJobManager:
    global _index_jobs
    if _index_jobs is None:
        _index_jobs = IndexJobManager(get_search_service())
    return _index_jobs


def require_api_key(
    authorization: str | None = Header(default=None),
    x_scene_search_key: str | None = Header(default=None),
) -> None:
    expected = settings.scene_api_key
    if not expected:
        return

    supplied = x_scene_search_key
    if authorization and authorization.lower().startswith("bearer "):
        supplied = authorization[7:].strip()

    if not supplied or not hmac.compare_digest(supplied, expected):
        raise HTTPException(status_code=401, detail="Invalid API key")


@app.get("/", include_in_schema=False)
def web_dashboard() -> FileResponse:
    return FileResponse(WEB_INDEX_PATH, media_type="text/html")


@app.get("/api/health")
def health() -> dict:
    return get_search_service().health()


@app.get("/api/search", dependencies=[Depends(require_api_key)])
def search(
    q: str = Query(min_length=1, max_length=500),
    scene_limit: int = Query(default=50, ge=1, le=100),
    item_limit: int = Query(default=20, ge=1, le=50),
) -> dict:
    if get_index_jobs().running:
        raise HTTPException(
            status_code=503,
            detail="Jellyfin indexing is currently using the embedding model",
        )

    try:
        return get_search_service().search(
            q,
            scene_limit=scene_limit,
            item_limit=item_limit,
            jellyfin_only=True,
        )
    except ValueError as exc:
        raise HTTPException(status_code=400, detail=str(exc)) from exc
    except Exception as exc:
        raise HTTPException(status_code=503, detail=str(exc)) from exc


@app.get(
    "/api/jellyfin/libraries",
    dependencies=[Depends(require_api_key)],
)
def jellyfin_libraries() -> dict:
    try:
        return {"libraries": get_index_jobs().libraries()}
    except Exception as exc:
        raise HTTPException(status_code=503, detail=str(exc)) from exc


@app.post(
    "/api/jellyfin/index",
    dependencies=[Depends(require_api_key)],
    status_code=202,
)
def jellyfin_index(request: IndexRequest) -> dict:
    try:
        return get_index_jobs().start(
            request.library_ids,
            force=request.force,
        )
    except RuntimeError as exc:
        message = str(exc)
        status = 409 if "already running" in message else 400
        raise HTTPException(status_code=status, detail=message) from exc
    except Exception as exc:
        raise HTTPException(status_code=503, detail=str(exc)) from exc


@app.get(
    "/api/jellyfin/index/status",
    dependencies=[Depends(require_api_key)],
)
def jellyfin_index_status() -> dict:
    return get_index_jobs().status()


@app.get(
    "/api/frames/{frame_id}/thumbnail",
    dependencies=[Depends(require_api_key)],
)
def frame_thumbnail(frame_id: str) -> FileResponse:
    init_db(settings.db_path)
    with connect(settings.db_path) as con:
        row = con.execute(
            """
            SELECT f.thumb_path
            FROM frames f
            JOIN media m ON m.id = f.media_id
            WHERE f.id = ? AND m.source_kind = 'jellyfin'
            """,
            (frame_id,),
        ).fetchone()

    if row is None:
        raise HTTPException(status_code=404, detail="Frame not found")

    path = Path(str(row["thumb_path"]))
    if not path.is_file():
        raise HTTPException(status_code=404, detail="Thumbnail file not found")

    return FileResponse(path, media_type="image/jpeg")


def main() -> None:
    import uvicorn

    uvicorn.run(
        "scene_search.api:app",
        host=settings.scene_api_host,
        port=settings.scene_api_port,
        workers=1,
    )


if __name__ == "__main__":
    main()