From 49f14108143418bd130e71ef47c5cfacae1c08ca Mon Sep 17 00:00:00 2001 From: Colin Powell Date: Thu, 26 Mar 2026 15:52:45 -0400 Subject: [PATCH] [books] Add an openlibrary source --- vrobbler/apps/books/sources/openlibrary.py | 132 ++++++++++++++++++ .../cleanup_duplicate_chartrecords.py | 87 ++++++++++++ 2 files changed, 219 insertions(+) create mode 100644 vrobbler/apps/books/sources/openlibrary.py create mode 100644 vrobbler/apps/charts/management/commands/cleanup_duplicate_chartrecords.py diff --git a/vrobbler/apps/books/sources/openlibrary.py b/vrobbler/apps/books/sources/openlibrary.py new file mode 100644 index 0000000..aecf4b5 --- /dev/null +++ b/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 diff --git a/vrobbler/apps/charts/management/commands/cleanup_duplicate_chartrecords.py b/vrobbler/apps/charts/management/commands/cleanup_duplicate_chartrecords.py new file mode 100644 index 0000000..bf7c505 --- /dev/null +++ b/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}"))