Merge pull request 'feat(connect): scan audiobookshelf and call a webhook after changes' (#3) from feature/connections into main

Reviewed-on: #3

README.md

@@ -25,6 +25,9 @@ the same mount, otherwise moving turns into copy + delete.
 | `WORDARR_EXTRA_EBOOK_EXTENSIONS` | – | extra ebook formats the scan should offer, e.g. `.kepub,.lit,.rtf` |
 | `WORDARR_HARDCOVER_TOKEN` | – | Hardcover API token, the best ebook source; without it Hardcover is skipped |
 | `WORDARR_GOOGLE_BOOKS_KEY` | – | optional, only needed when the keyless Google Books quota runs dry |
+| `WORDARR_ABS_URL` | – | Audiobookshelf base URL, e.g. `http://audiobookshelf`; after an import, relocate or retag it scans the libraries whose folders hold the files, so both containers must mount them under the same path |
+| `WORDARR_ABS_TOKEN` | – | API token of an Audiobookshelf admin (only admins may start a scan) |
+| `WORDARR_WEBHOOK_URL` | – | gets every history row as JSON (`event`, `request_id`, `title`, `media_type`, `source`, `dest`, `message`) |
 | `WORDARR_BUILD` | – | shown next to the version, e.g. the git sha; also a build arg of the image |
 
 The header shows the running version and, behind it, the build stamp (the newest

tests/test_import_flow.py

@@ -1220,3 +1220,53 @@ def test_history_records_what_happened_to_a_request(client):
     assert rows[2]["dest"] == dest
     assert rows[3]["message"]
     assert rows[0]["at"].endswith("+00:00")
+
+
+def test_connections_hear_about_committed_rows(client, monkeypatch):
+    import httpx
+
+    from wordarr import connect
+
+    calls = []
+
+    def handler(request):
+        calls.append((request.method, str(request.url)))
+        if request.url.path == "/api/libraries":
+            return httpx.Response(200, json={"libraries": [
+                {"id": "lib-audio", "name": "Audio",
+                 "folders": [{"fullPath": str(client.tmp_path / "library")}]},
+                {"id": "lib-pod", "name": "Pod", "folders": [{"fullPath": "/podcasts"}]},
+            ]})
+        if request.url.host == "hook":
+            return httpx.Response(500)  # a broken webhook must not stop the scan
+        return httpx.Response(200)
+
+    real_client = httpx.Client
+    monkeypatch.setattr(connect.httpx, "Client",
+                        lambda **kw: real_client(transport=httpx.MockTransport(handler), **kw))
+    monkeypatch.setattr(connect, "WEBHOOK_URL", "http://hook/in")
+    monkeypatch.setattr(connect, "ABS_URL", "http://abs")
+    monkeypatch.setattr(connect, "ABS_TOKEN", "t")
+
+    root = client.tmp_path / "library" / "conn"
+    lib_id = client.post("/api/libraries", json={
+        "name": "conn", "media_type": "ebook", "root_path": str(root),
+        "folder_template": "{Title}", "file_template": "{Title}",
+    }).json()["id"]
+    req_id = client.post("/api/requests", json={
+        "library_id": lib_id, "title": "Momo"}).json()["id"]
+    src = client.downloads / "Momo.epub"
+    src.write_bytes(b"x")
+    assert client.post("/api/import", json={"items": [{
+        "path": str(src), "is_dir": False, "files": [str(src)], "request_id": req_id,
+    }]}).json()["results"][0]["ok"]
+
+    assert calls == [
+        ("POST", "http://hook/in"),
+        ("GET", "http://abs/api/libraries"),
+        ("POST", "http://abs/api/libraries/lib-audio/scan"),
+    ]
+
+    calls.clear()
+    client.delete(f"/api/requests/{req_id}")
+    assert calls == [("POST", "http://hook/in")]  # nothing new on disk, no scan

wordarr/api/imports.py

@@ -1,11 +1,11 @@
 from pathlib import Path
 
-from fastapi import APIRouter, Depends, HTTPException
+from fastapi import APIRouter, BackgroundTasks, Depends, HTTPException
 from pydantic import BaseModel
 from sqlalchemy import select
 from sqlalchemy.orm import Session
 
-from .. import config
+from .. import config, connect
 from ..db import BookRequest, get_session, record
 from ..fsnames import decode_path, readable
 from ..importer import matcher, mover, scanner
@@ -58,8 +58,9 @@ class ImportIn(BaseModel):
 
 
 @router.post("")
-def do_import(data: ImportIn, session: Session = Depends(get_session)):
+def do_import(data: ImportIn, background: BackgroundTasks,
-    results = []
+              session: Session = Depends(get_session)):
+    results, rows = [], []
     download_root = Path(config.DOWNLOAD_DIR).resolve()
     for item in data.items:
         # the client hands back whatever the scan produced, base64 included
@@ -91,12 +92,12 @@ def do_import(data: ImportIn, session: Session = Depends(get_session)):
             try:
                 dest = mover.append_to_import(item.files, req, req.library)
             except Exception as exc:
-                record(session, "failed", req, source=shown, message=str(exc))
+                rows.append(record(session, "failed", req, source=shown, message=str(exc)))
                 session.commit()
                 results.append({"path": shown, "ok": False, "error": str(exc)})
                 continue
             req.imported_path = dest
-            record(session, "appended", req, source=shown, dest=dest)
+            rows.append(record(session, "appended", req, source=shown, dest=dest))
             session.commit()
             results.append({"path": shown, "ok": True, "dest": dest, "appended": True})
             continue
@@ -114,15 +115,16 @@ def do_import(data: ImportIn, session: Session = Depends(get_session)):
             dest = mover.import_item(item.path, item.files, item.is_dir, req, req.library)
         except Exception as exc:
             session.rollback()  # the next commit must not keep the folder's series
-            record(session, "failed", req, source=shown, message=str(exc))
+            rows.append(record(session, "failed", req, source=shown, message=str(exc)))
             session.commit()
             results.append({"path": shown, "ok": False, "error": str(exc)})
             continue
         req.status = "imported"
         req.imported_path = dest
-        record(session, "imported", req, source=shown, dest=dest)
+        rows.append(record(session, "imported", req, source=shown, dest=dest))
         session.commit()
         results.append({"path": shown, "ok": True, "dest": dest})
     if not results:
         raise HTTPException(400, "nothing to import")
+    connect.schedule(background, rows)
     return {"results": results}

wordarr/api/requests.py

@@ -1,11 +1,11 @@
 from pathlib import Path
 
-from fastapi import APIRouter, Depends, HTTPException
+from fastapi import APIRouter, BackgroundTasks, Depends, HTTPException
 from pydantic import BaseModel, ConfigDict
 from sqlalchemy import select
 from sqlalchemy.orm import Session
 
-from .. import config
+from .. import config, connect
 from ..db import BookRequest, Library, get_session, record
 from ..importer import mover, tagger
 
@@ -193,7 +193,8 @@ def update_request(request_id: int, data: RequestIn, session: Session = Depends(
 
 
 @router.post("/{request_id}/retag")
-def retag_request(request_id: int, session: Session = Depends(get_session)):
+def retag_request(request_id: int, background: BackgroundTasks,
+                  session: Session = Depends(get_session)):
     """Rewrite the audio tags of an already imported request from its current
     metadata (e.g. after adding series/volume/narrator later)."""
     req = session.get(BookRequest, request_id)
@@ -213,19 +214,22 @@ def retag_request(request_id: int, session: Session = Depends(get_session)):
             raise HTTPException(404, f"no audio files in {path}")
         for i, f in enumerate(files, 1):
             tagger.tag_audio(f, req, track=i, total=len(files))
-        record(session, "retagged", req, dest=str(path))
+        row = record(session, "retagged", req, dest=str(path))
         session.commit()
+        connect.schedule(background, [row])
         return {"ok": True, "files": len(files)}
     if path.is_file():
         tagger.tag_audio(path, req)
-        record(session, "retagged", req, dest=str(path))
+        row = record(session, "retagged", req, dest=str(path))
         session.commit()
+        connect.schedule(background, [row])
         return {"ok": True, "files": 1}
     raise HTTPException(404, f"imported path no longer exists: {path}")
 
 
 @router.post("/{request_id}/relocate")
-def relocate_request(request_id: int, session: Session = Depends(get_session)):
+def relocate_request(request_id: int, background: BackgroundTasks,
+                     session: Session = Depends(get_session)):
     """Re-apply the library's naming scheme to an imported title, e.g. after
     fixing a series that Audible had filed differently."""
     req = session.get(BookRequest, request_id)
@@ -241,18 +245,20 @@ def relocate_request(request_id: int, session: Session = Depends(get_session)):
     except FileExistsError as exc:
         raise HTTPException(409, str(exc))
     req.imported_path = dest
-    if dest != old:
+    rows = [record(session, "relocated", req, source=old, dest=dest)] if dest != old else []
-        record(session, "relocated", req, source=old, dest=dest)
     session.commit()
+    connect.schedule(background, rows)
     return {"ok": True, "moved": dest != old, "from": old, "dest": dest}
 
 
 @router.delete("/{request_id}")
-def delete_request(request_id: int, session: Session = Depends(get_session)):
+def delete_request(request_id: int, background: BackgroundTasks,
+                   session: Session = Depends(get_session)):
     req = session.get(BookRequest, request_id)
     if not req:
         raise HTTPException(404, "request not found")
-    record(session, "deleted", req, source=req.imported_path)
+    row = record(session, "deleted", req, source=req.imported_path)
     session.delete(req)
     session.commit()
+    connect.schedule(background, [row])
     return {"ok": True}

wordarr/connect.py

@@ -0,0 +1,60 @@
+"""Tell other services about history rows once they are committed: a generic
+webhook gets every row, Audiobookshelf a scan of the libraries holding new
+files. Runs as a background task, so a slow or dead service never fails the
+request that caused it; errors only reach the log."""
+import logging
+import os
+
+import httpx
+from fastapi import BackgroundTasks
+
+from .db import History
+
+ABS_URL = os.environ.get("WORDARR_ABS_URL", "").strip().rstrip("/")
+ABS_TOKEN = os.environ.get("WORDARR_ABS_TOKEN", "").strip()
+WEBHOOK_URL = os.environ.get("WORDARR_WEBHOOK_URL", "").strip()
+
+log = logging.getLogger(__name__)
+
+
+def schedule(background: BackgroundTasks, rows: list[History]) -> None:
+    """Queue the notification; it runs after the response is sent."""
+    events = [
+        {"event": r.event, "request_id": r.request_id, "title": r.title,
+         "media_type": r.media_type, "source": r.source, "dest": r.dest,
+         "message": r.message}
+        for r in rows
+    ]
+    if events:
+        background.add_task(notify, events)
+
+
+def notify(events: list[dict]) -> None:
+    with httpx.Client(timeout=10) as client:
+        if WEBHOOK_URL:
+            for e in events:
+                try:
+                    client.post(WEBHOOK_URL, json=e).raise_for_status()
+                except httpx.HTTPError as exc:
+                    log.warning("webhook %s failed: %s", WEBHOOK_URL, exc)
+        dests = [e["dest"] for e in events if e["dest"] and e["event"] != "failed"]
+        if ABS_URL and ABS_TOKEN and dests:
+            try:
+                _scan_audiobookshelf(client, dests)
+            except httpx.HTTPError as exc:
+                log.warning("audiobookshelf scan failed: %s", exc)
+
+
+def _scan_audiobookshelf(client: httpx.Client, dests: list[str]) -> None:
+    """Scan only the libraries whose folders contain a destination. That needs
+    both containers to mount the media under the same path."""
+    headers = {"Authorization": f"Bearer {ABS_TOKEN}"}
+    resp = client.get(f"{ABS_URL}/api/libraries", headers=headers)
+    resp.raise_for_status()
+    for lib in resp.json().get("libraries") or []:
+        folders = [str(f.get("fullPath") or "").rstrip("/")
+                   for f in lib.get("folders") or []]
+        if any(f and (d == f or d.startswith(f + "/")) for f in folders for d in dests):
+            client.post(f"{ABS_URL}/api/libraries/{lib.get('id')}/scan",
+                        headers=headers).raise_for_status()
+            log.info("audiobookshelf scans library %s", lib.get("name"))

wordarr/db.py

@@ -68,10 +68,12 @@ class History(Base):
     message: Mapped[str] = mapped_column(String, default="")
 
 
-def record(session, event: str, req: BookRequest, **fields: str) -> None:
+def record(session, event: str, req: BookRequest, **fields: str) -> History:
     """Add a history row; it is saved with the caller's next commit."""
-    session.add(History(event=event, request_id=req.id, title=req.title,
+    row = History(event=event, request_id=req.id, title=req.title,
-                        media_type=req.media_type, **fields))
+                  media_type=req.media_type, **fields)
+    session.add(row)
+    return row
 
 
 _engine = None