Create app.py

app.py

@@ -0,0 +1,552 @@
+#!/usr/bin/env python3
+# app.py
+# --------------------------------------------------
+# Gradio app for state-specific school fuzzy matching
+#
+# - Loads school masters from Hugging Face dataset (Apf-AI4Good/Schools)
+# - Uses normalization patterns from patterns.json (no hardcoded patterns)
+# - Public "Search" tab for users
+# - "Admin" tab to view/edit/save patterns (password protected for saving)
+# --------------------------------------------------
+
+import os
+import re
+import json
+import pandas as pd
+import gradio as gr
+from rapidfuzz import process, fuzz
+from datasets import load_dataset
+
+# ====================================================
+# CONFIG: columns, states, HF dataset, admin
+# ====================================================
+
+# Expected columns in the master CSVs (must match your HF CSVs)
+MASTER_SCHOOL_COL   = "School_Name__c"
+MASTER_DISTRICT_COL = "School_District__c"
+MASTER_BLOCK_COL    = "School_Block__c"       # optional
+MASTER_UDISE_COL    = "School_Udise_Code__c"
+MASTER_STATE_COL    = "School_State__c"       # optional
+
+# Hugging Face dataset that holds all state CSVs
+HF_SCHOOLS_DATASET = "Apf-AI4Good/Schools"
+
+# Map state keys to CSV filenames inside that dataset
+STATE_HF_FILES = {
+    "ARUNACHAL PRADESH": "Arunachal Pradesh.csv",
+    "ASSAM": "Assam.csv",
+    "BIHAR": "Bihar.csv",
+    "CHHATTISGARH": "Chhattisgarh.csv",
+    "JHARKHAND": "Jharkhand.csv",
+    "MADHYA PRADESH": "Madhya Pradesh.csv",
+    "MANIPUR": "Manipur.csv",
+    "MEGHALAYA": "Meghalaya.csv",
+    "MIZORAM": "Mizoram.csv",
+    "NAGALAND": "Nagaland.csv",
+    "ODISHA": "Odisha.csv",
+    "PUDUCHERRY": "Puducherry.csv",
+    "RAJASTHAN": "Rajasthan.csv",
+    "SIKKIM": "Sikkim.csv",
+    "TELANGANA": "Telangana.csv",
+    "TRIPURA": "Tripura.csv",
+    "UTTAR PRADESH": "Uttar Pradesh.csv",
+    "UTTARAKHAND": "Uttarakhand.csv"
+}
+
+
+# Default state for dropdown
+DEFAULT_STATE_KEY = "ARUNACHAL PRADESH"
+
+# Number of candidates to show in the table
+MAX_CANDIDATES = 5
+
+# JSON file to store patterns
+PATTERN_FILE = "patterns.json"
+
+# Admin password (CHANGE THIS!)
+ADMIN_PASSWORD = "secret"
+
+# Global master df (for currently selected state)
+master_df: pd.DataFrame | None = None
+
+# Global pattern config (loaded from patterns.json)
+pattern_config: dict | None = None
+
+
+# ====================================================
+# PATTERN CONFIG LOAD / SAVE (JSON)
+# ====================================================
+
+DEFAULT_PATTERN_CONFIG = {
+    "global": [],   # start empty; managed via Admin tab
+    "states": {}    # e.g. "KARNATAKA": [{pattern: "...", replacement: "..."}]
+}
+
+
+def load_pattern_config() -> dict:
+    """Load patterns from JSON file, or create empty file."""
+    if os.path.exists(PATTERN_FILE):
+        with open(PATTERN_FILE, "r", encoding="utf-8") as f:
+            return json.load(f)
+
+    # If file not present, seed with empty structure
+    with open(PATTERN_FILE, "w", encoding="utf-8") as f:
+        json.dump(DEFAULT_PATTERN_CONFIG, f, indent=2, ensure_ascii=False)
+    return DEFAULT_PATTERN_CONFIG.copy()
+
+
+def save_pattern_config(config: dict) -> None:
+    with open(PATTERN_FILE, "w", encoding="utf-8") as f:
+        json.dump(config, f, indent=2, ensure_ascii=False)
+
+
+def build_patterns_from_config(config: dict, state_key: str | None):
+    """
+    Convert JSON config into lists of (pattern, replacement) tuples
+    for global and state-specific patterns.
+    """
+    global_list = [(p["pattern"], p["replacement"]) for p in config.get("global", [])]
+
+    state_list = []
+    if state_key:
+        state_key_up = state_key.upper().strip()
+        state_patterns = config.get("states", {}).get(state_key_up, [])
+        state_list = [(p["pattern"], p["replacement"]) for p in state_patterns]
+
+    return global_list, state_list
+
+
+# ====================================================
+# NORMALIZATION
+# ====================================================
+
+def normalize_with_patterns_dynamic(s: str, state_key: str | None) -> str:
+    """Normalize using global + state patterns from pattern_config."""
+    global pattern_config
+
+    if not isinstance(s, str):
+        return ""
+    s = s.upper()
+
+    if pattern_config is None:
+        pattern_config = load_pattern_config()
+
+    global_patterns, state_patterns = build_patterns_from_config(pattern_config, state_key)
+
+    for pat, repl in global_patterns:
+        s = re.sub(pat, repl, s)
+    for pat, repl in state_patterns:
+        s = re.sub(pat, repl, s)
+
+    # Standard final cleaning
+    s = re.sub(r"[^A-Z0-9]+", " ", s)
+    s = re.sub(r"\s+", " ", s).strip()
+    return s
+
+
+# ====================================================
+# DATA LOADING FROM HF
+# ====================================================
+
+def load_master_for_state(state_key: str | None):
+    """
+    Load the master CSV for a state from Hugging Face Datasets,
+    set global master_df, and return District & Block dropdown configs.
+    """
+    global master_df
+
+    if not state_key:
+        master_df = None
+        return (
+            gr.Dropdown(choices=[], value=None),
+            gr.Dropdown(choices=[], value=None),
+        )
+
+    state_key_norm = state_key.upper().strip()
+    if state_key_norm not in STATE_HF_FILES:
+        master_df = None
+        return (
+            gr.Dropdown(choices=[], value=None),
+            gr.Dropdown(choices=[], value=None),
+        )
+
+    csv_filename = STATE_HF_FILES[state_key_norm]
+
+    # Load only that CSV file from the HF dataset repo
+    ds_dict = load_dataset(
+        HF_SCHOOLS_DATASET,
+        data_files={"train": csv_filename},
+    )
+    ds = ds_dict["train"]
+
+    # Convert to pandas and standardize
+    master_df = ds.to_pandas().fillna("")
+
+    # District choices
+    if MASTER_DISTRICT_COL in master_df.columns:
+        districts = sorted(master_df[MASTER_DISTRICT_COL].dropna().unique().tolist())
+        districts = ["All"] + districts
+    else:
+        districts = []
+
+    # Initial blocks (will be refined when district changes)
+    if MASTER_BLOCK_COL in master_df.columns:
+        blocks = ["All"]
+    else:
+        blocks = []
+
+    return (
+        gr.Dropdown(choices=districts, value="All" if districts else None),
+        gr.Dropdown(choices=blocks, value="All" if blocks else None),
+    )
+
+
+def update_blocks(district: str | None):
+    """
+    Update Block dropdown when District changes.
+    """
+    global master_df
+
+    if master_df is None or MASTER_BLOCK_COL not in master_df.columns:
+        return gr.Dropdown(choices=["All"], value="All")
+
+    df = master_df
+    if (
+        district
+        and district != "All"
+        and MASTER_DISTRICT_COL in df.columns
+    ):
+        df = df[df[MASTER_DISTRICT_COL] == district]
+
+    blocks = sorted(df[MASTER_BLOCK_COL].dropna().unique().tolist())
+    blocks = ["All"] + blocks if blocks else ["All"]
+    return gr.Dropdown(choices=blocks, value="All")
+
+
+# ====================================================
+# FUZZY SEARCH
+# ====================================================
+
+def search_candidates(query_name: str, state_key: str | None, district: str | None, block: str | None):
+    """
+    Given school name + state + district + block, return:
+    - candidates table (top N matches)
+    - best-candidate table (single row)
+    """
+    global master_df
+
+    if master_df is None:
+        return pd.DataFrame(), pd.DataFrame()
+
+    query_name = (query_name or "").strip()
+    if not query_name:
+        return pd.DataFrame(), pd.DataFrame()
+
+    df = master_df
+
+    # Filter by district
+    if (
+        district
+        and district != "All"
+        and MASTER_DISTRICT_COL in df.columns
+    ):
+        df = df[df[MASTER_DISTRICT_COL] == district]
+
+    # Filter by block
+    if (
+        block
+        and block != "All"
+        and MASTER_BLOCK_COL in df.columns
+    ):
+        df = df[df[MASTER_BLOCK_COL] == block]
+
+    if df.empty:
+        return pd.DataFrame(), pd.DataFrame()
+
+    state_for_patterns = (state_key or DEFAULT_STATE_KEY).upper().strip()
+
+    choices = df[MASTER_SCHOOL_COL].astype(str)
+
+    candidates_raw = process.extract(
+        query_name,
+        choices,
+        scorer=fuzz.token_set_ratio,
+        processor=lambda s: normalize_with_patterns_dynamic(s, state_for_patterns),
+        limit=MAX_CANDIDATES,
+    )  # (choice, score, key)
+
+    if not candidates_raw:
+        return pd.DataFrame(), pd.DataFrame()
+
+    rows = []
+    for choice_name, score, key in candidates_raw:
+        try:
+            row = df.loc[key]
+        except Exception:
+            continue
+
+        rows.append({
+            "School_Name": row.get(MASTER_SCHOOL_COL, ""),
+            "State": row.get(MASTER_STATE_COL, "") if MASTER_STATE_COL in df.columns else state_for_patterns,
+            "District": row.get(MASTER_DISTRICT_COL, "") if MASTER_DISTRICT_COL in df.columns else "",
+            "Block": row.get(MASTER_BLOCK_COL, "") if MASTER_BLOCK_COL in df.columns else "",
+            "UDISE_Code": row.get(MASTER_UDISE_COL, "") if MASTER_UDISE_COL in df.columns else "",
+            "Score": score,
+        })
+
+    if not rows:
+        return pd.DataFrame(), pd.DataFrame()
+
+    candidates_df = pd.DataFrame(rows)
+    best_df = candidates_df.head(1).copy()
+    return candidates_df, best_df
+
+
+# ====================================================
+# ADMIN / PATTERN CALLBACKS (PASSWORD PROTECTED SAVE)
+# ====================================================
+
+def load_state_patterns_for_editor(selected_state: str | None, new_state_name: str | None):
+    """
+    Load state-specific patterns into the editor dataframe.
+    Does NOT require password (view-only).
+    """
+    global pattern_config
+    if pattern_config is None:
+        pattern_config = load_pattern_config()
+
+    key = None
+    if new_state_name and new_state_name.strip():
+        key = new_state_name.strip().upper()
+    elif selected_state:
+        key = selected_state.strip().upper()
+
+    if not key:
+        return pd.DataFrame(columns=["pattern", "replacement"])
+
+    state_patterns = pattern_config.get("states", {}).get(key, [])
+    if not state_patterns:
+        return pd.DataFrame(columns=["pattern", "replacement"])
+
+    return pd.DataFrame(state_patterns)
+
+
+def save_global_patterns_from_editor(df: pd.DataFrame, password: str):
+    global pattern_config
+    if password != ADMIN_PASSWORD:
+        return "❌ Invalid admin password. Global patterns NOT saved."
+
+    if pattern_config is None:
+        pattern_config = load_pattern_config()
+
+    pattern_config["global"] = df.fillna("").to_dict(orient="records")
+    save_pattern_config(pattern_config)
+    return "✅ Global patterns saved to patterns.json"
+
+
+def save_state_patterns_from_editor(selected_state: str | None, new_state_name: str | None, df: pd.DataFrame, password: str):
+    global pattern_config
+    if password != ADMIN_PASSWORD:
+        return "❌ Invalid admin password. State patterns NOT saved."
+
+    if pattern_config is None:
+        pattern_config = load_pattern_config()
+
+    key = None
+    if new_state_name and new_state_name.strip():
+        key = new_state_name.strip().upper()
+    elif selected_state:
+        key = selected_state.strip().upper()
+
+    if not key:
+        return "⚠️ Please select a state or type a new state key."
+
+    pattern_config.setdefault("states", {})[key] = df.fillna("").to_dict(orient="records")
+    save_pattern_config(pattern_config)
+    return f"✅ Patterns for **{key}** saved to patterns.json"
+
+
+def load_global_patterns_for_editor():
+    """Load global patterns into editor (view-only, no password)."""
+    global pattern_config
+    if pattern_config is None:
+        pattern_config = load_pattern_config()
+    return pd.DataFrame(pattern_config.get("global", []))
+
+
+# ====================================================
+# BUILD GRADIO UI
+# ====================================================
+
+# Initialize pattern_config at startup
+pattern_config = load_pattern_config()
+
+with gr.Blocks(title="State School Fuzzy Matcher") as demo:
+    gr.Markdown(
+        """
+        # State School Fuzzy Matcher (RapidFuzz)
+
+        **Search tab** (public):
+        1. Select **State** (loads master CSV from Hugging Face dataset).
+        2. (Optional) choose **District** and **Block**.
+        3. Type a **School Name** (as on marksheet).
+        4. See top fuzzy-match candidates and the best candidate.
+
+        **Admin tab** (protected for saving):
+        - View / edit **global** and **state-specific** normalization patterns.  
+        - Saving changes requires an **admin password**.
+        """
+    )
+
+    # ------------------------------------------------
+    # TAB: SEARCH (PUBLIC)
+    # ------------------------------------------------
+    with gr.Tab("Search"):
+        with gr.Row():
+            state_dd = gr.Dropdown(
+                label="State",
+                choices=list(STATE_HF_FILES.keys()),
+                value=DEFAULT_STATE_KEY if DEFAULT_STATE_KEY in STATE_HF_FILES else None,
+                interactive=True,
+            )
+
+        with gr.Row():
+            district_dd = gr.Dropdown(label="District", choices=[], value=None, interactive=True)
+            block_dd = gr.Dropdown(label="Block", choices=[], value=None, interactive=True)
+
+        school_input = gr.Textbox(
+            label="Input School Name",
+            placeholder="Type school name from marksheet..."
+        )
+        search_btn = gr.Button("Find Candidates")
+
+        gr.Markdown("### Candidates (top matches)")
+        candidates_table = gr.Dataframe(
+            headers=["School_Name", "State", "District", "Block", "UDISE_Code", "Score"],
+            datatype=["str", "str", "str", "str", "str", "number"],
+            interactive=False
+        )
+
+        gr.Markdown("### Best Candidate")
+        best_table = gr.Dataframe(
+            headers=["School_Name", "State", "District", "Block", "UDISE_Code", "Score"],
+            datatype=["str", "str", "str", "str", "str", "number"],
+            interactive=False
+        )
+
+    # ------------------------------------------------
+    # TAB: ADMIN (PASSWORD-PROTECTED SAVE)
+    # ------------------------------------------------
+    with gr.Tab("Admin"):
+        gr.Markdown(
+            """
+            ### Admin – Pattern configuration
+
+            - **Global patterns**: applied to all states.  
+            - **State-specific patterns**: applied only for that state key.  
+
+            You can **view and edit** patterns freely, but **saving** requires the admin password.
+            """
+        )
+
+        admin_pwd = gr.Textbox(
+            label="Admin Password",
+            type="password",
+            placeholder="Enter admin password to save changes",
+        )
+
+        # ---- Global patterns editor ----
+        gr.Markdown("#### Global Patterns")
+        global_df = gr.Dataframe(
+            value=load_global_patterns_for_editor(),
+            headers=["pattern", "replacement"],
+            datatype=["str", "str"],
+            interactive=True,
+            row_count=(len(pattern_config.get("global", [])) or 1),
+        )
+        save_global_btn = gr.Button("💾 Save Global Patterns")
+        global_status = gr.Markdown("")
+
+        # ---- State patterns editor ----
+        gr.Markdown("#### State-specific Patterns")
+
+        existing_state_keys = sorted(pattern_config.get("states", {}).keys())
+        with gr.Row():
+            state_pattern_dd = gr.Dropdown(
+                label="Existing state key",
+                choices=existing_state_keys,
+                value=existing_state_keys[0] if existing_state_keys else None,
+                interactive=True,
+            )
+            new_state_tb = gr.Textbox(
+                label="Or type new state key",
+                placeholder="e.g. KARNATAKA or TAMIL NADU",
+            )
+
+        state_df = gr.Dataframe(
+            value=pd.DataFrame(columns=["pattern", "replacement"]),
+            headers=["pattern", "replacement"],
+            datatype=["str", "str"],
+            interactive=True,
+            row_count=3,
+        )
+        load_state_btn = gr.Button("Load patterns for selected / new state")
+        save_state_btn = gr.Button("💾 Save State Patterns")
+        state_status = gr.Markdown("")
+
+    # ====================================================
+    # WIRING EVENTS
+    # ====================================================
+
+    # Load master when state changes
+    state_dd.change(
+        fn=load_master_for_state,
+        inputs=state_dd,
+        outputs=[district_dd, block_dd],
+    )
+
+    # Initial load for default state (if any)
+    demo.load(
+        fn=load_master_for_state,
+        inputs=state_dd,
+        outputs=[district_dd, block_dd],
+    )
+
+    # Update blocks when district changes
+    district_dd.change(
+        fn=update_blocks,
+        inputs=district_dd,
+        outputs=block_dd,
+    )
+
+    # Search button
+    search_btn.click(
+        fn=search_candidates,
+        inputs=[school_input, state_dd, district_dd, block_dd],
+        outputs=[candidates_table, best_table],
+    )
+
+    # Admin: save global patterns (password checked inside)
+    save_global_btn.click(
+        fn=save_global_patterns_from_editor,
+        inputs=[global_df, admin_pwd],
+        outputs=global_status,
+    )
+
+    # Admin: load state patterns into editor
+    load_state_btn.click(
+        fn=load_state_patterns_for_editor,
+        inputs=[state_pattern_dd, new_state_tb],
+        outputs=state_df,
+    )
+
+    # Admin: save state patterns (password checked inside)
+    save_state_btn.click(
+        fn=save_state_patterns_from_editor,
+        inputs=[state_pattern_dd, new_state_tb, state_df, admin_pwd],
+        outputs=state_status,
+    )
+
+
+if __name__ == "__main__":
+    demo.launch()