Gov APIdata.stats.gov.cn ↗

"""
NBS China Data Portal API Client
Get your API key from: https://parse.bot/settings
"""

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


class ParseClient:
    """Client for the NBS China Data Portal API via Parse.bot"""

    def __init__(self, api_key: Optional[str] = None):
        self.base_url = "https://api.parse.bot"
        self.scraper_id = "ccd066ce-bff4-4b66-ba20-af1a8c09ae1c"
        self.api_key = api_key or os.getenv("PARSE_API_KEY")
        
        if not self.api_key:
            raise ValueError("API key required. Set PARSE_API_KEY env var or pass api_key parameter.")

    def _call(self, endpoint: str, method: str = "POST", **params) -> Dict[str, Any]:
        """
        Make an API call to the Parse.bot scraper.
        
        Args:
            endpoint: The API endpoint name
            method: HTTP method (GET or POST)
            **params: Query/body parameters
            
        Returns:
            Parsed JSON response
        """
        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)
        elif method == "POST":
            response = requests.post(url, headers=headers, json=params)
        else:
            raise ValueError(f"Unsupported method: {method}")
        
        response.raise_for_status()
        return response.json()

    def get_indicator_tree(
        self, 
        dbcode: str, 
        wdcode: str = "zb",
        parent_id: str = ""
    ) -> List[Dict[str, Any]]:
        """
        Browse the hierarchical tree of indicators, regions, or time periods.
        
        Args:
            dbcode: Database code (e.g., 'hgyd' for national monthly, 'fsnd' for provincial annual)
            wdcode: Dimension code ('zb' for indicators, 'reg' for regions, 'sj' for time)
            parent_id: Parent ID to drill down into (empty for root)
            
        Returns:
            List of tree nodes with id, name, isParent, etc.
        """
        return self._call(
            "get_indicator_tree",
            method="GET",
            dbcode=dbcode,
            wdcode=wdcode,
            parent_id=parent_id
        )

    def query_data(
        self,
        dbcode: str,
        wds: Optional[str] = None,
        dfwds: Optional[str] = None,
        rowcode: str = "zb",
        colcode: str = "sj"
    ) -> Dict[str, Any]:
        """
        Execute a custom data query against the NBS database.
        
        Args:
            dbcode: Database code
            wds: Fixed dimension filters as JSON string
            dfwds: Display filters as JSON string
            rowcode: Dimension for rows
            colcode: Dimension for columns
            
        Returns:
            Query result with datanodes and wdnodes
        """
        return self._call(
            "query_data",
            method="GET",
            dbcode=dbcode,
            wds=wds,
            dfwds=dfwds,
            rowcode=rowcode,
            colcode=colcode
        )

    def search_indicators(self, query: str, page: int = 0) -> Dict[str, Any]:
        """
        Search for statistical indicators by keyword.
        
        Args:
            query: Keyword to search (Chinese or English)
            page: Page number (default 0)
            
        Returns:
            Search results with pagecount and result array
        """
        return self._call(
            "search_indicators",
            method="GET",
            query=query,
            page=page
        )

    def get_provincial_data(
        self,
        indicator_code: str,
        limit: int = 10
    ) -> Dict[str, Any]:
        """
        Retrieve annual data for all 31 Chinese provinces for a specific indicator.
        
        Args:
            indicator_code: Indicator code (e.g., 'A020101' for Regional GDP)
            limit: Number of recent years to retrieve
            
        Returns:
            Provincial data with datanodes and wdnodes
        """
        return self._call(
            "get_provincial_data",
            method="GET",
            indicator_code=indicator_code,
            limit=limit
        )

    def get_money_supply(self, limit: int = 13) -> Dict[str, Any]:
        """
        Retrieve the most recent monthly Money Supply (M0, M1, M2) data.
        
        Args:
            limit: Number of recent months to retrieve
            
        Returns:
            Money supply data with datanodes and wdnodes
        """
        return self._call(
            "get_money_supply",
            method="GET",
            limit=limit
        )

    def get_regions(
        self,
        dbcode: str,
        parent_id: str = ""
    ) -> List[Dict[str, str]]:
        """
        Get list of region codes (provinces, cities) for a database.
        
        Args:
            dbcode: Database code
            parent_id: Parent region ID (empty for root)
            
        Returns:
            List of regions with id and name
        """
        return self._call(
            "get_regions",
            method="GET",
            dbcode=dbcode,
            parent_id=parent_id
        )

    def get_time_periods(
        self,
        dbcode: str,
        parent_id: str = ""
    ) -> List[Dict[str, str]]:
        """
        Get available time period codes for a database.
        
        Args:
            dbcode: Database code
            parent_id: Parent time ID (empty for root)
            
        Returns:
            List of time periods with id and name
        """
        return self._call(
            "get_time_periods",
            method="GET",
            dbcode=dbcode,
            parent_id=parent_id
        )


def main():
    """Practical workflow example: Search for GDP data and retrieve provincial statistics."""
    
    # Initialize client
    client = ParseClient()
    
    print("=" * 70)
    print("NBS China Data Portal API - Practical Workflow Example")
    print("=" * 70)
    
    # Step 1: Search for GDP indicator
    print("\n[Step 1] Searching for 'GDP' indicator...")
    search_results = client.search_indicators(query="GDP", page=0)
    
    if search_results.get("result"):
        print(f"Found {len(search_results['result'])} results (page count: {search_results.get('pagecount', 'N/A')})")
        for i, result in enumerate(search_results["result"][:3], 1):
            db_name = result.get("db", "Unknown")
            indicator_name = result.get("zb", "Unknown")
            print(f"  {i}. [{db_name}] {indicator_name}")
    
    # Step 2: Get available regions for provincial analysis
    print("\n[Step 2] Fetching available regions for provincial database (fsnd)...")
    regions = client.get_regions(dbcode="fsnd")
    
    print(f"Available regions: {len(regions)}")
    for region in regions[:5]:
        print(f"  - {region.get('name')} (ID: {region.get('id')})")
    if len(regions) > 5:
        print(f"  ... and {len(regions) - 5} more regions")
    
    # Step 3: Get time periods available for provincial annual data
    print("\n[Step 3] Fetching available time periods for provincial annual data...")
    time_periods = client.get_time_periods(dbcode="fsnd")
    
    print(f"Available time periods: {len(time_periods)}")
    for period in time_periods[-5:]:  # Show last 5 (most recent)
        print(f"  - {period.get('name')} (ID: {period.get('id')})")
    
    # Step 4: Get provincial GDP data
    print("\n[Step 4] Retrieving provincial GDP data (Regional GDP indicator A020101)...")
    provincial_data = client.get_provincial_data(
        indicator_code="A020101",
        limit=3
    )
    
    if provincial_data.get("datanodes"):
        print(f"Retrieved {len(provincial_data['datanodes'])} data points")
        
        # Parse and display sample data
        sample_points = provincial_data["datanodes"][:5]
        for point in sample_points:
            code = point.get("code", "Unknown")
            data_value = point.get("data", {}).get("data")
            if data_value:
                # Extract region info from code
                print(f"  Code: {code} | Value: {data_value:,.2f}")
    
    # Step 5: Get money supply data
    print("\n[Step 5] Retrieving recent Money Supply (M0, M1, M2) data...")
    money_supply = client.get_money_supply(limit=3)
    
    if money_supply.get("datanodes"):
        print(f"Retrieved {len(money_supply['datanodes'])} money supply data points")
        for point in money_supply["datanodes"][:3]:
            code = point.get("code", "Unknown")
            data_value = point.get("data", {}).get("data")
            if data_value:
                print(f"  {code}: {data_value:,.2f}")
    
    # Step 6: Explore indicator hierarchy
    print("\n[Step 6] Exploring indicator hierarchy (national monthly database)...")
    indicators = client.get_indicator_tree(
        dbcode="hgyd",
        wdcode="zb",
        parent_id=""
    )
    
    print(f"Top-level indicator categories: {len(indicators)}")
    for indicator in indicators[:5]:
        is_parent = indicator.get("isParent", False)
        parent_marker = " [has children]" if is_parent else ""
        print(f"  - {indicator.get('name')}{parent_marker} (ID: {indicator.get('id')})")
    
    print("\n" + "=" * 70)
    print("Workflow completed successfully!")
    print("=" * 70)


if __name__ == "__main__":
    main()