implement certificate in app

app.py

@@ -1,6 +1,12 @@
 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
@@ -10,24 +16,22 @@ 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 10
+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")
 
 ds = load_dataset(EXAM_DATASET_ID, split="train")
 
 DATASET_REPO_URL = "https://huggingface.co/datasets/agents-course/certificates"
-CERTIFIED_USERS_FILENAME = "certified_students.csv"
-CERTIFIED_USERS_DIR = "certificates"
-repo = Repository(
-    local_dir=CERTIFIED_USERS_DIR,
-    clone_from=DATASET_REPO_URL,
-    use_auth_token=os.getenv("HF_TOKEN"),
-)
 
 # Convert dataset to a list of dicts and randomly sort
 quiz_data = ds.to_pandas().to_dict("records")
-random.shuffle(quiz_data)
+# random.shuffle(quiz_data)
 
 # Limit to max questions if specified
 if EXAM_MAX_QUESTIONS:
@@ -69,43 +73,109 @@ def on_user_logged_in(token: gr.OAuthToken | None):
         ]
 
 
-def add_certified_user(hf_username, pass_percentage, submission_time):
-    """
-    Add the certified user to the database
-    """
-    print("ADD CERTIFIED USER")
-    repo.git_pull()
-    history = pd.read_csv(os.path.join(CERTIFIED_USERS_DIR, CERTIFIED_USERS_FILENAME))
-
-    # Check if this hf_username is already in our dataset:
-    check = history.loc[history["hf_username"] == hf_username]
-    if not check.empty:
-        history = history.drop(labels=check.index[0], axis=0)
-
-    new_row = pd.DataFrame(
-        {
-            "hf_username": hf_username,
-            "pass_percentage": pass_percentage,
-            "datetime": submission_time,
-        },
-        index=[0],
+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"
     )
-    history = pd.concat([new_row, history[:]]).reset_index(drop=True)
+    im = Image.open(certificate_path)
+    d = ImageDraw.Draw(im)
 
-    history.to_csv(
-        os.path.join(CERTIFIED_USERS_DIR, CERTIFIED_USERS_FILENAME), index=False
-    )
-    repo.push_to_hub(commit_message="Update certified users list")
+    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)
 
-def push_results_to_hub(user_answers, token: gr.OAuthToken | None):
-    """
-    Create a new dataset from user_answers and push it to the Hub.
-    Calculates grade and checks against passing threshold.
+    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/agents-course-finishers"
+
+    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>
     """
-    if token is None:
-        gr.Warning("Please log in to Hugging Face before pushing!")
-        return
+    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="agents-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/agents-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, 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),
+        )
 
     # Calculate grade
     correct_count = sum(1 for answer in user_answers if answer["is_correct"])
@@ -113,36 +183,54 @@ def push_results_to_hub(user_answers, token: gr.OAuthToken | None):
     grade = correct_count / total_questions if total_questions > 0 else 0
 
     if grade < float(EXAM_PASSING_SCORE):
-        gr.Warning(
-            f"Score {grade:.1%} below passing threshold of {float(EXAM_PASSING_SCORE):.1%}"
+        return (
+            gr.update(
+                visible=True,
+                value=f"You scored {grade:.1%}. Please try again to achieve at least "
+                f"{float(EXAM_PASSING_SCORE):.1%}",
+            ),
+            gr.update(visible=False),
+            gr.update(visible=False),
         )
-        return f"You scored {grade:.1%}. Please try again to achieve at least {float(EXAM_PASSING_SCORE):.1%}"
-
-    gr.Info("Submitting answers to the Hub. Please wait...", duration=2)
-
-    user_info = whoami(token=token.token)
-    repo_id = f"{EXAM_DATASET_ID}_student_responses"
-    submission_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
-
-    # filter down user answers to only include first character of the question and the answer
-    new_ds = Dataset.from_list(user_answers)
-    new_ds = new_ds.map(
-        lambda x: {
-            "username": user_info["name"],
-            "datetime": submission_time,
-            "grade": grade,
-        }
-    )
-    sanitized_name = user_info["name"].replace("-", "000")
-    new_ds.push_to_hub(repo_id=repo_id, split=sanitized_name)
 
-    # I'm adding a csv version
-    # The idea, if the user passed, we create a simple row in a csv
-    print("ADD CERTIFIED USER")
-    # Add this user to our database
-    add_certified_user(sanitized_name, grade, submission_time)
+    try:
+        # Generate certificate
+        certificate_img, _ = generate_certificate(
+            name=profile.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 = "https://huggingface.co/agents-course"
 
-    return f"Your responses have been submitted to the Hub! Final grade: {grade:.1%}"
+        # 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),
+        )
+
+    except Exception as e:
+        print(f"Error generating certificate: {e}")
+        return (
+            gr.update(
+                visible=True, value=f"🎉 Congratulations! You passed with {grade:.1%}!"
+            ),
+            gr.update(visible=False),
+            gr.update(visible=False),
+        )
 
 
 def handle_quiz(
@@ -153,12 +241,21 @@ def handle_quiz(
     token: gr.OAuthToken | None,
     profile: gr.OAuthProfile | None,
 ):
-    """
-    Handle quiz state transitions and store answers
-    """
+    """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
+        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]
@@ -184,34 +281,37 @@ def handle_quiz(
             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),  # hide radio choices
-            f"{'🎉 Passed! Click now on ✅ Submit to save your exam score!' if grade >= float(EXAM_PASSING_SCORE) else '❌ Did not pass'}",
-            question_idx,
-            user_answers,
-            gr.update(visible=False),  # start button visibility
-            gr.update(visible=False),  # next button visibility
-            gr.update(visible=True),  # submit button visibility
-            results_text,  # final results 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(  # properly update radio choices
+        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.",
-        question_idx,
-        user_answers,
-        gr.update(visible=False),  # start button visibility
-        gr.update(visible=True),  # next button visibility
-        gr.update(visible=False),  # submit button visibility
-        "",  # clear final markdown
+        "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
     ]
 
 
@@ -239,18 +339,19 @@ with gr.Blocks() as demo:
     with gr.Row(variant="panel"):
         question_text = gr.Markdown("")
         radio_choices = gr.Radio(
-            choices=[], label="Your Answer", scale=1.5, visible=False
+            choices=[], label="Your Answer", scale=1, visible=False
         )
 
     with gr.Row(variant="compact"):
         status_text = gr.Markdown("")
-        final_markdown = 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("Submit ✅", visible=False)
+        submit_btn = gr.Button("🎓 Get your certificate", visible=False)
 
     # Wire up the event handlers
     login_btn.click(
@@ -266,7 +367,8 @@ with gr.Blocks() as demo:
             status_text,
             question_idx,
             user_answers,
-            final_markdown,
+            certificate_img,
+            linkedin_btn,
             user_token,
         ],
     )
@@ -283,7 +385,8 @@ with gr.Blocks() as demo:
             start_btn,
             next_btn,
             submit_btn,
-            final_markdown,
+            certificate_img,
+            linkedin_btn,
         ],
     )
 
@@ -299,13 +402,19 @@ with gr.Blocks() as demo:
             start_btn,
             next_btn,
             submit_btn,
-            final_markdown,
+            certificate_img,
+            linkedin_btn,
         ],
     )
 
-    submit_btn.click(fn=push_results_to_hub, inputs=[user_answers])
+    submit_btn.click(
+        fn=push_results_to_hub,
+        inputs=[user_answers],
+        outputs=[status_text, certificate_img, linkedin_btn],
+    )
 
 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()