302 lines
9.8 KiB
Python
302 lines
9.8 KiB
Python
import logging
|
|
import time
|
|
|
|
from django.core.management.base import BaseCommand
|
|
from django.db import transaction
|
|
|
|
from books.constants import READCOMICSONLINE_URL
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
MISSING_ALL = [
|
|
"cover",
|
|
"summary",
|
|
"isbn",
|
|
"pages",
|
|
"language",
|
|
"publisher",
|
|
"publish_year",
|
|
]
|
|
|
|
MISSING_GROUPS = {
|
|
"cover": lambda b: not bool(b.cover),
|
|
"summary": lambda b: not b.summary,
|
|
"isbn": lambda b: not b.isbn_13 and not b.isbn_10,
|
|
"pages": lambda b: b.pages is None,
|
|
"language": lambda b: not b.language,
|
|
"publisher": lambda b: not b.publisher,
|
|
"publish_year": lambda b: b.first_publish_year is None,
|
|
}
|
|
|
|
|
|
def _book_matches(book, flags):
|
|
if not flags:
|
|
return False
|
|
for flag in flags:
|
|
fn = MISSING_GROUPS.get(flag)
|
|
if fn and fn(book):
|
|
return True
|
|
return False
|
|
|
|
|
|
class Command(BaseCommand):
|
|
help = "Backfill missing metadata on books from Google Books, OpenLibrary, and ComicVine"
|
|
|
|
def add_arguments(self, parser):
|
|
parser.add_argument(
|
|
"--commit",
|
|
action="store_true",
|
|
help="Commit changes to the database",
|
|
)
|
|
parser.add_argument(
|
|
"--batch-size",
|
|
type=int,
|
|
default=100,
|
|
help="Number of books to process per batch (default: 100)",
|
|
)
|
|
parser.add_argument(
|
|
"--sleep",
|
|
type=float,
|
|
default=0.5,
|
|
help="Seconds to sleep between API calls (default: 0.5)",
|
|
)
|
|
for flag in MISSING_ALL:
|
|
parser.add_argument(
|
|
f"--missing-{flag}",
|
|
dest="missing_flags",
|
|
action="append_const",
|
|
const=flag,
|
|
help=f"Process books missing {flag}",
|
|
)
|
|
parser.add_argument(
|
|
"--comics-only",
|
|
action="store_true",
|
|
help="Only process books with a readcomicsonline.ru URL",
|
|
)
|
|
parser.add_argument(
|
|
"--all",
|
|
action="store_true",
|
|
dest="all_missing",
|
|
help="Process books missing any metadata field",
|
|
)
|
|
|
|
def handle(self, *args, **options):
|
|
from books.models import Book
|
|
|
|
commit = options["commit"]
|
|
batch_size = options["batch_size"]
|
|
sleep_secs = options["sleep"]
|
|
flags = options.get("missing_flags") or []
|
|
comics_only = options["comics_only"]
|
|
all_missing = options["all_missing"]
|
|
|
|
if all_missing:
|
|
flags = MISSING_ALL
|
|
|
|
if not flags and not comics_only:
|
|
self.stdout.write(
|
|
"No filters specified. Use --all, --missing-*, or --comics-only."
|
|
)
|
|
return
|
|
|
|
qs = Book.objects.all()
|
|
if comics_only:
|
|
qs = qs.filter(readcomics_url__isnull=False)
|
|
|
|
if flags:
|
|
qs = [b for b in qs.iterator() if _book_matches(b, flags)]
|
|
else:
|
|
qs = list(qs)
|
|
|
|
total = len(qs)
|
|
self.stdout.write(f"Found {total} books to process")
|
|
|
|
if not commit:
|
|
self.stdout.write(
|
|
"Dry run — no API calls will be made. Use --commit to run lookups."
|
|
)
|
|
return
|
|
|
|
enriched = 0
|
|
skipped = 0
|
|
stats = {
|
|
"cover_fixed": 0,
|
|
"summary_fixed": 0,
|
|
"isbn_fixed": 0,
|
|
"pages_fixed": 0,
|
|
"language_fixed": 0,
|
|
"publisher_fixed": 0,
|
|
"publish_year_fixed": 0,
|
|
}
|
|
|
|
for batch_num, offset in enumerate(range(0, len(qs), batch_size)):
|
|
batch = qs[offset : offset + batch_size]
|
|
for book in batch:
|
|
result = self._enrich_book(book, sleep_secs)
|
|
if result:
|
|
enriched += 1
|
|
for key in stats:
|
|
if result.get(key):
|
|
stats[key] += 1
|
|
else:
|
|
skipped += 1
|
|
|
|
self.stdout.write(
|
|
f" Batch {batch_num + 1}: {offset + len(batch)}/{total} — "
|
|
f"enriched: {enriched}, skipped: {skipped}"
|
|
)
|
|
|
|
self.stdout.write(
|
|
f"\nResults (commit={commit}):\n"
|
|
f" Books enriched: {enriched}\n"
|
|
f" Books skipped: {skipped}\n"
|
|
f" Covers fixed: {stats['cover_fixed']}\n"
|
|
f" Summaries fixed:{stats['summary_fixed']}\n"
|
|
f" ISBNs fixed: {stats['isbn_fixed']}\n"
|
|
f" Pages fixed: {stats['pages_fixed']}\n"
|
|
f" Languages fixed:{stats['language_fixed']}\n"
|
|
f" Publishers fixed:{stats['publisher_fixed']}\n"
|
|
f" Publish yrs fixed: {stats['publish_year_fixed']}"
|
|
)
|
|
|
|
def _enrich_book(self, book, sleep_secs):
|
|
from books.sources.comicvine import (
|
|
lookup_comic_from_comicvine,
|
|
lookup_issue_by_comicvine_id,
|
|
)
|
|
from books.sources.google import lookup_book_from_google
|
|
from books.sources.openlibrary import lookup_book_from_openlibrary as lookup_book_from_ol
|
|
|
|
title = book.original_title or book.title
|
|
author_name = book.author.name if book.author else None
|
|
book_dict = {}
|
|
|
|
cv_data = None
|
|
if book.comicvine_id:
|
|
cv_data = lookup_issue_by_comicvine_id(str(book.comicvine_id))
|
|
if not cv_data:
|
|
cv_data = lookup_comic_from_comicvine(title)
|
|
if cv_data:
|
|
book_dict.update(cv_data)
|
|
|
|
ol_data = lookup_book_from_ol(title, author=author_name)
|
|
time.sleep(sleep_secs)
|
|
google_data = lookup_book_from_google(title)
|
|
|
|
if ol_data:
|
|
for k, v in ol_data.items():
|
|
book_dict.setdefault(k, v)
|
|
|
|
if google_data:
|
|
for k, v in google_data.items():
|
|
if v:
|
|
book_dict.setdefault(k, v)
|
|
|
|
if not book_dict:
|
|
return None
|
|
|
|
changed = self._apply(book, book_dict, title)
|
|
return changed
|
|
|
|
def _apply(self, book, data, title):
|
|
changed = {
|
|
"cover_fixed": False,
|
|
"summary_fixed": False,
|
|
"isbn_fixed": False,
|
|
"pages_fixed": False,
|
|
"language_fixed": False,
|
|
"publisher_fixed": False,
|
|
"publish_year_fixed": False,
|
|
}
|
|
|
|
update_fields = []
|
|
|
|
cover_url = data.pop("cover_url", "")
|
|
|
|
if data.get("summary") and not book.summary:
|
|
book.summary = data["summary"]
|
|
update_fields.append("summary")
|
|
changed["summary_fixed"] = True
|
|
|
|
if data.get("isbn_13") and not book.isbn_13:
|
|
book.isbn_13 = data["isbn_13"]
|
|
update_fields.append("isbn_13")
|
|
changed["isbn_fixed"] = True
|
|
|
|
if data.get("isbn_10") and not book.isbn_10:
|
|
book.isbn_10 = data["isbn_10"]
|
|
update_fields.append("isbn_10")
|
|
changed["isbn_fixed"] = True
|
|
|
|
if data.get("pages") and book.pages is None:
|
|
book.pages = data["pages"]
|
|
update_fields.append("pages")
|
|
changed["pages_fixed"] = True
|
|
|
|
if data.get("language") and not book.language:
|
|
book.language = data["language"]
|
|
update_fields.append("language")
|
|
changed["language_fixed"] = True
|
|
|
|
if data.get("publisher") and not book.publisher:
|
|
book.publisher = data["publisher"]
|
|
update_fields.append("publisher")
|
|
changed["publisher_fixed"] = True
|
|
|
|
if data.get("first_publish_year") and book.first_publish_year is None:
|
|
book.first_publish_year = data["first_publish_year"]
|
|
update_fields.append("first_publish_year")
|
|
changed["publish_year_fixed"] = True
|
|
|
|
if data.get("openlibrary_id") and not book.openlibrary_id:
|
|
book.openlibrary_id = data["openlibrary_id"]
|
|
update_fields.append("openlibrary_id")
|
|
|
|
if data.get("comicvine_id") and not book.comicvine_id:
|
|
book.comicvine_id = data["comicvine_id"]
|
|
update_fields.append("comicvine_id")
|
|
|
|
if data.get("issue_number") and book.issue_number is None:
|
|
book.issue_number = data["issue_number"]
|
|
update_fields.append("issue_number")
|
|
|
|
if data.get("volume_number") and book.volume_number is None:
|
|
book.volume_number = data["volume_number"]
|
|
update_fields.append("volume_number")
|
|
|
|
if data.get("volume") and not book.volume:
|
|
book.volume = data["volume"]
|
|
update_fields.append("volume")
|
|
|
|
if data.get("volume_comicvine_id") and not book.volume_comicvine_id:
|
|
book.volume_comicvine_id = data["volume_comicvine_id"]
|
|
update_fields.append("volume_comicvine_id")
|
|
|
|
if update_fields:
|
|
book.save(update_fields=update_fields)
|
|
self.stdout.write(f" [ENRICHED] {book} — {', '.join(update_fields)}")
|
|
|
|
if cover_url and not book.cover:
|
|
book.save_image_from_url(cover_url)
|
|
if book.cover:
|
|
changed["cover_fixed"] = True
|
|
self.stdout.write(f" [COVER] {book} — cover saved from source")
|
|
|
|
genres = data.pop("genres", data.pop("generes", []))
|
|
if genres:
|
|
existing = set(book.genre.names())
|
|
new_genres = [g for g in genres if g not in existing]
|
|
if new_genres:
|
|
book.genre.add(*new_genres)
|
|
self.stdout.write(f" [GENRES] {book} — added {len(new_genres)} genres")
|
|
|
|
tags = data.pop("tags", [])
|
|
if tags:
|
|
existing_tags = set(book.tags.names())
|
|
new_tags = [t for t in tags if t not in existing_tags]
|
|
if new_tags:
|
|
book.tags.add(*new_tags)
|
|
self.stdout.write(f" [TAGS] {book} — added {', '.join(new_tags)}")
|
|
|
|
return changed if any(changed.values()) else None
|