nuzlocke-tracker/backend/scripts/fetch_moves_abilities.py

#!/usr/bin/env python3
"""Fetch moves and abilities from PokeAPI and save as seed data JSON files.

Usage:
    cd backend && uv run python scripts/fetch_moves_abilities.py

This script fetches all moves and abilities from PokeAPI, extracts their names
and introduced generation, and saves them to the seed data directory.
"""

import asyncio
import json
import re
from pathlib import Path

import httpx

DATA_DIR = Path(__file__).parent.parent / "src" / "app" / "seeds" / "data"
POKEAPI_BASE = "https://pokeapi.co/api/v2"

# Map generation names to numbers
GEN_MAP = {
    "generation-i": 1,
    "generation-ii": 2,
    "generation-iii": 3,
    "generation-iv": 4,
    "generation-v": 5,
    "generation-vi": 6,
    "generation-vii": 7,
    "generation-viii": 8,
    "generation-ix": 9,
}


def title_case_name(name: str) -> str:
    """Convert a hyphenated PokeAPI name to title case.

    Examples:
        'thunder-punch' -> 'Thunder Punch'
        'self-destruct' -> 'Self-Destruct'
    """
    return " ".join(word.capitalize() for word in name.split("-"))


async def fetch_all_moves(client: httpx.AsyncClient) -> list[dict]:
    """Fetch all moves from PokeAPI."""
    moves = []

    # First, get the list of all moves
    print("Fetching move list...")
    url = f"{POKEAPI_BASE}/move?limit=10000"
    resp = await client.get(url)
    resp.raise_for_status()
    data = resp.json()

    move_urls = [m["url"] for m in data["results"]]
    print(f"Found {len(move_urls)} moves")

    # Fetch each move's details in batches
    batch_size = 50
    for i in range(0, len(move_urls), batch_size):
        batch = move_urls[i : i + batch_size]
        print(f"Fetching moves {i + 1}-{min(i + batch_size, len(move_urls))}...")

        tasks = [client.get(url) for url in batch]
        responses = await asyncio.gather(*tasks, return_exceptions=True)

        for resp in responses:
            if isinstance(resp, Exception):
                print(f"  Error fetching move: {resp}")
                continue
            if resp.status_code != 200:
                print(f"  HTTP {resp.status_code} for {resp.url}")
                continue

            move_data = resp.json()
            gen_name = move_data["generation"]["name"]
            introduced_gen = GEN_MAP.get(gen_name)

            if introduced_gen is None:
                print(f"  Unknown generation: {gen_name} for move {move_data['name']}")
                continue

            # Get type if available
            move_type = None
            if move_data.get("type"):
                move_type = move_data["type"]["name"]

            moves.append(
                {
                    "name": title_case_name(move_data["name"]),
                    "introduced_gen": introduced_gen,
                    "type": move_type,
                }
            )

    # Sort by name for consistent ordering
    moves.sort(key=lambda m: m["name"])
    return moves


async def fetch_all_abilities(client: httpx.AsyncClient) -> list[dict]:
    """Fetch all abilities from PokeAPI."""
    abilities = []

    # First, get the list of all abilities
    print("Fetching ability list...")
    url = f"{POKEAPI_BASE}/ability?limit=10000"
    resp = await client.get(url)
    resp.raise_for_status()
    data = resp.json()

    ability_urls = [a["url"] for a in data["results"]]
    print(f"Found {len(ability_urls)} abilities")

    # Fetch each ability's details in batches
    batch_size = 50
    for i in range(0, len(ability_urls), batch_size):
        batch = ability_urls[i : i + batch_size]
        print(f"Fetching abilities {i + 1}-{min(i + batch_size, len(ability_urls))}...")

        tasks = [client.get(url) for url in batch]
        responses = await asyncio.gather(*tasks, return_exceptions=True)

        for resp in responses:
            if isinstance(resp, Exception):
                print(f"  Error fetching ability: {resp}")
                continue
            if resp.status_code != 200:
                print(f"  HTTP {resp.status_code} for {resp.url}")
                continue

            ability_data = resp.json()
            gen_name = ability_data["generation"]["name"]
            introduced_gen = GEN_MAP.get(gen_name)

            if introduced_gen is None:
                print(
                    f"  Unknown generation: {gen_name} for ability {ability_data['name']}"
                )
                continue

            abilities.append(
                {
                    "name": title_case_name(ability_data["name"]),
                    "introduced_gen": introduced_gen,
                }
            )

    # Sort by name for consistent ordering
    abilities.sort(key=lambda a: a["name"])
    return abilities


async def main():
    print("Fetching moves and abilities from PokeAPI...")
    print()

    async with httpx.AsyncClient(timeout=30.0) as client:
        # Fetch moves
        moves = await fetch_all_moves(client)
        print()

        # Fetch abilities
        abilities = await fetch_all_abilities(client)
        print()

    # Write moves to JSON
    moves_path = DATA_DIR / "moves.json"
    with open(moves_path, "w") as f:
        json.dump(moves, f, indent=2)
        f.write("\n")
    print(f"Wrote {len(moves)} moves to {moves_path}")

    # Write abilities to JSON
    abilities_path = DATA_DIR / "abilities.json"
    with open(abilities_path, "w") as f:
        json.dump(abilities, f, indent=2)
        f.write("\n")
    print(f"Wrote {len(abilities)} abilities to {abilities_path}")

    print()
    print("Done!")


if __name__ == "__main__":
    asyncio.run(main())