Duck APIduck.ai ↗

"""
Duck.ai Models API Client
Retrieve AI model data and service status from Duck.ai (DuckDuckGo's private AI chat).
Get your API key from: https://parse.bot/settings
"""

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


class ParseClient:
    """Client for interacting with Duck.ai Models API via Parse.bot"""

    def __init__(self, api_key: Optional[str] = None):
        """
        Initialize the Parse API 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 = "ec7d8282-525d-433a-9943-342788add371"
        self.api_key = api_key or os.getenv("PARSE_API_KEY")
        
        if not self.api_key:
            raise ValueError("API key is required. Set PARSE_API_KEY environment variable or pass api_key parameter.")

    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_models', 'get_status')
            method: HTTP method ('GET' or 'POST')
            **params: Additional parameters to pass to the endpoint
            
        Returns:
            Response data as a dictionary
            
        Raises:
            requests.RequestException: If the API call fails
        """
        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:
            payload = params if params else {}
            response = requests.post(url, headers=headers, json=payload)
        
        response.raise_for_status()
        return response.json()

    def get_models(self) -> Dict[str, Any]:
        """
        Fetch all available AI models from Duck.ai.
        
        Returns information about model capabilities, providers, access tiers,
        supported tools, and file type support.
        
        Returns:
            Dictionary containing:
            - models: List of model objects with configurations
            - attachment_limits: File/image upload limits per subscription tier
            - javascript: Pre-formatted JavaScript export string
            - total_models: Count of models returned
        """
        return self._call("get_models", method="GET")

    def get_status(self) -> Dict[str, Any]:
        """
        Fetch the current Duck.ai service status.
        
        Returns:
            Dictionary containing:
            - status: Primary status code (0 = operational)
            - secondaryStatus: Secondary status code
            - statusV2: V2 status code
        """
        return self._call("get_status", method="GET")


def analyze_model_ecosystem(models: List[Dict[str, Any]]) -> Dict[str, Any]:
    """
    Analyze the Duck.ai model ecosystem to extract insights.
    
    Args:
        models: List of model objects from get_models response
        
    Returns:
        Dictionary with ecosystem analysis
    """
    analysis = {
        "total_models": len(models),
        "providers": {},
        "access_tiers": {},
        "tools_coverage": {},
        "image_support": 0,
        "pdf_support": 0,
        "reasoning_support": 0,
    }
    
    for model in models:
        # Provider stats
        provider = model.get("provider", "unknown").upper()
        if provider not in analysis["providers"]:
            analysis["providers"][provider] = {
                "count": 0,
                "models": [],
                "max_access_tier": "free"
            }
        analysis["providers"][provider]["count"] += 1
        analysis["providers"][provider]["models"].append(model.get("name"))
        
        # Access tier stats
        tiers = model.get("accessTier", [])
        for tier in tiers:
            if tier not in analysis["access_tiers"]:
                analysis["access_tiers"][tier] = 0
            analysis["access_tiers"][tier] += 1
        
        # Tools coverage
        tools = model.get("supportedTools", [])
        for tool in tools:
            if tool not in analysis["tools_coverage"]:
                analysis["tools_coverage"][tool] = []
            analysis["tools_coverage"][tool].append(model.get("name"))
        
        # Feature support
        if model.get("supportsImageUpload"):
            analysis["image_support"] += 1
        
        file_types = model.get("supportedFileTypes", [])
        if "application/pdf" in file_types:
            analysis["pdf_support"] += 1
        
        reasoning = model.get("supportedReasoningEffort", [])
        if len(reasoning) > 1:  # More than just "none"
            analysis["reasoning_support"] += 1
    
    return analysis


def filter_models_by_tier(models: List[Dict[str, Any]], tier: str) -> List[Dict[str, Any]]:
    """
    Filter models available for a specific access tier.
    
    Args:
        models: List of model objects
        tier: Access tier ('free', 'plus', 'pro', 'internal')
        
    Returns:
        List of models available in the specified tier
    """
    return [m for m in models if tier in m.get("accessTier", [])]


def find_best_model_for_task(models: List[Dict[str, Any]], 
                             requires_images: bool = False,
                             requires_pdf: bool = False,
                             requires_web_search: bool = False,
                             tier: str = "pro") -> Optional[Dict[str, Any]]:
    """
    Find the best model that matches specific requirements.
    
    Args:
        models: List of model objects
        requires_images: Whether image upload capability is needed
        requires_pdf: Whether PDF file support is needed
        requires_web_search: Whether web search tool is needed
        tier: Minimum access tier required
        
    Returns:
        Best matching model or None if no match found
    """
    candidates = filter_models_by_tier(models, tier)
    
    for model in candidates:
        if requires_images and not model.get("supportsImageUpload"):
            continue
        
        if requires_pdf and "application/pdf" not in model.get("supportedFileTypes", []):
            continue
        
        if requires_web_search and "WebSearch" not in model.get("supportedTools", []):
            continue
        
        return model
    
    return None


def main():
    """Practical workflow: Check service health, analyze models, and find optimal models for tasks."""
    
    # Initialize the client
    client = ParseClient()
    
    print("\n" + "=" * 80)
    print("DUCK.AI MODELS ANALYSIS WORKFLOW")
    print("=" * 80)
    
    # Step 1: Check service status
    print("\n[1/5] Checking Duck.ai service status...")
    try:
        status_response = client.get_status()
        status_code = int(status_response.get("status", -1))
        
        if status_code == 0:
            print("✓ Duck.ai service is OPERATIONAL")
        else:
            print(f"✗ Service degraded - Status code: {status_code}")
            print(f"  Secondary status: {status_response.get('secondaryStatus', 'unknown')}")
            return
    except Exception as e:
        print(f"✗ Failed to check service status: {e}")
        return
    
    # Step 2: Fetch all available models
    print("\n[2/5] Fetching available AI models...")
    try:
        models_response = client.get_models()
        models = models_response.get("models", [])
        total_models = models_response.get("total_models", 0)
        attachment_limits = models_response.get("attachment_limits", {})
        
        print(f"✓ Retrieved {total_models} models from Duck.ai")
    except Exception as e:
        print(f"✗ Failed to fetch models: {e}")
        return
    
    # Step 3: Analyze ecosystem
    print("\n[3/5] Analyzing model ecosystem...")
    analysis = analyze_model_ecosystem(models)
    
    print("\n  Provider Distribution:")
    for provider, data in sorted(analysis["providers"].items()):
        model_names = ", ".join(data["models"][:2])
        suffix = f" (+{len(data['models'])-2} more)" if len(data["models"]) > 2 else ""
        print(f"    • {provider:15} {data['count']:2} models  ({model_names}{suffix})")
    
    print("\n  Feature Support:")
    print(f"    • Models with image upload:    {analysis['image_support']}/{total_models}")
    print(f"    • Models with PDF support:     {analysis['pdf_support']}/{total_models}")
    print(f"    • Models with reasoning effort: {analysis['reasoning_support']}/{total_models}")
    
    print("\n  Available Tools:")
    for tool, model_names in sorted(analysis["tools_coverage"].items()):
        print(f"    • {tool:25} {len(model_names)} models")
    
    print("\n  Access Tier Distribution:")
    for tier, count in sorted(analysis["access_tiers"].items(), 
                               key=lambda x: {"free": 0, "plus": 1, "pro": 2, "internal": 3}.get(x[0], 99)):
        print(f"    • {tier:10} {count:2} models")
    
    # Step 4: Find models for specific use cases
    print("\n[4/5] Finding optimal models for specific use cases...")
    
    use_cases = [
        {
            "name": "Document Analysis (PDF + Images)",
            "requires_pdf": True,
            "requires_images": True,
            "tier": "pro"
        },
        {
            "name": "Web Research with AI",
            "requires_web_search": True,
            "tier": "plus"
        },
        {
            "name": "Free Tier Analysis",
            "tier": "free"
        }
    ]
    
    for use_case in use_cases:
        best_model = find_best_model_for_task(
            models,
            requires_images=use_case.get("requires_images", False),
            requires_pdf=use_case.get("requires_pdf", False),
            requires_web_search=use_case.get("requires_web_search", False),
            tier=use_case["tier"]
        )
        
        status = "✓" if best_model else "✗"
        if best_model:
            print(f"  {status} {use_case['name']:40} → {best_model.get('name')}")
        else:
            print(f"  {status} {use_case['name']:40} → No matching model")
    
    # Step 5: Display subscription tier information
    print("\n[5/5] Subscription tier limits...")
    
    tier_order = ["free", "plus", "pro"]
    for tier in tier_order:
        if tier in attachment_limits:
            limits = attachment_limits[tier]
            files = limits.get("files", {})
            images = limits.get("images", {})
            
            tier_display = tier.upper()
            print(f"\n  {tier_display} Tier:")
            print(f"    Files:   {files.get('maxPerConversation', '?')} per conversation, " 
                  f"max {files.get('maxFileSizeMB', '?')}MB, " 
                  f"{files.get('maxPagesPerFile', '?')} pages per file")
            print(f"    Images:  {images.get('maxPerTurn', '?')} per turn, " 
                  f"{images.get('maxPerConversation', '?')} per conversation")
    
    # Summary
    print("\n" + "=" * 80)
    print("WORKFLOW COMPLETE")
    print(f"Total models analyzed: {total_models}")
    print(f"Providers: {len(analysis['providers'])}")
    print(f"Tools available: {len(analysis['tools_coverage'])}")
    print("=" * 80 + "\n")


if __name__ == "__main__":
    main()