Spaces:
Sleeping
Sleeping
| # #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) | |
| 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() |