app.py

from datetime import datetime
from time import sleep
from typing import Dict, List, Tuple
from openai import OpenAI
import gradio as gr
import pandas as pd
import numpy as np
import dotenv
import json
import os

dotenv.load_dotenv()

openai = OpenAI(
    base_url="https://openrouter.ai/api/v1", api_key=os.environ.get("API_KEY")
)

MBTI_TYPES = [
    "ISTJ",
    "ISFJ",
    "INFJ",
    "INTJ",
    "ISTP",
    "ISFP",
    "INFP",
    "INTP",
    "ESTP",
    "ESFP",
    "ENFP",
    "ENTP",
    "ESTJ",
    "ESFJ",
    "ENFJ",
    "ENTJ",
]


class User:
    def __init__(self, username: str, name: str):
        self.username = username
        self.name = name
        self.scores_history: List[Dict] = []
        self.recommendations_history: List[Dict] = []

    def add_scores(self, scores: "UTBKScores", mbti: str):
        score_entry = {
            "timestamp": datetime.now().isoformat(),
            "scores": scores.__dict__,
            "mbti": mbti,
        }
        self.scores_history.append(score_entry)

    def add_recommendations(self, recommendations: pd.DataFrame):
        rec_entry = {
            "timestamp": datetime.now().isoformat(),
            "recommendations": recommendations.to_dict("records"),
        }
        self.recommendations_history.append(rec_entry)


class UserManager:
    def __init__(self):
        self.users: Dict[str, User] = {}

    def get_or_create_user(self, oauth_profile: gr.OAuthProfile) -> User:
        if oauth_profile.username not in self.users:
            self.users[oauth_profile.username] = User(
                username=oauth_profile.username, name=oauth_profile.name
            )
        return self.users[oauth_profile.username]

    def save_user_interaction(
        self,
        oauth_profile: gr.OAuthProfile,
        scores: "UTBKScores",
        mbti: str,
        recommendations: pd.DataFrame,
    ):
        user = self.get_or_create_user(oauth_profile)
        user.add_scores(scores, mbti)
        user.add_recommendations(recommendations)


class UTBKScores:
    def __init__(self, scores_dict: Dict):
        self.penalaran_umum = scores_dict["penalaran_umum"]
        self.pengetahuan_umum = scores_dict["pengetahuan_umum"]
        self.pemahaman_bacaan = scores_dict["pemahaman_bacaan"]
        self.pengetahuan_kuantitatif = scores_dict["pengetahuan_kuantitatif"]
        self.literasi_indonesia = scores_dict["literasi_indonesia"]
        self.literasi_inggris = scores_dict["literasi_inggris"]
        self.penalaran_matematika = scores_dict["penalaran_matematika"]


class UniversityRecommender:
    def get_initial_recommendations(self, scores: UTBKScores, mbti: str) -> Dict:
        prompt = f"""Based on the following UTBK scores and MBTI personality type, suggest 10 suitable study programs 
        and Indonesian universities. Return the response in the following JSON format:
        {{
            "recommendations": [
                {{
                    "program": "program_name",
                    "university": "university_name",
                    "acceptance_rate": "estimated_acceptance_rate_as_decimal",
                    "required_scores": {{
                        "penalaran_umum": minimum_score,
                        ... (other scores)
                    }},
                    "ideal_mbti_types": ["TYPE1", "TYPE2", "TYPE3"],
                    "program_description": "brief_description"
                }}
            ]
        }}

        UTBK Scores:
        {scores.__dict__}
        MBTI: {mbti}

        RETURN ***JUST*** THE JSON. DO NOT WRAP IT IN A CODE BLOCK. PROVIDE THE JSON AS-IS, NOTHING ELSE.
        """

        response = openai.chat.completions.create(
            model="meta-llama/llama-4-scout",
            messages=[
                {
                    "role": "system",
                    "content": "You are a knowledgeable educational consultant who specializes in Indonesian universities.",
                },
                {"role": "user", "content": prompt},
            ],
        )

        try:
            return json.loads(response.choices[0].message.content)
        except json.JSONDecodeError:
            print("Failed to get a pred (malformation), retrying...")
            sleep(0.1)
            return self.get_initial_recommendations(scores, mbti)

    def calculate_compatibility(
        self, recommendation: Dict, scores: UTBKScores, mbti: str
    ) -> float:
        score_diffs = []
        for score_type, required_score in recommendation["required_scores"].items():
            actual_score = getattr(scores, score_type)
            score_diffs.append(max(0, (actual_score - required_score) / 100))
        score_compatibility = np.mean(score_diffs)

        mbti_compatibility = 1.0 if mbti in recommendation["ideal_mbti_types"] else 0.5
        acceptance_rate = float(recommendation["acceptance_rate"])

        final_compatibility = (
            0.4 * score_compatibility + 0.3 * mbti_compatibility + 0.3 * acceptance_rate
        )

        return final_compatibility

    def process_recommendations(
        self, raw_recommendations: Dict, scores: UTBKScores, mbti: str
    ) -> pd.DataFrame:
        results = []
        for rec in raw_recommendations["recommendations"]:
            compatibility = self.calculate_compatibility(rec, scores, mbti)
            results.append(
                {
                    "Program": rec["program"],
                    "University": rec["university"],
                    "Compatibility": f"{compatibility*100:.1f}%",
                    "Acceptance Rate": f"{float(rec['acceptance_rate'])*100:.1f}%",
                    "Description": rec["program_description"],
                }
            )

        return pd.DataFrame(results).sort_values("Compatibility", ascending=False)


def create_input_interface():
    with gr.Row():
        with gr.Column():
            gr.Markdown("## UTBK Scores")
            scores = {
                "penalaran_umum": gr.Slider(0, 100, value=50, label="Penalaran Umum"),
                "pengetahuan_umum": gr.Slider(
                    0, 100, value=50, label="Pengetahuan Umum"
                ),
                "pemahaman_bacaan": gr.Slider(
                    0, 100, value=50, label="Pemahaman Bacaan"
                ),
                "pengetahuan_kuantitatif": gr.Slider(
                    0, 100, value=50, label="Pengetahuan Kuantitatif"
                ),
                "literasi_indonesia": gr.Slider(
                    0, 100, value=50, label="Literasi Indonesia"
                ),
                "literasi_inggris": gr.Slider(
                    0, 100, value=50, label="Literasi Inggris"
                ),
                "penalaran_matematika": gr.Slider(
                    0, 100, value=50, label="Penalaran Matematika"
                ),
            }

        with gr.Column():
            gr.Markdown("## Personality")
            mbti = gr.Dropdown(choices=MBTI_TYPES, label="MBTI Type")

    return scores, mbti


def create_output_interface():
    with gr.Row():
        results_df = gr.Dataframe(
            headers=[
                "Program",
                "University",
                "Compatibility",
                "Acceptance Rate",
                "Description",
            ],
            label="Recommended Programs",
        )
    return results_df


def process_input(*args):
    recommender = UniversityRecommender()

    scores_dict = {
        "penalaran_umum": args[0],
        "pengetahuan_umum": args[1],
        "pemahaman_bacaan": args[2],
        "pengetahuan_kuantitatif": args[3],
        "literasi_indonesia": args[4],
        "literasi_inggris": args[5],
        "penalaran_matematika": args[6],
    }
    mbti = args[7]

    scores = UTBKScores(scores_dict)
    raw_recommendations = recommender.get_initial_recommendations(scores, mbti)
    return recommender.process_recommendations(raw_recommendations, scores, mbti)


def create_recommendations_from_dict(
    self, recommendations_dict: List[Dict], oauth_profile: gr.OAuthProfile | None
) -> pd.DataFrame:
    return pd.DataFrame(recommendations_dict)


def create_interface():
    user_manager = UserManager()

    with gr.Blocks(title="University Program Recommender") as interface:
        gr.LoginButton()
        gr.Markdown("# Indonesian University Program Recommender")

        # User info section
        user_info = gr.Markdown()

        # Input Section
        scores, mbti = create_input_interface()

        # Add Load Last Recommendation button
        load_last_btn = gr.Button("Load Last Recommendation")

        # Submit Button
        submit_btn = gr.Button("Get New Recommendations")

        # Output Section
        results_df = create_output_interface()

        # History Tab
        with gr.Tab("History"):
            history_df = gr.Dataframe(
                headers=["Timestamp", "MBTI", "Scores", "Recommendations"],
                label="Your Previous Recommendations",
            )

        def load_last_recommendation(
            oauth_profile: gr.OAuthProfile | None,
        ) -> pd.DataFrame:
            if oauth_profile is None:
                return pd.DataFrame()

            user = user_manager.get_or_create_user(oauth_profile)
            if user.recommendations_history:
                last_recommendation = user.recommendations_history[-1]
                return pd.DataFrame(last_recommendation["recommendations"])
            return pd.DataFrame()

        def process_with_user(oauth_profile: gr.OAuthProfile | None, *args):
            if oauth_profile is None:
                return pd.DataFrame()

            results = process_input(*args)

            scores_dict = {
                "penalaran_umum": args[0],
                "pengetahuan_umum": args[1],
                "pemahaman_bacaan": args[2],
                "pengetahuan_kuantitatif": args[3],
                "literasi_indonesia": args[4],
                "literasi_inggris": args[5],
                "penalaran_matematika": args[6],
            }
            scores = UTBKScores(scores_dict)
            mbti = args[7]

            user_manager.save_user_interaction(oauth_profile, scores, mbti, results)

            return results

        def update_user_info(
            oauth_profile: gr.OAuthProfile | None,
        ) -> Tuple[str, pd.DataFrame]:
            if oauth_profile is None:
                return (
                    "Not logged in. Please login to use the recommender.",
                    pd.DataFrame(),
                )

            # Load last recommendation on login
            last_rec = load_last_recommendation(oauth_profile)
            return (
                f"Logged in as: {oauth_profile.username} ({oauth_profile.name})",
                last_rec,
            )

        def show_history(oauth_profile: gr.OAuthProfile | None) -> pd.DataFrame:
            if oauth_profile is None:
                return pd.DataFrame()

            user = user_manager.get_or_create_user(oauth_profile)
            history_data = []

            for score_entry, rec_entry in zip(
                user.scores_history, user.recommendations_history
            ):
                history_data.append(
                    {
                        "Timestamp": score_entry["timestamp"],
                        "MBTI": score_entry["mbti"],
                        "Scores": str(score_entry["scores"]),
                        "Recommendations": str(rec_entry["recommendations"]),
                    }
                )

            return pd.DataFrame(history_data)

        input_components = list(scores.values()) + [mbti]

        submit_btn.click(
            fn=process_with_user,
            inputs=input_components,
            outputs=[results_df],
        )

        load_last_btn.click(
            fn=load_last_recommendation,
            outputs=[results_df],
        )

        interface.load(update_user_info, outputs=[user_info, results_df])
        interface.load(show_history, outputs=[history_df])

    return interface


if __name__ == "__main__":
    interface = create_interface()
    interface.launch()