feat(series): watch series and suggest new parts once a day

tests/test_import_flow.py

@@ -1359,3 +1359,68 @@ def test_recent_logs_newest_first(client):
     rows = client.get("/api/system/logs").json()
     assert [r["message"] for r in rows[:2]] == ["second", "first"]
     assert rows[0]["level"] == "WARNING"
+
+
+def test_watched_series_suggests_only_new_parts(client, monkeypatch):
+    import asyncio
+    import datetime
+
+    from wordarr import db
+    from wordarr.api import series
+    from wordarr.metadata import SERIES_PROVIDERS
+    from wordarr.metadata.base import MetadataResult
+
+    def part(volume, external_id=""):
+        return MetadataResult(media_type="ebook", title=f"Band {volume}", volume=volume,
+                              series="Saga", external_id=external_id, source="hardcover")
+
+    hits = [part(1, "a1"), part(2, "a2")]
+    calls = []
+
+    async def lookup(series_id, title, language, source):
+        calls.append(series_id)
+        if hits is None:
+            raise RuntimeError("offline")
+        return hits
+
+    monkeypatch.setitem(SERIES_PROVIDERS, "ebook", lookup)
+    lib = client.post("/api/libraries", json={
+        "name": "Saga", "media_type": "ebook", "root_path": str(client.tmp_path)}).json()
+    client.post("/api/requests", json={"library_id": lib["id"], "title": "Band 3", "volume": 3})
+    watch = {"media_type": "ebook", "library_id": lib["id"], "series_id": "s1",
+             "source": "hardcover", "title": "Saga"}
+
+    resp = client.post("/api/series", json=watch)
+    assert resp.status_code == 200
+    assert resp.json()["new"] == []  # what exists today is not news
+    assert client.post("/api/series", json=watch).status_code == 409
+
+    hits += [part(3), part(4, "a4"), part(5, "a5")]  # 3 is requested already
+    [w] = client.post("/api/series/check").json()
+    new = {p["title"]: p["id"] for p in w["new"]}
+    assert set(new) == {"Band 4", "Band 5"}
+
+    client.post("/api/series/parts", json={"ids": [new["Band 4"]], "action": "request"})
+    client.post("/api/series/parts", json={"ids": [new["Band 5"]], "action": "ignore"})
+    missing = client.get("/api/requests?status=missing").json()
+    assert {r["title"] for r in missing} == {"Band 3", "Band 4"}
+    assert client.post("/api/series/check").json()[0]["new"] == []
+
+    hits = None
+    [w] = client.post("/api/series/check").json()
+    assert w["error"] == "offline"
+    assert client.post("/api/series", json={**watch, "series_id": "s2"}).status_code == 502
+
+    # the daily run only picks up what was checked more than a day ago
+    hits, calls[:] = [], []
+    asyncio.run(series.check_due())
+    assert calls == []
+    with db.SessionLocal() as session:
+        session.get(db.WatchedSeries, w["id"]).checked_at = (
+            datetime.datetime.now(datetime.UTC) - datetime.timedelta(days=2))
+        session.commit()
+    asyncio.run(series.check_due())
+    assert calls == ["s1"]
+
+    assert client.delete(f"/api/series/{w['id']}").status_code == 200
+    assert client.get("/api/series").json() == []

wordarr/api/series.py

@@ -0,0 +1,164 @@
+import asyncio
+import datetime
+import logging
+
+from fastapi import APIRouter, Depends, HTTPException
+from pydantic import BaseModel
+from sqlalchemy import or_, select
+from sqlalchemy.orm import Session
+
+from .. import db
+from ..db import BookRequest, Library, SeriesPart, WatchedSeries, get_session
+from ..metadata import SERIES_PROVIDERS
+
+router = APIRouter(prefix="/api/series", tags=["series"])
+log = logging.getLogger(__name__)
+
+PART_FIELDS = ("title", "authors", "narrator", "external_id", "year", "series",
+               "volume", "cover_url")
+
+
+def _now() -> datetime.datetime:
+    return datetime.datetime.now(datetime.UTC)
+
+
+def _key(r) -> str:
+    return r.external_id or f"{r.title.strip().lower()}|{r.volume}"
+
+
+async def check_watch(session: Session, w: WatchedSeries) -> None:
+    """Look the series up again. The first lookup only records what exists, so
+    just later parts become suggestions; so do parts nobody has requested yet."""
+    try:
+        hits = await SERIES_PROVIDERS[w.media_type](w.series_id, w.title, w.language, w.source)
+    except Exception as exc:
+        log.warning("series %r lookup failed: %s", w.title, exc)
+        w.error, w.checked_at = str(exc), _now()
+        return
+    initial = not w.parts
+    reqs = session.scalars(select(BookRequest).where(BookRequest.media_type == w.media_type)).all()
+    requested = {r.external_id for r in reqs if r.external_id}
+    requested |= {(r.title.strip().lower(), r.volume) for r in reqs}
+    seen = {p.key for p in w.parts}
+    for hit in hits:
+        if (key := _key(hit)) in seen:
+            continue
+        seen.add(key)
+        known = initial or hit.external_id in requested or (
+            hit.title.strip().lower(), hit.volume) in requested
+        w.parts.append(SeriesPart(key=key, state="known" if known else "new",
+                                  **{f: getattr(hit, f) for f in PART_FIELDS}))
+    w.error, w.checked_at = "", _now()
+
+
+async def check_due() -> None:
+    cutoff = (_now() - datetime.timedelta(days=1)).replace(tzinfo=None)  # SQLite stores naive
+    with db.SessionLocal() as session:
+        due = session.scalars(select(WatchedSeries).where(
+            or_(WatchedSeries.checked_at.is_(None), WatchedSeries.checked_at < cutoff))).all()
+        for w in due:
+            await check_watch(session, w)
+            session.commit()
+        if due:
+            log.info("checked %d watched series", len(due))
+
+
+async def run_daily() -> None:
+    """Wakes hourly and checks what is a day old, so restarts do not reset the clock."""
+    while True:
+        try:
+            await check_due()
+        except Exception:
+            log.exception("series check failed")
+        await asyncio.sleep(3600)
+
+
+def _out(w: WatchedSeries) -> dict:
+    checked = w.checked_at and w.checked_at.replace(tzinfo=datetime.UTC).isoformat()
+    return {
+        "id": w.id, "media_type": w.media_type, "library_id": w.library_id,
+        "library_name": w.library.name, "series_id": w.series_id, "source": w.source,
+        "title": w.title, "checked_at": checked, "error": w.error,
+        "new": [{"id": p.id, **{f: getattr(p, f) for f in PART_FIELDS}}
+                for p in w.parts if p.state == "new"],
+    }
+
+
+def _all(session: Session) -> list[dict]:
+    return [_out(w) for w in session.scalars(select(WatchedSeries).order_by(WatchedSeries.title))]
+
+
+@router.get("")
+def list_watches(session: Session = Depends(get_session)):
+    return _all(session)
+
+
+class WatchIn(BaseModel):
+    media_type: str
+    library_id: int
+    series_id: str
+    source: str = ""
+    title: str
+    language: str = ""
+
+
+@router.post("")
+async def create_watch(data: WatchIn, session: Session = Depends(get_session)):
+    if data.media_type not in SERIES_PROVIDERS:
+        raise HTTPException(400, f"no series lookup for media_type: {data.media_type}")
+    lib = session.get(Library, data.library_id)
+    if not lib:
+        raise HTTPException(404, "library not found")
+    if lib.media_type != data.media_type:
+        raise HTTPException(400, "library has a different media type")
+    if session.scalar(select(WatchedSeries).where(
+            WatchedSeries.media_type == data.media_type,
+            WatchedSeries.source == data.source,
+            WatchedSeries.series_id == data.series_id)):
+        raise HTTPException(409, "series is already watched")
+    w = WatchedSeries(**data.model_dump())
+    session.add(w)
+    await check_watch(session, w)
+    if w.error:  # without a first lookup every part would turn up as new later
+        session.rollback()
+        raise HTTPException(502, f"metadata provider error: {w.error}")
+    session.commit()
+    return _out(w)
+
+
+@router.delete("/{watch_id}")
+def delete_watch(watch_id: int, session: Session = Depends(get_session)):
+    w = session.get(WatchedSeries, watch_id)
+    if not w:
+        raise HTTPException(404, "series not found")
+    session.delete(w)
+    session.commit()
+    return {"ok": True}
+
+
+@router.post("/check")
+async def check_all(session: Session = Depends(get_session)):
+    for w in session.scalars(select(WatchedSeries)).all():
+        await check_watch(session, w)
+        session.commit()
+    return _all(session)
+
+
+class PartsIn(BaseModel):
+    ids: list[int]
+    action: str  # request | ignore
+
+
+@router.post("/parts")
+def handle_parts(data: PartsIn, session: Session = Depends(get_session)):
+    if data.action not in ("request", "ignore"):
+        raise HTTPException(400, f"invalid action: {data.action}")
+    parts = session.scalars(select(SeriesPart).where(
+        SeriesPart.id.in_(data.ids), SeriesPart.state == "new")).all()
+    for p in parts:
+        if data.action == "request":
+            session.add(BookRequest(media_type=p.watch.media_type, library_id=p.watch.library_id,
+                                    **{f: getattr(p, f) for f in PART_FIELDS}))
+        p.state = "requested" if data.action == "request" else "ignored"
+    session.commit()
+    return {"done": len(parts)}

wordarr/db.py

@@ -23,6 +23,8 @@ class Library(Base):
     language: Mapped[str] = mapped_column(String, default="")
 
     requests: Mapped[list["BookRequest"]] = relationship(back_populates="library")
+    watches: Mapped[list["WatchedSeries"]] = relationship(
+        back_populates="library", cascade="all, delete-orphan")
 
 
 class BookRequest(Base):
@@ -68,6 +70,47 @@ class History(Base):
     message: Mapped[str] = mapped_column(String, default="")
 
 
+class WatchedSeries(Base):
+    """A series looked up again now and then; new parts become suggestions."""
+
+    __tablename__ = "watched_series"
+
+    id: Mapped[int] = mapped_column(primary_key=True)
+    media_type: Mapped[str] = mapped_column(String)
+    library_id: Mapped[int] = mapped_column(ForeignKey("libraries.id"))
+    series_id: Mapped[str] = mapped_column(String)
+    source: Mapped[str] = mapped_column(String, default="")
+    title: Mapped[str] = mapped_column(String)
+    language: Mapped[str] = mapped_column(String, default="")
+    checked_at: Mapped[datetime.datetime | None] = mapped_column(default=None)
+    error: Mapped[str] = mapped_column(String, default="")
+
+    library: Mapped[Library] = relationship(back_populates="watches")
+    parts: Mapped[list["SeriesPart"]] = relationship(
+        back_populates="watch", cascade="all, delete-orphan")
+
+
+class SeriesPart(Base):
+    """Every part a watch has seen, so a later lookup can tell what is new."""
+
+    __tablename__ = "series_parts"
+
+    id: Mapped[int] = mapped_column(primary_key=True)
+    watch_id: Mapped[int] = mapped_column(ForeignKey("watched_series.id"))
+    key: Mapped[str] = mapped_column(String)
+    state: Mapped[str] = mapped_column(String)  # known | new | requested | ignored
+    title: Mapped[str] = mapped_column(String)
+    authors: Mapped[str] = mapped_column(String, default="")
+    narrator: Mapped[str] = mapped_column(String, default="")
+    external_id: Mapped[str] = mapped_column(String, default="")
+    year: Mapped[int | None] = mapped_column(default=None)
+    series: Mapped[str] = mapped_column(String, default="")
+    volume: Mapped[int | None] = mapped_column(default=None)
+    cover_url: Mapped[str] = mapped_column(String, default="")
+
+    watch: Mapped[WatchedSeries] = relationship(back_populates="parts")
+
+
 def record(session, event: str, req: BookRequest, **fields: str) -> History:
     """Add a history row; it is saved with the caller's next commit."""
     row = History(event=event, request_id=req.id, title=req.title,

wordarr/main.py

@@ -1,3 +1,4 @@
+import asyncio
 from contextlib import asynccontextmanager
 from pathlib import Path
 
@@ -5,7 +6,7 @@ from fastapi import FastAPI
 from fastapi.staticfiles import StaticFiles
 
 from . import db, logs, version
-from .api import fs, history, imports, libraries, requests, search, system
+from .api import fs, history, imports, libraries, requests, search, series, system
 
 logs.install()
 
@@ -13,7 +14,9 @@ logs.install()
 @asynccontextmanager
 async def lifespan(app: FastAPI):
     db.init_db()
+    watcher = asyncio.create_task(series.run_daily())
     yield
+    watcher.cancel()
 
 
 app = FastAPI(title="wordarr", lifespan=lifespan)
@@ -25,6 +28,7 @@ app.include_router(imports.router)
 app.include_router(fs.router)
 app.include_router(history.router)
 app.include_router(system.router)
+app.include_router(series.router)
 
 
 @app.get("/api/version")