app.py

from __future__ import annotations
import os
import re
import typing as T
import numpy as np
import pandas as pd
from dataclasses import dataclass

from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.cluster import KMeans
from sklearn.metrics.pairwise import cosine_similarity

import gradio as gr


CANONICAL_DISCIPLINES = [
    "Computer Engineering",
    "Computer Science",
    "Software Engineering",
    "Information Systems",
    "Data Science",
    "Artificial Intelligence",
    "Electrical Engineering",
    "Electronics Engineering",
    "Communication Engineering",
    "Mechanical Engineering",
    "Civil Engineering",
    "Biomedical Engineering",
    "Mechatronics",
    "Chemical Engineering",
    "Industrial Engineering",
    "Architecture",
    "Business Administration",
    "Accounting",
    "Marketing",
    "Finance",
    "Economics",
]

# Keyword rules for direct mapping (Arabic + English). Order matters.
RULES: list[tuple[str, str]] = [
    # AI / Data / CS
    (r"\b(data\s*science|تحليل\s*البيانات|علم\s*البيانات)\b", "Data Science"),
    (r"\b(artificial\s*intelligence|ذكاء\s*اصطناعي|ذكاء\s*إصطناعي|AI)\b", "Artificial Intelligence"),
    (r"\b(software\s*engineering|هندسة\s*البرمجيات)\b", "Software Engineering"),
    (r"\b(information\s*systems|نظم\s*المعلومات)\b", "Information Systems"),
    (r"\b(computer\s*science|علوم?\s*الحاسوب|حاسبات|CS)\b", "Computer Science"),
    (r"\b(computer\s*engineering|هندسة\s*الحاسبات|كمبيوتر)\b", "Computer Engineering"),
    # EE / Comm / Electronics
    (r"\b(communications?\s*engineering|اتصالات)\b", "Communication Engineering"),
    (r"\b(electrical\s*engineering|كهرب(اء|ائية))\b", "Electrical Engineering"),
    (r"\b(electronics?\s*engineering|إلكترونيات)\b", "Electronics Engineering"),
    # Other engineering
    (r"\b(mechatronics?|ميكاترونكس)\b", "Mechatronics"),
    (r"\b(mechanical\s*engineering|ميكانيكا)\b", "Mechanical Engineering"),
    (r"\b(civil\s*engineering|مدني)\b", "Civil Engineering"),
    (r"\b(biomedical\s*engineering|هندسة\s*طبية)\b", "Biomedical Engineering"),
    (r"\b(chemical\s*engineering|كيميائية)\b", "Chemical Engineering"),
    (r"\b(industrial\s*engineering|انتاج|صناعية)\b", "Industrial Engineering"),
    (r"\b(architecture|هندسة\s*معمارية|عمارة)\b", "Architecture"),
    # Business
    (r"\b(business\s*administration|ادارة\s*اعمال)\b", "Business Administration"),
    (r"\b(accounting|محاسبة)\b", "Accounting"),
    (r"\b(marketing|تسويق)\b", "Marketing"),
    (r"\b(finance|تمويل)\b", "Finance"),
    (r"\b(economics|اقتصاد)\b", "Economics"),
]

STOPWORDS_AR = {
    "جامعة", "كلية", "قسم", "تخصص", "مشروع", "مشاريع", "عن", "في", "من", "على", "و",
}

STOPWORDS_EN = {
    'a', 'about', 'above', 'after', 'again', 'against', 'all', 'am', 'an', 'and',
    'any', 'are', 'aren', "aren't", 'as', 'at', 'be', 'because', 'been', 'before',
    'being', 'below', 'between', 'both', 'but', 'by', 'can', 'cannot', 'could',
    'couldn', "couldn't", 'did', 'didn', "didn't", 'do', 'does', 'doesn',
    "doesn't", 'doing', 'don', "don't", 'down', 'during', 'each', 'few', 'for',
    'from', 'further', 'had', 'hadn', "hadn't", 'has', 'hasn', "hasn't", 'have',
    'haven', "haven't", 'having', 'he', 'her', 'here', 'hers', 'herself', 'him',
    'himself', 'his', 'how', 'i', 'if', 'in', 'into', 'is', 'isn', "isn't", 'it',
    "it's", 'its', 'itself', 'just', 'll', 'm', 'ma', 'me', 'mightn', "mightn't",
    'more', 'most', 'mustn', "mustn't", 'my', 'myself', 'no', 'nor', 'not', 'now',
    'o', 'of', 'off', 'on', 'once', 'only', 'or', 'other', 'our', 'ours',
    'ourselves', 'out', 'over', 'own', 're', 's', 'same', 'shan', "shan't", 'she',
    "she's", 'should', "should've", 'shouldn', "shouldn't", 'so', 'some', 'such',
    't', 'than', 'that', "that'll", 'the', 'their', 'theirs', 'them', 'themselves',
    'then', 'there', 'these', 'they', 'this', 'those', 'through', 'to', 'too',
    'under', 'until', 'up', 've', 'very', 'was', 'wasn', "wasn't", 'we', 'were',
    'weren', "weren't", 'what', 'when', 'where', 'which', 'while', 'who', 'whom',
    'why', 'will', 'with', 'won', "won't", 'wouldn', "wouldn't", 'y', 'you',
    "you'd", "you'll", "you're", "you've", 'your', 'yours', 'yourself', 'yourselves'
}

## -------------------
## Data Structures
## -------------------

@dataclass
class Models:
    vectorizer: TfidfVectorizer
    kmeans: KMeans
    canonical_matrix: np.ndarray  # TF-IDF vectors for canonical labels

@dataclass
class AppState:
    df: pd.DataFrame
    models: Models
    dep_dict: dict[str, list[str]]


def _normalize_text(s: str) -> str:
    if not isinstance(s, str):
        return ""
    s = s.strip().lower()
    s = re.sub(r"[\u0610-\u061A\u064B-\u065F\u06D6-\u06ED]", "", s)  # remove Arabic diacritics
    s = re.sub(r"[\W_]+", " ", s)
    words = s.split()
    # Filter out stopwords from both Arabic and English sets
    filtered_words = [word for word in words if word not in STOPWORDS_AR and word not in STOPWORDS_EN]
    return " ".join(filtered_words)

def rule_based_map(text: str) -> str | None:
    t = _normalize_text(text)
    for pat, label in RULES:
        if re.search(pat, t, flags=re.IGNORECASE):
            return label
    return None

def build_department_dict(df: pd.DataFrame) -> dict[str, list[str]]:
    mapping: dict[str, list[str]] = {}
    for uni, group in df.groupby("university"):
        deps = (
            group["department"].astype(str).fillna("")
            .apply(lambda x: x.strip())
            .replace("", np.nan)
            .dropna()
            .unique()
            .tolist()
        )
        mapping[str(uni)] = sorted(list(set(deps)), key=lambda s: s.lower())
    return mapping

def train_kmeans(df: pd.DataFrame, n_clusters: int | None = None) -> Models:
    # Use combined text to better infer discipline
    combo = (
        df["department"].astype(str).fillna("") + " " +
        df["description"].astype(str).fillna("") + " " +
        df["keywords"].astype(str).fillna("")
    ).apply(_normalize_text)

    # If dataset is tiny set clusters to min(len(CANONICAL_DISCIPLINES), unique departments)
    if n_clusters is None:
        n_clusters = min(len(CANONICAL_DISCIPLINES), max(2, df['department'].nunique()))

    vectorizer = TfidfVectorizer(ngram_range=(1, 2), min_df=1, max_df=0.9)
    X = vectorizer.fit_transform(combo)

    kmeans = KMeans(n_clusters=n_clusters, random_state=42, n_init=10)
    kmeans.fit(X)

    # Build canonical label matrix to map clusters to closest discipline later
    canonical_texts = [
        _normalize_text(lbl) + " " + lbl.replace("Engineering", " Eng ")
        for lbl in CANONICAL_DISCIPLINES
    ]
    canonical_matrix = vectorizer.transform(canonical_texts)

    return Models(vectorizer=vectorizer, kmeans=kmeans, canonical_matrix=canonical_matrix)

def infer_discipline(text_fields: list[str], models: Models) -> str:
    # Try rules first
    for t in text_fields:
        m = rule_based_map(t)
        if m:
            return m

    # Fallback to KMeans + nearest canonical
    merged = _normalize_text(" ".join([t for t in text_fields if isinstance(t, str)]))
    if not merged.strip():
        return "Unknown"

    vec = models.vectorizer.transform([merged])
    cluster_idx = models.kmeans.predict(vec)[0]
    # Find canonical label closest to this vector
    sims = cosine_similarity(vec, models.canonical_matrix)[0]
    best_idx = int(np.argmax(sims))
    return CANONICAL_DISCIPLINES[best_idx]

def add_discipline_column(df: pd.DataFrame, models: Models) -> pd.DataFrame:
    texts = (
        df[["department", "description", "keywords"]]
        .astype(str)
        .fillna("")
        .values
        .tolist()
    )
    labels = [infer_discipline(row, models) for row in texts]
    out = df.copy()
    out["discipline"] = labels
    return out

def load_dataset(csv_file_path: str | None) -> pd.DataFrame:
    if not csv_file_path or not os.path.exists(csv_file_path):
        raise FileNotFoundError("CSV file not found. Please upload or set a valid path.")

    df = pd.read_csv(csv_file_path)
    
    # Check for expected columns, be flexible with case/spacing
    required = ["title", "description", "keywords", "university", "faculty", "department"]
    df.columns = [c.strip().lower() for c in df.columns] # Normalize column names
    
    missing = [c for c in required if c not in df.columns]
    if missing:
        raise ValueError(f"CSV missing required columns: {missing}")

    # Clean data
    for c in required:
        df[c] = df[c].astype(str).fillna("").str.strip()
    return df

# Initialize from a default path if provided via env
DEFAULT_CSV = os.getenv("PROJECTS_CSV_PATH", "projects_100.csv")
_state: AppState | None = None

def _init_state(csv_path: str) -> AppState:
    df = load_dataset(csv_path)
    models = train_kmeans(df)
    df_with_discipline = add_discipline_column(df, models)
    dep_dict = build_department_dict(df_with_discipline)
    return AppState(df=df_with_discipline, models=models, dep_dict=dep_dict)

def refresh_data(csv_file_obj):
    """(Re)load CSV and rebuild models + dropdowns."""
    global _state
    if csv_file_obj is None:
        return "Please upload a file.", gr.Dropdown(choices=[]), gr.Dropdown(choices=[]), gr.Dataset(headers=[], samples=[])

    try:
        csv_path = csv_file_obj.name
        _state = _init_state(csv_path)
    except Exception as e:
        return f"Error: {e}", gr.Dropdown(choices=[]), gr.Dropdown(choices=[]), gr.Dataset(headers=[], samples=[])


    universities = sorted(_state.dep_dict.keys())
    first_uni = universities[0] if universities else None
    
    deps = _state.dep_dict.get(first_uni, []) if first_uni else []
    first_dep = deps[0] if deps else None

    # Example preview dataset (first 5 rows)
    preview = _state.df[["title", "university", "faculty", "department", "discipline"]].head(5)
    
    return (
        f"Loaded {len(_state.df)} projects.",
        gr.Dropdown(choices=universities, value=first_uni),
        gr.Dropdown(choices=deps, value=first_dep),
        gr.Dataset(samples=preview.values.tolist(), headers=list(preview.columns))
    )

def update_departments(university: str):
    if not _state or not university:
        return gr.Dropdown(choices=[], value=None)
    deps = _state.dep_dict.get(university, [])
    return gr.Dropdown(choices=deps, value=(deps[0] if deps else None))

def query_projects(university: str, department: str):
    if not _state:
        return "Please load a file first.", pd.DataFrame(), pd.DataFrame()

    if not university or not department:
        return "Please select a university and department.", pd.DataFrame(), pd.DataFrame()

    # Determine the discipline of the chosen department
    subset = _state.df[
        (_state.df["university"].str.lower() == str(university).lower()) &
        (_state.df["department"].str.lower() == str(department).lower())
    ]

    discipline = subset.iloc[0]["discipline"] if not subset.empty else infer_discipline([department], _state.models)

    # Filter projects from the same university and discipline
    same_uni = _state.df[
        (_state.df["university"].str.lower() == str(university).lower()) &
        (_state.df["discipline"] == discipline)
    ]

    # Filter projects from other universities but the same discipline
    other_unis = _state.df[
        (_state.df["university"].str.lower() != str(university).lower()) &
        (_state.df["discipline"] == discipline)
    ]

    msg = f"Unified Discipline: **{discipline}**\n\nProjects from the same university: {len(same_uni)} | From other universities: {len(other_unis)}"
    
    cols = ["title", "description", "keywords", "university", "faculty", "department", "discipline"]
    return msg, same_uni[cols].reset_index(drop=True), other_unis[cols].reset_index(drop=True)

def classify_ad_hoc(university: str, faculty: str, department: str, title: str, description: str, keywords: str):
    if not _state:
        return "Please load a file first.", pd.DataFrame(), pd.DataFrame()

    discipline = infer_discipline([department, description, keywords, title], _state.models)

    # Find similar projects based on the inferred discipline
    same_uni = _state.df[
        (_state.df["university"].str.lower() == str(university).lower()) &
        (_state.df["discipline"] == discipline)
    ]

    other_unis = _state.df[
        (_state.df["university"].str.lower() != str(university).lower()) &
        (_state.df["discipline"] == discipline)
    ]

    info = f"Your project was classified as: **{discipline}**"
    cols = ["title", "description", "keywords", "university", "faculty", "department", "discipline"]
    return info, same_uni[cols].reset_index(drop=True), other_unis[cols].reset_index(drop=True)

def build_app():
    with gr.Blocks(title="University Project Discipline Classifier", theme=gr.themes.Soft()) as demo:
        gr.Markdown("""
        # 🔎 Classify Graduation Projects by **Unified Discipline**
        **Upload a CSV file** with the required columns. After uploading, choose the university and department to view:
        1. Projects from the **same university** with the same unified discipline.
        2. Projects from **other universities** with the same discipline (thanks to clustering).
        """)

        with gr.Row():
            csv_file = gr.File(label="Projects File (CSV)", file_types=[".csv"])
            load_btn = gr.Button("Load / Reload Data")

        status = gr.Markdown("No file loaded yet.")
        preview = gr.Dataset(components=[], headers=[], samples=[], label="Data Preview (first 5 rows)")

        with gr.Tab("Search by Discipline"):
            with gr.Row():
                uni_dd = gr.Dropdown(label="University", choices=[])
                dep_dd = gr.Dropdown(label="Department / Specialization", choices=[])
                search_btn = gr.Button("Search")

            result_msg = gr.Markdown()
            same_uni_tbl = gr.Dataframe(label="Projects from the Same University & Discipline", interactive=False)
            other_unis_tbl = gr.Dataframe(label="Projects from Other Universities (Same Discipline)", interactive=False)

        with gr.Tab("Classify a New Project"):
            gr.Markdown("## Try Classifying a New Project (without saving)")
            with gr.Row():
                ah_uni = gr.Textbox(label="University")
                ah_fac = gr.Textbox(label="Faculty")
                ah_dep = gr.Textbox(label="Department / Specialization")
            ah_title = gr.Textbox(label="Project Title")
            ah_desc = gr.Textbox(label="Description", lines=3)
            ah_keys = gr.Textbox(label="Keywords (comma-separated)", info="e.g., deep learning, Python, IoT")
            classify_btn = gr.Button("Classify Project & Show Similar Projects")
            info_box = gr.Markdown()

        load_btn.click(
            fn=refresh_data,
            inputs=[csv_file],
            outputs=[status, uni_dd, dep_dd, preview]
        )

        uni_dd.change(
            fn=update_departments,
            inputs=[uni_dd],
            outputs=[dep_dd]
        )
        
        search_btn.click(
            fn=query_projects,
            inputs=[uni_dd, dep_dd],
            outputs=[result_msg, same_uni_tbl, other_unis_tbl]
        )

        classify_btn.click(
            fn=classify_ad_hoc,
            inputs=[ah_uni, ah_fac, ah_dep, ah_title, ah_desc, ah_keys],
            outputs=[info_box, same_uni_tbl, other_unis_tbl]
        )

    return demo


if __name__ == "__main__":
    # Try to preload if a default CSV exists
    try:
        if os.path.exists(DEFAULT_CSV):
            print(f"Loading default data from: {DEFAULT_CSV}")
            _state = _init_state(DEFAULT_CSV)
            print("Default data loaded successfully.")
        else:
            print(f"Default CSV '{DEFAULT_CSV}' not found. Please upload a file in the app.")
            _state = None
    except Exception as e:
        print(f"Initial load failed: {e}")
        _state = None

    app = build_app()
    # For local dev, set share=True if you want a public link
    app.launch(server_name="0.0.0.0", server_port=int(os.getenv("PORT", 7860)))