Create app.py

app.py

@@ -0,0 +1,404 @@
+
+
+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)))