polylan-submitter/polylan_submitter/games/services/planning.py

import logging
import re
import unicodedata

import httpx
from django.utils.text import slugify

from ..models import EventGame, Game, PlanningSlot
from .steam_store import search_store

logger = logging.getLogger(__name__)

# "Cosmoteer: Starship Architect & Commander" is still "Cosmoteer"
SUBTITLE_SEPARATOR = re.compile(r"\s*(?::|\s-\s|\s–\s|\s—\s)")


def normalize_name(name: str) -> str:
    """'Towerfall: Ascension' and 'TowerFall Ascension' are the same game:
    no accents, no case, letters and digits only."""
    # Before NFKD, which turns ™ into "TM"
    name = re.sub(r"[™®©]", "", name)
    name = unicodedata.normalize("NFKD", name)
    name = "".join(c for c in name if not unicodedata.combining(c))
    return re.sub(r"[^0-9a-z]", "", name.casefold())


def _steam_match(name: str) -> dict | None:
    """The Steam store app named name (or name + a subtitle), if there's one.
    Strict on purpose: "Jass" or "Crochet" must not end up as some random game."""
    wanted = normalize_name(name)
    if not wanted:
        return None
    # Steam's search chokes on punctuation ("Puyo Puyo — Tetris" finds nothing)
    term = " ".join(re.sub(r"[^\w\s]", " ", name).split())
    for item in search_store(term):
        steam_name = item.get("name") or ""
        if wanted in (
            normalize_name(steam_name),
            normalize_name(SUBTITLE_SEPARATOR.split(steam_name, maxsplit=1)[0]),
        ):
            return item
    return None


def _unique_path(name: str) -> str:
    base = slugify(name)[:90] or "game"
    path, n = base, 2
    while Game.objects.filter(path=path).exists():
        path, n = f"{base}-{n}", n + 1
    return path


def create_missing_steam_games(event, names: list[str]) -> list[Game]:
    """For each name without a Game, look it up on the Steam store and create
    the Game + its EventGame when found. Returns the created games."""
    known = {normalize_name(game.name) for game in Game.objects.all()}
    created = []
    for name in dict.fromkeys(names):  # unique, keeps order
        if normalize_name(name) in known:
            continue
        try:
            item = _steam_match(name)
        except (httpx.HTTPError, ValueError) as exc:
            logger.warning("Steam store search failed for %r: %s", name, exc)
            continue
        if item is None:
            continue

        if Game.objects.filter(steam_app_id=item["id"]).exists():
            # Already there under another name, the planning won't match it
            logger.warning(
                "%r is Steam app %s, already a Game under another name",
                name,
                item["id"],
            )
            continue

        # Named like in the planning, so the slot matches it
        game = Game.objects.create(
            name=name, steam_app_id=item["id"], path=_unique_path(name)
        )
        EventGame.objects.get_or_create(event=event, game=game)
        known.add(normalize_name(name))
        created.append(game)
    return created


def slot_game_names(entries: list[dict]) -> list[str]:
    """Names to look for in a planning: titles of single slots, game names of
    multi-anims (their title, like "Fils rouges", isn't a game)."""
    names = []
    for entry in entries:
        if not entry.get("is_multi_anims"):
            names.append(entry["title"])
        names += [a["name"] for a in entry.get("aandts") or []]
    return names


def _game_out(name: str, game: Game | None) -> dict:
    if game is None:
        return {"name": name}
    return {
        "name": game.name,
        "library_image": game.library_image_url,
        "store_url": game.store_url,
        "steam_app_id": game.steam_app_id,
    }


def planning_with_games(slots) -> list[dict]:
    """Slots as PlanningSlotOut dicts, with their games matched by name to ours."""
    games_by_name = {normalize_name(game.name): game for game in Game.objects.all()}

    def match(name: str) -> Game | None:
        return games_by_name.get(normalize_name(name))

    out = []
    for slot in slots:
        if slot.is_multi_anims:
            games = [_game_out(name, match(name)) for name in slot.game_names]
        else:
            # The title is the usual match, the legacy game names are a fallback
            # ("Golf it!" slot running the "Golf It!" game)
            game = next(
                filter(None, map(match, [slot.title, *slot.game_names])), None
            )
            games = [_game_out(slot.title, game)]

        out.append(
            {
                "id": slot.id,
                "title": slot.title,
                "begin": slot.begin,
                "end": slot.end,
                "column": slot.column,
                "color": slot.color,
                "is_multi_anims": slot.is_multi_anims,
                "games": games,
            }
        )
    return out


def import_planning(event, entries: list[dict]) -> tuple[int, int, int]:
    """Sync the event planning with the legacy planning entries (its "list").
    Returns (created, updated, deleted)."""
    created = updated = 0
    for entry in entries:
        _, was_created = PlanningSlot.objects.update_or_create(
            event=event,
            source_pk=entry["pk"],
            defaults={
                "title": entry["title"],
                "begin": entry["begin"],
                "end": entry["end"],
                "column": entry.get("colonneS") or 0,
                "color": entry.get("color_code") or "",
                "is_multi_anims": bool(entry.get("is_multi_anims")),
                "game_names": [a["name"] for a in entry.get("aandts") or []],
            },
        )
        if was_created:
            created += 1
        else:
            updated += 1

    deleted, _ = (
        PlanningSlot.objects.filter(event=event)
        .exclude(source_pk__in=[entry["pk"] for entry in entries])
        .delete()
    )
    return created, updated, deleted