scraper.py

#!/usr/bin/env python3
"""
Audiobookshelf Metadata Enricher

Multi-pass Audible search with fuzzy matching and interactive fallback.

.env file (same directory as script):
    ABS_URL=http://localhost:13378
    ABS_TOKEN=your_token
    AUDIBLE_REGION=us

Usage:
    python scraper.py                          # Process all new books
    python scraper.py --force                  # Reprocess everything
    python scraper.py --author "Tolkien"       # Only Tolkien books
    python scraper.py --series "Dune"          # Only Dune series
    python scraper.py --title "The Hobbit"     # Specific book
    python scraper.py --dry-run                # Preview only
    python scraper.py --interactive            # Prompt on failures
    python scraper.py -i --force --author "Flanagan"
"""

import argparse
import json
import logging
import os
import re
import sys
import time
from pathlib import Path

try:
  from dotenv import load_dotenv
  load_dotenv(Path(__file__).parent / ".env")
except ImportError:
  pass

import requests

from audible import Audible, auto_pick, clean_title

# ============================================================
# Config
# ============================================================

ABS_URL = os.environ.get("ABS_URL", "")
ABS_TOKEN = os.environ.get("ABS_TOKEN", "")
AUDIBLE_REGION = os.environ.get("AUDIBLE_REGION", "us")

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(message)s",
    handlers=[
        logging.StreamHandler(),
        logging.FileHandler(Path(__file__).parent / "enrich.log"),
    ],
)
log = logging.getLogger(__name__)


# ============================================================
# Audiobookshelf Client
# ============================================================

class ABSClient:
  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):
    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):
    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 update_metadata(self, item_id: str, metadata: dict):
    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):
    r = self.s.post(f"{self.base_url}/api/items/{item_id}/cover",
                    json={"url": image_url})
    return r.status_code == 200


# ============================================================
# Processed Tracker
# ============================================================

class ProcessedTracker:
  def __init__(self, path: Path):
    self.path = path
    self.data = self._load()

  def _load(self) -> dict:
    if self.path.exists():
      try:
        return json.loads(self.path.read_text())
      except Exception:
        return {}
    return {}

  def save(self):
    self.path.write_text(json.dumps(self.data, indent=2))

  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.save()


# ============================================================
# Interactive Helpers
# ============================================================

def show_results(results: list):
  for i, r in enumerate(results):
    parts = [f"  [{i}] {r.get('title', '?')}"]
    if r.get("author"):
      parts.append(f"by {r['author']}")
    if r.get("narrator"):
      parts.append(f"(narr. {r['narrator']})")
    if r.get("series"):
      seq = r.get("series_sequence", "")
      parts.append(f"[{r['series']}{' #' + seq if seq else ''}]")
    if r.get("runtime"):
      parts.append(f"- {r['runtime']}")
    print(" ".join(parts))


def interactive_pick(title: str, author: str, results: list, audible: Audible):
  while True:
    print(f"\n--- No auto-match for: \"{title}\" by {author} ---")
    if results:
      show_results(results)
    else:
      print("  (no results)")

    print(
        "\nOptions: [0-9] pick | [s] search | [a] ASIN | [k] skip | [q] quit")
    choice = input("> ").strip().lower()

    if choice == 'q':
      sys.exit(0)
    elif choice == 'k':
      return None
    elif choice == 'a':
      asin = input("ASIN: ").strip().upper()
      if re.match(r'^[A-Z0-9]{10}$', asin):
        book = audible.get_by_asin(asin)
        if book:
          print(f"  Found: {book.get('title', asin)}")
          return book
        print("  Not found.")
      else:
        print("  Invalid ASIN (10 alphanumeric chars).")
    elif choice == 's':
      query = input("Search: ").strip()
      if query:
        results = audible.search(query)
        if results:
          show_results(results)
        else:
          print("  No results.")
    elif choice.isdigit():
      idx = int(choice)
      if 0 <= idx < len(results):
        return results[idx]
      print(f"  Pick 0-{len(results)-1}.")
    else:
      print("  Invalid.")


# ============================================================
# Enricher
# ============================================================

class Enricher:
  def __init__(self, abs_client, audible, dry_run=False, force=False,
               interactive=False, filter_author=None, filter_series=None,
               filter_title=None):
    self.abs = abs_client
    self.audible = audible
    self.dry_run = dry_run
    self.force = force
    self.interactive = interactive
    self.filter_author = filter_author
    self.filter_series = filter_series
    self.filter_title = filter_title
    self.tracker = ProcessedTracker(
        Path(__file__).parent / "processed.json")
    self.stats = {"total": 0, "matched": 0,
                  "skipped": 0, "failed": 0, "no_match": 0}

  def run(self, library_name: str = None):
    libraries = self.abs.get_libraries()
    book_libs = [l for l in libraries if l.get("mediaType") == "book"]

    if library_name:
      book_libs = [l for l in book_libs if l["name"].lower()
                   == library_name.lower()]
      if not book_libs:
        log.error(f"Library '{library_name}' not found")
        return

    for lib in book_libs:
      log.info(f"Processing library: {lib['name']}")
      items = self.abs.get_library_items(lib["id"])
      log.info(f"Found {len(items)} items")

      items = self._filter(items)
      if self.filter_author or self.filter_series or self.filter_title:
        log.info(f"Filtered to {len(items)} items")

      for item in items:
        self.stats["total"] += 1
        if not self.force and self.tracker.is_done(item["id"]):
          t = item.get("media", {}).get(
              "metadata", {}).get("title", "?")
          log.info(f"  Skipping (processed): {t}")
          self.stats["skipped"] += 1
          continue
        self._process(item)
        time.sleep(1)

    self._print_stats()

  def _filter(self, items):
    if not (self.filter_author or self.filter_series or self.filter_title):
      return items
    out = []
    for item in items:
      m = item.get("media", {}).get("metadata", {})
      title = m.get("title", "").lower()
      author = m.get("authorName", "").lower()
      if not author and m.get("authors"):
        author = m["authors"][0].get("name", "").lower()
      series = ""
      if m.get("series") and isinstance(m["series"], list) and m["series"]:
        series = m["series"][0].get("name", "").lower()
      if self.filter_author and self.filter_author.lower() not in author:
        continue
      if self.filter_series and self.filter_series.lower() not in series:
        continue
      if self.filter_title and self.filter_title.lower() not in title:
        continue
      out.append(item)
    return out

  def _process(self, item):
    meta = item.get("media", {}).get("metadata", {})
    title = meta.get("title", "Unknown")
    author = meta.get("authorName", "")
    if not author and meta.get("authors"):
      author = meta["authors"][0].get("name", "")
    item_id = item["id"]

    log.info(f"Processing: {title} by {author}")

    # Collect all results across passes
    all_results = []

    # Pass 1: full title + author
    r1 = self.audible.search(f"{title} {author}")
    idx = auto_pick(title, author, r1)
    if idx is not None:
      return self._apply(item_id, title, r1[idx])
    all_results.extend(r1)

    # Pass 2: cleaned title + author
    cleaned = clean_title(title)
    if cleaned and cleaned != title:
      log.info(f"  Pass 2: '{cleaned} {author}'")
      r2 = self.audible.search(f"{cleaned} {author}")
      idx = auto_pick(cleaned, author, r2)
      if idx is not None:
        return self._apply(item_id, title, r2[idx])
      self._merge_results(all_results, r2)

    # Pass 3: cleaned title only
    if cleaned:
      log.info(f"  Pass 3: '{cleaned}'")
      r3 = self.audible.search(cleaned)
      idx = auto_pick(cleaned, author, r3)
      if idx is not None:
        return self._apply(item_id, title, r3[idx])
      self._merge_results(all_results, r3)

    # Interactive fallback
    if self.interactive:
      picked = interactive_pick(title, author, all_results, self.audible)
      if picked:
        return self._apply(item_id, title, picked)
      self.tracker.mark(item_id, "skipped_interactive")
      self.stats["no_match"] += 1
      return

    log.warning(f"  No match: {title}")
    self.tracker.mark(item_id, "no_match")
    self.stats["no_match"] += 1

  def _merge_results(self, target: list, new: list):
    seen = {r.get("asin") for r in target if r.get("asin")}
    for r in new:
      if r.get("asin") not in seen:
        target.append(r)
        if r.get("asin"):
          seen.add(r["asin"])

  def _apply(self, item_id: str, original_title: str, match: dict):
    details = {}
    if match.get("url"):
      details = self.audible.get_details(match["url"])
      time.sleep(0.5)

    merged = {**match, **details}

    if self.dry_run:
      preview = {k: v for k, v in merged.items()
                 if k in ("title", "author", "narrator", "series",
                          "series_sequence", "asin", "cover_url")}
      log.info(
          f"  [DRY RUN] Would update:\n{json.dumps(preview, indent=2)}")
      self.tracker.mark(item_id, "dry_run", merged.get("title", ""))
      self.stats["matched"] += 1
      return

    update = {}
    for src, dst in [
        ("title", "title"), ("author",
                             "authorName"), ("narrator", "narratorName"),
        ("description", "description"), ("publisher", "publisher"),
        ("asin", "asin"), ("language", "language"),
    ]:
      if merged.get(src):
        update[dst] = merged[src]

    if merged.get("genres"):
      update["genres"] = merged["genres"]

    if merged.get("series"):
      update["series"] = [{
          "name": merged["series"],
          "sequence": merged.get("series_sequence", ""),
      }]

    if update:
      ok, msg = self.abs.update_metadata(item_id, update)
      if ok:
        log.info(f"  Updated metadata: {original_title}")
      else:
        log.error(f"  Failed: {msg}")
        self.tracker.mark(item_id, "failed", merged.get("title", ""))
        self.stats["failed"] += 1
        return

    cover = merged.get("cover_url")
    if cover and self.abs.set_cover(item_id, cover):
      log.info(f"  Updated cover: {original_title}")

    self.tracker.mark(item_id, "matched", merged.get("title", ""))
    self.stats["matched"] += 1

  def _print_stats(self):
    log.info("=" * 50)
    log.info("Enrichment Complete")
    log.info(f"  Total:      {self.stats['total']}")
    log.info(f"  Matched:    {self.stats['matched']}")
    log.info(f"  Skipped:    {self.stats['skipped']}")
    log.info(f"  No Match:   {self.stats['no_match']}")
    log.info(f"  Failed:     {self.stats['failed']}")
    log.info("=" * 50)


# ============================================================
# CLI
# ============================================================

def main():
  p = argparse.ArgumentParser(
      description="Enrich Audiobookshelf metadata from Audible")
  p.add_argument("--abs-url", default=ABS_URL)
  p.add_argument("--abs-token", default=ABS_TOKEN)
  p.add_argument("--region", default=AUDIBLE_REGION,
                 choices=["us", "uk", "de", "fr"])
  p.add_argument("--library", default=None, help="Process only this library")
  p.add_argument("--author", default=None, help="Filter by author (partial)")
  p.add_argument("--series", default=None, help="Filter by series (partial)")
  p.add_argument("--title", default=None, help="Filter by title (partial)")
  p.add_argument("--dry-run", action="store_true",
                 help="Preview without changes")
  p.add_argument("--force", action="store_true", help="Reprocess all")
  p.add_argument("--interactive", "-i", action="store_true",
                 help="Prompt on failures")
  args = p.parse_args()

  if not args.abs_url or not args.abs_token:
    print("Error: ABS_URL and ABS_TOKEN required (set in .env or pass as args)")
    sys.exit(1)

  enricher = Enricher(
      abs_client=ABSClient(args.abs_url, args.abs_token),
      audible=Audible(args.region),
      dry_run=args.dry_run,
      force=args.force,
      interactive=args.interactive,
      filter_author=args.author,
      filter_series=args.series,
      filter_title=args.title,
  )
  enricher.run(args.library)


if __name__ == "__main__":
  main()