Org APIzppa.org.zm ↗

"""
Zambia Public Procurement Authority (ZPPA) OCDS API Client

Access Zambia's public procurement data through the ZPPA Open Contracting Data Standard (OCDS) system.
Get your API key from: https://parse.bot/settings
"""

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


class ParseClient:
    """Client for interacting with the ZPPA OCDS API via Parse."""
    
    def __init__(self, api_key: Optional[str] = None):
        """
        Initialize the Parse client.
        
        Args:
            api_key: API key for authentication. If not provided, reads from PARSE_API_KEY env var.
        """
        self.base_url = "https://api.parse.bot"
        self.scraper_id = "9194eb38-d322-4509-9d51-933120aa8121"
        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 endpoint.
        
        Args:
            endpoint: The endpoint name (e.g., 'get_current_tenders')
            method: HTTP method ('GET' or 'POST')
            **params: Parameters to pass to the endpoint
            
        Returns:
            Dictionary containing the API response data
        """
        url = f"{self.base_url}/scraper/{self.scraper_id}/{endpoint}"
        headers = {
            "X-API-Key": self.api_key,
            "Content-Type": "application/json"
        }
        
        if method.upper() == "GET":
            response = requests.get(url, headers=headers, params=params)
        else:  # POST
            response = requests.post(url, headers=headers, json=params)
        
        response.raise_for_status()
        result = response.json()
        
        if result.get("status") != "success":
            raise Exception(f"API error: {result}")
        
        return result.get("data", {})
    
    def get_current_tenders(self, page: int = 1) -> Dict[str, Any]:
        """
        Fetch a paginated list of procurement records from the ZPPA OCDS system.
        
        Args:
            page: Page number to fetch (default 1). Each page contains 20 records.
            
        Returns:
            Dictionary with 'tenders', 'page', 'total_pages', and 'total_records'
        """
        return self._call("get_current_tenders", method="GET", page=page)
    
    def get_tender_details(self, record_id: str) -> Dict[str, Any]:
        """
        Fetch full OCDS compiled release details for a specific procurement record.
        
        Args:
            record_id: MongoDB ObjectId of the record (e.g., '69f3df4ab4275d46715e26b5')
            
        Returns:
            Dictionary with 'details' containing full OCDS compiled release data
        """
        return self._call("get_tender_details", method="GET", record_id=record_id)
    
    def search_tenders(self, query: str, page: int = 1) -> Dict[str, Any]:
        """
        Search OCDS procurement records by OCID.
        
        Args:
            query: OCID search term (matches numeric portion, e.g., '10000137')
            page: Page number to fetch (default 1). Each page contains 20 records.
            
        Returns:
            Dictionary with 'results', 'query', 'page', 'total_pages', and 'total_results'
        """
        return self._call("search_tenders", method="GET", query=query, page=page)


def format_timestamp(epoch_ms: int) -> str:
    """Convert epoch milliseconds to readable date string."""
    return datetime.fromtimestamp(epoch_ms / 1000).strftime("%Y-%m-%d %H:%M:%S")


def main():
    """
    Practical workflow: Search for a tender, retrieve its details,
    and analyze procurement information.
    """
    # Initialize the client
    client = ParseClient()
    
    print("=" * 70)
    print("ZPPA OCDS Procurement Data Analysis")
    print("=" * 70)
    
    # Step 1: Search for a specific tender by OCID
    print("\n[1] Searching for tender with OCID containing '10000137'...")
    search_results = client.search_tenders(query="10000137")
    
    total_results = search_results.get("total_results", 0)
    print(f"    Found {total_results} matching record(s)")
    
    results = search_results.get("results", [])
    if not results:
        print("    No records found. Fetching recent tenders instead...")
        
        # Step 2: If search yields nothing, get current/recent tenders
        print("\n[2] Fetching most recent procurement records (page 1)...")
        recent = client.get_current_tenders(page=1)
        tenders = recent.get("tenders", [])
        total_records = recent.get("total_records", 0)
        
        print(f"    Total records in system: {total_records}")
        print(f"    Records on this page: {len(tenders)}")
        
        # Use first few records for analysis
        results = tenders[:3]
    
    # Step 3: Retrieve and analyze details for each result
    print(f"\n[3] Retrieving detailed information for {len(results)} tender(s)...")
    
    for idx, tender_summary in enumerate(results, 1):
        tender_id = tender_summary.get("id")
        ocid = tender_summary.get("ocid")
        status = tender_summary.get("status")
        creation_date = format_timestamp(tender_summary.get("creation_date", 0))
        process_date = format_timestamp(tender_summary.get("process_date", 0))
        
        print(f"\n    --- Tender {idx} ---")
        print(f"    OCID: {ocid}")
        print(f"    Status: {status}")
        print(f"    Created: {creation_date}")
        print(f"    Processed: {process_date}")
        
        # Fetch full details
        details_response = client.get_tender_details(record_id=tender_id)
        details = details_response.get("details", {})
        
        # Extract key procurement information
        tender_info = details.get("tender", {})
        title = tender_info.get("title", "N/A")
        description = tender_info.get("description", "N/A")
        procurement_method = tender_info.get("procurement_method", "N/A")
        number_of_tenderers = tender_info.get("number_of_tenderers", 0)
        
        value = tender_info.get("value", {})
        amount = value.get("amount", "N/A")
        currency = value.get("currency", "N/A")
        
        procuring_entity = tender_info.get("procuring_entity", {})
        entity_name = procuring_entity.get("name", "N/A")
        
        buyer = details.get("buyer", {})
        buyer_name = buyer.get("name", "N/A")
        
        # Print procurement details
        print(f"\n    Title: {title}")
        print(f"    Description: {description}")
        print(f"    Procuring Entity: {entity_name}")
        print(f"    Buyer: {buyer_name}")
        print(f"    Procurement Method: {procurement_method}")
        print(f"    Budget: {amount:,} {currency}" if isinstance(amount, (int, float)) else f"    Budget: {amount} {currency}")
        print(f"    Number of Tenderers: {number_of_tenderers}")
        
        # Extract and display items
        items = tender_info.get("items", [])
        if items:
            print(f"    Items ({len(items)}):")
            for item in items[:2]:  # Show first 2 items
                item_desc = item.get("description", "N/A")
                item_id = item.get("id", "N/A")
                print(f"      - {item_desc} (ID: {item_id})")
            if len(items) > 2:
                print(f"      ... and {len(items) - 2} more")
        
        # Display tenderers
        tenderers = tender_info.get("tenderers", [])
        if tenderers:
            print(f"    Tenderers ({len(tenderers)}):")
            for tenderer in tenderers[:3]:  # Show first 3 tenderers
                tenderer_name = tenderer.get("name", "N/A")
                print(f"      - {tenderer_name}")
            if len(tenderers) > 3:
                print(f"      ... and {len(tenderers) - 3} more")
    
    print("\n" + "=" * 70)
    print("Analysis complete!")
    print("=" * 70)


if __name__ == "__main__":
    main()