app.py

from __future__ import annotations

import os
from collections.abc import Callable
from typing import Any

import gradio as gr
import pandas as pd
import requests

from agent import answer_question

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


def space_code_url(space_id: str | None) -> tuple[str, str]:
    if not space_id:
        return "", "SPACE_ID is not configured; agent_code will be empty."
    return f"https://huggingface.co/spaces/{space_id}/tree/main", ""


def _fetch_questions(api_url: str) -> list[dict[str, Any]]:
    response = requests.get(f"{api_url}/questions", timeout=30)
    response.raise_for_status()
    payload = response.json()
    if not isinstance(payload, list) or not payload:
        raise ValueError("The questions endpoint returned an empty or invalid payload.")
    return [item for item in payload if isinstance(item, dict)]


def run_and_submit_all(profile: gr.OAuthProfile | None):
    if not profile:
        return "Please log in to Hugging Face first.", None

    username = str(profile.username).strip()
    api_url = os.getenv("GAIA_API_URL", DEFAULT_API_URL).rstrip("/")
    agent_code, warning = space_code_url(os.getenv("SPACE_ID"))

    try:
        questions = _fetch_questions(api_url)
    except Exception as exc:
        return f"Error fetching questions: {exc}", None

    rows: list[dict[str, Any]] = []
    answers: list[dict[str, str]] = []

    for index, item in enumerate(questions, start=1):
        task_id = str(item.get("task_id") or "").strip()
        question = str(item.get("question") or "")
        file_name = str(item.get("file_name") or "").strip()
        if not task_id or not question:
            continue

        try:
            answer = answer_question(question, file_name=file_name)
            if answer is None:
                status = "skipped_attachment"
                displayed_answer = ""
            else:
                status = "answered"
                displayed_answer = answer
                answers.append({"task_id": task_id, "submitted_answer": answer})
        except Exception as exc:
            status = f"error: {type(exc).__name__}: {exc}"
            displayed_answer = ""

        rows.append(
            {
                "#": index,
                "task_id": task_id,
                "file_name": file_name,
                "status": status,
                "submitted_answer": displayed_answer,
                "question": question,
            }
        )

    frame = pd.DataFrame(rows)
    if not answers:
        return "No answers were produced; nothing was submitted.", frame

    payload = {
        "username": username,
        "agent_code": agent_code,
        "answers": answers,
    }

    try:
        response = requests.post(f"{api_url}/submit", json=payload, timeout=90)
        response.raise_for_status()
        result = response.json()
    except Exception as exc:
        status = f"Submission failed: {exc}"
        if warning:
            status = f"{warning}\n{status}"
        return status, frame

    status = (
        "Submission successful!\n"
        f"User: {result.get('username', username)}\n"
        f"Submitted answers: {len(answers)}/{len(rows)}\n"
        f"Overall score: {result.get('score', 'N/A')}% "
        f"({result.get('correct_count', '?')}/{result.get('total_attempted', '?')} correct)\n"
        f"Message: {result.get('message', 'No message received.')}"
    )
    if warning:
        status = f"{warning}\n{status}"
    return status, frame


def build_demo(login_button_factory: Callable[[], Any] | None = None) -> gr.Blocks:
    with gr.Blocks() as demo:
        gr.Markdown("# GAIA Agent Evaluation Runner")
        gr.Markdown(
            "File-attachment questions are skipped. YouTube questions use Gemini; "
            "all other questions use one LangChain/OpenAI agent."
        )
        (login_button_factory or gr.LoginButton)()
        run_button = gr.Button("Run Evaluation & Submit")
        status_output = gr.Textbox(label="Status", lines=7, interactive=False)
        results_table = gr.DataFrame(label="Question results", wrap=True)
        run_button.click(
            fn=run_and_submit_all,
            outputs=[status_output, results_table],
        )
    return demo


if __name__ == "__main__":
    build_demo().launch(debug=True, share=False)