Gov APIblockchain.gov.in ↗

"""
India Blockchain Land Records API Client

This module provides a Python client for accessing blockchain statistics and
case study information from India's National Informatics Centre Blockchain platform.

Get your API key from: https://parse.bot/settings
"""

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


class ParseClient:
    """Client for interacting with the India Blockchain Land Records API via Parse."""

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

        Args:
            api_key: API key for Parse. If not provided, reads from PARSE_API_KEY env var.
        """
        self.base_url = "https://api.parse.bot"
        self.scraper_id = "7738d2cf-8e2b-4fe6-a637-53fc4a450434"
        self.api_key = api_key or os.getenv("PARSE_API_KEY")

        if not self.api_key:
            raise ValueError("API key must be provided or set in PARSE_API_KEY environment variable")

    def _call(self, endpoint: str, method: str = "POST", **params) -> dict[str, Any]:
        """
        Make an API call to the Parse scraper.

        Args:
            endpoint: The endpoint name (e.g., 'get_statistics')
            method: HTTP method - "GET" or "POST"
            **params: Query/body parameters for the endpoint

        Returns:
            Parsed JSON response from the API

        Raises:
            requests.RequestException: If the API call fails
            ValueError: If the response cannot be parsed
        """
        url = f"{self.base_url}/scraper/{self.scraper_id}/{endpoint}"
        headers = {"X-API-Key": self.api_key, "Content-Type": "application/json"}

        try:
            if method == "GET":
                response = requests.get(url, headers=headers, params=params, timeout=10)
            elif method == "POST":
                response = requests.post(url, headers=headers, json=params, timeout=10)
            else:
                raise ValueError(f"Unsupported HTTP method: {method}")

            response.raise_for_status()
            return response.json()
        except requests.RequestException as e:
            print(f"API call failed: {e}")
            raise

    def get_statistics(self) -> dict[str, Any]:
        """
        Fetch overall blockchain statistics from the India Blockchain platform.

        Returns statistics about blockchain products, states, departments,
        documents, verifications, and per-chain breakdowns.

        Returns:
            Dictionary containing blockchain statistics
        """
        return self._call("get_statistics", method="GET")

    def get_chain_statistics(self, chain: str) -> dict[str, Any]:
        """
        Fetch statistics for a specific blockchain chain.

        Args:
            chain: Chain identifier. Must be one of:
                   cbse, certificate, document, icjs, judiciary, property, sslcpuc

        Returns:
            Dictionary containing chain-specific statistics with transactions
            and verifications breakdown by state and department
        """
        return self._call("get_chain_statistics", method="GET", chain=chain)

    def get_case_study(self, case_study: str) -> dict[str, Any]:
        """
        Fetch and parse a case study from the India Blockchain platform.

        Args:
            case_study: Case study identifier. Must be one of:
                        BloodBank, GSTChain, LandRegistration, PDS, RemoteVoting

        Returns:
            Dictionary containing case study title and structured content sections
        """
        return self._call("get_case_study", method="GET", case_study=case_study)


def format_indian_number(num_str: str) -> int:
    """Convert Indian numeral format string to integer."""
    return int(num_str.replace(",", ""))


def main():
    """Practical workflow demonstrating the India Blockchain Land Records API."""

    # Initialize the client
    client = ParseClient()

    print("=" * 80)
    print("India Blockchain Land Records API - Practical Workflow")
    print("=" * 80)

    # Step 1: Get overall statistics
    print("\n📊 Step 1: Fetching overall blockchain statistics...")
    stats = client.get_statistics()

    print(f"\n✓ Statistics as of: {stats['date']}")
    print(f"  • Blockchain Products: {stats['blockchain_products']}")
    print(f"  • States/Organizations: {stats['states_organizations']}")
    print(f"  • Departments Onboarded: {stats['departments_onboarded']}")
    print(f"  • Total Documents: {stats['total_documents']}")
    print(f"  • Total Verifications: {stats['total_verifications']}")

    # Display sample states and departments
    print(f"\n  Sample States/Organizations:")
    for state in stats["states"][:3]:
        print(f"    - {state}")

    print(f"\n  Sample Departments:")
    for dept in stats["departments"][:3]:
        print(f"    - {dept}")

    # Step 2: Analyze transaction applications
    print("\n📈 Step 2: Transaction Applications Overview...")
    if stats.get("transaction_applications"):
        for app in stats["transaction_applications"]:
            transaction_count = format_indian_number(app["count"])
            print(f"\n  {app['name']}:")
            print(f"    • Total Transactions: {app['count']} ({transaction_count:,})")

    # Step 3: Get detailed statistics for each transaction application chain
    print("\n🔗 Step 3: Analyzing Property Chain in detail...")
    property_stats = client.get_chain_statistics("property")

    print(f"\n✓ Property Chain Statistics (Updated: {property_stats['date']})")

    # Aggregate statistics by department
    dept_transactions = {}
    for trans in property_stats.get("transactions", []):
        dept_key = f"{trans['state_or_organization']} - {trans['department']}"
        dept_transactions[dept_key] = int(trans["transaction_count"])

    # Sort by transaction count
    sorted_depts = sorted(dept_transactions.items(), key=lambda x: x[1], reverse=True)

    print("\n  Top 5 Departments by Transaction Count:")
    for i, (dept, count) in enumerate(sorted_depts[:5], 1):
        print(f"    {i}. {dept}: {count:,} transactions")

    # Step 4: Analyze verification data
    print("\n✓ Verification Data:")
    if property_stats.get("verifications"):
        top_verification = property_stats["verifications"][0]
        total_reads = format_indian_number(top_verification["total_reads"])
        reads_24h = format_indian_number(top_verification["reads_last_24h"])

        print(f"  Top Department: {top_verification['state_or_organization']} - {top_verification['department']}")
        print(f"    • Total Reads (All-Time): {total_reads:,}")
        print(f"    • Reads (Last 24h): {reads_24h:,}")
        print(f"    • Daily Average: {total_reads // 365:,}")

    # Step 5: Get case study information
    print("\n📚 Step 5: Fetching Land Registration Case Study...")
    case_study = client.get_case_study("LandRegistration")

    print(f"\n✓ Case Study: {case_study['title']}")
    print(f"  Sections available: {len(case_study['sections'])}")

    for section in case_study["sections"]:
        print(f"\n  📖 {section['heading']}:")
        for para in section["content"][:1]:  # Show first paragraph of each section
            # Truncate long paragraphs for display
            preview = para[:100] + "..." if len(para) > 100 else para
            print(f"     {preview}")

    # Step 6: Compare chains
    print("\n" + "=" * 80)
    print("📊 Step 6: Comparing Multiple Chains")
    print("=" * 80)

    chains_to_compare = ["property", "document"]
    chain_comparison = {}

    for chain_name in chains_to_compare:
        try:
            chain_data = client.get_chain_statistics(chain_name)
            transaction_count = sum(
                int(t["transaction_count"]) for t in chain_data.get("transactions", [])
            )
            verification_count = sum(
                format_indian_number(v["total_reads"]) for v in chain_data.get("verifications", [])
            )

            chain_comparison[chain_name] = {
                "transactions": transaction_count,
                "verifications": verification_count,
            }
        except Exception as e:
            print(f"  ⚠ Could not fetch {chain_name} chain: {e}")

    print("\nChain Comparison Summary:")
    for chain, metrics in chain_comparison.items():
        print(f"\n  {chain.upper()} Chain:")
        print(f"    • Total Transactions: {metrics['transactions']:,}")
        print(f"    • Total Verifications: {metrics['verifications']:,}")

    print("\n" + "=" * 80)
    print("✅ Workflow completed successfully!")
    print("=" * 80)


if __name__ == "__main__":
    main()