Update app.py

app.py

@@ -78,7 +78,7 @@ config = Config()
 # =============================================================================
 def retry_with_backoff(max_retries=None, base_delay=None):
     """Decorator for automatic retry with exponential backoff"""
-    max_retries = max_retries or config.MAX_RETRIES
+    max_retries = max_retries or Config.MAX_RETRIES
     base_delay = base_delay or config.BASE_RETRY_DELAY
     
     def decorator(func):
@@ -1397,11 +1397,11 @@ def get_youtube_transcript(video_url: str) -> str:
         transcriber = aai.Transcriber()
         print(f"   Submitting to AssemblyAI...")
         
-        config = aai.TranscriptionConfig(
+        config_obj = aai.TranscriptionConfig(
             speech_model=aai.SpeechModel.best,
         )
         
-        transcript = transcriber.transcribe(video_url, config=config)
+        transcript = transcriber.transcribe(video_url, config=config_obj)
         
         # Wait for completion
         print(f"   Initial status: {transcript.status}")
@@ -1424,12 +1424,21 @@ def get_youtube_transcript(video_url: str) -> str:
                 print(f"   Status after {elapsed}s: {transcript.status}")
             except Exception as refresh_err:
                 print(f"   Warning: Could not refresh status: {refresh_err}")
-                # Continue anyway, maybe it finished
                 break
         
         # Check final status
         if transcript.status == aai.TranscriptStatus.error:
             error_msg = getattr(transcript, 'error', 'Unknown error')
+            
+            # ===== NEW: Check for network block =====
+            if "text/html" in error_msg or "HTML document" in error_msg:
+                raise RuntimeError(
+                    "YouTube access blocked. "
+                    "If a local video file was provided, use analyze_image or audio_transcription_tool instead. "
+                    "Or try downloading the video first."
+                )
+            # ===== END NEW =====
+            
             raise RuntimeError(f"AssemblyAI transcription failed: {error_msg}")
         
         if transcript.status != aai.TranscriptStatus.completed:
@@ -1454,29 +1463,23 @@ def get_youtube_transcript(video_url: str) -> str:
         telemetry.record_call("get_youtube_transcript", time.time() - start_time, True)
         return f"YouTube Transcript:\n{truncate_if_needed(result_text)}"
         
-    except aai.types.TranscriptError as e:
+    except Exception as e:
         telemetry.record_call("get_youtube_transcript", time.time() - start_time, False)
         error_msg = str(e)
         
         suggestions = []
-        if "not found" in error_msg.lower():
+        if "text/html" in error_msg.lower() or "html document" in error_msg.lower():
+            suggestions.append("YouTube blocked on HuggingFace. Use the local .mp4 file instead with audio_transcription_tool or analyze_image")
+        elif "not found" in error_msg.lower():
             suggestions.append("Video may be private or deleted")
-        if "quota" in error_msg.lower() or "limit" in error_msg.lower():
+        elif "quota" in error_msg.lower() or "limit" in error_msg.lower():
             suggestions.append("AssemblyAI quota exceeded")
-        if "timeout" in error_msg.lower():
+        elif "timeout" in error_msg.lower():
             suggestions.append("Video may be too long (try shorter video)")
         
-        suggestion_text = " | ".join(suggestions) if suggestions else "Check video URL and API quota"
+        suggestion_text = " | ".join(suggestions) if suggestions else "Check video URL is valid and public"
         
         raise ToolError("get_youtube_transcript", e, suggestion_text)
-    
-    except TimeoutError as e:
-        telemetry.record_call("get_youtube_transcript", time.time() - start_time, False)
-        raise ToolError("get_youtube_transcript", e, "Video too long or AssemblyAI overloaded. Try shorter video.")
-    
-    except Exception as e:
-        telemetry.record_call("get_youtube_transcript", time.time() - start_time, False)
-        raise ToolError("get_youtube_transcript", e, "Check video URL is valid and public")
 
 
 class BrowseInput(BaseModel):
@@ -1711,7 +1714,76 @@ def scrape_and_retrieve(url: str, query: str) -> str:
         telemetry.record_call("scrape_and_retrieve", time.time() - start_time, False)
         raise ToolError("scrape_and_retrieve", e)
 
+class VideoAnalysisInput(BaseModel):
+    file_path: str = Field(description="Path to video file (.mp4, .mov, etc.)")
+    query: str = Field(description="What to find in the video")
 
+@tool(args_schema=VideoAnalysisInput)
+def analyze_video(file_path: str, query: str) -> str:
+    """
+    Analyze video using Gemini Vision (supports video).
+    
+    Use for:
+    - Counting objects/people/animals in video
+    - Describing what happens
+    - Finding specific moments
+    - Visual Q&A about video content
+    """
+    start_time = time.time()
+    
+    try:
+        print(f"🎥 Analyzing video: {file_path}")
+        print(f"   Query: {query[:100]}...")
+        
+        video_path = find_file(file_path)
+        if not video_path and os.path.exists(file_path):
+            video_path = Path(file_path)
+        
+        if not video_path or not video_path.exists():
+            raise FileNotFoundError(f"Video not found: {file_path}")
+        
+        GOOGLE_API_KEY = os.getenv("GEMINI_API_KEY")
+        if not GOOGLE_API_KEY:
+            raise ValueError("GEMINI_API_KEY not set")
+        
+        # Upload video to Gemini
+        print(f"   Uploading video to Gemini...")
+        import google.generativeai as genai
+        genai.configure(api_key=GOOGLE_API_KEY)
+        
+        video_file = genai.upload_file(path=str(video_path))
+        
+        print(f"   Waiting for processing...")
+        while video_file.state.name == "PROCESSING":
+            time.sleep(2)
+            video_file = genai.get_file(video_file.name)
+        
+        if video_file.state.name == "FAILED":
+            raise RuntimeError("Video processing failed")
+        
+        # Analyze with Gemini
+        print(f"   Analyzing with Gemini...")
+        model = genai.GenerativeModel("gemini-2.0-flash-exp")
+        
+        response = model.generate_content([
+            video_file,
+            query
+        ])
+        
+        result = response.text
+        
+        # Clean up
+        genai.delete_file(video_file.name)
+        
+        print(f"✓ Analysis complete: {len(result)} chars")
+        
+        telemetry.record_call("analyze_video", time.time() - start_time, True)
+        return f"Video Analysis:\n{truncate_if_needed(result)}"
+        
+    except Exception as e:
+        telemetry.record_call("analyze_video", time.time() - start_time, False)
+        raise ToolError("analyze_video", e, "Check video file path and Gemini API")
+        
 class FinalAnswerInput(BaseModel):
     answer: str = Field(description="Final answer - exact, no fluff")
 
@@ -1741,8 +1813,9 @@ defined_tools = [
     
     # Core tools
     search_tool,
-    wikipedia_search,  # ← ADD THIS NEW TOOL
+    wikipedia_search,
     calculator,
+    analyze_video,
     code_interpreter,
     
     # File operations
@@ -1998,6 +2071,19 @@ Turn 5: final_answer_tool("3")
 REMEMBER: wikipedia_search() wants just the SUBJECT NAME!
 ═══════════════════════════════════════════════════════════════
 
+**YOUTUBE VIDEO QUESTIONS** (Network restrictions):
+⚠️ YouTube URLs may be blocked on HuggingFace Spaces!
+If question mentions YouTube AND a local .mp4 file exists:
+→ Use analyze_image tool on the local video file instead
+→ Or use audio_transcription_tool for audio content
+
+Example:
+Q: "In video https://youtube.com/..., what happens?"
+[FILE: task_123.mp4]
+✅ CORRECT: analyze_image("files/task_123.mp4", "what happens in video")
+❌ WRONG: get_youtube_transcript("https://youtube.com/...")
+
+
 🚨 ANTI-LOOP RULES:
 ═══════════════════════════════════════════════════════════════
 
@@ -2159,7 +2245,7 @@ REMEMBER: One tool per turn. No reasoning without tools. Exact answer format.
         
             # Check if we should force reflection
             consecutive_errors = state.get('consecutive_errors', 0)
-            should_reflect = (current_turn > 5 and current_turn % REFLECT_EVERY_N_TURNS == 0) or consecutive_errors >= 3         
+            should_reflect = (current_turn > 5 and current_turn % Config.REFLECT_EVERY_N_TURNS == 0) or consecutive_errors >= 3         
             
             # Force tool usage
             if len(messages_to_send) >= 2: