541 lines
21 KiB
Python
541 lines
21 KiB
Python
import logging
|
|
import time
|
|
|
|
from django.core.management.base import BaseCommand
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
MISSING_ALL = [
|
|
"cover",
|
|
"screenshot",
|
|
"summary",
|
|
"rating",
|
|
"release_date",
|
|
"release_year",
|
|
"igdb_id",
|
|
"hltb_id",
|
|
]
|
|
|
|
MISSING_GROUPS = {
|
|
"cover": lambda g: _image_missing_or_broken(g, "cover"),
|
|
"screenshot": lambda g: _image_missing_or_broken(g, "screenshot"),
|
|
"summary": lambda g: not g.summary,
|
|
"rating": lambda g: g.rating is None,
|
|
"release_date": lambda g: g.release_date is None,
|
|
"release_year": lambda g: g.release_year is None,
|
|
"igdb_id": lambda g: g.igdb_id is None,
|
|
"hltb_id": lambda g: g.hltb_id is None,
|
|
}
|
|
|
|
|
|
def _image_missing_or_broken(game, field_name) -> bool:
|
|
field = getattr(game, field_name)
|
|
if not bool(field):
|
|
return True
|
|
try:
|
|
return not field.storage.exists(field.name)
|
|
except Exception:
|
|
return True
|
|
|
|
|
|
def _game_matches(game, flags):
|
|
if not flags:
|
|
return False
|
|
for flag in flags:
|
|
fn = MISSING_GROUPS.get(flag)
|
|
if fn and fn(game):
|
|
return True
|
|
return False
|
|
|
|
|
|
class Command(BaseCommand):
|
|
help = "Backfill missing metadata on video games from IGDB and HowLongToBeat"
|
|
|
|
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 games 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)",
|
|
)
|
|
parser.add_argument(
|
|
"--force",
|
|
action="store_true",
|
|
help="Re-fetch metadata even if data already exists",
|
|
)
|
|
parser.add_argument(
|
|
"--game-id",
|
|
type=int,
|
|
help="Only process a specific game by ID",
|
|
)
|
|
parser.add_argument(
|
|
"--fix-broken-images",
|
|
action="store_true",
|
|
help="Check and refetch broken/deleted game images (cover, screenshot, hltb_cover)",
|
|
)
|
|
for flag in MISSING_ALL:
|
|
parser.add_argument(
|
|
f"--missing-{flag}",
|
|
dest="missing_flags",
|
|
action="append_const",
|
|
const=flag,
|
|
help=f"Process games missing {flag}",
|
|
)
|
|
parser.add_argument(
|
|
"--all",
|
|
action="store_true",
|
|
dest="all_missing",
|
|
help="Process games missing any metadata field",
|
|
)
|
|
|
|
def handle(self, *args, **options):
|
|
from videogames.models import VideoGame
|
|
|
|
commit = options["commit"]
|
|
batch_size = options["batch_size"]
|
|
sleep_secs = options["sleep"]
|
|
force = options["force"]
|
|
game_id = options["game_id"]
|
|
fix_broken_images = options.get("fix_broken_images", False)
|
|
flags = options.get("missing_flags") or []
|
|
all_missing = options["all_missing"]
|
|
|
|
if all_missing:
|
|
flags = MISSING_ALL
|
|
fix_broken_images = True
|
|
|
|
if not flags and not game_id and not force and not fix_broken_images:
|
|
self.stdout.write(
|
|
"No filters specified. Use --all, --missing-*, --game-id, --force, or --fix-broken-images."
|
|
)
|
|
return
|
|
|
|
if game_id:
|
|
qs = VideoGame.objects.filter(id=game_id)
|
|
else:
|
|
qs = VideoGame.objects.all()
|
|
|
|
if flags:
|
|
qs = [g for g in qs.iterator() if _game_matches(g, flags)]
|
|
else:
|
|
qs = list(qs)
|
|
|
|
total = len(qs)
|
|
self.stdout.write(f"Found {total} games to process")
|
|
|
|
if not commit:
|
|
self.stdout.write(
|
|
"Dry run — no API calls will be made. Use --commit to run lookups."
|
|
)
|
|
return
|
|
|
|
title_mismatches = []
|
|
enriched = 0
|
|
skipped = 0
|
|
stats = {
|
|
"cover_fixed": 0,
|
|
"screenshot_fixed": 0,
|
|
"summary_fixed": 0,
|
|
"rating_fixed": 0,
|
|
"release_date_fixed": 0,
|
|
"release_year_fixed": 0,
|
|
"igdb_id_found": 0,
|
|
"hltb_id_found": 0,
|
|
"images_fixed": 0,
|
|
}
|
|
|
|
enriched_any = bool(flags or game_id or force)
|
|
|
|
if enriched_any:
|
|
for batch_num, offset in enumerate(range(0, len(qs), batch_size)):
|
|
batch = qs[offset : offset + batch_size]
|
|
for game in batch:
|
|
result = self._enrich_game(game, sleep_secs, force)
|
|
self._check_retroarch_name(game, title_mismatches)
|
|
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}"
|
|
)
|
|
|
|
if fix_broken_images:
|
|
broken_stats = self._fix_broken_images(qs, sleep_secs)
|
|
stats["images_fixed"] = broken_stats["images_fixed"]
|
|
|
|
self.stdout.write(
|
|
f"\nResults (commit={commit}):\n"
|
|
f" Games enriched: {enriched}\n"
|
|
f" Games skipped: {skipped}\n"
|
|
f" Covers fixed: {stats['cover_fixed']}\n"
|
|
f" Screenshots fixed: {stats['screenshot_fixed']}\n"
|
|
f" Summaries fixed: {stats['summary_fixed']}\n"
|
|
f" Ratings fixed: {stats['rating_fixed']}\n"
|
|
f" Release dates fixed: {stats['release_date_fixed']}\n"
|
|
f" Release years fixed: {stats['release_year_fixed']}\n"
|
|
f" IGDB IDs found: {stats['igdb_id_found']}\n"
|
|
f" HLtB IDs found: {stats['hltb_id_found']}"
|
|
)
|
|
if fix_broken_images:
|
|
self.stdout.write(f" Broken images fixed: {stats['images_fixed']}")
|
|
|
|
if title_mismatches:
|
|
self.stdout.write("\nTitle vs retroarch_name mismatches (not auto-fixed):")
|
|
for retroarch_name, title, game_id in title_mismatches:
|
|
self.stdout.write(
|
|
f" Game #{game_id}: retroarch_name={retroarch_name!r} vs title={title!r}"
|
|
)
|
|
|
|
def _clean_retroarch_name(self, name):
|
|
if not name:
|
|
return ""
|
|
name = name.strip()
|
|
if "(" in name:
|
|
name = name.split("(")[0].strip()
|
|
return name
|
|
|
|
def _check_retroarch_name(self, game, mismatches):
|
|
if not game.retroarch_name:
|
|
return
|
|
cleaned = self._clean_retroarch_name(game.retroarch_name)
|
|
if cleaned.lower() != game.title.lower():
|
|
mismatches.append((game.retroarch_name, game.title, game.id))
|
|
if "retroarch-mismatch" not in game.tags.names():
|
|
game.tags.add("retroarch-mismatch")
|
|
self.stdout.write(
|
|
f" [TAG] {game} — tagged as retroarch-mismatch"
|
|
)
|
|
|
|
def _enrich_game(self, game, sleep_secs, force):
|
|
from videogames.igdb import lookup_game_id_from_gdb, lookup_game_from_igdb
|
|
from videogames.howlongtobeat import lookup_game_from_hltb
|
|
|
|
search_name = self._clean_retroarch_name(game.retroarch_name) or game.title
|
|
|
|
changed = {}
|
|
|
|
if not game.hltb_id:
|
|
hltb_data = None
|
|
if search_name:
|
|
hltb_data = lookup_game_from_hltb(search_name, search_by_title=True)
|
|
time.sleep(sleep_secs)
|
|
if not hltb_data and game.title and game.title != search_name:
|
|
hltb_data = lookup_game_from_hltb(game.title, search_by_title=True)
|
|
time.sleep(sleep_secs)
|
|
if hltb_data:
|
|
result = self._apply_hltb_data(game, hltb_data, force)
|
|
if result:
|
|
changed.update(result)
|
|
|
|
igdb_data = None
|
|
if not game.igdb_id and search_name:
|
|
igdb_id = lookup_game_id_from_gdb(search_name)
|
|
time.sleep(sleep_secs)
|
|
if igdb_id:
|
|
igdb_data = lookup_game_from_igdb(str(igdb_id))
|
|
time.sleep(sleep_secs)
|
|
elif game.igdb_id:
|
|
igdb_data = lookup_game_from_igdb(str(game.igdb_id))
|
|
time.sleep(sleep_secs)
|
|
|
|
if igdb_data:
|
|
igdb_title = igdb_data.get("title", "")
|
|
igdb_title_clean = self._clean_retroarch_name(igdb_title)
|
|
if igdb_title_clean.lower() == search_name.lower():
|
|
if game.igdb_id is None and igdb_data.get("igdb_id"):
|
|
game.igdb_id = int(igdb_data["igdb_id"])
|
|
game.save(update_fields=["igdb_id"])
|
|
changed["igdb_id_found"] = True
|
|
self.stdout.write(f" [IGDB_ID] {game} — found IGDB ID {game.igdb_id}")
|
|
result = self._apply_igdb_data(game, igdb_data, force)
|
|
if result:
|
|
changed.update(result)
|
|
else:
|
|
self.stdout.write(
|
|
f" [IGDB] {game} — title mismatch (IGDB: {igdb_title!r} vs expected: {search_name!r}), re-searching…"
|
|
)
|
|
resolved = False
|
|
for candidate in (search_name, game.title if game.title != search_name else None):
|
|
if not candidate:
|
|
continue
|
|
new_id = lookup_game_id_from_gdb(candidate)
|
|
time.sleep(sleep_secs)
|
|
if not new_id:
|
|
continue
|
|
new_data = lookup_game_from_igdb(str(new_id))
|
|
time.sleep(sleep_secs)
|
|
if not new_data:
|
|
continue
|
|
new_title = new_data.get("title", "")
|
|
new_title_clean = self._clean_retroarch_name(new_title)
|
|
if new_title_clean.lower() == candidate.lower():
|
|
game.igdb_id = int(new_id)
|
|
if new_title and new_title != game.title:
|
|
game.title = new_title
|
|
changed["title_updated"] = True
|
|
self.stdout.write(f" [TITLE] {game} — updated title to {new_title!r} from IGDB")
|
|
game.save(update_fields=["igdb_id"] + (["title"] if changed.get("title_updated") else []))
|
|
changed["igdb_id_found"] = True
|
|
self.stdout.write(f" [IGDB_ID] {game} — re-found IGDB ID {game.igdb_id}")
|
|
if "igdb-mismatch" in game.tags.names():
|
|
game.tags.remove("igdb-mismatch")
|
|
self.stdout.write(f" [TAG] {game} — removed igdb-mismatch tag")
|
|
result = self._apply_igdb_data(game, new_data, force)
|
|
if result:
|
|
changed.update(result)
|
|
resolved = True
|
|
break
|
|
|
|
if not resolved and "igdb-mismatch" not in game.tags.names():
|
|
game.tags.add("igdb-mismatch")
|
|
self.stdout.write(f" [TAG] {game} — tagged igdb-mismatch")
|
|
|
|
# If retroarch-mismatch tag exists but no longer applies, remove it
|
|
if "retroarch-mismatch" in game.tags.names():
|
|
cleaned = self._clean_retroarch_name(game.retroarch_name or "")
|
|
if cleaned.lower() == game.title.lower():
|
|
game.tags.remove("retroarch-mismatch")
|
|
self.stdout.write(f" [TAG] {game} — removed retroarch-mismatch tag (title now matches)")
|
|
|
|
return changed if changed else None
|
|
|
|
def _apply_igdb_data(self, game, data, force):
|
|
from django.core.files.base import ContentFile
|
|
|
|
import requests
|
|
|
|
changed = {
|
|
"cover_fixed": False,
|
|
"screenshot_fixed": False,
|
|
"summary_fixed": False,
|
|
"rating_fixed": False,
|
|
"release_date_fixed": False,
|
|
}
|
|
|
|
update_fields = []
|
|
|
|
if data.get("alternative_name") and not game.alternative_name:
|
|
game.alternative_name = data["alternative_name"]
|
|
update_fields.append("alternative_name")
|
|
|
|
if data.get("summary") and (not game.summary or force):
|
|
game.summary = data["summary"]
|
|
update_fields.append("summary")
|
|
changed["summary_fixed"] = True
|
|
|
|
if data.get("rating") is not None and (game.rating is None or force):
|
|
game.rating = data["rating"]
|
|
update_fields.append("rating")
|
|
changed["rating_fixed"] = True
|
|
|
|
if data.get("rating_count") is not None and (game.rating_count is None or force):
|
|
game.rating_count = data["rating_count"]
|
|
update_fields.append("rating_count")
|
|
|
|
if data.get("release_date") and (game.release_date is None or force):
|
|
game.release_date = data["release_date"]
|
|
update_fields.append("release_date")
|
|
changed["release_date_fixed"] = True
|
|
|
|
if update_fields:
|
|
game.save(update_fields=update_fields)
|
|
self.stdout.write(f" [IGDB] {game} — {', '.join(update_fields)}")
|
|
|
|
cover_url = data.get("cover_url")
|
|
if cover_url:
|
|
r = requests.get(cover_url)
|
|
if r.status_code == 200:
|
|
fname = f"{game.title}_{game.uuid}.jpg"
|
|
game.cover.save(fname, ContentFile(r.content), save=True)
|
|
changed["cover_fixed"] = True
|
|
self.stdout.write(f" [COVER] {game} — cover saved from IGDB")
|
|
|
|
screenshot_url = data.get("screenshot_url")
|
|
if screenshot_url:
|
|
r = requests.get(screenshot_url)
|
|
if r.status_code == 200:
|
|
fname = f"{game.title}_{game.uuid}.jpg"
|
|
game.screenshot.save(fname, ContentFile(r.content), save=True)
|
|
changed["screenshot_fixed"] = True
|
|
self.stdout.write(f" [SCREENSHOT] {game} — screenshot saved from IGDB")
|
|
|
|
genres = data.get("genres", [])
|
|
if genres:
|
|
existing = set(game.genre.names())
|
|
new_genres = [g for g in genres if g not in existing]
|
|
if new_genres:
|
|
game.genre.add(*new_genres)
|
|
self.stdout.write(f" [GENRES] {game} — added {len(new_genres)} genres")
|
|
|
|
platforms = data.get("platforms", [])
|
|
if platforms:
|
|
existing = set(game.platforms.values_list("name", flat=True))
|
|
new_platforms = [p for p in platforms if p not in existing]
|
|
if new_platforms:
|
|
from videogames.models import VideoGamePlatform
|
|
|
|
for name in new_platforms:
|
|
p, _ = VideoGamePlatform.objects.get_or_create(name=name)
|
|
game.platforms.add(p)
|
|
self.stdout.write(f" [PLATFORMS] {game} — added {len(new_platforms)} platforms")
|
|
|
|
if "igdb-enriched" not in game.tags.names():
|
|
game.tags.add("igdb-enriched")
|
|
self.stdout.write(f" [TAG] {game} — tagged igdb-enriched")
|
|
|
|
return changed if any(changed.values()) else None
|
|
|
|
def _apply_hltb_data(self, game, data, force):
|
|
from django.core.files.base import ContentFile
|
|
|
|
import requests
|
|
|
|
changed = {
|
|
"hltb_id_found": False,
|
|
"release_year_fixed": False,
|
|
}
|
|
|
|
update_fields = []
|
|
|
|
hltb_title = data.get("title", "")
|
|
if hltb_title and hltb_title != game.title:
|
|
game.title = hltb_title
|
|
update_fields.append("title")
|
|
self.stdout.write(f" [TITLE] {game} — updated title to {hltb_title!r}")
|
|
|
|
if data.get("hltb_id") and (game.hltb_id is None or force):
|
|
game.hltb_id = data["hltb_id"]
|
|
update_fields.append("hltb_id")
|
|
changed["hltb_id_found"] = True
|
|
self.stdout.write(f" [HLTB_ID] {game} — found HLtB ID {data['hltb_id']}")
|
|
|
|
if data.get("release_year") and (game.release_year is None or force):
|
|
game.release_year = data["release_year"]
|
|
update_fields.append("release_year")
|
|
changed["release_year_fixed"] = True
|
|
|
|
if data.get("main_story_time") and (game.main_story_time is None or force):
|
|
game.main_story_time = data["main_story_time"]
|
|
update_fields.append("main_story_time")
|
|
|
|
if data.get("main_extra_time") and (game.main_extra_time is None or force):
|
|
game.main_extra_time = data["main_extra_time"]
|
|
update_fields.append("main_extra_time")
|
|
|
|
if data.get("completionist_time") and (game.completionist_time is None or force):
|
|
game.completionist_time = data["completionist_time"]
|
|
update_fields.append("completionist_time")
|
|
|
|
if data.get("hltb_score") is not None and (game.hltb_score is None or force):
|
|
game.hltb_score = data["hltb_score"]
|
|
update_fields.append("hltb_score")
|
|
|
|
if update_fields:
|
|
game.save(update_fields=update_fields)
|
|
self.stdout.write(f" [HLTB] {game} — {', '.join(update_fields)}")
|
|
|
|
cover_url = data.get("cover_url")
|
|
if cover_url:
|
|
headers = {"User-Agent": "Vrobbler 0.11.12"}
|
|
r = requests.get(cover_url, headers=headers)
|
|
if r.status_code == 200:
|
|
fname = f"{game.title}_cover_{game.uuid}.jpg"
|
|
game.hltb_cover.save(fname, ContentFile(r.content), save=True)
|
|
self.stdout.write(f" [HLTB_COVER] {game} — cover saved from HLtB")
|
|
|
|
platforms = data.get("platforms", [])
|
|
if platforms:
|
|
existing = set(game.platforms.values_list("name", flat=True))
|
|
new_platforms = [p for p in platforms if p not in existing]
|
|
if new_platforms:
|
|
from videogames.models import VideoGamePlatform
|
|
|
|
for name in new_platforms:
|
|
p, _ = VideoGamePlatform.objects.get_or_create(name=name)
|
|
game.platforms.add(p)
|
|
self.stdout.write(f" [PLATFORMS] {game} — added {len(new_platforms)} platforms")
|
|
|
|
if "hltb-enriched" not in game.tags.names():
|
|
game.tags.add("hltb-enriched")
|
|
self.stdout.write(f" [TAG] {game} — tagged hltb-enriched")
|
|
|
|
return changed if any(changed.values()) else None
|
|
|
|
def _fix_broken_images(self, games, sleep_secs):
|
|
from django.core.files.base import ContentFile
|
|
|
|
import requests
|
|
|
|
from videogames.igdb import lookup_game_from_igdb
|
|
from videogames.howlongtobeat import lookup_game_from_hltb
|
|
|
|
stats = {"cover_fixed": 0, "screenshot_fixed": 0, "images_fixed": 0}
|
|
|
|
for game in games:
|
|
for field_name, source in [
|
|
("cover", "igdb"),
|
|
("screenshot", "igdb"),
|
|
("hltb_cover", "hltb"),
|
|
]:
|
|
field = getattr(game, field_name)
|
|
if not field.name:
|
|
continue
|
|
if field.storage.exists(field.name):
|
|
continue
|
|
|
|
self.stdout.write(
|
|
f" [IMAGE] {game} — {field_name} is broken (file missing), refetching…"
|
|
)
|
|
|
|
if source == "igdb" and game.igdb_id:
|
|
data = lookup_game_from_igdb(str(game.igdb_id))
|
|
time.sleep(sleep_secs)
|
|
if not data:
|
|
continue
|
|
url = data.get("cover_url" if field_name == "cover" else "screenshot_url")
|
|
if not url:
|
|
continue
|
|
r = requests.get(url)
|
|
if r.status_code != 200:
|
|
continue
|
|
fname = f"{game.title}_{game.uuid}.jpg"
|
|
getattr(game, field_name).save(fname, ContentFile(r.content), save=True)
|
|
stats["images_fixed"] += 1
|
|
self.stdout.write(f" [IMAGE] {game} — {field_name} refetched from IGDB")
|
|
|
|
elif source == "hltb" and game.hltb_id:
|
|
data = lookup_game_from_hltb(str(game.hltb_id))
|
|
time.sleep(sleep_secs)
|
|
if not data:
|
|
continue
|
|
url = data.get("cover_url")
|
|
if not url:
|
|
continue
|
|
headers = {"User-Agent": "Vrobbler 0.11.12"}
|
|
r = requests.get(url, headers=headers)
|
|
if r.status_code != 200:
|
|
continue
|
|
fname = f"{game.title}_cover_{game.uuid}.jpg"
|
|
game.hltb_cover.save(fname, ContentFile(r.content), save=True)
|
|
stats["images_fixed"] += 1
|
|
self.stdout.write(f" [IMAGE] {game} — hltb_cover refetched from HLtB")
|
|
|
|
return stats
|