app.py

import os
import gradio as gr
import requests
import inspect
import pandas as pd
from smolagents import (
    CodeAgent, 
    LiteLLMModel,
    DuckDuckGoSearchTool, 
    LogLevel, 
    load_tool,
    PythonInterpreterTool
)
from dotenv import load_dotenv
from smolagents import Tool
import base64
import anthropic
from PIL import Image
import io

class SimpleExcelTool(Tool):
    name = "SimpleExcelTool"
    description = "Load a downloaded Excel file associated with a task ID and perform basic operations like reading data"
    inputs = {
        "task_id": {
            "type": "string",
            "description": "Task ID for which the Excel file has been downloaded"
        },
        "operation": {
            "type": "string",
            "description": "Operation to perform on the Excel file (currently only 'read' is supported)",
            "nullable": True
        }
    }
    output_type = "string"

    def forward(self, task_id: str, operation: str = "read") -> str:
        try:
            filename = f"{task_id}_downloaded_file"
            df = pd.read_excel(filename, engine="openpyxl")
            
            if operation == "read":
                return df.head().to_string()
            else:
                return f"Unsupported operation: {operation}"
        except Exception as e:
            return f"Error reading Excel file: {str(e)}"

class ImageAnalysisTool(Tool):
    name = "ImageAnalysisTool"
    description = "Analyze a downloaded image file associated with a task ID using Claude Vision. Provide a detailed description of what's in the image."
    inputs = {
        "task_id": {
            "type": "string",
            "description": "Task ID for which the image file has been downloaded"
        },
        "prompt": {
            "type": "string",
            "description": "Optional specific question or aspect to analyze about the image",
            "nullable": True
        }
    }
    output_type = "string"

    def __init__(self):
        super().__init__()
        self.client = anthropic.Client(api_key="")

    def forward(self, task_id: str, prompt: str = "Describe what you see in this image in detail.") -> str:
        try:
            filename = f"{task_id}_downloaded_file"

            with open(filename, 'rb') as img_file:
                img_bytes = img_file.read()

            img = Image.open(io.BytesIO(img_bytes))
            if img.mode != 'RGB':
                img = img.convert('RGB')

            img_byte_arr = io.BytesIO()
            img.save(img_byte_arr, format='JPEG')
            img_byte_arr = img_byte_arr.getvalue()

            base64_image = base64.b64encode(img_byte_arr).decode('utf-8')

            message = self.client.messages.create(
                model="claude-3-7-sonnet-20250219",
                max_tokens=1000,
                messages=[{
                    "role": "user",
                    "content": [
                        {
                            "type": "image",
                            "source": {
                                "type": "base64",
                                "media_type": "image/jpeg",
                                "data": base64_image
                            }
                        },
                        {
                            "type": "text",
                            "text": prompt
                        }
                    ]
                }]
            )

            return message.content[0].text

        except Exception as e:
            return f"Error analyzing image: {str(e)}"


# New: TaskFileDownloaderTool
class TaskFileDownloaderTool(Tool):
    name = "TaskFileDownloaderTool"
    description = "Download a specific file associated with a given task ID and save it locally"
    inputs = {
        "task_id": {
            "type": "string",
            "description": "Task ID for which to download the associated file"
        }
    }
    output_type = "string"

    def forward(self, task_id: str) -> str:
        try:
            download_url = f"{DEFAULT_API_URL}/files/{task_id}"
            response = requests.get(download_url)
            response.raise_for_status()

            filename = f"{task_id}_downloaded_file"
            with open(filename, "wb") as f:
                f.write(response.content)
            
            return f"File downloaded successfully and saved as: {filename}"
        except Exception as e:
            return f"Error downloading file: {str(e)}"

# New: FileOpenerTool
class FileOpenerTool(Tool):
    name = "FileOpenerTool"
    description = "Open a downloaded file associated with a task ID and read its contents as plain text."
    inputs = {
        "task_id": {
            "type": "string",
            "description": "Task ID for which the file has been downloaded"
        },
        "num_lines": {
            "type": "integer",
            "description": "Number of lines to read from the file",
            "nullable": True
        }
    }
    output_type = "string"

    def forward(self, task_id: str, num_lines: int = 10) -> str:
        try:
            filename = f"{task_id}_downloaded_file"
            if not os.path.exists(filename):
                return f"Error: File {filename} does not exist."

            with open(filename, "r", encoding="utf-8", errors="ignore") as file:
                lines = []
                for _ in range(num_lines):
                    line = file.readline()
                    if not line:
                        break
                    lines.append(line.strip())

            return "\n".join(lines)
        except Exception as e:
            return f"Error reading file: {str(e)}"

# New: SpeechToTextTool
import mlx_whisper

class SpeechToTextTool(Tool):
    name = "SpeechToTextTool"
    description = "Transcribe a downloaded MP3 audio file associated with a task ID into text."
    inputs = {
        "task_id": {
            "type": "string",
            "description": "Task ID for which the MP3 audio file has been downloaded"
        }
    }
    output_type = "string"

    def __init__(self):
        super().__init__()

    def forward(self, task_id: str) -> str:
        try:
            filename = f"{task_id}_downloaded_file"
            if not os.path.exists(filename):
                return f"Error: Audio file {filename} does not exist."

            result = mlx_whisper.transcribe(filename)
            return result["text"]
        except Exception as e:
            return f"Error transcribing audio file: {str(e)}"



import wikipedia

class WikipediaSearchTool(Tool):
    name = "WikipediaSearchTool"
    description = "Search Wikipedia for a query and return a brief summary."
    inputs = {
        "query": {
            "type": "string",
            "description": "Query to search on Wikipedia"
        }
    }
    output_type = "string"

    def __init__(self):
        super().__init__()
        wikipedia.set_lang("en")  # Ensure English Wikipedia

    def forward(self, query: str) -> str:
        try:
            summary = wikipedia.summary(query, sentences=3000)
            return summary
        except wikipedia.exceptions.DisambiguationError as e:
            return f"Disambiguation error. Possible options: {e.options[:5]}"
        except wikipedia.exceptions.PageError:
            return f"Page not found for query: {query}"
        except Exception as e:
            return f"Error searching Wikipedia: {str(e)}"

def format_transcript(transcript_data):
    return "\n".join([f"{item['start']}: {item['text']}" for item in transcript_data])

import os
from youtube_transcript_api import YouTubeTranscriptApi
import yt_dlp
import mlx_whisper

class YouTubeTranscriptTool(Tool):
    name = "YouTubeTranscriptTool"
    description = "Fetches or transcribes the text from a YouTube video ID."
    inputs = {
        "video_id": {
            "type": "string",
            "description": "YouTube Video ID (the part after 'watch?v=')"
        }
    }
    output_type = "string"

    def __init__(self):
        super().__init__()

    def forward(self, video_id: str) -> str:
        try:
            transcript_list = YouTubeTranscriptApi.list_transcripts(video_id)

            try:
                # First try manually created transcript
                transcript = transcript_list.find_manually_created_transcript(['en'])
            except Exception:
                # If not found, try auto-generated transcript
                transcript = transcript_list.find_generated_transcript(['en'])

            transcript_data = transcript.fetch()

            # Format nicely
            text = format_transcript(transcript_data)

            return text

        except Exception as e:
            print(f"No direct transcript found: {e}")
            print("Trying to download and transcribe audio with Whisper...")

            # Step 1: Download audio using yt_dlp
            audio_filename = f"{video_id}.mp3"
            try:
                ydl_opts = {
                    'format': 'bestaudio/best',
                    'outtmpl': audio_filename,
                    'postprocessors': [{
                        'key': 'FFmpegExtractAudio',
                        'preferredcodec': 'mp3',
                        'preferredquality': '192',
                    }],
                    'quiet': True,
                }
                with yt_dlp.YoutubeDL(ydl_opts) as ydl:
                    ydl.download([f"https://www.youtube.com/watch?v={video_id}"])

                # Step 2: Transcribe audio using mlx_whisper
                result = mlx_whisper.transcribe(audio_filename)
                return result["text"]

            except Exception as download_error:
                return f"Error downloading or transcribing YouTube audio: {str(download_error)}"
            finally:
                if os.path.exists(audio_filename):
                    os.remove(audio_filename)  # Clean up downloaded file
# Load environment variables
load_dotenv()

# (Keep Constants as is)
# --- Constants ---
DEFAULT_API_URL = "https://agents-course-unit4-scoring.hf.space"

# --- Basic Agent Definition ---
class BasicAgent:
    def __init__(self):
        print("Initializing Agent with tools...")

        # Initialize the model using Claude via LiteLLM
        self.model = LiteLLMModel(
            model_id="ollama_chat/qwen2:7b",
            api_base="http://127.0.0.1:11434",
            temperature=0.7,
            max_tokens=4096
        )

        # Initialize tools
        youtube_transcript_tool = YouTubeTranscriptTool()
        excel_tool = SimpleExcelTool()
        image_analysis_tool = ImageAnalysisTool()
        file_opener_tool = FileOpenerTool()
        speech_to_text_tool = SpeechToTextTool()
        task_file_downloader_tool = TaskFileDownloaderTool()
        wikipedia_search_tool = WikipediaSearchTool()

        self.tools = [
            DuckDuckGoSearchTool(),
            wikipedia_search_tool,
            youtube_transcript_tool,
            PythonInterpreterTool(),
            excel_tool,
            image_analysis_tool,
            file_opener_tool,
            speech_to_text_tool,
            task_file_downloader_tool
        ]

        # Initialize the agent
        self.agent = CodeAgent(
            tools=self.tools,
            model=self.model,
            verbosity_level=LogLevel.INFO
        )

        print("Agent initialized successfully")

    def __call__(self, question: str, task_id: str) -> str:
        print(f"Agent received question: {question[:100]}...")
        try:
            # Step 1: Download the file associated with the task first
            download_result = self.tools[-1](task_id=task_id)  # TaskFileDownloaderTool is the last in self.tools
            print(download_result)

            # Step 2: Create a comprehensive prompt for the agent
            prompt = f"""Please answer the following question. Use the available tools (web search) 
            to gather relevant information before providing a comprehensive answer.
            
            Question: {question}
            Task_id: {task_id}
            
            Instructions:
            1. Search for relevant information using web search.
            2. Look for relevant YouTube content if applicable.
            3. If the task requires working with an Excel or image file:
            - First, download the file associated with the task ID using the file download tool.
            - Then, perform analysis on the downloaded file.
            4. Extract and analyze data from Excel files after downloading.
            5. Convert images to text after downloading the image file.
            6. Convert attached mp3 to text as seepch to text
            7. Make Wikipedia search on facts and for a query and return a brief summary
            78. Synthesize all gathered and analyzed information into a clear, well-structured final answer.

            Answer:"""

            # Step 3: Get response from the agent
            response = self.agent.run(prompt)
            print(f"Agent generated response: {response[:100]}...")
            return response

        except Exception as e:
            error_msg = f"Error generating answer: {str(e)}"
            print(error_msg)
            return error_msg

def run_and_submit_all(profile: gr.OAuthProfile | None):
    """
    Fetches all questions, runs the BasicAgent on them, submits all answers,
    and displays the results.
    """
    # --- Determine HF Space Runtime URL and Repo URL ---
    space_id = os.getenv("SPACE_ID") # Get the SPACE_ID for sending link to the code

    if profile:
        username = f"{profile.username}"
        print(f"User logged in: {username}")
    else:
        print("User not logged in.")
        return "Please Login to Hugging Face with the button.", None

    api_url = DEFAULT_API_URL
    questions_url = f"{api_url}/questions"
    submit_url = f"{api_url}/submit"

    # 1. Instantiate Agent
    try:
        agent = BasicAgent()
    except Exception as e:
        print(f"Error instantiating agent: {e}")
        return f"Error initializing agent: {e}", None
        
    agent_code = f"https://huggingface.co/spaces/{space_id}/tree/main"
    print(agent_code)

    # 2. Fetch Questions
    print(f"Fetching questions from: {questions_url}")
    try:
        response = requests.get(questions_url, timeout=15)
        response.raise_for_status()
        questions_data = response.json()

        print(questions_data)
        
        if not questions_data:
             print("Fetched questions list is empty.")
             return "Fetched questions list is empty or invalid format.", None
        print(f"Fetched {len(questions_data)} questions.")
    except requests.exceptions.RequestException as e:
        print(f"Error fetching questions: {e}")
        return f"Error fetching questions: {e}", None
    except requests.exceptions.JSONDecodeError as e:
         print(f"Error decoding JSON response from questions endpoint: {e}")
         print(f"Response text: {response.text[:500]}")
         return f"Error decoding server response for questions: {e}", None
    except Exception as e:
        print(f"An unexpected error occurred fetching questions: {e}")
        return f"An unexpected error occurred fetching questions: {e}", None

    # 3. Run Agent
    results_log = []
    answers_payload = []
    print(f"Running agent on {len(questions_data)} questions...")
    
    for item in questions_data:
        task_id = item.get("task_id")
        question_text = item.get("question")
        if not task_id or question_text is None:
            print(f"Skipping item with missing task_id or question: {item}")
            continue

        try:
            submitted_answer = agent(question_text, task_id)
            answers_payload.append({"task_id": task_id, "submitted_answer": submitted_answer})
            results_log.append({
                "Task ID": task_id,
                "Question": question_text,
                "Submitted Answer": submitted_answer
            })
        except Exception as e:
            print(f"Error running agent on task {task_id}: {e}")
            results_log.append({
                "Task ID": task_id,
                "Question": question_text,
                "Submitted Answer": f"AGENT ERROR: {e}"
            })

    if not answers_payload:
        print("Agent did not produce any answers to submit.")
        return "Agent did not produce any answers to submit.", pd.DataFrame(results_log)

    # 4. Prepare Submission
    submission_data = {
        "username": username.strip(),
        "agent_code": agent_code,
        "answers": answers_payload
    }
    status_update = f"Agent finished. Submitting {len(answers_payload)} answers for user '{username}'..."
    print(status_update)

    # 5. Submit
    print(f"Submitting {len(answers_payload)} answers to: {submit_url}")
    try:
        response = requests.post(submit_url, json=submission_data, timeout=60)
        response.raise_for_status()
        result_data = response.json()
        final_status = (
            f"Submission Successful!\n"
            f"User: {result_data.get('username')}\n"
            f"Overall Score: {result_data.get('score', 'N/A')}% "
            f"({result_data.get('correct_count', '?')}/{result_data.get('total_attempted', '?')} correct)\n"
            f"Message: {result_data.get('message', 'No message received.')}"
        )
        print("Submission successful.")
        results_df = pd.DataFrame(results_log)
        return final_status, results_df
    except requests.exceptions.HTTPError as e:
        error_detail = f"Server responded with status {e.response.status_code}."
        try:
            error_json = e.response.json()
            error_detail += f" Detail: {error_json.get('detail', e.response.text)}"
        except requests.exceptions.JSONDecodeError:
            error_detail += f" Response: {e.response.text[:500]}"
        status_message = f"Submission Failed: {error_detail}"
        print(status_message)
        results_df = pd.DataFrame(results_log)
        return status_message, results_df
    except requests.exceptions.Timeout:
        status_message = "Submission Failed: The request timed out."
        print(status_message)
        results_df = pd.DataFrame(results_log)
        return status_message, results_df
    except requests.exceptions.RequestException as e:
        status_message = f"Submission Failed: Network error - {e}"
        print(status_message)
        results_df = pd.DataFrame(results_log)
        return status_message, results_df
    except Exception as e:
        status_message = f"An unexpected error occurred during submission: {e}"
        print(status_message)
        results_df = pd.DataFrame(results_log)
        return status_message, results_df

# --- Build Gradio Interface using Blocks ---
with gr.Blocks() as demo:
    gr.Markdown("# Advanced Agent Evaluation Runner")
    gr.Markdown(
        """
        **Instructions:**
        1. Make sure you have set up your environment variables:
           - HF_TOKEN: Your Hugging Face API token
           - YOUTUBE_API_KEY: Your YouTube API key (optional)
        2. Log in to your Hugging Face account using the button below
        3. Click 'Run Evaluation & Submit All Answers' to process all questions
        
        The agent will use:
        - Web search (DuckDuckGo)
        - YouTube search (if API key provided)
        - Mistral-7B-Instruct LLM
        """
    )

    gr.LoginButton()
    run_button = gr.Button("Run Evaluation & Submit All Answers")
    status_output = gr.Textbox(label="Run Status / Submission Result", lines=5, interactive=False)
    results_table = gr.DataFrame(label="Questions and Agent Answers", wrap=True)

    run_button.click(
        fn=run_and_submit_all,
        outputs=[status_output, results_table]
    )

if __name__ == "__main__":
    print("\n" + "-"*30 + " App Starting " + "-"*30)
    
    # Check for required environment variables
    hf_token = os.getenv("HF_TOKEN")
    if not hf_token:
        print("⚠️ Warning: HF_TOKEN not found in environment variables")
    
    youtube_api_key = os.getenv("YOUTUBE_API_KEY")
    if not youtube_api_key:
        print("ℹ️ Note: YOUTUBE_API_KEY not found. YouTube search will be disabled")
    
    # Check for SPACE_HOST and SPACE_ID
    space_host_startup = os.getenv("SPACE_HOST")
    space_id_startup = os.getenv("SPACE_ID")

    if space_host_startup:
        print(f"✅ SPACE_HOST found: {space_host_startup}")
        print(f"   Runtime URL should be: https://{space_host_startup}.hf.space")
    else:
        print("ℹ️ SPACE_HOST environment variable not found (running locally?).")

    if space_id_startup:
        print(f"✅ SPACE_ID found: {space_id_startup}")
        print(f"   Repo URL: https://huggingface.co/spaces/{space_id_startup}")
        print(f"   Repo Tree URL: https://huggingface.co/spaces/{space_id_startup}/tree/main")
    else:
        print("ℹ️ SPACE_ID environment variable not found (running locally?)")

    print("-"*(60 + len(" App Starting ")) + "\n")
    print("Launching Gradio Interface for Advanced Agent Evaluation...")
    demo.launch(debug=True, share=False)