app.py

"""
app.py
This script provides the Gradio web interface to run the evaluation.
This version focuses on robust image detection and processing.
"""

import os
import re
import gradio as gr
import requests
import pandas as pd
from urllib.parse import urlparse
import mimetypes
from typing import Optional, Tuple

from agent import create_agent_executor

# --- Constants ---
DEFAULT_API_URL = "https://agents-course-unit4-scoring.hf.space"

# --- Helper function to parse the agent's output ---
def parse_final_answer(agent_response: str) -> str:
    # Remove the FINAL ANSWER pattern search entirely
    lines = [line for line in agent_response.split('\n') if line.strip()]
    if lines: return lines[-1].strip()
    return "Could not parse answer."

def detect_file_type_robust(url: str) -> Tuple[str, dict]:
    """
    Robust file type detection with multiple validation methods.
    Returns (file_type, metadata_dict)
    """
    if not url or not url.strip():
        return "unknown", {"error": "Empty URL"}
    
    url = url.strip()
    metadata = {"original_url": url}
    
    # Normalize URL
    if not url.startswith(('http://', 'https://')):
        return "unknown", {"error": "Invalid URL format - must start with http/https"}
    
    try:
        parsed = urlparse(url)
        metadata["domain"] = parsed.netloc
        metadata["path"] = parsed.path
    except Exception as e:
        return "unknown", {"error": f"URL parsing failed: {e}"}
    
    # Method 1: File extension analysis
    url_lower = url.lower()
    image_extensions = {'.jpg', '.jpeg', '.png', '.gif', '.bmp', '.webp', '.svg', '.tiff', '.ico'}
    
    # Check for image extensions
    for ext in image_extensions:
        if url_lower.endswith(ext) or ext in url_lower.split('?')[0]:  # Handle query params
            metadata["detection_method"] = "file_extension"
            metadata["extension"] = ext
            return "image", metadata
    
    # Method 2: Content-Type header check
    try:
        print(f"Checking content type for: {url}")
        response = requests.head(url, timeout=10, allow_redirects=True)
        content_type = response.headers.get('content-type', '').lower()
        metadata["content_type"] = content_type
        metadata["status_code"] = response.status_code
        
        if response.status_code == 200:
            if any(img_type in content_type for img_type in ['image/', 'image/jpeg', 'image/png', 'image/gif', 'image/webp']):
                metadata["detection_method"] = "content_type"
                return "image", metadata
        else:
            metadata["error"] = f"HTTP {response.status_code}"
            
    except requests.RequestException as e:
        metadata["error"] = f"Network error: {e}"
        print(f"Network error checking {url}: {e}")
    
    # Method 3: Domain-based detection for common image hosts
    image_domains = {
        'imgur.com', 'i.imgur.com',
        'cdn.discordapp.com', 'media.discordapp.net',
        'pbs.twimg.com', 'abs.twimg.com',
        'i.redd.it', 'preview.redd.it',
        'images.unsplash.com',
        'via.placeholder.com',
        'picsum.photos'
    }
    
    domain_lower = metadata.get("domain", "").lower()
    if any(img_domain in domain_lower for img_domain in image_domains):
        metadata["detection_method"] = "domain_based"
        return "image", metadata
    
    # Method 4: Guess from MIME types
    try:
        mime_type, _ = mimetypes.guess_type(url)
        if mime_type and mime_type.startswith('image/'):
            metadata["detection_method"] = "mime_guess"
            metadata["mime_type"] = mime_type
            return "image", metadata
    except Exception:
        pass
    
    return "unknown", metadata

def create_structured_prompt(question_text: str, file_url: str = None) -> str:
    """
    Create a structured prompt that provides clear task analysis for the agent.
    """
    if not file_url:
        return f"""TASK: {question_text}

ANALYSIS: This is a text-only question with no attachments.
APPROACH: Use available tools (web search, Wikipedia, etc.) as needed to answer accurately."""

    file_type, metadata = detect_file_type_robust(file_url)
    
    if file_type == "image":
        return f"""TASK: {question_text}

ATTACHMENT ANALYSIS:
- Type: Image file detected
- URL: {file_url}
- Detection method: {metadata.get('detection_method', 'unknown')}
- Metadata: {metadata}

REASONING REQUIRED:
1. This question involves an image that needs to be analyzed
2. You must examine the image content to answer the question
3. The image URL should be processed directly by your vision capabilities

APPROACH: Process the image URL directly with your vision model, then provide a comprehensive answer based on what you see."""

    else:
        error_info = metadata.get('error', 'Unknown file type')
        return f"""TASK: {question_text}

ATTACHMENT ANALYSIS:
- URL: {file_url}
- Type: Could not identify as supported file type
- Error: {error_info}
- Metadata: {metadata}

REASONING REQUIRED:
1. There is an attachment but it's not a recognized image format
2. You should attempt to process it as a regular web resource
3. Use web search or other tools to gather information about the URL content

APPROACH: Use web search or other available tools to gather information about this resource."""

def run_and_submit_all(profile: gr.OAuthProfile | None):
    """
    Fetches all questions, runs the agent on them, submits all answers,
    and displays the results.
    """
    if not profile:
        return "Please log in to Hugging Face with the button above to submit.", None
    
    username = profile.username
    print(f"User logged in: {username}")
    
    space_id = os.getenv("SPACE_ID")
    agent_code = f"https://huggingface.co/spaces/{space_id}/tree/main"
    questions_url = f"{DEFAULT_API_URL}/questions"
    submit_url = f"{DEFAULT_API_URL}/submit"
    
    # 1. Instantiate Agent
    print("Initializing your custom agent...")
    try:
        agent_executor = create_agent_executor(provider="groq")
    except Exception as e:
        return f"Fatal Error: Could not initialize agent. Check logs. Details: {e}", None

    # 2. Fetch Questions
    print(f"Fetching questions from: {questions_url}")
    try:
        response = requests.get(questions_url, timeout=20)
        response.raise_for_status()
        questions_data = response.json()
        print(f"Fetched {len(questions_data)} questions.")
    
    except Exception as e:
        return f"Error fetching questions: {e}", pd.DataFrame()
        
    # DEBUG: Print format of each question
    print("\n=== QUESTION FORMATS DEBUG ===")
    for i, item in enumerate(questions_data):
        print(f"Question {i+1} keys: {list(item.keys())}")
        print(f"Question {i+1} full data: {item}")
        print("-" * 50)
    print("=== END DEBUG ===\n")
    

    # 3. Run your Agent
    results_log, answers_payload = [], []
    print(f"Running agent on {len(questions_data)} questions...")
    
    for i, item in enumerate(questions_data):
        task_id = item.get("task_id")
        question_text = item.get("question")
        if not task_id or question_text is None: 
            continue
        
        print(f"\n--- Running Task {i+1}/{len(questions_data)} (ID: {task_id}) ---")
        
        # Get file URL if it exists
        file_url = f"{DEFAULT_API_URL}/files/{task_id}" if item.get("has_file") else None

        if file_url is None:
            print ("No File Url")
        # Create structured prompt with robust file analysis
        structured_prompt = create_structured_prompt(question_text, file_url)
        
        if file_url:
            file_type, metadata = detect_file_type_robust(file_url)
            print(f"File analysis: {file_url}")
            print(f"  - Type: {file_type}")
            print(f"  - Detection method: {metadata.get('detection_method', 'unknown')}")
            if metadata.get('error'):
                print(f"  - Error: {metadata['error']}")
        
        print(f"Structured Prompt for Agent:\n{structured_prompt}")

        try:
            # Pass the structured prompt to the agent
            result = agent_executor.invoke({"messages": [("user", structured_prompt)]})
            
            raw_answer = result['messages'][-1].content
            submitted_answer = parse_final_answer(raw_answer)
            
            print(f"Raw LLM Response: '{raw_answer}'")
            print(f"PARSED FINAL ANSWER: '{submitted_answer}'")

            answers_payload.append({"task_id": task_id, "submitted_answer": submitted_answer})
            results_log.append({
                "Task ID": task_id, 
                "Question": question_text, 
                "File URL": file_url or "None",
                "File Type": detect_file_type_robust(file_url)[0] if file_url else "None",
                "Detection Method": detect_file_type_robust(file_url)[1].get('detection_method', 'N/A') if file_url else "N/A",
                "Submitted Answer": submitted_answer
            })
            
        except Exception as e:
            print(f"!! AGENT ERROR on task {task_id}: {e}")
            error_msg = f"AGENT RUNTIME ERROR: {e}"
            answers_payload.append({"task_id": task_id, "submitted_answer": error_msg})
            results_log.append({
                "Task ID": task_id, 
                "Question": question_text, 
                "File URL": file_url or "None",
                "File Type": detect_file_type_robust(file_url)[0] if file_url else "None",
                "Detection Method": "Error",
                "Submitted Answer": error_msg
            })

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

    # 4. Prepare and 5. Submit
    submission_data = {"username": username, "agent_code": agent_code, "answers": answers_payload}
    print(f"\nSubmitting {len(answers_payload)} answers for user '{username}'...")
    try:
        response = requests.post(submit_url, json=submission_data, timeout=60)
        response.raise_for_status()
        result_data = response.json()
        final_status = (f"Submission Successful!\nUser: {result_data.get('username')}\n"
                       f"Overall Score: {result_data.get('score', 'N/A')}%\n"
                       f"Processed {len([r for r in results_log if 'ERROR' not in r['Submitted Answer']])} successful tasks")
        return final_status, pd.DataFrame(results_log)
    except Exception as e:
        status_message = f"Submission Failed: {e}"
        print(status_message)
        return status_message, pd.DataFrame(results_log)

# --- Gradio UI ---
with gr.Blocks(title="Image-Capable Agent Evaluation") as demo:
    gr.Markdown("# Image-Capable Agent Evaluation Runner")
    gr.Markdown("This agent can process images and perform web searches using Groq's vision-capable models.")
    
    gr.LoginButton()
    run_button = gr.Button("Run Evaluation & Submit All Answers", variant="primary")
    status_output = gr.Textbox(label="Run Status / Submission Result", lines=6, interactive=False)
    results_table = gr.DataFrame(
        label="Questions and Agent Answers", 
        wrap=True, 
        row_count=10,
        column_widths=[80, 200, 120, 100, 80, 200]
    )
    
    run_button.click(fn=run_and_submit_all, outputs=[status_output, results_table])

if __name__ == "__main__":
    print("\n" + "-"*30 + " Image Agent App Starting " + "-"*30)
    demo.launch()