Edu APIaustlii.edu.au ↗

"""
AustLII Legal Case Scraper - ParseClient Implementation

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

import os
import requests
from typing import Optional, Any


class ParseClient:
    """Client for interacting with the AustLII Legal Case Scraper API."""
    
    def __init__(self, api_key: Optional[str] = None):
        """
        Initialize the ParseClient.
        
        Args:
            api_key: API key for Parse API. If not provided, reads from PARSE_API_KEY environment variable.
        """
        self.base_url = "https://api.parse.bot"
        self.scraper_id = "6c24f681-35dd-4b59-b1c9-216b3fc16289"
        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 a request to the Parse API.
        
        Args:
            endpoint: The API endpoint name
            method: HTTP method (GET or POST)
            **params: Parameters to pass to the endpoint
            
        Returns:
            The JSON response from the API
        """
        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 HTTP method: {method}")
        
        response.raise_for_status()
        return response.json()
    
    def list_databases(self) -> dict[str, Any]:
        """
        List all available legal databases on AustLII organized by jurisdiction and category.
        
        Returns:
            Dictionary containing list of databases with their metadata
        """
        return self._call("list_databases", method="GET")
    
    def get_database_years(self, path: str) -> dict[str, Any]:
        """
        List all available years for a specific database.
        
        Args:
            path: The database path (e.g. 'cgi-bin/viewdb/au/cases/cth/HCA')
            
        Returns:
            Dictionary containing list of available years for the database
        """
        return self._call("get_database_years", method="GET", path=path)
    
    def list_cases(self, path: str, year: Optional[str] = None) -> dict[str, Any]:
        """
        List cases for a specific database and year.
        
        Args:
            path: The database path (e.g. 'cgi-bin/viewdb/au/cases/cth/HCA')
            year: The year (e.g. '2024'). If not provided, lists all cases
            
        Returns:
            Dictionary containing list of cases for the specified database and year
        """
        params = {"path": path}
        if year:
            params["year"] = year
        return self._call("list_cases", method="GET", **params)
    
    def get_case(self, url: str) -> dict[str, Any]:
        """
        Retrieve detailed metadata and download links for a specific case document.
        
        Args:
            url: The URL of the case document page
            
        Returns:
            Dictionary containing case metadata and download links
        """
        return self._call("get_case", method="GET", url=url)


def main():
    """Practical workflow demonstrating the ParseClient with real use case."""
    
    # Initialize the client
    client = ParseClient()
    
    print("=" * 90)
    print("AustLII Legal Case Scraper - Practical Usage Example")
    print("=" * 90)
    
    # Step 1: List all available databases
    print("\n[Step 1] Fetching all available legal databases from AustLII...")
    try:
        databases_response = client.list_databases()
        databases = databases_response.get("data", {}).get("databases", [])
    except Exception as e:
        print(f"Error fetching databases: {e}")
        return
    
    print(f"✓ Found {len(databases)} databases across all jurisdictions\n")
    
    # Group databases by jurisdiction for display
    jurisdictions = {}
    for db in databases:
        jurisdiction = db.get("jurisdiction", "Unknown")
        if jurisdiction not in jurisdictions:
            jurisdictions[jurisdiction] = []
        jurisdictions[jurisdiction].append(db)
    
    # Display first few jurisdictions
    print("Available Jurisdictions:")
    for i, (jurisdiction, dbs) in enumerate(list(jurisdictions.items())[:3]):
        print(f"  • {jurisdiction}: {len(dbs)} database(s)")
    
    # Find the High Court of Australia database as an example
    print("\n[Step 2] Searching for High Court of Australia database...")
    hca_db = None
    for db in databases:
        if "High Court of Australia" in db.get("database_name", ""):
            hca_db = db
            break
    
    if not hca_db:
        print("High Court of Australia database not found. Using first available database.")
        hca_db = databases[0] if databases else None
    
    if not hca_db:
        print("✗ No databases found!")
        return
    
    print(f"✓ Selected: {hca_db['database_name']}")
    print(f"  Jurisdiction: {hca_db['jurisdiction']}")
    print(f"  Category: {hca_db['category']}")
    
    # Step 3: Get available years for the selected database
    db_path = hca_db["path"]
    print(f"\n[Step 3] Fetching available years for: {hca_db['database_name']}...")
    try:
        years_response = client.get_database_years(path=db_path)
        years = years_response.get("data", {}).get("years", [])
    except Exception as e:
        print(f"Error fetching years: {e}")
        return
    
    print(f"✓ Found {len(years)} years with available cases")
    print(f"  Years available: {', '.join(years[:5])}{'...' if len(years) > 5 else ''}")
    
    # Step 4: List cases for the most recent year
    if years:
        recent_year = years[0]
        print(f"\n[Step 4] Listing cases from {recent_year}...")
        try:
            cases_response = client.list_cases(path=db_path, year=recent_year)
            cases = cases_response.get("data", {}).get("cases", [])
            total_count = cases_response.get("data", {}).get("count", 0)
        except Exception as e:
            print(f"Error fetching cases: {e}")
            return
        
        print(f"✓ Found {total_count} cases in {recent_year}")
        
        # Step 5: Get detailed information for multiple cases
        print(f"\n[Step 5] Retrieving details for the first {min(3, len(cases))} cases...")
        print("-" * 90)
        
        for i, case in enumerate(cases[:3], 1):
            try:
                case_details = client.get_case(url=case["url"])
                case_data = case_details.get("data", {})
                
                print(f"\nCase {i}:")
                print(f"  Title: {case_data.get('title', 'N/A')[:80]}")
                
                downloads = case_data.get("downloads", {})
                priority_url = case_data.get("priority_download_url", "N/A")
                
                print(f"  Priority Download Format: {priority_url.split('.')[-1].upper() if priority_url != 'N/A' else 'N/A'}")
                
                available_formats = [fmt.replace('_', ' ').title() for fmt in downloads.keys() if downloads[fmt]]
                print(f"  Available Formats: {', '.join(available_formats)}")
                
            except Exception as e:
                print(f"\n  ✗ Error retrieving case details: {e}")
        
        print("\n" + "-" * 90)
    else:
        print("✗ No years found for this database.")
    
    print("\n" + "=" * 90)
    print("✓ Workflow completed successfully!")
    print("=" * 90)


if __name__ == "__main__":
    main()