Create searchschool.py

searchschool.py

@@ -0,0 +1,180 @@
+# searchschool.py
+import os
+import pandas as pd
+from huggingface_hub import hf_hub_download
+from rapidfuzz import process, fuzz
+
+# ====================================================
+# CONFIG: columns, states, HF dataset
+# ====================================================
+MASTER_SCHOOL_COL   = "School_Name__c"
+MASTER_DISTRICT_COL = "School_District__c"
+MASTER_BLOCK_COL    = "School_Block__c"
+MASTER_UDISE_COL    = "School_Udise_Code__c"
+MASTER_STATE_COL    = "School_State__c"
+
+HF_SCHOOLS_DATASET = "Apf-AI4Good/Schools"
+
+# Map state keys to CSV filenames inside that dataset
+STATE_HF_FILES = {
+    "ARUNACHAL PRADESH": "Arunachal Pradesh.csv",
+    # add more states as you upload them
+}
+
+DEFAULT_STATE_KEY = "ARUNACHAL PRADESH"
+MAX_CANDIDATES = 5
+
+# global cache
+master_df = None
+
+# You will import normalize_with_patterns_dynamic from admin_patterns when needed
+# to avoid circular imports, main app passes runtime normalization in search_candidates
+try:
+    from admin_patterns import normalize_with_patterns_dynamic
+except Exception:
+    # if admin_patterns isn't importable at module import time, we will import inside functions
+    normalize_with_patterns_dynamic = None
+
+
+def load_master_for_state(state_key: str | None):
+    """
+    Load the master CSV for a state from Hugging Face Hub (dataset repo),
+    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)  # gr referenced in app; kept for signature
+
+    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]
+
+    # Download the CSV file from the dataset repo
+    local_path = hf_hub_download(
+        repo_id=HF_SCHOOLS_DATASET,
+        repo_type="dataset",
+        filename=csv_filename,
+    )
+
+    master_df = pd.read_csv(local_path, dtype=str).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
+    blocks = ["All"] if MASTER_BLOCK_COL in master_df.columns else []
+
+    # Return gr-compatible Dropdown values (constructed in app)
+    # To avoid importing gr here (keeping logic separate), return lists and let app assemble Dropdowns if needed.
+    # However, in our app we directly return gr.Dropdown — so keep compatibility.
+    import gradio as gr  # local import to avoid circular imports at top
+    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
+
+    import gradio as gr
+    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")
+
+
+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, normalize_with_patterns_dynamic
+
+    # import normalize function if not loaded yet (avoids circular import)
+    if normalize_with_patterns_dynamic is None:
+        from admin_patterns import normalize_with_patterns_dynamic  # local import
+        normalize_with_patterns_dynamic = normalize_with_patterns_dynamic
+
+    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,
+        })
+
+    candidates_df = pd.DataFrame(rows)
+    best_df = candidates_df.head(1).copy()
+    return candidates_df, best_df