[books] Add an openlibrary source

vrobbler/apps/books/sources/openlibrary.py

@@ -0,0 +1,132 @@
+import logging
+
+logger = logging.getLogger(__name__)
+
+
+def lookup_book_from_openlibrary(title: str, author: str | None = None) -> dict:
+    try:
+        from olclient import OpenLibrary
+    except ImportError:
+        logger.warning("openlibrary-client not installed")
+        return {}
+
+    ol = OpenLibrary()
+
+    try:
+        work_obj = ol.Work.search(title)
+    except Exception as e:
+        logger.warning(f"Could not find work for {title}: {e}")
+        return {}
+
+    if not work_obj:
+        return {}
+
+    olid = work_obj.identifiers.get("olid", [None])[0]
+    if not olid:
+        return {}
+
+    try:
+        work_json = ol.Work.get(olid)
+    except Exception as e:
+        logger.warning(f"Could not get work JSON for {olid}: {e}")
+        return {}
+
+    book_dict: dict = {"title": work_json.title, "openlibrary_id": olid}
+
+    if description := getattr(work_json, "description", None):
+        if isinstance(description, dict):
+            book_dict["summary"] = description.get("value", "")
+        else:
+            book_dict["summary"] = str(description)
+
+    if subjects := getattr(work_json, "subjects", None):
+        book_dict["genres"] = subjects[:10]
+
+    covers = getattr(work_json, "covers", None)
+    if covers:
+        book_dict["cover_url"] = (
+            f"https://covers.openlibrary.org/b/id/{covers[0]}-L.jpg"
+        )
+
+    try:
+        editions_data = ol.get(f"{olid}/editions.json")
+        if editions_data and (entries := editions_data.get("entries")):
+            if entries:
+                edition = entries[0]
+                if isbn_list := edition.get("isbn_10"):
+                    book_dict["isbn_10"] = isbn_list[0]
+                if isbn_list := edition.get("isbn_13"):
+                    book_dict["isbn_13"] = isbn_list[0]
+                if publishers := edition.get("publishers"):
+                    book_dict["publisher"] = publishers[0]
+                if pub_date := edition.get("publish_date"):
+                    book_dict["publish_date"] = pub_date
+                    if len(pub_date) >= 4:
+                        try:
+                            book_dict["first_publish_year"] = int(pub_date[:4])
+                        except ValueError:
+                            pass
+                if pages := edition.get("number_of_pages_median"):
+                    book_dict["pages"] = pages
+                if edition_covers := edition.get("covers"):
+                    if edition_covers and not book_dict.get("cover_url"):
+                        book_dict["cover_url"] = (
+                            f"https://covers.openlibrary.org/b/id/{edition_covers[0]}-L.jpg"
+                        )
+    except Exception as e:
+        logger.warning(f"Could not get editions for {olid}: {e}")
+
+    if author:
+        book_dict["authors"] = [author]
+    elif work_obj.authors:
+        book_dict["authors"] = [
+            a.get("name", "") for a in work_obj.authors if a.get("name")
+        ]
+
+    return book_dict
+
+
+def lookup_author_from_openlibrary(name: str) -> dict:
+    try:
+        from olclient import OpenLibrary
+    except ImportError:
+        logger.warning("openlibrary-client not installed")
+        return {}
+
+    ol = OpenLibrary()
+
+    try:
+        author_list = ol.Author.search(name)
+    except Exception as e:
+        logger.warning(f"Could not find author {name}: {e}")
+        return {}
+
+    if not author_list or not isinstance(author_list, list):
+        return {}
+
+    author_json = author_list[0]
+    author_olid = author_json.get("key", "").replace("/authors/", "")
+
+    if not author_olid:
+        return {}
+
+    author_dict: dict = {
+        "name": author_json.get("name", name),
+        "openlibrary_id": author_olid,
+    }
+
+    if author_json.get("bio"):
+        bio = author_json.get("bio")
+        if isinstance(bio, dict):
+            author_dict["bio"] = bio.get("value", "")
+        else:
+            author_dict["bio"] = str(bio)
+
+    if wikipedia := author_json.get("wikipedia"):
+        author_dict["wikipedia_url"] = wikipedia
+
+    author_dict["author_headshot_url"] = (
+        f"https://covers.openlibrary.org/a/olid/{author_olid}-L.jpg"
+    )
+
+    return author_dict

vrobbler/apps/charts/management/commands/cleanup_duplicate_chartrecords.py

@@ -0,0 +1,87 @@
+import logging
+
+from django.core.management.base import BaseCommand
+from django.db.models import Count, Max
+
+logger = logging.getLogger(__name__)
+
+MEDIA_FIELDS = [
+    "artist",
+    "album",
+    "track",
+    "tv_series",
+    "video",
+    "podcast",
+    "podcast_episode",
+    "board_game",
+    "trail",
+    "geo_location",
+    "food",
+    "book",
+]
+
+
+class Command(BaseCommand):
+    help = "Clean up duplicate ChartRecords"
+
+    def add_arguments(self, parser):
+        parser.add_argument(
+            "--dry-run",
+            action="store_true",
+            help="Show what would be deleted without actually deleting",
+        )
+
+    def handle(self, *args, **options):
+        from charts.models import ChartRecord
+
+        dry_run = options["dry_run"]
+
+        total_deleted = 0
+
+        for media_field in MEDIA_FIELDS:
+            field_id = f"{media_field}_id"
+
+            duplicates = (
+                ChartRecord.objects.filter(**{f"{field_id}__isnull": False})
+                .values("user_id", "year", "month", "week", "day", field_id)
+                .annotate(
+                    count=Count("id"),
+                    max_id=Max("id"),
+                )
+                .filter(count__gt=1)
+            )
+
+            for dup in duplicates:
+                max_id = dup["max_id"]
+                to_delete = list(
+                    ChartRecord.objects.filter(
+                        user_id=dup["user_id"],
+                        year=dup["year"],
+                        month=dup["month"],
+                        week=dup["week"],
+                        day=dup["day"],
+                        **{field_id: dup[field_id]},
+                    )
+                    .exclude(id=max_id)
+                    .values_list("id", flat=True)
+                )
+
+                if dry_run:
+                    self.stdout.write(
+                        f"Would delete {len(to_delete)} duplicate ChartRecords "
+                        f"({media_field}, user={dup['user_id']}, year={dup['year']}, "
+                        f"month={dup['month']}, week={dup['week']}, day={dup['day']})"
+                    )
+                else:
+                    deleted, _ = ChartRecord.objects.filter(id__in=to_delete).delete()
+                    total_deleted += deleted
+                    self.stdout.write(
+                        f"Deleted {deleted} duplicate ChartRecords "
+                        f"({media_field}, user={dup['user_id']}, year={dup['year']}, "
+                        f"month={dup['month']}, week={dup['week']}, day={dup['day']})"
+                    )
+
+        if dry_run:
+            self.stdout.write(self.style.WARNING("Dry run complete - no changes made"))
+        else:
+            self.stdout.write(self.style.SUCCESS(f"Total deleted: {total_deleted}"))