Daedalus and Talos integration test

backend/scripts/fetch_moves_abilities.py

@@ -0,0 +1,187 @@
+#!/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())