BluescarfAI/Design-To-code-Agent
0
1# #crew_tool.py2import json3import os4import requests5from crewai.tools import tool6from dotenv import load_dotenv7 8# Load environment variables9load_dotenv()10 11def get_landing_ai_api_key():12 """13 Get Landing AI API key from environment variables.14 Tries multiple sources for compatibility with different deployment environments.15 16 Returns:17 str: The API key if found, None otherwise18 """19 # Try different environment variable names for flexibility20 api_key = (21 os.getenv("LANDING_AI_API_KEY") or # HuggingFace secrets or production22 os.getenv("LANDING_AI_API_KEY_LOCAL") or # Local development alternative23 os.getenv("LANDINGAI_API_KEY") or # Alternative naming24 os.getenv("LANDING_API_KEY") # Another alternative25 )26 27 return api_key28 29def safe_format_with_indicators(text):30 """Replace emojis with text indicators"""31 if not isinstance(text, str):32 text = str(text)33 34 emoji_replacements = {35 '🔋': '[BATTERY]',36 '✅': '[SUCCESS]',37 '❌': '[ERROR]',38 '📄': '[DOCUMENT]',39 '📊': '[CHART]',40 '⚠️': '[WARNING]',41 '🔍': '[SEARCH]',42 '💡': '[INSIGHT]',43 '🎯': '[TARGET]',44 '✨': '[HIGHLIGHT]',45 '🚀': '[LAUNCH]',46 '📈': '[GROWTH]',47 '📉': '[DECLINE]',48 '🔧': '[TOOL]',49 '💻': '[CODE]',50 '🌟': '[STAR]',51 '🎨': '[DESIGN]',52 '📱': '[MOBILE]',53 '🖥️': '[DESKTOP]',54 '🔗': '[LINK]',55 '📝': '[NOTE]',56 '🏆': '[ACHIEVEMENT]',57 '💰': '[MONEY]',58 '🎉': '[CELEBRATION]',59 '🔒': '[SECURE]',60 '🔓': '[UNLOCKED]',61 '⭐': '[RATING]',62 '❤️': '[FAVORITE]',63 '👍': '[THUMBS_UP]',64 '👎': '[THUMBS_DOWN]',65 '🔥': '[HOT]',66 '❄️': '[COLD]',67 '🌡️': '[TEMPERATURE]'68 }69 70 for emoji, replacement in emoji_replacements.items():71 text = text.replace(emoji, replacement)72 73 return text74 75def format_api_response(response_data):76 """77 Format API response and replace emojis with safe indicators.78 79 Args:80 response_data: Raw API response (dict, str, or other)81 82 Returns:83 str: Formatted response with safe emoji replacements84 """85 try:86 if isinstance(response_data, dict):87 # Convert dict to readable format88 formatted_parts = []89 90 for key, value in response_data.items():91 if isinstance(value, (str, int, float)):92 safe_value = safe_format_with_indicators(str(value))93 formatted_parts.append(f"{key}: {safe_value}")94 elif isinstance(value, list):95 safe_items = [safe_format_with_indicators(str(item)) for item in value]96 formatted_parts.append(f"{key}: {', '.join(safe_items)}")97 elif isinstance(value, dict):98 # Handle nested dictionaries99 nested_items = []100 for nested_key, nested_value in value.items():101 safe_nested = safe_format_with_indicators(str(nested_value))102 nested_items.append(f"{nested_key}: {safe_nested}")103 formatted_parts.append(f"{key}: {'; '.join(nested_items)}")104 105 result = "\n".join(formatted_parts)106 return safe_format_with_indicators(result)107 else:108 return safe_format_with_indicators(str(response_data))109 110 except Exception as e:111 error_msg = f"Error formatting response: {str(e)}"112 return safe_format_with_indicators(error_msg)113 114@tool("LandingAI Document Analysis")115def landing_ai_document_analysis(file_path: str, file_type: str = "image") -> str:116 """117 Analyze images or PDFs using LandingAI's document analysis API.118 119 Args:120 file_path (str): Path to the image or PDF file to analyze121 file_type (str): Type of file, either "image" or "pdf"122 123 Returns:124 str: Analysis results from the API125 """126 try:127 # Get API key with improved error handling128 api_key = get_landing_ai_api_key()129 130 if not api_key:131 error_msg = "Landing AI API key not found. Please ensure LANDING_AI_API_KEY is set in environment variables or HuggingFace secrets."132 return safe_format_with_indicators(f"[ERROR] {error_msg}")133 134 # Validate file exists135 if not os.path.exists(file_path):136 error_msg = f"File not found: {file_path}"137 return safe_format_with_indicators(f"[ERROR] {error_msg}")138 139 # API endpoint140 url = "https://api.va.landing.ai/v1/tools/agentic-document-analysis"141 142 # Prepare the file for upload based on file_type143 try:144 with open(file_path, "rb") as file_obj:145 if file_type.lower() == "pdf":146 files = {"pdf": file_obj}147 else:148 files = {"image": file_obj}149 150 # Prepare headers with authentication151 headers = {"Authorization": f"Basic {api_key}"}152 153 # Make the API request with timeout154 response = requests.post(155 url, 156 files=files, 157 headers=headers,158 timeout=60 # 60 second timeout159 )160 161 # Handle different response status codes162 if response.status_code == 200:163 try:164 response_json = response.json()165 formatted_result = format_api_response(response_json)166 return safe_format_with_indicators(formatted_result)167 168 except json.JSONDecodeError:169 # If response isn't JSON, format the text response170 safe_text = safe_format_with_indicators(response.text)171 return safe_format_with_indicators(safe_text)172 173 elif response.status_code == 401:174 error_msg = "Authentication failed. Please check your Landing AI API key."175 return safe_format_with_indicators(f"[ERROR] {error_msg}")176 177 elif response.status_code == 403:178 error_msg = "Access forbidden. Your API key may not have the required permissions."179 return safe_format_with_indicators(f"[ERROR] {error_msg}")180 181 elif response.status_code == 429:182 error_msg = "Rate limit exceeded. Please try again later."183 return safe_format_with_indicators(f"[ERROR] {error_msg}")184 185 elif response.status_code == 500:186 error_msg = "Server error from Landing AI. Please try again later."187 return safe_format_with_indicators(f"[ERROR] {error_msg}")188 189 else:190 error_msg = f"API request failed with status code {response.status_code}: {response.text}"191 return safe_format_with_indicators(f"[ERROR] {error_msg}")192 193 except FileNotFoundError:194 error_msg = f"File not found: {file_path}"195 return safe_format_with_indicators(f"[ERROR] {error_msg}")196 197 except requests.exceptions.Timeout:198 error_msg = "Request timed out. The file may be too large or the service is slow."199 return safe_format_with_indicators(f"[ERROR] {error_msg}")200 201 except requests.exceptions.ConnectionError:202 error_msg = "Connection error. Please check your internet connection."203 return safe_format_with_indicators(f"[ERROR] {error_msg}")204 205 except requests.exceptions.RequestException as e:206 error_msg = f"Request failed: {str(e)}"207 return safe_format_with_indicators(f"[ERROR] {error_msg}")208 209 except Exception as e:210 error_msg = f"Unexpected error in document analysis: {str(e)}"211 return safe_format_with_indicators(f"[ERROR] {error_msg}")212 213def test_landing_ai_connection():214 """215 Test function to verify Landing AI API connection.216 This can be called to verify the setup before processing documents.217 218 Returns:219 bool: True if API key is available and valid format, False otherwise220 """221 api_key = get_landing_ai_api_key()222 223 if not api_key:224 print("❌ Landing AI API key not found in environment variables")225 print("Available environment variables:")226 for key in os.environ.keys():227 if 'LANDING' in key.upper() or 'API' in key.upper():228 print(f" - {key}: {'SET' if os.environ[key] else 'EMPTY'}")229 return False230 231 # Basic format validation (adjust based on Landing AI key format)232 if len(api_key) < 10:233 print("❌ Landing AI API key appears to be too short")234 return False235 236 print("✅ Landing AI API key found and appears valid")237 return True238 239# Test the connection when module is imported (optional)240if __name__ == "__main__":241 test_landing_ai_connection()