app.py

import os
from datetime import datetime
import random
import requests
from io import BytesIO
from datetime import date
import tempfile
from PIL import Image, ImageDraw, ImageFont
from huggingface_hub import upload_file

import pandas as pd
from huggingface_hub import HfApi, hf_hub_download, Repository
from huggingface_hub.repocard import metadata_load

import gradio as gr
from datasets import load_dataset, Dataset
from huggingface_hub import whoami

import asyncio
from functools import partial

EXAM_DATASET_ID = os.getenv("EXAM_DATASET_ID") or "agents-course/unit_1_quiz"
EXAM_MAX_QUESTIONS = os.getenv("EXAM_MAX_QUESTIONS") or 1
EXAM_PASSING_SCORE = os.getenv("EXAM_PASSING_SCORE") or 0.8
CERTIFYING_ORG_LINKEDIN_ID = os.getenv("CERTIFYING_ORG_LINKEDIN_ID", "000000")
COURSE_TITLE = os.getenv("COURSE_TITLE", "AI Agents Fundamentals")
COURSE_URL = os.getenv("COURSE_URL", "https://huggingface.co/reasoning-course")

ds = load_dataset(EXAM_DATASET_ID, split="train")

DATASET_REPO_URL = "https://huggingface.co/datasets/reasoning-course/certificates"

# Convert dataset to a list of dicts and randomly sort
quiz_data = ds.to_pandas().to_dict("records")
random.shuffle(quiz_data)

# Limit to max questions if specified
if EXAM_MAX_QUESTIONS:
    quiz_data = quiz_data[: int(EXAM_MAX_QUESTIONS)]


def on_user_logged_in(token: gr.OAuthToken | None):
    """
    If the user has a valid token, show Start button.
    Otherwise, keep the login button visible.
    """
    if token is not None:
        return [
            gr.update(visible=False),  # login_btn
            gr.update(visible=True),  # start_btn
            gr.update(visible=False),  # next_btn
            gr.update(visible=False),  # submit_btn
            "",  # question_text
            gr.update(choices=[], visible=False),  # radio_choices
            "Click 'Start' to begin the quiz",  # status_text
            0,  # question_idx
            [],  # user_answers
            gr.update(visible=False),  # certificate_img
            gr.update(visible=False),  # linkedin_btn
            token,  # user_token
        ]
    else:
        return [
            gr.update(visible=True),  # login_btn
            gr.update(visible=False),  # start_btn
            gr.update(visible=False),  # next_btn
            gr.update(visible=False),  # submit_btn
            "",  # question_text
            gr.update(choices=[], visible=False),  # radio_choices
            "",  # status_text
            0,  # question_idx
            [],  # user_answers
            gr.update(visible=False),  # certificate_img
            gr.update(visible=False),  # linkedin_btn
            None,  # user_token
        ]


def generate_certificate(name: str, profile_url: str):
    """Generate certificate image and PDF."""
    certificate_path = os.path.join(
        os.path.dirname(__file__), "templates", "certificate.png"
    )
    im = Image.open(certificate_path)
    d = ImageDraw.Draw(im)

    name_font = ImageFont.truetype("Quattrocento-Regular.ttf", 100)
    date_font = ImageFont.truetype("Quattrocento-Regular.ttf", 48)

    name = name.title()
    d.text((1000, 740), name, fill="black", anchor="mm", font=name_font)

    d.text((1480, 1170), str(date.today()), fill="black", anchor="mm", font=date_font)

    pdf = im.convert("RGB")
    pdf.save("certificate.pdf")

    return im, "certificate.pdf"


def create_linkedin_button(username: str, cert_url: str | None) -> str:
    """Create LinkedIn 'Add to Profile' button HTML."""
    current_year = date.today().year
    current_month = date.today().month

    # Use the dataset certificate URL if available, otherwise fallback to default
    certificate_url = cert_url or "https://huggingface.co/reasoning-course"

    linkedin_params = {
        "startTask": "CERTIFICATION_NAME",
        "name": COURSE_TITLE,
        "organizationName": "Hugging Face",
        "organizationId": CERTIFYING_ORG_LINKEDIN_ID,
        "organizationIdissueYear": str(current_year),
        "issueMonth": str(current_month),
        "certUrl": certificate_url,
        "certId": username,  # Using username as cert ID
    }

    # Build the LinkedIn button URL
    base_url = "https://www.linkedin.com/profile/add?"
    params = "&".join(
        f"{k}={requests.utils.quote(v)}" for k, v in linkedin_params.items()
    )
    button_url = base_url + params

    message = f"""
        <a href="{button_url}" target="_blank" style="display: block; margin-top: 20px; text-align: center;">
            <img src="https://download.linkedin.com/desktop/add2profile/buttons/en_US.png" 
                 alt="LinkedIn Add to Profile button">
        </a>
    """
    return message


async def upload_certificate_to_hub(username: str, certificate_img) -> str:
    """Upload certificate to the dataset hub and return the URL asynchronously."""
    # Save image to temporary file
    with tempfile.NamedTemporaryFile(suffix=".png", delete=False) as tmp:
        certificate_img.save(tmp.name)

        try:
            # Run upload in a thread pool since upload_file is blocking
            loop = asyncio.get_event_loop()
            upload_func = partial(
                upload_file,
                path_or_fileobj=tmp.name,
                path_in_repo=f"certificates/{username}/{date.today()}.png",
                repo_id="reasoning-course/certificates",
                repo_type="dataset",
                token=os.getenv("HF_TOKEN"),
            )
            await loop.run_in_executor(None, upload_func)

            # Construct the URL to the image
            cert_url = (
                f"https://huggingface.co/datasets/reasoning-course/certificates/"
                f"resolve/main/certificates/{username}/{date.today()}.png"
            )

            # Clean up temp file
            os.unlink(tmp.name)
            return cert_url

        except Exception as e:
            print(f"Error uploading certificate: {e}")
            os.unlink(tmp.name)
            return None


async def push_results_to_hub(
    user_answers,
    custom_name: str | None,
    token: gr.OAuthToken | None,
    profile: gr.OAuthProfile | None,
):
    """Handle quiz completion and certificate generation."""
    if token is None or profile is None:
        gr.Warning("Please log in to Hugging Face before submitting!")
        return (
            gr.update(visible=True, value="Please login first"),
            gr.update(visible=False),
            gr.update(visible=False),
            gr.update(visible=False),  # hide custom name input
        )

    # Calculate grade
    correct_count = sum(1 for answer in user_answers if answer["is_correct"])
    total_questions = len(user_answers)
    grade = correct_count / total_questions if total_questions > 0 else 0

    if grade < float(EXAM_PASSING_SCORE):
        return (
            gr.update(visible=True, value=f"You scored {grade:.1%}..."),
            gr.update(visible=False),
            gr.update(visible=False),
            gr.update(visible=False),  # hide custom name input
        )

    try:
        # Use custom name if provided, otherwise use profile name
        name = (
            custom_name.strip() if custom_name and custom_name.strip() else profile.name
        )

        # Generate certificate
        certificate_img, _ = generate_certificate(
            name=name, profile_url=profile.picture
        )

        # Start certificate upload asynchronously
        gr.Info("Uploading your certificate...")
        cert_url = await upload_certificate_to_hub(profile.username, certificate_img)

        if cert_url is None:
            gr.Warning("Certificate upload failed, but you still passed!")
            cert_url = COURSE_URL

        # Create LinkedIn button
        linkedin_button = create_linkedin_button(profile.username, cert_url)

        result_message = f"""
        🎉 Congratulations! You passed with a score of {grade:.1%}!
        
        {linkedin_button}
        """

        return (
            gr.update(visible=True, value=result_message),
            gr.update(visible=True, value=certificate_img),
            gr.update(visible=True),
            gr.update(visible=True),  # show custom name input
        )

    except Exception as e:
        print(f"Error generating certificate: {e}")
        return (
            gr.update(visible=True, value=f"🎉 You passed with {grade:.1%}!"),
            gr.update(visible=False),
            gr.update(visible=False),
            gr.update(visible=False),  # hide custom name input
        )


def handle_quiz(
    question_idx,
    user_answers,
    selected_answer,
    is_start,
    token: gr.OAuthToken | None,
    profile: gr.OAuthProfile | None,
):
    """Handle quiz state transitions and store answers"""
    if token is None or profile is None:
        gr.Warning("Please log in to Hugging Face before starting the quiz!")
        return [
            "",  # question_text
            gr.update(choices=[], visible=False),  # radio choices
            "Please login first",  # status_text
            question_idx,  # question_idx
            user_answers,  # user_answers
            gr.update(visible=True),  # start button
            gr.update(visible=False),  # next button
            gr.update(visible=False),  # submit button
            gr.update(visible=False),  # certificate image
            gr.update(visible=False),  # linkedin button
        ]

    if not is_start and question_idx < len(quiz_data):
        current_q = quiz_data[question_idx]
        correct_reference = current_q["correct_answer"]
        correct_reference = f"answer_{correct_reference}".lower()
        is_correct = selected_answer == current_q[correct_reference]
        user_answers.append(
            {
                "question": current_q["question"],
                "selected_answer": selected_answer,
                "correct_answer": current_q[correct_reference],
                "is_correct": is_correct,
                "correct_reference": correct_reference,
            }
        )
        question_idx += 1

    if question_idx >= len(quiz_data):
        correct_count = sum(1 for answer in user_answers if answer["is_correct"])
        grade = correct_count / len(user_answers)
        results_text = (
            f"**Quiz Complete!**\n\n"
            f"Your score: {grade:.1%}\n"
            f"Passing score: {float(EXAM_PASSING_SCORE):.1%}\n\n"
        )
        has_passed = grade >= float(EXAM_PASSING_SCORE)
        return [
            "",  # question_text
            gr.update(choices=[], visible=False),  # radio choices
            f"{'🎉 Passed! Click now on 🎓 Get your certificate!' if has_passed else '❌ Did not pass'}",  # status_text
            question_idx,  # question_idx
            user_answers,  # user_answers
            gr.update(visible=False),  # start button
            gr.update(visible=False),  # next button
            gr.update(
                visible=True,
                value=f"🎓 Get your certificate" if has_passed else "❌ Did not pass",
                interactive=has_passed,
            ),  # submit button
            gr.update(visible=False),  # certificate image
            gr.update(visible=False),  # linkedin button
        ]

    # Show next question
    q = quiz_data[question_idx]
    return [
        f"## Question {question_idx + 1} \n### {q['question']}",  # question_text
        gr.update(  # radio choices
            choices=[q["answer_a"], q["answer_b"], q["answer_c"], q["answer_d"]],
            value=None,
            visible=True,
        ),
        "Select an answer and click 'Next' to continue.",  # status_text
        question_idx,  # question_idx
        user_answers,  # user_answers
        gr.update(visible=False),  # start button
        gr.update(visible=True),  # next button
        gr.update(visible=False),  # submit button
        gr.update(visible=False),  # certificate image
        gr.update(visible=False),  # linkedin button
    ]


def success_message(response):
    # response is whatever push_results_to_hub returned
    return f"{response}\n\n**Success!**"


with gr.Blocks() as demo:
    demo.title = f"Dataset Quiz for {EXAM_DATASET_ID}"

    # State variables
    question_idx = gr.State(value=0)
    user_answers = gr.State(value=[])
    user_token = gr.State(value=None)

    with gr.Row(variant="compact"):
        gr.Markdown(
f"""# Welcome to the Exam and Certification of {COURSE_TITLE}
Upon completion of {EXAM_MAX_QUESTIONS} questions, you can collect and share your certificate!

## Instructions
1. Log in then click 'Start' to begin. 
2. Answer each question and click 'Next' 
3. Click 'Claim your certificae' to get a certificate and share it with the world! 🌍"

## Questions:
""")

    with gr.Row(variant="panel"):
        question_text = gr.Markdown("")
        radio_choices = gr.Radio(
            choices=[], label="Your Answer", scale=1, visible=False
        )

    with gr.Row(variant="compact"):
        status_text = gr.Markdown("")
        certificate_img = gr.Image(type="pil", visible=False)
        linkedin_btn = gr.HTML(visible=False)

    with gr.Row(variant="compact"):
        login_btn = gr.LoginButton(visible=True)
        start_btn = gr.Button("Start ⏭️", visible=True)
        next_btn = gr.Button("Next ⏭️", visible=False)
        submit_btn = gr.Button("🎓 Get your certificate", visible=False)

    with gr.Row(variant="panel"):
        custom_name_input = gr.Textbox(
            label="Custom Name for Certificate",
            placeholder="Enter name as you want it to appear on the certificate",
            info="Leave empty to use your Hugging Face profile name",
            visible=False,
            value=None,
        )

    # Wire up the event handlers
    login_btn.click(
        fn=on_user_logged_in,
        inputs=None,
        outputs=[
            login_btn,
            start_btn,
            next_btn,
            submit_btn,
            question_text,
            radio_choices,
            status_text,
            question_idx,
            user_answers,
            certificate_img,
            linkedin_btn,
            user_token,
        ],
    )

    start_btn.click(
        fn=handle_quiz,
        inputs=[question_idx, user_answers, gr.State(""), gr.State(True)],
        outputs=[
            question_text,
            radio_choices,
            status_text,
            question_idx,
            user_answers,
            start_btn,
            next_btn,
            submit_btn,
            certificate_img,
            linkedin_btn,
        ],
    )

    next_btn.click(
        fn=handle_quiz,
        inputs=[question_idx, user_answers, radio_choices, gr.State(False)],
        outputs=[
            question_text,
            radio_choices,
            status_text,
            question_idx,
            user_answers,
            start_btn,
            next_btn,
            submit_btn,
            certificate_img,
            linkedin_btn,
        ],
    )

    submit_btn.click(
        fn=push_results_to_hub,
        inputs=[
            user_answers,
            custom_name_input,
        ],
        outputs=[
            status_text,
            certificate_img,
            linkedin_btn,
            custom_name_input,
        ],
    )

    custom_name_input.submit(
        fn=push_results_to_hub,
        inputs=[user_answers, custom_name_input],
        outputs=[status_text, certificate_img, linkedin_btn, custom_name_input],
    )

if __name__ == "__main__":
    # Note: If testing locally, you'll need to run `huggingface-cli login` or set HF_TOKEN
    # environment variable for the login to work locally.
    demo.queue()  # Enable queuing for async operations
    demo.launch()