validation.py

import json
import os

import requests

from model import AssistantModel

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

def _get_response(url: str):
    try:
        response = requests.get(url, timeout=15)
        response.raise_for_status()
    except requests.exceptions.RequestException as e:
        print(f"Error fetching questions: {e}")
        return None
    except Exception as e:
        print(f"An unexpected error occurred fetching questions: {e}")
        return None
    return response

def _get_response_json(url: str):
    try:
        response = _get_response(url)
        questions_data = response.json()
        if not questions_data:
            print("Fetched questions list is empty.")
            return {}, None
        print(f"Fetched {len(questions_data)} questions.")
    except requests.exceptions.JSONDecodeError as e:
         print(f"Error decoding JSON response from questions endpoint: {e}")
         print(f"Response text: {response.text[:500]}")
         return {}, None

    return questions_data


def load_questions() -> None:
    questions_url = f"{DEFAULT_API_URL}/questions"

    questions_data = _get_response_json(questions_url)

    with open(r'./dataset/questions.json', 'w') as f:
        json.dump(questions_data, f, indent=2)


def load_files() -> None:
    with open(r'./dataset/questions.json', 'r') as f:
        questions_data = json.load(f)

    for q in questions_data:
        if q["file_name"] != '':
            files_url = f'{DEFAULT_API_URL}/files/{q["task_id"]}'

            print(f"Fetching file from: {files_url}")

            file_data = _get_response(files_url)

            with open(f'./dataset/{q["file_name"]}', 'wb') as f:
                f.write(file_data.content)
                print(f"File {q['file_name']} downloaded successfully.")


def run_validation() -> None:
    with open(r'./dataset/questions.json', 'r') as f:
        questions_data = json.load(f)

    results_file = r'./dataset/results.json'
    if os.path.exists(results_file):
        with open(results_file, 'r') as f:
            results = json.load(f)
    else:
        results = {}

    model = AssistantModel()

    for question in questions_data:
        print('Task ID:', question['task_id'])
        try:
            if 'youtube.com' in question['question']:
                raise Exception('Youtube is not supported')

            if question['file_name'].endswith('.mp3'):
                raise Exception('MP3 file is not supported')

            answer = model.ask_question(question['question'], question['file_name'])
        except Exception as e:
            print(f"Error processing question {question['task_id']}: {e}")
            answer = "N/A"
        results[question['task_id']]=answer
        print('Answer:', answer)

        with open(r'./dataset/results.json', 'w') as f:
            json.dump(results, f, indent=2)


def prepare_submission() -> None:
    with open(r'./dataset/results.json', 'r') as f:
        results = json.load(f)

    lines = [f'{{"task_id": "{id}", "model_answer":"{a}"}}' for id, a in results.items()]

    with open(r'./dataset/submission.jsonl', 'w') as f:
        f.write('\n'.join(lines))


def post_submission() -> None:
    with open(r'./dataset/results.json', 'r') as f:
        results = json.load(f)

    space_id = os.environ['SPACE_ID']

    submission_json = {
        "username": os.environ['HF_USERNAME'],
        "agent_code": f"https://huggingface.co/spaces/{space_id}/tree/main",
        "answers": [{"task_id": id, "submitted_answer": a} for id, a in results.items()]
    }

    print(json.dumps(submission_json, indent=2))

    submit_url = f"{DEFAULT_API_URL}/submit"

    try:
        response = requests.post(submit_url, json=submission_json, 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(final_status)
        print("Submission successful.")
    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)
    except requests.exceptions.Timeout:
        status_message = "Submission Failed: The request timed out."
        print(status_message)
    except requests.exceptions.RequestException as e:
        status_message = f"Submission Failed: Network error - {e}"
        print(status_message)
    except Exception as e:
        status_message = f"An unexpected error occurred during submission: {e}"
        print(status_message)


if __name__ == '__main__':
    # load_questions()
    # load_files()

    # run_validation()

    # prepare_submission()

    post_submission()