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)))