abs.py

"""
abs.py — Audiobookshelf API client and processed book tracker.
"""

import json
import time
from pathlib import Path

import requests


class ABSClient:
  """Audiobookshelf API client."""

  def __init__(self, base_url: str, token: str):
    self.base_url = base_url.rstrip("/")
    self.s = requests.Session()
    self.s.headers["Authorization"] = f"Bearer {token}"

  def get_libraries(self) -> list:
    r = self.s.get(f"{self.base_url}/api/libraries")
    r.raise_for_status()
    return r.json()["libraries"]

  def get_library_items(self, library_id: str) -> list:
    items, page = [], 0
    while True:
      r = self.s.get(
          f"{self.base_url}/api/libraries/{library_id}/items",
          params={"limit": 50, "page": page},
      )
      r.raise_for_status()
      data = r.json()
      items.extend(data["results"])
      if len(items) >= data["total"]:
        break
      page += 1
    return items

  def get_all_books(self, library_name: str = None) -> list:
    """Get all book items, optionally filtered to one library."""
    libraries = self.get_libraries()
    book_libs = [lib for lib in libraries if lib.get(
        "mediaType") == "book"]
    if library_name:
      book_libs = [lib for lib in book_libs
                   if lib["name"].lower() == library_name.lower()]
    items = []
    for lib in book_libs:
      items.extend(self.get_library_items(lib["id"]))
    return items

  def update_metadata(self, item_id: str, metadata: dict) -> tuple[bool, str]:
    r = self.s.patch(
        f"{self.base_url}/api/items/{item_id}/media",
        json={"metadata": metadata},
    )
    return r.status_code == 200, r.text

  def set_cover(self, item_id: str, image_url: str) -> bool:
    r = self.s.post(
        f"{self.base_url}/api/items/{item_id}/cover",
        json={"url": image_url},
    )
    return r.status_code == 200


class ProcessedTracker:
  """Tracks which books have been processed."""

  def __init__(self, path: Path):
    self.path = path
    self.data = {}
    if path.exists():
      try:
        self.data = json.loads(path.read_text())
      except Exception:
        self.data = {}

  def is_done(self, item_id: str) -> bool:
    return item_id in self.data

  def mark(self, item_id: str, status: str, match: str = ""):
    self.data[item_id] = {
        "status": status,
        "match": match,
        "time": time.strftime("%Y-%m-%d %H:%M:%S"),
    }
    self.path.write_text(json.dumps(self.data, indent=2))