app.py

import os
import tempfile
from collections.abc import Callable
from dataclasses import asdict
from typing import Any

import gradio as gr
import pandas as pd
import requests

from agent import (
    acquire_attachment,
    build_default_services,
    evaluate_items,
    retry_call,
    solve_task,
)

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


def space_code_url(space_id: str | None) -> tuple[str, str]:
    """Return a verifiable Space code URL and any local configuration warning."""
    if not space_id:
        return "", "SPACE_ID is not configured; agent_code will be empty for this local run."
    return f"https://huggingface.co/spaces/{space_id}/tree/main", ""


def _result_frame(results: list[Any]) -> pd.DataFrame:
    columns = [
        "task_id",
        "question",
        "submitted_answer",
        "route",
        "status",
        "diagnostics",
    ]
    return pd.DataFrame([asdict(result) for result in results], columns=columns)


def run_and_submit_all(profile: gr.OAuthProfile | None):
    """Fetch, solve, and submit every valid evaluation task."""
    if not profile:
        return "Please Login to Hugging Face with the button.", None

    username = str(profile.username).strip()
    api_url = DEFAULT_API_URL
    questions_url = f"{api_url}/questions"
    submit_url = f"{api_url}/submit"
    agent_code, configuration_warning = space_code_url(os.getenv("SPACE_ID"))

    try:

        def fetch_questions():
            fetched = requests.get(questions_url, timeout=15)
            fetched.raise_for_status()
            return fetched

        response = retry_call(
            fetch_questions,
            attempts=3,
            delay_seconds=1,
        )
        questions_data = response.json()
        if not isinstance(questions_data, list) or not questions_data:
            return "Fetched questions list is empty or invalid format.", None
    except Exception as exc:
        return f"Error fetching questions: {exc}", None

    try:
        services = build_default_services()
    except Exception as exc:
        return f"Error initializing solver providers: {exc}", None

    prepared_items = []
    for item in questions_data:
        prepared = dict(item)
        if item.get("task_id") and item.get("file_name"):
            prepared["file_url"] = f"{api_url}/files/{item['task_id']}"
        prepared_items.append(prepared)

    def solve(context):
        return solve_task(context, services)

    def fallback(context):
        return services.synthesize(
            context.question,
            "No reliable external evidence was available. Give the best concise answer.",
        )

    with tempfile.TemporaryDirectory(prefix="gaia-attachments-") as attachment_dir:

        def prepare(context):
            return acquire_attachment(
                context,
                http_get=lambda url: requests.get(url, timeout=30),
                directory=attachment_dir,
            )

        batch = evaluate_items(
            prepared_items,
            username=username,
            agent_code=agent_code,
            prepare=prepare,
            solve=solve,
            fallback=fallback,
        )
    results_frame = _result_frame(batch.results)
    if not batch.payload["answers"]:
        return "Agent did not produce any answers to submit.", results_frame

    try:

        def submit_answers():
            submitted = requests.post(submit_url, json=batch.payload, timeout=60)
            submitted.raise_for_status()
            return submitted

        response = retry_call(
            submit_answers,
            attempts=3,
            delay_seconds=1,
        )
        result_data = response.json()
        final_status = (
            "Submission Successful!\n"
            f"User: {result_data.get('username', username)}\n"
            f"Overall Score: {result_data.get('score', 'N/A')}% "
            f"({result_data.get('correct_count', '?')}/"
            f"{result_data.get('total_attempted', '?')} correct)\n"
            f"Message: {result_data.get('message', 'No message received.')}"
        )
        if configuration_warning:
            final_status = f"{configuration_warning}\n{final_status}"
        return final_status, results_frame
    except Exception as exc:
        status = f"Submission Failed: {exc}"
        if configuration_warning:
            status = f"{configuration_warning}\n{status}"
        return status, results_frame


def build_demo(
    login_button_factory: Callable[[], Any] | None = None,
) -> gr.Blocks:
    """Build the existing authenticated Gradio evaluation interface."""
    with gr.Blocks() as demo:
        gr.Markdown("# GAIA Agent Evaluation Runner")
        gr.Markdown("Log in with Hugging Face, then run the complete evaluation and submission.")
        (login_button_factory or 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],
        )
    return demo


if __name__ == "__main__":
    code_url, warning = space_code_url(os.getenv("SPACE_ID"))
    if warning:
        print(warning)
    else:
        print(f"Space code URL: {code_url}")
    build_demo().launch(debug=True, share=False)