app.py

import os
import tempfile
from pathlib import Path

import gradio as gr
import pandas as pd
import requests
import spaces

DEFAULT_API_URL = "https://agents-course-unit4-scoring.hf.space"
DEFAULT_SPACE_ID = "Miladsaeedi70/Final_Assignment_Template"
SPACE_OWNER = os.getenv("SPACE_OWNER", "Miladsaeedi70").strip()

_AGENT_INSTANCE = None


def get_agent():
    """Import and initialize the production OpenAI agent lazily."""

    global _AGENT_INSTANCE

    if _AGENT_INSTANCE is not None:
        return _AGENT_INSTANCE

    from agent import GaiaAgent

    _AGENT_INSTANCE = GaiaAgent()
    return _AGENT_INSTANCE


def validate_profile(
    profile: gr.OAuthProfile | None,
) -> tuple[str | None, str | None]:
    """Return (username, error_message) for the authenticated Space user."""

    if profile is None:
        return None, "Please log in to Hugging Face first."

    username = str(profile.username).strip()
    if not username:
        return None, "Hugging Face login did not return a username."

    if SPACE_OWNER and username.lower() != SPACE_OWNER.lower():
        return (
            None,
            "This public Space is restricted to its owner to prevent "
            "unauthorized OpenAI API usage.",
        )

    return username, None


@spaces.GPU
def test_zero_gpu() -> str:
    """
    Small ZeroGPU probe required by the Space hardware configuration.

    The full GAIA evaluation is intentionally not decorated because GPT-4.1
    runs through the OpenAI API and does not use the allocated Hugging Face GPU.
    """
    return "ZeroGPU function executed successfully."


def run_preflight(
    profile: gr.OAuthProfile | None,
) -> str:
    """Validate authentication, dependencies, API key, and model access."""

    username, error_message = validate_profile(profile)
    if error_message:
        return error_message

    try:
        agent = get_agent()
        result = agent.health_check()
    except Exception as error:
        return (
            "Preflight failed: "
            f"{type(error).__name__}: {error}"
        )

    checks = [
        f"User: {username}",
        f"Text model: {result['text_model']}",
        f"Vision model: {result['vision_model']}",
        f"Audio model: {result['audio_model']}",
        f"Text response: {result['text_response']}",
        f"Stockfish available: {result['stockfish_available']}",
        f"FFmpeg available: {result['ffmpeg_available']}",
    ]

    if result["text_response"].strip().upper() != "OK":
        checks.append(
            "Warning: the model responded, but not with the expected exact word OK."
        )

    return "Preflight completed.\n" + "\n".join(checks)


def download_task_attachment(
    api_url: str,
    task_id: str,
    file_name: str,
    output_directory: Path,
) -> str:
    """Download one GAIA attachment and return its local path."""

    safe_name = Path(file_name).name
    output_path = output_directory / f"{task_id}_{safe_name}"

    response = requests.get(
        f"{api_url}/files/{task_id}",
        timeout=120,
    )
    response.raise_for_status()

    if not response.content:
        raise RuntimeError("The attachment response was empty.")

    content_type = response.headers.get("Content-Type", "").lower()
    if "application/json" in content_type:
        try:
            payload = response.json()
        except ValueError:
            payload = {}

        detail = payload.get("detail")
        if detail:
            raise RuntimeError(f"Attachment API error: {detail}")

    output_path.write_bytes(response.content)
    return str(output_path)


def run_and_submit_all(
    profile: gr.OAuthProfile | None,
):
    """Run the LangGraph agent on all GAIA questions and submit answers."""

    username, error_message = validate_profile(profile)
    if error_message:
        return error_message, None

    print(f"User logged in: {username}")

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

    space_id = os.getenv("SPACE_ID", DEFAULT_SPACE_ID)
    agent_code = f"https://huggingface.co/spaces/{space_id}/tree/main"

    try:
        agent = get_agent()
    except Exception as error:
        message = (
            "Agent initialization failed: "
            f"{type(error).__name__}: {error}"
        )
        print(message)
        return message, None

    try:
        response = requests.get(questions_url, timeout=30)
        response.raise_for_status()
        questions_data = response.json()
    except Exception as error:
        message = (
            "Could not fetch the questions: "
            f"{type(error).__name__}: {error}"
        )
        print(message)
        return message, None

    if not isinstance(questions_data, list) or not questions_data:
        return "The questions endpoint returned no questions.", None

    results_log: list[dict] = []
    answers_payload: list[dict] = []

    with tempfile.TemporaryDirectory(prefix="gaia_attachments_") as directory:
        attachment_directory = Path(directory)

        for question_number, item in enumerate(questions_data, start=1):
            task_id = str(item.get("task_id", "")).strip()
            question_text = str(item.get("question", "")).strip()
            file_name = str(item.get("file_name", "") or "").strip()

            if not task_id or not question_text:
                print(f"Skipping invalid question item: {item}")
                continue

            print("\n" + "=" * 80)
            print(f"QUESTION {question_number}/{len(questions_data)}")
            print(f"Task ID: {task_id}")
            print(f"Attachment: {file_name or 'None'}")
            print(f"Question: {question_text}")
            print("=" * 80)

            input_file: str | None = None
            submitted_answer = ""
            error_text = ""

            try:
                if file_name:
                    input_file = download_task_attachment(
                        api_url=api_url,
                        task_id=task_id,
                        file_name=file_name,
                        output_directory=attachment_directory,
                    )
                    print(f"Downloaded attachment: {input_file}")

                submitted_answer = agent(
                    question=question_text,
                    input_file=input_file,
                )

            except Exception as error:
                error_text = f"{type(error).__name__}: {error}"
                print(f"Agent error for {task_id}: {error_text}")
                submitted_answer = ""

            submitted_answer = str(submitted_answer or "").strip()

            answers_payload.append(
                {
                    "task_id": task_id,
                    "submitted_answer": submitted_answer,
                }
            )

            results_log.append(
                {
                    "Task ID": task_id,
                    "Question": question_text,
                    "Attachment": file_name,
                    "Submitted Answer": submitted_answer,
                    "Error": error_text,
                }
            )

            print(f"Submitted answer: {submitted_answer or '[blank]'}")

    if not answers_payload:
        return (
            "The agent did not produce any submission records.",
            pd.DataFrame(results_log),
        )

    submission_data = {
        "username": username,
        "agent_code": agent_code,
        "answers": answers_payload,
    }

    try:
        response = requests.post(
            submit_url,
            json=submission_data,
            timeout=180,
        )
        response.raise_for_status()
        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.')}"
        )

        return final_status, pd.DataFrame(results_log)

    except requests.HTTPError as error:
        response_text = error.response.text[:1000] if error.response else ""
        message = (
            "Submission failed: "
            f"HTTP {getattr(error.response, 'status_code', 'unknown')} - "
            f"{response_text}"
        )
        return message, pd.DataFrame(results_log)

    except Exception as error:
        message = (
            "Submission failed: "
            f"{type(error).__name__}: {error}"
        )
        return message, pd.DataFrame(results_log)


with gr.Blocks() as demo:
    gr.Markdown("# GAIA Final Assignment Agent")
    gr.Markdown(
        """
Log in with Hugging Face, then run the complete 20-question evaluation.
The Space downloads task attachments, runs the LangGraph agent, and submits
only the final answers to the course scorer.

Only the Space owner can run the evaluation, which protects the private
OpenAI API key used by this public Space.
        """
    )

    gr.LoginButton()

    zero_gpu_button = gr.Button(
        "1. Test ZeroGPU",
    )
    preflight_button = gr.Button(
        "2. Test OpenAI Configuration",
    )
    run_button = gr.Button(
        "3. Run Evaluation & Submit All Answers",
        variant="primary",
    )

    status_output = gr.Textbox(
        label="Preflight / Submission Status",
        lines=9,
        interactive=False,
    )

    results_table = gr.DataFrame(
        label="Questions and Agent Answers",
        wrap=True,
    )

    zero_gpu_button.click(
        fn=test_zero_gpu,
        outputs=status_output,
    )

    preflight_button.click(
        fn=run_preflight,
        outputs=status_output,
    )

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


demo.queue(default_concurrency_limit=1)


if __name__ == "__main__":
    print("Starting GAIA Final Assignment Space")
    print("SPACE_ID:", os.getenv("SPACE_ID", DEFAULT_SPACE_ID))
    print("SPACE_OWNER:", SPACE_OWNER or "[not restricted]")
    print("OPENAI_API_KEY configured:", bool(os.getenv("OPENAI_API_KEY")))
    demo.launch()