src/lib/processing/repo.py

"""Processing log repository — the only file that reads processing_log."""

from __future__ import annotations

from src.stage1_inventory.db import connect

from .models import StageRun

_COLUMNS = (
    "id", "book_id", "stage", "started_at", "finished_at",
    "success", "pages_done", "engine", "provider", "backend",
    "error", "duration_ms",
)


def _row_to_run(r) -> StageRun:
    # Tolerate older DBs where provider/backend aren't yet present —
    # _COLUMN_ADDITIONS will add them on the next connect, but a stale row
    # cursor can still surface during the same call.
    def _opt(key: str):
        try:
            return r[key]
        except (KeyError, IndexError):
            return None

    return StageRun(
        id=int(r["id"]),
        book_id=r["book_id"], stage=r["stage"],
        started_at=r["started_at"], finished_at=r["finished_at"],
        success=r["success"] if r["success"] is None else int(r["success"]),
        pages_done=r["pages_done"], engine=r["engine"],
        provider=_opt("provider"), backend=_opt("backend"),
        error=r["error"], duration_ms=r["duration_ms"],
    )


def query(
    *,
    book_id: str | None = None,
    stage: str | None = None,
    outcome: str | None = None,    # "success" | "failed" | "in-flight" | None
    limit: int = 100,
) -> list[StageRun]:
    """Filterable query over processing_log. Newest first."""
    where: list[str] = []
    args: list = []
    if book_id:
        where.append("book_id = ?"); args.append(book_id)
    if stage:
        where.append("stage = ?"); args.append(stage)
    if outcome == "success":
        where.append("success = 1")
    elif outcome == "failed":
        where.append("success = 0")
    elif outcome == "in-flight":
        where.append("success IS NULL")

    sql = (
        f"SELECT {','.join(_COLUMNS)} FROM processing_log "
        + (("WHERE " + " AND ".join(where)) if where else "")
        + " ORDER BY id DESC LIMIT ?"
    )
    args.append(int(limit))
    with connect() as conn:
        rows = conn.execute(sql, args).fetchall()
    return [_row_to_run(r) for r in rows]


def by_book(book_id: str, *, limit: int = 30) -> list[StageRun]:
    return query(book_id=book_id, limit=limit)


def distinct_book_ids() -> list[str]:
    with connect() as conn:
        return [r[0] for r in conn.execute(
            "SELECT DISTINCT book_id FROM processing_log ORDER BY book_id"
        ).fetchall()]