tools/custom_tool.py

# #crew_tool.py
import json
import os
import requests
from crewai.tools import tool
from dotenv import load_dotenv

# Load environment variables
load_dotenv()

def get_landing_ai_api_key():
    """
    Get Landing AI API key from environment variables.
    Tries multiple sources for compatibility with different deployment environments.
    
    Returns:
        str: The API key if found, None otherwise
    """
    # Try different environment variable names for flexibility
    api_key = (
        os.getenv("LANDING_AI_API_KEY") or           # HuggingFace secrets or production
        os.getenv("LANDING_AI_API_KEY_LOCAL") or     # Local development alternative
        os.getenv("LANDINGAI_API_KEY") or            # Alternative naming
        os.getenv("LANDING_API_KEY")                 # Another alternative
    )
    
    return api_key

def safe_format_with_indicators(text):
    """Replace emojis with text indicators"""
    if not isinstance(text, str):
        text = str(text)
        
    emoji_replacements = {
        '🔋': '[BATTERY]',
        '✅': '[SUCCESS]',
        '❌': '[ERROR]',
        '📄': '[DOCUMENT]',
        '📊': '[CHART]',
        '⚠️': '[WARNING]',
        '🔍': '[SEARCH]',
        '💡': '[INSIGHT]',
        '🎯': '[TARGET]',
        '✨': '[HIGHLIGHT]',
        '🚀': '[LAUNCH]',
        '📈': '[GROWTH]',
        '📉': '[DECLINE]',
        '🔧': '[TOOL]',
        '💻': '[CODE]',
        '🌟': '[STAR]',
        '🎨': '[DESIGN]',
        '📱': '[MOBILE]',
        '🖥️': '[DESKTOP]',
        '🔗': '[LINK]',
        '📝': '[NOTE]',
        '🏆': '[ACHIEVEMENT]',
        '💰': '[MONEY]',
        '🎉': '[CELEBRATION]',
        '🔒': '[SECURE]',
        '🔓': '[UNLOCKED]',
        '⭐': '[RATING]',
        '❤️': '[FAVORITE]',
        '👍': '[THUMBS_UP]',
        '👎': '[THUMBS_DOWN]',
        '🔥': '[HOT]',
        '❄️': '[COLD]',
        '🌡️': '[TEMPERATURE]'
    }
    
    for emoji, replacement in emoji_replacements.items():
        text = text.replace(emoji, replacement)
    
    return text

def format_api_response(response_data):
    """
    Format API response and replace emojis with safe indicators.
    
    Args:
        response_data: Raw API response (dict, str, or other)
        
    Returns:
        str: Formatted response with safe emoji replacements
    """
    try:
        if isinstance(response_data, dict):
            # Convert dict to readable format
            formatted_parts = []
            
            for key, value in response_data.items():
                if isinstance(value, (str, int, float)):
                    safe_value = safe_format_with_indicators(str(value))
                    formatted_parts.append(f"{key}: {safe_value}")
                elif isinstance(value, list):
                    safe_items = [safe_format_with_indicators(str(item)) for item in value]
                    formatted_parts.append(f"{key}: {', '.join(safe_items)}")
                elif isinstance(value, dict):
                    # Handle nested dictionaries
                    nested_items = []
                    for nested_key, nested_value in value.items():
                        safe_nested = safe_format_with_indicators(str(nested_value))
                        nested_items.append(f"{nested_key}: {safe_nested}")
                    formatted_parts.append(f"{key}: {'; '.join(nested_items)}")
            
            result = "\n".join(formatted_parts)
            return safe_format_with_indicators(result)
        else:
            return safe_format_with_indicators(str(response_data))
            
    except Exception as e:
        error_msg = f"Error formatting response: {str(e)}"
        return safe_format_with_indicators(error_msg)

@tool("LandingAI Document Analysis")
def landing_ai_document_analysis(file_path: str, file_type: str = "image") -> str:
    """
    Analyze images or PDFs using LandingAI's document analysis API.
    
    Args:
        file_path (str): Path to the image or PDF file to analyze
        file_type (str): Type of file, either "image" or "pdf"
    
    Returns:
        str: Analysis results from the API
    """
    try:
        # Get API key with improved error handling
        api_key = get_landing_ai_api_key()
        
        if not api_key:
            error_msg = "Landing AI API key not found. Please ensure LANDING_AI_API_KEY is set in environment variables or HuggingFace secrets."
            return safe_format_with_indicators(f"[ERROR] {error_msg}")
        
        # Validate file exists
        if not os.path.exists(file_path):
            error_msg = f"File not found: {file_path}"
            return safe_format_with_indicators(f"[ERROR] {error_msg}")
        
        # API endpoint
        url = "https://api.va.landing.ai/v1/tools/agentic-document-analysis"
        
        # Prepare the file for upload based on file_type
        try:
            with open(file_path, "rb") as file_obj:
                if file_type.lower() == "pdf":
                    files = {"pdf": file_obj}
                else:
                    files = {"image": file_obj}
                    
                # Prepare headers with authentication
                headers = {"Authorization": f"Basic {api_key}"}
                
                # Make the API request with timeout
                response = requests.post(
                    url, 
                    files=files, 
                    headers=headers,
                    timeout=60  # 60 second timeout
                )
                
                # Handle different response status codes
                if response.status_code == 200:
                    try:
                        response_json = response.json()
                        formatted_result = format_api_response(response_json)
                        return safe_format_with_indicators(formatted_result)
                        
                    except json.JSONDecodeError:
                        # If response isn't JSON, format the text response
                        safe_text = safe_format_with_indicators(response.text)
                        return safe_format_with_indicators(safe_text)
                
                elif response.status_code == 401:
                    error_msg = "Authentication failed. Please check your Landing AI API key."
                    return safe_format_with_indicators(f"[ERROR] {error_msg}")
                
                elif response.status_code == 403:
                    error_msg = "Access forbidden. Your API key may not have the required permissions."
                    return safe_format_with_indicators(f"[ERROR] {error_msg}")
                
                elif response.status_code == 429:
                    error_msg = "Rate limit exceeded. Please try again later."
                    return safe_format_with_indicators(f"[ERROR] {error_msg}")
                
                elif response.status_code == 500:
                    error_msg = "Server error from Landing AI. Please try again later."
                    return safe_format_with_indicators(f"[ERROR] {error_msg}")
                
                else:
                    error_msg = f"API request failed with status code {response.status_code}: {response.text}"
                    return safe_format_with_indicators(f"[ERROR] {error_msg}")
        
        except FileNotFoundError:
            error_msg = f"File not found: {file_path}"
            return safe_format_with_indicators(f"[ERROR] {error_msg}")
        
        except requests.exceptions.Timeout:
            error_msg = "Request timed out. The file may be too large or the service is slow."
            return safe_format_with_indicators(f"[ERROR] {error_msg}")
        
        except requests.exceptions.ConnectionError:
            error_msg = "Connection error. Please check your internet connection."
            return safe_format_with_indicators(f"[ERROR] {error_msg}")
        
        except requests.exceptions.RequestException as e:
            error_msg = f"Request failed: {str(e)}"
            return safe_format_with_indicators(f"[ERROR] {error_msg}")
    
    except Exception as e:
        error_msg = f"Unexpected error in document analysis: {str(e)}"
        return safe_format_with_indicators(f"[ERROR] {error_msg}")

def test_landing_ai_connection():
    """
    Test function to verify Landing AI API connection.
    This can be called to verify the setup before processing documents.
    
    Returns:
        bool: True if API key is available and valid format, False otherwise
    """
    api_key = get_landing_ai_api_key()
    
    if not api_key:
        print("❌ Landing AI API key not found in environment variables")
        print("Available environment variables:")
        for key in os.environ.keys():
            if 'LANDING' in key.upper() or 'API' in key.upper():
                print(f"  - {key}: {'SET' if os.environ[key] else 'EMPTY'}")
        return False
    
    # Basic format validation (adjust based on Landing AI key format)
    if len(api_key) < 10:
        print("❌ Landing AI API key appears to be too short")
        return False
    
    print("✅ Landing AI API key found and appears valid")
    return True

# Test the connection when module is imported (optional)
if __name__ == "__main__":
    test_landing_ai_connection()