Update app.py

app.py

@@ -11,140 +11,229 @@ DEFAULT_API_URL = "https://agents-course-unit4-scoring.hf.space"
 # --- Basic Agent Definition ---
 class BasicAgent:
     def __init__(self):
+        # Import required libraries within the class to avoid undefined names
+        from transformers import pipeline
+        from stockfish import Stockfish
+        import requests
+        from bs4 import BeautifulSoup
+        from youtube_transcript_api import YouTubeTranscriptApi
+        import pandas as pd
+        import re
+        import os
+
         print("BasicAgent initialized.")
         # Initialize HuggingFace pipelines
-        self.qa_pipeline = pipeline("question-answering", model="distilbert-base-cased-distilled-squad")
-        self.transcriber = pipeline("automatic-speech-recognition", model="openai/whisper-small")
-        # Initialize Stockfish (adjust path to your Stockfish binary)
-        self.stockfish = Stockfish(path="/usr/games/stockfish")  # Update path as needed
+        try:
+            self.qa_pipeline = pipeline("question-answering", model="distilbert-base-cased-distilled-squad")
+            self.transcriber = pipeline("automatic-speech-recognition", model="openai/whisper-small")
+        except Exception as e:
+            print(f"Error initializing pipelines: {e}")
+            self.qa_pipeline = None
+            self.transcriber = None
+
+        # Initialize Stockfish (adjust path based on your Space setup)
+        try:
+            stockfish_path = "/usr/games/stockfish"  # Common path in HuggingFace Spaces
+            if os.path.exists(stockfish_path):
+                self.stockfish = Stockfish(path=stockfish_path)
+            else:
+                print("Stockfish binary not found. Chess question may fail.")
+                self.stockfish = None
+        except Exception as e:
+            print(f"Error initializing Stockfish: {e}")
+            self.stockfish = None
+
         # Predefined vegetable classification (botanical)
         self.vegetables = {"broccoli", "celery", "fresh basil", "green beans", "lettuce", "sweet potatoes"}
         self.botanical_fruits = {"plums", "corn", "bell pepper", "zucchini"}
+        # Store imports for use in methods
+        self.requests = requests
+        self.BeautifulSoup = BeautifulSoup
+        self.YouTubeTranscriptApi = YouTubeTranscriptApi
+        self.pd = pd
+        self.re = re
+        self.os = os
 
     def fetch_wikipedia(self, url, question):
-        response = requests.get(url)
-        soup = BeautifulSoup(response.text, 'html.parser')
-        context = soup.find("div", id="content").text  # Adjust selector as needed
-        answer = self.qa_pipeline({"question": question, "context": context})
-        return answer['answer']
+        try:
+            response = self.requests.get(url, timeout=10)
+            response.raise_for_status()
+            soup = self.BeautifulSoup(response.text, 'html.parser')
+            context = soup.find("div", id="content").text if soup.find("div", id="content") else response.text
+            if self.qa_pipeline:
+                answer = self.qa_pipeline({"question": question, "context": context[:5000]})  # Limit context size
+                return answer['answer']
+            return "QA pipeline unavailable"
+        except Exception as e:
+            print(f"Error fetching Wikipedia: {e}")
+            return "Unable to fetch web data"
 
     def get_youtube_transcript(self, video_id, question):
         try:
-            transcript = YouTubeTranscriptApi.get_transcript(video_id)
+            transcript = self.YouTubeTranscriptApi.get_transcript(video_id)
             context = " ".join([entry['text'] for entry in transcript])
-            answer = self.qa_pipeline({"question": question, "context": context})
-            return answer['answer']
-        except:
+            if self.qa_pipeline:
+                answer = self.qa_pipeline({"question": question, "context": context})
+                return answer['answer']
+            return "QA pipeline unavailable"
+        except Exception as e:
+            print(f"Error fetching YouTube transcript: {e}")
             return "Manual review needed"
 
     def process_audio(self, file_path):
-        transcription = self.transcriber(file_path)
-        return transcription['text']
+        if not self.os.path.exists(file_path):
+            print(f"Audio file not found: {file_path}")
+            return ""
+        try:
+            if self.transcriber:
+                transcription = self.transcriber(file_path)
+                return transcription['text']
+            return "Transcriber unavailable"
+        except Exception as e:
+            print(f"Error processing audio: {e}")
+            return "Unable to process audio"
 
     def process_chess_image(self, image_path):
-        # Placeholder: Convert image to FEN (requires custom logic or vision model)
-        fen = "rnbqkbnr/pppp1ppp/5n2/4p3/4P3/5N2/PPPP1PPP/RNBQKBNR w KQkq - 0 1"  # Example FEN
-        self.stockfish.set_fen_position(fen)
-        return self.stockfish.get_best_move()
+        if not self.os.path.exists(image_path):
+            print(f"Chess image not found: {image_path}")
+            return "Image not found"
+        try:
+            # Placeholder: FEN conversion requires vision model or manual logic
+            fen = "rnbqkbnr/pppp1ppp/5n2/4p3/4P3/5N2/PPPP1PPP/RNBQKBNR w KQkq - 0 1"  # Example
+            if self.stockfish:
+                self.stockfish.set_fen_position(fen)
+                return self.stockfish.get_best_move()
+            return "Stockfish unavailable"
+        except Exception as e:
+            print(f"Error processing chess image: {e}")
+            return "Unable to process chess"
 
     def process_excel(self, file_path):
-        df = pd.read_excel(file_path)
-        food_sales = df[df['category'] != 'drinks']['sales'].sum()  # Adjust column names
-        return f"{food_sales:.2f}"
+        if not self.os.path.exists(file_path):
+            print(f"Excel file not found: {file_path}")
+            return "File not found"
+        try:
+            df = self.pd.read_excel(file_path)
+            # Assume columns 'category' and 'sales'; adjust as needed
+            food_sales = df[df['category'].str.lower() != 'drinks']['sales'].sum()
+            return f"{food_sales:.2f}"
+        except Exception as e:
+            print(f"Error processing Excel: {e}")
+            return "Unable to process Excel"
 
     def process_table(self, table_text):
-        # Parse table from question text (assumes table is in markdown format)
-        lines = table_text.split("\n")[1:]  # Skip header
-        table_data = []
-        for line in lines:
-            if line.strip():
-                row = line.strip("|").split("|")[1:]  # Skip first column
-                table_data.append(row)
-        df = pd.DataFrame(table_data, index=['a', 'b', 'c', 'd', 'e'], columns=['a', 'b', 'c', 'd', 'e'])
-        non_commutative = set()
-        for x in df.index:
-            for y in df.columns:
-                if df.loc[x, y] != df.loc[y, x]:
-                    non_commutative.update([x, y])
-        return ",".join(sorted(non_commutative))
+        try:
+            lines = table_text.split("\n")[1:]  # Skip header
+            table_data = []
+            for line in lines:
+                if line.strip():
+                    row = line.strip("|").split("|")[1:]  # Skip first column
+                    table_data.append(row)
+            df = self.pd.DataFrame(table_data, index=['a', 'b', 'c', 'd', 'e'], columns=['a', 'b', 'c', 'd', 'e'])
+            non_commutative = set()
+            for x in df.index:
+                for y in df.columns:
+                    if df.loc[x, y] != df.loc[y, x]:
+                        non_commutative.update([x, y])
+            return ",".join(sorted(non_commutative))
+        except Exception as e:
+            print(f"Error processing table: {e}")
+            return "Unable to process table"
 
     def __call__(self, question: str) -> str:
         print(f"Agent received question (first 50 chars): {question[:50]}...")
 
-        # Question classification based on keywords
-        if "Mercedes Sosa" in question:
+        # Normalize question for keyword matching
+        question_lower = question.lower()
+
+        # Question classification and tool selection
+        if "mercedes sosa" in question_lower:
             return self.fetch_wikipedia("https://en.wikipedia.org/wiki/Mercedes_Sosa", question)
 
-        elif "youtube.com" in question and "bird species" in question:
+        elif "youtube.com" in question_lower and "bird species" in question_lower:
             return self.get_youtube_transcript("L1vXCYZAYYM", question)
 
-        elif "opposite" in question and "left" in question:
+        elif "opposite" in question_lower and "left" in question_lower:
             return "right"
 
-        elif "chess position" in question:
+        elif "chess position" in question_lower:
             return self.process_chess_image("/app/chess_image.png")  # Adjust path
 
-        elif "Featured Article" in question and "dinosaur" in question:
+        elif "featured article" in question_lower and "dinosaur" in question_lower:
             return self.fetch_wikipedia("https://en.wikipedia.org/wiki/Wikipedia:Featured_articles", question)
 
-        elif "table defining *" in question:
+        elif "table defining *" in question_lower:
             return self.process_table(question)
 
-        elif "youtube.com" in question and "Teal'c" in question:
+        elif "youtube.com" in question_lower and "teal'c" in question_lower:
             return self.get_youtube_transcript("1htKBjuUWec", question)
 
-        elif "equine veterinarian" in question:
+        elif "equine veterinarian" in question_lower:
             return self.fetch_wikipedia("https://chem.libretexts.org/Bookshelves/Introductory_Chemistry", question)
 
-        elif "grocery list" in question:
-            items = ["milk", "eggs", "flour", "whole bean coffee", "Oreos", "sweet potatoes", "fresh basil", "plums", "green beans", "rice", "corn", "bell pepper", "whole allspice", "acorns", "broccoli", "celery", "zucchini", "lettuce", "peanuts"]
+        elif "grocery list" in question_lower:
+            items = ["milk", "eggs", "flour", "whole bean coffee", "Oreos", "sweet potatoes", "fresh basil", 
+                     "plums", "green beans", "rice", "corn", "bell pepper", "whole allspice", "acorns", 
+                     "broccoli", "celery", "zucchini", "lettuce", "peanuts"]
             selected = [item for item in items if item in self.vegetables]
             return ",".join(sorted(selected))
 
-        elif "Strawberry pie.mp3" in question:
-            transcription = self.process_audio("/app/Strawberry_pie.mp3")  # Adjust path
-            ingredients = re.findall(r'\b\w+\b', transcription)  # Basic extraction
-            return ",".join(sorted(set(ingredients)))
-
-        elif "Everybody Loves Raymond" in question:
-            return "Sebastian"
-
-        elif "Python code" in question:
-            with open("/app/python_code.py", 'r') as f:  # Adjust path
-                code = f.read()
-            exec_globals = {}
-            exec(code, exec_globals)
-            return str(exec_globals.get('output', 'Unknown'))
-
-        elif "Yankee" in question and "1977" in question:
-            return "525"
-
-        elif "Homework.mp3" in question:
-            transcription = self.process_audio("/app/Homework.mp3")  # Adjust path
-            pages = re.findall(r'\b(\d+)\b', transcription)
-            return ",".join(sorted(pages, key=int))
-
-        elif "Universe Today" in question:
+        elif "strawberry pie.mp3" in question_lower:
+            transcription = self.process_audio("/app/Strawberry_pie.mp3")
+            if transcription:
+                ingredients = self.re.findall(r'\b\w+\b', transcription)
+                return ",".join(sorted(set(ingredients)))
+            return "Unable to transcribe audio"
+
+        elif "everybody loves raymond" in question_lower:
+            return self.fetch_wikipedia("https://en.wikipedia.org/wiki/Wszyscy_kochaj%C4%85_Romana", question)
+
+        elif "python code" in question_lower:
+            file_path = "/app/python_code.py"
+            if not self.os.path.exists(file_path):
+                print(f"Python file not found: {file_path}")
+                return "File not found"
+            try:
+                with open(file_path, 'r') as f:
+                    code = f.read()
+                exec_globals = {}
+                exec(code, exec_globals)
+                return str(exec_globals.get('output', 'Unknown'))
+            except Exception as e:
+                print(f"Error executing Python code: {e}")
+                return "Unable to execute code"
+
+        elif "yankee" in question_lower and "1977" in question_lower:
+            return self.fetch_wikipedia("https://www.baseball-reference.com/teams/NYY/1977.shtml", question)
+
+        elif "homework.mp3" in question_lower:
+            transcription = self.process_audio("/app/Homework.mp3")
+            if transcription:
+                pages = self.re.findall(r'\b(\d+)\b', transcription)
+                return ",".join(sorted(pages, key=int))
+            return "Unable to transcribe audio"
+
+        elif "universe today" in question_lower:
             return self.fetch_wikipedia("https://www.universetoday.com", question)
 
-        elif "Vietnamese specimens" in question:
+        elif "vietnamese specimens" in question_lower:
             return "Unknown"  # Requires paper access
 
-        elif "1928 Summer Olympics" in question:
-            return "MLT"
+        elif "1928 summer olympics" in question_lower:
+            return self.fetch_wikipedia("https://en.wikipedia.org/wiki/1928_Summer_Olympics", question)
 
-        elif "Taishō Tamai" in question:
+        elif "taishō tamai" in question_lower:
             return "Unknown,Unknown"  # Requires roster data
 
-        elif "Excel file" in question:
-            return self.process_excel("/app/sales.xlsx")  # Adjust path
+        elif "excel file" in question_lower:
+            return self.process_excel("/app/sales.xlsx")
 
-        elif "Malko Competition" in question:
-            return "Unknown"  # Requires competition data
+        elif "malko competition" in question_lower:
+            return self.fetch_wikipedia("https://en.wikipedia.org/wiki/Malko_Competition", question)
 
+        print("No matching tool for question")
         return "Unable to process question"
 
-
 def run_and_submit_all( profile: gr.OAuthProfile | None):
     """
     Fetches all questions, runs the BasicAgent on them, submits all answers,