website/tools/content/igdb.py

import argparse
import datetime
import json
import re
import sys
import time
import urllib.error
import urllib.parse
import urllib.request

from envfile import env_value
from jsonfile import PROJECT_ROOT

TOKEN_URL = "https://id.twitch.tv/oauth2/token"
GAMES_URL = "https://api.igdb.com/v4/games"
EXTERNAL_GAMES_URL = "https://api.igdb.com/v4/external_games"
IMAGE_URL = "https://images.igdb.com/igdb/image/upload/{size}/{image_id}.jpg"
TOKEN_CACHE_PATH = PROJECT_ROOT / "tools" / "content" / ".cache" / "igdb_token.json"
TOKEN_EXPIRY_MARGIN_SECONDS = 300
SUMMARY_MAX_LENGTH = 600
GAME_FIELDS = ", ".join([
    "name",
    "slug",
    "url",
    "summary",
    "first_release_date",
    "total_rating",
    "genres.name",
    "platforms.name",
    "involved_companies.developer",
    "involved_companies.publisher",
    "involved_companies.company.name",
    "cover.image_id",
    "artworks.image_id",
    "screenshots.image_id",
])
IGDB_SLUG_PATTERN = re.compile(r"igdb\.com/games/([\w-]+)")
STORE_ID_PATTERNS = {
    "store.steampowered.com": re.compile(r"store\.steampowered\.com/app/(\d+)"),
    "xbox.com": re.compile(r"xbox\.com/.*?/([0-9A-Za-z]{12})(?:[/?#]|$)"),
}
LINK_NOISE_SEGMENTS = {"store", "products", "product", "games", "game", "app", "p", "us", "en-us"}
SEGMENT_ID_PATTERN = re.compile(r"\d+|[0-9A-Z]{12}")
EPIC_SUFFIX_PATTERN = re.compile(r"-(?=[0-9a-f]*\d)[0-9a-f]{6}$")
NAME_NOISE_PATTERN = re.compile(r"\b(switch|ps4|ps5|pc|xbox)\b", re.I)


class IgdbError(ValueError):
    pass


def credentials() -> tuple[str, str]:
    client_id = env_value("IGDB_CLIENT_ID")
    client_secret = env_value("IGDB_CLIENT_SECRET")
    if not client_id or not client_secret:
        raise IgdbError("Set IGDB_CLIENT_ID and IGDB_CLIENT_SECRET in .env to look up games")
    return client_id, client_secret


def cached_token() -> str | None:
    try:
        cached = json.loads(TOKEN_CACHE_PATH.read_text(encoding="utf-8"))
    except (OSError, ValueError):
        return None
    return cached["token"] if cached.get("expiresAt", 0) > time.time() else None


def fetch_token(client_id: str, client_secret: str) -> str:
    parameters = urllib.parse.urlencode({"client_id": client_id, "client_secret": client_secret, "grant_type": "client_credentials"})
    request = urllib.request.Request(f"{TOKEN_URL}?{parameters}", method="POST")
    try:
        with urllib.request.urlopen(request, timeout=15) as response:
            result = json.load(response)
    except urllib.error.HTTPError as error:
        raise IgdbError(f"Twitch refused the IGDB keys ({error.code}). Check IGDB_CLIENT_ID and IGDB_CLIENT_SECRET") from error
    TOKEN_CACHE_PATH.parent.mkdir(parents=True, exist_ok=True)
    expires_at = time.time() + result["expires_in"] - TOKEN_EXPIRY_MARGIN_SECONDS
    TOKEN_CACHE_PATH.write_text(json.dumps({"token": result["access_token"], "expiresAt": expires_at}), encoding="utf-8")
    return result["access_token"]


def query(url: str, body: str) -> list[dict]:
    client_id, client_secret = credentials()
    token = cached_token() or fetch_token(client_id, client_secret)
    headers = {"Client-ID": client_id, "Authorization": f"Bearer {token}", "Accept": "application/json"}
    request = urllib.request.Request(url, data=body.encode("utf-8"), headers=headers, method="POST")
    try:
        with urllib.request.urlopen(request, timeout=15) as response:
            return json.load(response)
    except urllib.error.HTTPError as error:
        if error.code == 401:
            TOKEN_CACHE_PATH.unlink(missing_ok=True)
        raise IgdbError(f"IGDB returned {error.code}: {error.read().decode('utf-8', errors='replace')[:200]}") from error


def quoted(text: str) -> str:
    return '"' + text.replace("\\", "\\\\").replace('"', '\\"') + '"'


def game_by_id(identifier: str) -> dict | None:
    results = query(GAMES_URL, f"fields {GAME_FIELDS}; where id = {int(identifier)};")
    return results[0] if results else None


def game_by_slug(slug: str) -> dict | None:
    results = query(GAMES_URL, f"fields {GAME_FIELDS}; where slug = {quoted(slug)};")
    return results[0] if results else None


def game_by_store_link(link: str) -> dict | None:
    for host, pattern in STORE_ID_PATTERNS.items():
        match = pattern.search(link)
        if not match:
            continue
        uid = match.group(1).upper() if host == "xbox.com" else match.group(1)
        matches = query(EXTERNAL_GAMES_URL, f"fields game, url; where uid = {quoted(uid)}; limit 50;")
        same_store = [item for item in matches if host in (item.get("url") or "")]
        chosen = same_store or (matches if len(matches) == 1 else [])
        if chosen and chosen[0].get("game"):
            return game_by_id(str(chosen[0]["game"]))
    return None


def search_games(name: str) -> list[dict]:
    return query(GAMES_URL, f"search {quoted(name)}; fields {GAME_FIELDS}; limit 10;")


def search_game(name: str) -> dict | None:
    results = search_games(name)
    exact = [game for game in results if game.get("name", "").lower() == name.lower()]
    return (exact or results or [None])[0]


def name_from_link(link: str) -> str:
    segments = [segment for segment in urllib.parse.urlparse(link).path.split("/") if segment]
    words = [segment for segment in segments if segment.lower() not in LINK_NOISE_SEGMENTS and not SEGMENT_ID_PATTERN.fullmatch(segment)]
    guess = EPIC_SUFFIX_PATTERN.sub("", words[-1]) if words else ""
    guess = NAME_NOISE_PATTERN.sub("", re.sub(r"[-_]+", " ", guess))
    return " ".join(guess.split())


def find_game(link: str | None, name: str | None) -> dict:
    game = None
    if link and IGDB_SLUG_PATTERN.search(link):
        game = game_by_slug(IGDB_SLUG_PATTERN.search(link).group(1))
    if not game and link:
        game = game_by_store_link(link)
    search_name = name or (name_from_link(link) if link else "")
    if not game and search_name:
        game = search_game(search_name)
    if not game:
        raise IgdbError(f"Could not find {name or link} on IGDB. Try an IGDB link like https://www.igdb.com/games/hades-ii")
    return game


def image_url(image: dict | None, size: str) -> str | None:
    return IMAGE_URL.format(size=size, image_id=image["image_id"]) if image and image.get("image_id") else None


def first_image(game: dict, key: str) -> dict | None:
    images = game.get(key) or []
    return images[0] if images else None


def company_names(game: dict, role: str) -> list[str]:
    return [item["company"]["name"] for item in game.get("involved_companies") or [] if item.get(role) and item.get("company")]


def short_summary(text: str | None) -> str | None:
    if not text or len(text) <= SUMMARY_MAX_LENGTH:
        return text
    return text[:SUMMARY_MAX_LENGTH].rsplit(" ", 1)[0] + "..."


def release_date(game: dict) -> str | None:
    seconds = game.get("first_release_date")
    if not isinstance(seconds, (int, float)):
        return None
    return datetime.datetime.fromtimestamp(seconds, tz=datetime.timezone.utc).date().isoformat()


def details(game: dict) -> dict:
    wide_image = first_image(game, "artworks") or first_image(game, "screenshots")
    return {
        "igdbId": str(game["id"]),
        "name": game.get("name"),
        "igdbUrl": game.get("url"),
        "released": release_date(game),
        "platforms": [platform["name"] for platform in game.get("platforms") or [] if platform.get("name")],
        "developers": company_names(game, "developer"),
        "publishers": company_names(game, "publisher"),
        "genres": [genre["name"] for genre in game.get("genres") or [] if genre.get("name")],
        "summary": short_summary(game.get("summary")),
        "score": round(game["total_rating"]) if game.get("total_rating") is not None else None,
        "imageSource": image_url(wide_image, "t_1080p"),
        "coverSource": image_url(game.get("cover"), "t_cover_big"),
    }


def main() -> int:
    parser = argparse.ArgumentParser(description="Look up a video game on IGDB")
    parser.add_argument("text", help="a store link, an IGDB link or a game name")
    arguments = parser.parse_args()
    is_link = arguments.text.startswith("http")
    try:
        result = details(find_game(arguments.text if is_link else None, None if is_link else arguments.text))
    except (OSError, ValueError) as error:
        print(f"Error: {error}", file=sys.stderr)
        return 1
    print(json.dumps(result, indent="\t", ensure_ascii=False))
    return 0


if __name__ == "__main__":
    sys.exit(main())