Update app.py

app.py

@@ -6,17 +6,9 @@ import pandas as pd
 
 from langgraph.graph import StateGraph, END
 from typing import TypedDict
-import re
 import string
-from duckduckgo_search import DDGS
-
-from pytube import YouTube, exceptions as pytube_exceptions
-from pytube.innertube import _default_clients
-_default_clients["WEB"]["clientVersion"] = "2.20240401.00.00"
-import cv2
-from ultralytics import YOLO
-import tempfile
-import os
+
+
 
 
 # (Keep Constants as is)
@@ -36,7 +28,6 @@ DEFAULT_API_URL = "https://agents-course-unit4-scoring.hf.space"
 
 class SuperSmartAgent:
     def __init__(self):
-        self.model = YOLO("yolov8n.pt")
         self.graph = self._build_graph()
 
     def _build_graph(self):
@@ -134,168 +125,6 @@ class SuperSmartAgent:
 
         ###################
         
-        def contains_youtube_url(text):
-            # This regex is good for identifying the presence of a YouTube URL
-            return re.search(r"(https?://)?(www\.)?(youtube\.com|youtu\.be)/", text) is not None
-        
-        def extract_youtube_url(text):
-            # A more focused regex to *extract* the full YouTube URL
-            # This pattern attempts to match common YouTube URL formats and capture the entire URL
-            youtube_regex = (
-                r'(https?://)?(www\.)?'
-                '(?:youtube\.com|youtu\.be)/'  # Non-capturing group for domain
-                '(?:watch\?v=|embed/|v/|.+\?v=)?' # Non-capturing group for path
-                '([a-zA-Z0-9_-]{11})' # Capturing group for the 11-character video ID
-                '(?:[&?].*)?' # Optional: matches any query parameters after the video ID
-            )
-            match = re.search(youtube_regex, text)
-            if match:
-                # Reconstruct the standard YouTube watch URL from the video ID
-                video_id = match.group(3) # The 11-character video ID
-                return f"https://www.youtube.com/watch?v={video_id}"
-            return None
-
-        
-        def check_special_tools(state):
-            q = state["question"].lower()
-        
-            if "search the web" in q or "look up" in q:
-                state["use_tool"] = "web"
-            elif contains_youtube_url(q) or "in the video" in q or "from the video" in q:
-                state["use_tool"] = "video"
-            else:
-                state["use_tool"] = None
-        
-            return state
-
-        def use_web_search(state):
-            question = state.get("question", "")
-            query = question.lower().replace("search the web for", "").replace("look up", "").strip()
-            print(f"Searching DuckDuckGo for: {query}")
-        
-            try:
-                with DDGS() as ddgs:
-                    results = ddgs.text(query, max_results=3)
-        
-                if results:
-                    snippets = [f"{r['title']}: {r.get('body', '')}" for r in results]
-                    state["response"] = "\n".join(snippets)
-                else:
-                    state["response"] = "No results found."
-            except Exception as e:
-                state["response"] = f"Web search error: {e}"
-            return state
-        
-        def use_video_analysis(state):
-            if self.model is None:
-                state["response"] = "Video analysis skipped: YOLO model failed to load during agent initialization."
-                return state
-
-            # Use the new helper function to extract a clean YouTube URL
-            url = extract_youtube_url(state["question"])
-
-            if not url:
-                state["response"] = "Could not find a valid YouTube video URL in the question."
-                print("No valid YouTube URL found in question.")
-                return state
-
-            print(f"URL being passed to pytube: {url}")
-
-            temp_path = None
-
-            try:
-                #yt = YouTube(url)
-                yt = YouTube(url, use_oauth=False, allow_oauth_cache=True)
-                print(f"YouTube video title: {yt.title}")
-
-                # Select a stream. Prioritize progressive MP4.
-                stream = yt.streams.filter(file_extension="mp4", progressive=True).order_by('resolution').desc().first()
-
-                if not stream:
-                    print("No progressive MP4 stream found. Trying to find any MP4 stream.")
-                    # Fallback to any MP4 stream, but note that it might not be progressive (audio+video)
-                    stream = yt.streams.filter(file_extension="mp4").order_by('resolution').desc().first()
-
-                if not stream:
-                    state["response"] = "No suitable MP4 video stream found for download."
-                    print("No suitable MP4 video stream found after trying progressive and general MP4 filters.")
-                    return state
-
-                print(f"Selected stream: {stream.resolution}, {stream.mime_type}")
-
-                with tempfile.TemporaryDirectory() as tmpdirname:
-                    temp_path = os.path.join(tmpdirname, "video.mp4")
-                    print(f"Downloading video to {temp_path}...")
-                    stream.download(filename=temp_path)
-                    print("Video downloaded successfully.")
-
-                    cap = cv2.VideoCapture(temp_path)
-                    if not cap.isOpened():
-                        state["response"] = "Error: Could not open downloaded video file with OpenCV."
-                        print(f"Error opening video file: {temp_path}")
-                        return state
-
-                    frames = []
-                    frame_count = 0
-                    max_frames_to_process = 5 # Limit processing to save time and resources
-
-                    while True:
-                        ret, frame = cap.read()
-                        if not ret or frame_count >= max_frames_to_process:
-                            break
-                        frames.append(frame)
-                        frame_count += 1
-                    cap.release()
-                    print(f"Captured {len(frames)} frames for analysis.")
-
-                    if not frames:
-                        state["response"] = "No frames could be read from the video for analysis."
-                        print("OpenCV failed to read any frames.")
-                        return state
-
-                    object_counts = []
-                    for i, frame in enumerate(frames):
-                        try:
-                            results = self.model(frame, verbose=False)
-                            count = len(results[0].boxes)
-                            object_counts.append(count)
-                            print(f"Frame {i+1}/{len(frames)}: Detected {count} objects.")
-                        except Exception as yolo_e:
-                            print(f"Error during object detection on frame {i+1}: {yolo_e}")
-                            object_counts.append("Error")
-
-                    state["response"] = f"Objects detected per frame (first {len(object_counts)} frames): {object_counts}"
-
-            except pytube_exceptions.AgeRestrictedError:
-                state["response"] = "Error: Video is age-restricted and cannot be downloaded without further authentication."
-                print("Pytube Error: AgeRestrictedError")
-            except pytube_exceptions.LiveStreamError:
-                state["response"] = "Error: Cannot process live YouTube streams."
-                print("Pytube Error: LiveStreamError")
-            except pytube_exceptions.VideoUnavailable:
-                state["response"] = "Error: YouTube video is unavailable or private."
-                print("Pytube Error: VideoUnavailable")
-            except pytube_exceptions.PytubeError as pe:
-                state["response"] = f"Pytube specific error during video download: {pe}"
-                print(f"Pytube Error: {pe}")
-            except requests.exceptions.Timeout:
-                state["response"] = "Error processing video: Video download timed out."
-                print("Request Timeout during video download.")
-            except Exception as e:
-                import traceback
-                error_traceback = traceback.format_exc()
-                state["response"] = f"An unexpected error occurred during video analysis: {e}. Check logs for details."
-                print(f"General Error in use_video_analysis: {e}\n{error_traceback}")
-            finally:
-                if temp_path and os.path.exists(temp_path):
-                    try:
-                        os.remove(temp_path)
-                        print(f"Cleaned up temporary video file: {temp_path}")
-                    except OSError as e:
-                        print(f"Error during cleanup of temporary file {temp_path}: {e}")
-
-            return state
-
 
 
 
@@ -310,11 +139,8 @@ class SuperSmartAgent:
             response: str
         
         builder = StateGraph(AgentState)
-    
-        # Nodes
-        builder.add_node("check_special_tools", check_special_tools)
-        builder.add_node("use_web_search", use_web_search)
-        builder.add_node("use_video_analysis", use_video_analysis)
+
+        # --- Nodes ---
         builder.add_node("check_reversed", check_reversed)
         builder.add_node("fix_question", fix_question)
         builder.add_node("check_riddle_or_trick", check_riddle_or_trick)
@@ -322,30 +148,20 @@ class SuperSmartAgent:
         builder.add_node("check_python_suitability", check_python_suitability)
         builder.add_node("generate_code", generate_code)
         builder.add_node("fallback", fallback)
-        
-        # Entry point: special tool detection first
-        builder.set_entry_point("check_special_tools")
 
-        # Tool-based branching
-        builder.add_conditional_edges(
-            "check_special_tools",
-            lambda s: "use_web_search" if s.get("use_tool") == "web"
-            else "use_video_analysis" if s.get("use_tool") == "video"
-            else "check_reversed"
-        )
-        builder.add_edge("use_web_search", END)
-        builder.add_edge("use_video_analysis", END)
-        
-        # Reasoning pipeline (non-tool questions)
+        # --- Entry point ---
+        builder.set_entry_point("check_reversed")
+
+        # --- Reasoning pipeline ---
         builder.add_edge("check_reversed", "fix_question")
         builder.add_edge("fix_question", "check_riddle_or_trick")
-        
+
         builder.add_conditional_edges(
             "check_riddle_or_trick",
             lambda s: "solve_riddle" if s.get("is_riddle") else "check_python_suitability"
         )
         builder.add_edge("solve_riddle", END)
-        
+
         builder.add_conditional_edges(
             "check_python_suitability",
             lambda s: "generate_code" if s.get("is_python") else "fallback"
@@ -355,8 +171,6 @@ class SuperSmartAgent:
 
         
   
-
-            
         graph = builder.compile()
         return graph