P Bandai APIp-bandai.jp ↗

"""
Premium Bandai Japan API Client

Search and retrieve product information from Premium Bandai (p-bandai.jp).
Get your API key from: https://parse.bot/settings
"""

import os
import requests
from typing import Optional, Dict, Any, List


class ParseClient:
    """Client for interacting with the Premium Bandai Japan API via Parse."""

    def __init__(self, api_key: Optional[str] = None):
        """Initialize the Parse API client.

        Args:
            api_key: API key for Parse. Defaults to PARSE_API_KEY environment variable.
        """
        self.base_url = "https://api.parse.bot"
        self.scraper_id = "ea02a033-95e2-40f9-89c9-4c13960430ec"
        self.api_key = api_key or os.getenv("PARSE_API_KEY")
        
        if not self.api_key:
            raise ValueError(
                "API key required. Pass as argument or set PARSE_API_KEY environment variable."
            )

    def _call(self, endpoint: str, method: str = "GET", **params) -> Dict[str, Any]:
        """Make a request to the Parse API.

        Args:
            endpoint: API endpoint name
            method: HTTP method (GET or POST)
            **params: Parameters to send with the request

        Returns:
            Response JSON as dictionary
        """
        url = f"{self.base_url}/scraper/{self.scraper_id}/{endpoint}"
        headers = {
            "X-API-Key": self.api_key,
            "Content-Type": "application/json"
        }

        if method == "GET":
            response = requests.get(url, headers=headers, params=params)
        else:  # POST
            response = requests.post(url, headers=headers, json=params)

        response.raise_for_status()
        return response.json()

    def search_products(self, query: str, page: int = 1) -> Dict[str, Any]:
        """Search for products on p-bandai.jp using keyword queries.

        Args:
            query: Search keyword (e.g., 'Gundam', 'ガンダム')
            page: Page number for pagination (default: 1)

        Returns:
            Dictionary with page, total_count, and items list
        """
        return self._call("search_products", method="GET", query=query, page=page)

    def get_product_details(self, product_id: str) -> Dict[str, Any]:
        """Retrieve product details for a specific item.

        Args:
            product_id: Product ID (numeric string, e.g., '1000242716')

        Returns:
            Dictionary with product details (name, price, images, labels, url)
        """
        return self._call("get_product_details", method="GET", product_id=product_id)

    def search_new_products(self, page: int = 1) -> Dict[str, Any]:
        """Fetch newly listed products sorted by newest first.

        Args:
            page: Page number for pagination (default: 1)

        Returns:
            Dictionary with items list of new products
        """
        return self._call("search_new_products", method="GET", page=page)

    def search_upcoming_products(self, page: int = 1) -> Dict[str, Any]:
        """Fetch products that are not yet available for sale (pre-order items).

        Args:
            page: Page number for pagination (default: 1)

        Returns:
            Dictionary with items list of upcoming products
        """
        return self._call("search_upcoming_products", method="GET", page=page)

    def filter_by_character(self, char_id: str, page: int = 1) -> Dict[str, Any]:
        """Filter products by character series ID.

        Args:
            char_id: Character ID (e.g., 'c2933' for アークナイツ)
            page: Page number for pagination (default: 1)

        Returns:
            Dictionary with items list filtered by character
        """
        return self._call("filter_by_character", method="GET", char_id=char_id, page=page)

    def filter_by_genre(self, genre_id: str, page: int = 1) -> Dict[str, Any]:
        """Filter products by merchandise genre ID.

        Args:
            genre_id: Genre ID (e.g., '0013' for おもちゃ・キャラクター玩具)
            page: Page number for pagination (default: 1)

        Returns:
            Dictionary with items list filtered by genre
        """
        return self._call("filter_by_genre", method="GET", genre_id=genre_id, page=page)

    def get_search_filters(self, query: Optional[str] = None) -> Dict[str, Any]:
        """Retrieve available filter facets for characters and genres.

        Args:
            query: Optional search query to scope the available filters

        Returns:
            Dictionary with characters and genres arrays
        """
        params = {}
        if query:
            params["query"] = query
        return self._call("get_search_filters", method="GET", **params)


def format_price(price_str: str) -> str:
    """Format price string for display."""
    return price_str if price_str else "N/A"


def main():
    """Main workflow demonstrating realistic API usage patterns."""
    client = ParseClient()

    print("=" * 80)
    print("Premium Bandai Product Discovery & Analysis Tool")
    print("=" * 80)

    # Workflow: Find RG Gundam models, compare prices, check availability
    search_keyword = "RG Gundam"
    
    print(f"\n[Phase 1] Searching for '{search_keyword}' products...")
    try:
        search_response = client.search_products(query=search_keyword, page=1)
        items = search_response.get("data", {}).get("items", [])
        total_count = search_response.get("data", {}).get("total_count", 0)
        
        print(f"✓ Found {len(items)} products (Total available: {total_count})")
    except Exception as e:
        print(f"✗ Search failed: {e}")
        return

    if not items:
        print("No products found. Exiting.")
        return

    # Workflow Phase 2: Collect detailed information for top results
    print(f"\n[Phase 2] Gathering detailed information for top 5 results...")
    detailed_products: List[Dict[str, Any]] = []
    
    for idx, item in enumerate(items[:5], 1):
        product_id = item.get("product_id")
        name = item.get("name", "Unknown")
        price = format_price(item.get("price", ""))
        
        print(f"\n  {idx}. {name[:60]}...")
        print(f"     Listed Price: {price}")
        
        if product_id:
            try:
                details_response = client.get_product_details(product_id=product_id)
                details = details_response.get("data", {})
                
                images = details.get("images", [])
                labels = details.get("labels", [])
                product_url = details.get("product_url", "")
                
                detailed_products.append({
                    "product_id": product_id,
                    "name": name,
                    "price": price,
                    "num_images": len(images),
                    "labels": labels,
                    "url": product_url
                })
                
                print(f"     Images: {len(images)}")
                if labels:
                    print(f"     Status: {', '.join(labels)}")
                    
            except Exception as e:
                print(f"     ✗ Could not fetch details: {e}")

    # Workflow Phase 3: Analyze detailed results
    print(f"\n[Phase 3] Product Analysis Summary")
    print("-" * 80)
    
    if detailed_products:
        print(f"Successfully retrieved {len(detailed_products)} complete product details:\n")
        
        for product in detailed_products:
            print(f"  • {product['name'][:50]}")
            print(f"    Price: {product['price']} | Images: {product['num_images']}")
            if product['labels']:
                print(f"    Availability: {product['labels'][0]}")
            print()

    # Workflow Phase 4: Check for upcoming related products
    print("[Phase 4] Checking for upcoming RG Gundam products...")
    try:
        upcoming_response = client.search_upcoming_products(page=1)
        upcoming_items = upcoming_response.get("data", {}).get("items", [])
        
        # Filter for gundam-related items
        gundam_upcoming = [
            item for item in upcoming_items 
            if "gundam" in item.get("name", "").lower()
        ][:3]
        
        if gundam_upcoming:
            print(f"✓ Found {len(gundam_upcoming)} upcoming Gundam items:\n")
            for item in gundam_upcoming:
                print(f"  • {item.get('name', 'Unknown')[:60]}")
                print(f"    Price: {format_price(item.get('price', ''))}\n")
        else:
            print("No upcoming Gundam-specific items found on current page.")
            
    except Exception as e:
        print(f"✗ Could not fetch upcoming products: {e}")

    # Workflow Phase 5: Explore available filters
    print("\n[Phase 5] Available Search Filters")
    print("-" * 80)
    try:
        filters_response = client.get_search_filters(query=search_keyword)
        filters = filters_response.get("data", {})
        characters = filters.get("characters", [])
        genres = filters.get("genres", [])
        
        if characters:
            print(f"\nCharacter Series ({len(characters)} available):")
            for char in characters[:3]:
                print(f"  • {char.get('name')} (ID: {char.get('id')})")
        
        if genres:
            print(f"\nGenre Categories ({len(genres)} available):")
            for genre in genres[:3]:
                print(f"  • {genre.get('name')} (ID: {genre.get('id')})")
                
    except Exception as e:
        print(f"✗ Could not fetch search filters: {e}")

    # Workflow Phase 6: Deep dive - filter by first genre
    print("\n[Phase 6] Genre-based Product Discovery")
    print("-" * 80)
    try:
        filters_response = client.get_search_filters()
        genres = filters_response.get("data", {}).get("genres", [])
        
        if genres:
            target_genre = genres[0]
            genre_id = target_genre.get("id")
            genre_name = target_genre.get("name", "Unknown")
            
            print(f"\nFetching products in '{genre_name}' genre...")
            genre_response = client.filter_by_genre(genre_id=genre_id, page=1)
            genre_items = genre_response.get("data", {}).get("items", [])
            total_in_genre = genre_response.get("data", {}).get("total_count", 0)
            
            print(f"✓ Found {len(genre_items)} items in this genre (Total: {total_in_genre})")
            
            if genre_items:
                print("\nSample products from this genre:")
                for item in genre_items[:3]:
                    print(f"  • {item.get('name', 'Unknown')[:55]}")
                    print(f"    {format_price(item.get('price', ''))}")
                    
    except Exception as e:
        print(f"✗ Could not fetch genre products: {e}")

    print("\n" + "=" * 80)
    print("Analysis Complete!")
    print("=" * 80)


if __name__ == "__main__":
    main()