sync_mod.py

"""Upsert (or remove) mod entries in data/mods.json from each mod repo's metadata.yml.

Usage:
    python scripts/sync_mod.py <repo> [<repo> ...] [--ref master] [--sha <sha>] [--remove]
    python scripts/sync_mod.py --all
"""

import argparse
import json
import os
import re
import sys
from datetime import datetime, timezone
from pathlib import Path
from urllib.parse import quote

import requests
import yaml

ORG = "DarkbordermanModding"
DATA_PATH = Path(__file__).resolve().parent.parent / "data" / "mods.json"
TOKEN = os.environ.get("GITHUB_TOKEN") or os.environ.get("TOKEN")

# Repo-name prefix -> display name
GAME_NAMES = {
    "CursaderKing3": "Crusader Kings III",
    "CrusaderKings3": "Crusader Kings III",
    "HeartsOfIron4": "Hearts of Iron IV",
    "MountAndBladeBannerlord": "Mount & Blade II: Bannerlord",
    "ProjectZomboid": "Project Zomboid",
    "Rimworld": "RimWorld",
    "SkyrimSpecialEdition": "Skyrim Special Edition",
}

# Thumbnail candidates tried when metadata names none
PREVIEW_FALLBACKS = ["preview.png", "preview.jpg", "thumbnail.png"]

# Steam visibility: 0 public, 1 friends-only, 2 private, 3 unlisted
LINKABLE_VISIBILITY = {0, 3}

# Fields that do not count as a content change
VOLATILE_FIELDS = {"updated_at", "commit"}

REPO_PATTERN = re.compile(r"^[A-Za-z0-9._-]+$")


def api_get(path):
    headers = {"Accept": "application/vnd.github+json"}
    if TOKEN:
        headers["Authorization"] = f"Bearer {TOKEN}"
    resp = requests.get(f"https://api.github.com{path}", headers=headers, timeout=30)
    resp.raise_for_status()
    return resp.json()


def raw_url(repo, ref, path):
    return f"https://raw.githubusercontent.com/{ORG}/{repo}/{ref}/{quote(path)}"


def fetch_raw(repo, ref, path):
    """Return file text, or None if the file does not exist."""
    resp = requests.get(raw_url(repo, ref, path), timeout=30)
    if resp.status_code == 404:
        return None
    resp.raise_for_status()
    return resp.text


def raw_exists(repo, ref, path):
    return requests.head(raw_url(repo, ref, path), timeout=30).status_code == 200


def first(*values):
    for value in values:
        if value not in (None, ""):
            return value
    return None


def game_of(repo):
    prefix = repo.split("-", 1)[0]
    return GAME_NAMES.get(prefix, prefix)


def title_from_repo(repo):
    name = repo.split("-", 1)[1] if "-" in repo else repo
    return re.sub(r"(?<=[a-z0-9])(?=[A-Z])", " ", name)


def clean_title(title):
    # Drop leading version tags such as "[0.23.229] " or "[B42.20 SP/MP] "
    return re.sub(r"^\s*(\[[^\]]*\]\s*)+", "", title).strip()


def build_entry(repo, ref, sha):
    info = api_get(f"/repos/{ORG}/{repo}")
    ref = ref or info["default_branch"]
    sha = sha or api_get(f"/repos/{ORG}/{repo}/commits/{ref}")["sha"]

    text = fetch_raw(repo, sha, "metadata.yml")
    meta = (yaml.safe_load(text) if text else None) or {}
    page = meta.get("page") or {}
    github = meta.get("github") or {}
    workshop = meta.get("workshop") or {}
    english = (workshop.get("localizations") or {}).get("english") or {}

    if page.get("hidden"):
        return None

    title = first(page.get("title"), english.get("title"), workshop.get("title"))
    title = clean_title(title) if title else title_from_repo(repo)

    summary = first(
        page.get("summary"),
        github.get("description"),
        workshop.get("description"),
        english.get("description"),
        info.get("description"),
    ) or ""

    description = ""
    description_path = first(page.get("description_path"), english.get("description_path"))
    if description_path:
        description = (fetch_raw(repo, sha, description_path) or "").strip()

    thumbnail = None
    preview_path = first(page.get("preview_path"), workshop.get("preview_path"))
    for candidate in [preview_path] if preview_path else PREVIEW_FALLBACKS:
        if raw_exists(repo, sha, candidate):
            thumbnail = raw_url(repo, ref, candidate)
            break

    workshop_url = None
    workshop_id = workshop.get("publishedfield_id")
    if workshop_id and workshop.get("visibility", 0) in LINKABLE_VISIBILITY:
        workshop_url = f"https://steamcommunity.com/sharedfiles/filedetails/?id={workshop_id}"

    return {
        "repo": repo,
        "game": page.get("game") or game_of(repo),
        "title": title,
        "summary": summary.strip(),
        "description": description,
        "thumbnail": thumbnail,
        "source_url": info["html_url"],
        "workshop_url": workshop_url,
        "updated_at": datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ"),
        "commit": sha[:7],
    }


def stable(entry):
    return {k: v for k, v in entry.items() if k not in VOLATILE_FIELDS}


def load_data():
    if DATA_PATH.exists():
        return json.loads(DATA_PATH.read_text(encoding="utf-8"))
    return {"mods": {}}


def save_data(data):
    DATA_PATH.parent.mkdir(parents=True, exist_ok=True)
    text = json.dumps(data, indent=2, ensure_ascii=False, sort_keys=True)
    DATA_PATH.write_text(text + "\n", encoding="utf-8", newline="\n")


def main():
    parser = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
    parser.add_argument("repos", nargs="*", help="mod repo names")
    parser.add_argument("--all", action="store_true", help="re-sync every repo already in mods.json")
    parser.add_argument("--ref", help="branch to read (default: the repo's default branch)")
    parser.add_argument("--sha", help="commit to read (default: head of --ref)")
    parser.add_argument("--remove", action="store_true", help="remove the repos instead of syncing")
    args = parser.parse_args()

    data = load_data()
    mods = data.setdefault("mods", {})
    repos = sorted(mods) if args.all else args.repos
    if not repos:
        parser.error("give at least one repo, or --all")
    if args.sha and len(repos) > 1:
        parser.error("--sha only applies to a single repo")

    failed = []
    for repo in repos:
        if not REPO_PATTERN.match(repo):
            sys.exit(f"invalid repo name: {repo!r}")

        try:
            entry = None if args.remove else build_entry(repo, args.ref, args.sha)
        except requests.HTTPError as error:
            print(f"failed {repo}: {error}", file=sys.stderr)
            failed.append(repo)
            continue
        old = mods.get(repo)

        if entry is None:
            if old is not None:
                del mods[repo]
                print(f"removed {repo}")
            else:
                print(f"skipped {repo} (not listed)")
        elif old is not None and stable(old) == stable(entry):
            print(f"unchanged {repo}")
        else:
            mods[repo] = entry
            print(f"{'updated' if old else 'added'} {repo}")

    save_data(data)
    if failed:
        sys.exit(f"failed to sync: {', '.join(failed)}")


if __name__ == "__main__":
    main()