diff --git "a/app.py" "b/app.py"
--- "a/app.py"
+++ "b/app.py"
@@ -1,1733 +1,1838 @@
-"""
-app.py – School Data Fetcher
-Progressive disclosure UI: each step appears only when it is needed.
-"""
-
-import gradio as gr
-import subprocess
-import sys
-import os
-import re
-import json
-import tempfile
-import pandas as pd
-from district_mapper import (
- build_mapper, build_udise_geo_lookup,
- apply_geo_decode, apply_district_backmap,
- NON_ACTUAL_STATES,
-)
-from hf_store import (
- push_state_file, pull_all_complete_states,
- ALL_REQUIRED_CATEGORIES,
- pull_district_reference, pull_mapping_rules,
- push_district_reference, push_mapping_rules,
- update_district_reference_add, update_district_reference_rename,
- delete_district_reference, upsert_mapping_rule, delete_mapping_rule,
- get_flagged_districts,
- get_hf_credentials, seed_district_reference_from_csv,
- seed_mapping_rules_from_excel,
-)
-
-# ── Auto-push helper (called after successful scrape) ─────────────────────────
-def _auto_push_to_dataset(state_name: str, categories: list) -> str:
- """Automatically push scraped Excel to HF dataset after scrape succeeds."""
- token, repo = get_hf_credentials()
- if not token or not repo:
- return "⚠️ No HF credentials — raw data not pushed to dataset."
- excel_path = get_excel_file(state_name) if state_name else None
- if not excel_path or not os.path.exists(excel_path):
- return "⚠️ Excel not found — push skipped."
- try:
- df = pd.read_excel(excel_path)
- cat_nums = [c.split(" - ")[0].strip() for c in (categories or [])]
- if len(cat_nums) < 7 and "All Categories (1 to 7)" not in categories:
- return "⚠️ Partial scrape completed. Excel downloaded, but NOT pushed to dataset. You must select all 7 categories to push."
- push_state_file(df=df, state_name=state_name, categories_scraped=cat_nums, token=token, repo=repo)
- return f"✅ Auto-pushed raw data for {state_name} to the cloud. Check the Master Sheet tab for any new districts flagged."
- except Exception as ex:
- return f"❌ Auto-push failed: {ex}"
-
-# ── State list ────────────────────────────────────────────────────────────────
-STATE_LABEL_TO_ID = {
- "ANDAMAN & NICOBAR ISLANDS": 135, "ANDHRA PRADESH": 128,
- "ARUNACHAL PRADESH": 112, "ASSAM": 118, "BIHAR": 110,
- "CHANDIGARH": 104, "CHHATTISGARH": 122,
- "DADRA & NAGAR HAVELI AND DAMAN & DIU": 138, "DELHI": 107,
- "GOA": 130, "GUJARAT": 124, "HARYANA": 106, "HIMACHAL PRADESH": 102,
- "IAF EC SOCIETY": 163, "JAMMU & KASHMIR": 101, "JHARKHAND": 120,
- "KARNATAKA": 129, "KENDRIYA VIDYALAYA SANGHATHAN": 192,
- "KERALA": 132, "LADAKH": 137, "LAKSHADWEEP": 131,
- "MADHYA PRADESH": 123, "MAHARASHTRA": 127, "MANIPUR": 114,
- "MEGHALAYA": 117, "MIZORAM": 115, "MSRVVP": 161,
- "NAGALAND": 113, "NAVODAYA VIDYALAYA SAMITI": 193,
- "NAVY EDUCATION SOCIETY": 162, "ODISHA": 121, "PUDUCHERRY": 134,
- "PUNJAB": 103, "RAJASTHAN": 108, "SIKKIM": 111, "TAMILNADU": 133,
- "TELANGANA": 136, "TEST STATE": 199, "TRIPURA": 116,
- "UTTAR PRADESH": 109, "UTTARAKHAND": 105, "WEST BENGAL": 119,
-}
-
-ALL_CATEGORIES = {
- "3": "3 - Pr. with Up.Pr. sec. and H.Sec.",
- "5": "5 - Up. Pr. Secondary and Higher Sec",
- "6": "6 - Pr. Up Pr. and Secondary Only",
- "7": "7 - Upper Pr. and Secondary",
- "8": "8 - Secondary Only",
- "10": "10 - Secondary with Higher Secondary",
- "11": "11 - Higher Secondary only/Jr. College",
-}
-
-OUTPUT_DIR = os.path.join(os.path.dirname(os.path.abspath(__file__)), "output")
-EXCEL_DIR = os.path.join(os.path.dirname(os.path.abspath(__file__)), "output_excel")
-_proc: dict = {"current": None}
-
-# Analysis results are computed on demand (empty until user uploads files)
-_EMPTY_ANALYSIS: dict = {}
-
-
-def _state_prefix(s: str) -> str:
- return re.sub(r"[^a-z0-9]+", "_", s.lower()).strip("_")
-
-def get_output_file(state: str) -> str:
- return os.path.join(OUTPUT_DIR, f"{_state_prefix(state)}_schools_by_category.json")
-
-def get_excel_file(state: str) -> str:
- return os.path.join(EXCEL_DIR, f"{_state_prefix(state)}_Schools.xlsx")
-
-def check_missed_schools(json_path: str):
- if not os.path.exists(json_path):
- return None
- try:
- import json
- with open(json_path, "r", encoding="utf-8") as f:
- data = json.load(f)
- if not isinstance(data, list):
- return None
- def is_invalid(r):
- if r.get("status") == "captcha_failed_all_retries":
- return True
- details = (r.get("response") or {}).get("error", {}).get("errorDetails", {}).get("details", "")
- return isinstance(details, str) and "invalid captcha" in details.lower()
- return sum(1 for r in data if is_invalid(r))
- except Exception:
- return None
-
-
-# ── HTML helpers ──────────────────────────────────────────────────────────────
-
-def _progress_html(current: int, total: int) -> str:
- pct = round(current / total * 100) if total > 0 else 0
- return f"""
-
-
- Progress
- {current} / {total} ({pct}%)
-
-
-
- {"" + str(pct) + "%" if pct > 8 else ""}
-
-
-
- {safe}
-
"""
-
-def _banner_html(icon: str, title: str, body: str, kind: str) -> str:
- """Large status banner shown after a run completes."""
- colors = {
- "success": "var(--color-green-500)",
- "warning": "var(--color-yellow-500)",
- "nodata": "var(--color-blue-500)",
- "stopped": "var(--color-red-500)",
- }
- border_col = colors.get(kind, "var(--border-color-primary)")
- return f"""
-
-
{icon} {title}
-
{body}
-
Ready.
"
-
-def _stopped_html() -> str:
- return "⏹ Stopped by user.
"
-
-
-# ── Log filter ─────────────────────────────────────────────────────────────────
-NOISE = [
- "test session starts", "platform ", "cachedir:", "rootdir:",
- "configfile:", "plugins:", "collected ", "live log call",
- "PASSED", "FAILED", "=== 1 passed", "=== 1 failed", "no tests ran",
- "tests/test_", "====================", "short test summary",
- "UserWarning", "warnings.warn",
-]
-def _is_noise(line: str) -> bool:
- return any(m in line for m in NOISE)
-
-def _clean(line: str) -> str:
- return re.sub(r"(INFO|WARNING|ERROR|DEBUG)\s+\S+:\S+:\d+\s+", "", line).strip()
-
-
-# ── Core streamer ─────────────────────────────────────────────────────────────
-# Outputs (9 values):
-# 0 progress_html
-# 1 phase_html
-# 2 result_html
-# 3 download_btn ← gr.update(value) — raw JSON
-# 4 downloads_row ← gr.update(visible)
-# 5 excel_dl_btn ← gr.update(value) — auto-generated Excel
-# 6 retry_row ← gr.update(visible)
-# 7 map_row ← gr.update(visible) — show Map Districts after success
-# 8 stop_row ← gr.update(visible)
-
-def _stream(pytest_args: list, state: str, max_retries: int, mode: str, target_categories: str = None):
- out_file = get_output_file(state)
- excel_file = get_excel_file(state)
-
- if mode == "scrape" and os.path.exists(excel_file):
- try: os.remove(excel_file)
- except Exception: pass
-
- env = os.environ.copy()
- env["KYS_MAX_RETRIES"] = str(int(max_retries))
- if target_categories:
- env["KYS_TARGET_CATEGORIES"] = target_categories
-
- try:
- process = subprocess.Popen(
- [sys.executable, "-m"] + pytest_args,
- stdout=subprocess.PIPE, stderr=subprocess.STDOUT,
- encoding="utf-8", errors="replace", text=True, bufsize=1, universal_newlines=True,
- cwd=os.path.dirname(os.path.abspath(__file__)),
- env=env,
- )
- except Exception as e:
- import traceback
- err_banner = _banner_html("❌", "Error starting process", str(e), "stopped")
- yield (_idle_html(), _phase_html(f"❌ Error: {e}"), _result_html(err_banner, ""), traceback.format_exc(),
- gr.update(visible=False), gr.update(visible=False),
- gr.update(visible=False, value=None), gr.update(visible=False),
- gr.update(visible=False), gr.update(visible=False), gr.update(visible=False))
- return
-
- _proc["current"] = process
- current, total = 0, 0
- phase_text = "Initialising…"
- log_lines = []
- banner_rendered = ""
- stats_rendered = ""
- is_done = False
- has_failures = False
- success_count = 0
- no_data_count = 0
- failed_count = 0
-
- re_p1 = re.compile(r"Found (\d+) districts")
- re_p2 = re.compile(r"\[(\d+)/(\d+)\] NOW SEARCHING")
- re_retry = re.compile(r"\[(\d+)/(\d+)\] RETRY")
- re_dc = re.compile(r"District: (.+?)\s*\|\s*Category: (.+?)(?:\s*\(id=\d+\))?$")
- re_done_s = re.compile(r"DONE\.\s+Total:\s*(\d+)\s*\|.*?Success:\s*(\d+).*?No Data:\s*(\d+).*?Captcha Failed:\s*(\d+)")
- re_done_r = re.compile(r"RETRY DONE\.\s+Attempted:\s*(\d+)\s*\|.*?Resolved:\s*(\d+).*?Still captcha-failed:\s*(\d+)")
-
- excel_cache = [None]
- def _auto_export_excel():
- """Auto-generate Excel from JSON after scrape completes."""
- excel = get_excel_file(state)
- if excel_cache[0] == excel and os.path.exists(excel):
- return gr.update(value=excel, visible=True)
-
- try:
- proc = subprocess.run(
- [sys.executable, "export_to_excel.py", "--state", state],
- capture_output=True, encoding="utf-8", errors="replace", text=True,
- cwd=os.path.dirname(os.path.abspath(__file__))
- )
- if proc.returncode == 0 and os.path.exists(excel):
- excel_cache[0] = excel
- return gr.update(value=excel, visible=True)
- except Exception:
- pass
- return gr.update(value=None, visible=False)
-
- def _emit():
- prog = _progress_html(current, total) if total > 0 else _idle_html()
- phase = _phase_html(phase_text)
- res = _result_html(banner_rendered, stats_rendered)
- stop_vis = gr.update(visible=not is_done)
-
- if is_done:
- if success_count == 0 and no_data_count > 0 and failed_count == 0:
- dl_row_vis, retry_vis = gr.update(visible=False), gr.update(visible=False)
- elif not has_failures:
- # All data complete — show downloads, hide retry
- dl_row_vis, retry_vis = gr.update(visible=True), gr.update(visible=False)
- else:
- # Some failures — show retry, hide downloads
- dl_row_vis, retry_vis = gr.update(visible=False), gr.update(visible=True)
- else:
- dl_row_vis, retry_vis = gr.update(visible=False), gr.update(visible=False)
-
- dl_val = gr.update(value=out_file) if (is_done and not has_failures and success_count > 0 and os.path.exists(out_file)) else gr.update(value=None)
-
- # Auto-export excel only when scrape succeeds
- if is_done and not has_failures and success_count > 0:
- excel_update = _auto_export_excel()
- else:
- excel_update = gr.update(value=None, visible=False)
-
- return (prog, phase, res, dl_val, dl_row_vis, excel_update, retry_vis, stop_vis)
-
- yield _emit()
- for raw in iter(process.stdout.readline, ""):
- line = raw.rstrip()
- if not line: continue
- clean = _clean(line)
- m = re_p1.search(line)
- if m: phase_text = f"📍 {m.group(1)} districts found. Collecting data…"; total = current = 0
- m2 = re_p2.search(line) or re_retry.search(line)
- if m2:
- current, total = int(m2.group(1)), int(m2.group(2))
- md = re_dc.search(clean)
- phase_text = (f"🔍 [{current}/{total}] {md.group(1).strip()} · {md.group(2).strip()}" if md else f"🔍 [{current}/{total}] Processing…")
- if "PASS Call" in line or "[PASS" in line: phase_text = phase_text.replace("🔍", "✅")
- elif "NO DATA" in line: phase_text = phase_text.replace("🔍", "○")
- elif "ALL ROUNDS FAILED" in line: phase_text = phase_text.replace("🔍", "✗")
- ms = re_done_s.search(line)
- if ms:
- t = int(ms.group(1)); success_count = int(ms.group(2)); no_data_count = int(ms.group(3)); failed_count = int(ms.group(4))
- current = total = t; is_done = True; has_failures = failed_count > 0
- stats_rendered = _stats_html(success_count, no_data_count, failed_count, t)
- if success_count == 0 and failed_count == 0:
- phase_text = "○ No schools found in selected categories."; banner_rendered = _banner_html("○", "No Schools Found", "The selected categories have no schools.", "nodata")
- elif not has_failures:
- phase_text = "✅ All data retrieved!"; banner_rendered = _banner_html("✅", "All Data Retrieved!", "All school data fetched successfully.", "success")
- else:
- phase_text = f"⚠️ {failed_count} record(s) incomplete."; banner_rendered = _banner_html("⚠️", "Some Records Are Incomplete", f"{failed_count} record(s) could not be fetched. Please proceed to Step 2 to fetch missing data.", "warning")
- mr = re_done_r.search(line)
- if mr:
- attempted = int(mr.group(1)); resolved = int(mr.group(2)); still_failed = int(mr.group(3))
- success_count = resolved; failed_count = still_failed; no_data_count = 0; current = total = attempted; is_done = True; has_failures = still_failed > 0
- stats_rendered = _stats_html(resolved, 0, still_failed, attempted)
- if not has_failures:
- phase_text = "✅ All data retrieved!"; banner_rendered = _banner_html("✅", "All Data Retrieved!", "All missing data fetched.", "success")
- else:
- phase_text = f"⚠️ {still_failed} record(s) still incomplete."; banner_rendered = _banner_html("⚠️", "Some Records Still Incomplete", f"{still_failed} record(s) failed. You can run Step 2 again or proceed to Step 3.", "warning")
- yield _emit()
-
- process.stdout.close(); process.wait(); _proc["current"] = None
- if process.returncode not in (0, 1):
- is_done = True; phase_text = "Stopped by user."; banner_rendered = _banner_html("⏹", "Process Stopped", "The process was stopped.", "stopped"); yield _emit()
- else:
- fc = check_missed_schools(out_file)
- if fc is not None:
- if fc == 0 and success_count > 0:
- is_done, has_failures = True, False; phase_text = "✅ All data retrieved!"; banner_rendered = _banner_html("✅", "All Data Retrieved!", "All school data fetched successfully.", "success")
- elif fc > 0:
- is_done, has_failures = True, True; failed_count = fc; phase_text = f"⚠️ {fc} record(s) incomplete."; banner_rendered = _banner_html("⚠️", "Some Records Are Incomplete", f"{fc} record(s) failed. Please proceed to Step 2 to fetch missing data.", "warning")
- yield _emit()
-
-
-# ── Button handlers ───────────────────────────────────────────────────────────
-
-def _err_yield(msg: str):
- # 8 outputs matching stream_outputs
- return (_idle_html(), _phase_html(msg), "",
- gr.update(value=None), gr.update(visible=False),
- gr.update(value=None, elem_classes=["download-highlight", "hide-file"]),
- gr.update(visible=False), gr.update(visible=False))
-
-def ui_main_scrape(state, selected_cats, max_retries):
- if not state: yield _err_yield("⚠️ Please select a state first."); return
- if not selected_cats: yield _err_yield("⚠️ Please select at least one category."); return
- cat_ids = [c.split(" - ")[0] for c in selected_cats]
- yield from _stream(["pytest", "tests/test_scrape_districts_by_category.py", "-v", "-s", "--state", state], state, max_retries, "scrape", ",".join(cat_ids))
-
-def ui_retry(state, max_retries):
- if not state: yield _err_yield("⚠️ Please select a state first."); return
- out = get_output_file(state)
- if not os.path.exists(out): yield _err_yield("⚠️ No data file found."); return
- fc = check_missed_schools(out)
- if fc == 0:
- # Auto-export Excel for complete data
- excel_path = get_excel_file(state)
- excel_update = gr.update(value=None, elem_classes=["download-highlight", "hide-file"])
- try:
- proc = subprocess.run([sys.executable, "export_to_excel.py", "--state", state],
- capture_output=True, encoding="utf-8", errors="replace", text=True,
- cwd=os.path.dirname(os.path.abspath(__file__)))
- if proc.returncode == 0 and os.path.exists(excel_path):
- excel_update = gr.update(value=excel_path, visible=True)
- except Exception: pass
- banner = _banner_html("✅", "All Data Complete!", "Nothing missing — your Excel is ready to download below.", "success")
- yield (_idle_html(), _phase_html("✅ All data is complete!"), _result_html(banner, ""),
- gr.update(value=out), gr.update(visible=True),
- excel_update,
- gr.update(visible=False), gr.update(visible=False))
- return
- yield from _stream(["pytest", "tests/test_retry_districts_by_category.py", "-v", "-s", "--state", state], state, max_retries, "retry")
-
-def ui_stop():
- proc = _proc.get("current")
- if proc and proc.poll() is None:
- proc.terminate()
- try: proc.wait(timeout=5)
- except subprocess.TimeoutExpired: proc.kill()
- _proc["current"] = None
- return _stopped_html()
- return _phase_html("ℹ️ No process is currently running.")
-
-
-# ── Dashboard helpers ─────────────────────────────────────────────────────────
-def _metric_card(label, value, color="#6366f1"):
- return f"""
- """
-
-def _dash_headline_html():
- ns = _ANALYSIS.get("National Summary", pd.DataFrame())
- if ns.empty:
- return "Analysis file not loaded.
"
- vals = dict(zip(ns["Metric"], ns["Count"]))
- old = f"{vals.get('Total Old Schools', 0):,}"
- new = f"{vals.get('Total New Schools', 0):,}"
- added= f"{vals.get('Total Schools Added', 0):,}"
- # compute deleted
- deleted_df = _ANALYSIS.get("All Deleted Schools", pd.DataFrame())
- deleted = f"{len(deleted_df):,}" if not deleted_df.empty else "0"
- return f"""
-
- {_metric_card('Total Schools (Last Year)', old, '#6366f1')}
- {_metric_card('Total Schools (This Year)', new, '#6366f1')}
- {_metric_card('✅ Schools Added', added, '#22c55e')}
- {_metric_card('❌ Schools Removed', deleted, '#ef4444')}
-
"""
-
-def _truncate_udise(val, n=3):
- """Truncate long UDISE lists to first n codes + count."""
- if pd.isna(val) or val == "":
- return ""
- codes = [c.strip() for c in str(val).split(",") if c.strip()]
- if len(codes) <= n:
- return ", ".join(codes)
- shown = ", ".join(codes[:n])
- return f"{shown} ... (+{len(codes)-n} more)"
-
-def _full_udise(val):
- if pd.isna(val) or val == "":
- return ""
- return str(val).strip()
-
-def _filter_state(sheet_name, state_col="State", state=None):
- df = _ANALYSIS.get(sheet_name, pd.DataFrame()).copy()
- if df.empty:
- return df
- if state and state != "All States":
- df = df[df[state_col] == state]
- return df.reset_index(drop=True)
-
-def _prep_with_udise(df, udise_col):
- """Return two versions: one with truncated UDISEs for display, one full."""
- if df.empty or udise_col not in df.columns:
- return df
- df = df.copy()
- df[udise_col] = df[udise_col].apply(_truncate_udise)
- return df
-
-def dash_update(state, analysis):
- """Returns all table data when state filter changes."""
- def _fs(sheet, sc="State"):
- df = analysis.get(sheet, pd.DataFrame()).copy()
- if df.empty: return df
- if state and state != "All States":
- df = df[df[sc] == state]
- return df.reset_index(drop=True)
-
- dm = _prep_with_udise(_fs("District Mapping"), "List_of_UDISEs")
- cs = _prep_with_udise(_fs("Complex Splits"), "List_of_UDISEs")
- bm = _prep_with_udise(_fs("Block Mapping"), "List_of_UDISEs")
- ct = _prep_with_udise(_fs("Category Transitions"), "List_of_UDISEs")
- ad = _fs("All Added Schools", sc="School_State__c")
- dl = _fs("All Deleted Schools", sc="School_State__c")
- return dm, cs, bm, ct, ad, dl
-
-
-def _get_udise_for_row(sheet_name, row_idx, analysis):
- """Look up the full UDISE list for a given row index from live analysis."""
- df = analysis.get(sheet_name, pd.DataFrame())
- if df.empty or row_idx >= len(df) or "List_of_UDISEs" not in df.columns:
- return ""
- return _full_udise(df.iloc[row_idx]["List_of_UDISEs"])
-
-
-def on_select_dm(state, last_row, analysis, evt: gr.SelectData):
- row = evt.index[0]
- if row == last_row: return "", -1
- # filter to current state then get that row's udise
- def _fs(sheet):
- df = analysis.get(sheet, pd.DataFrame()).copy()
- if state and state != "All States": df = df[df["State"] == state]
- return df.reset_index(drop=True)
- df = _fs("District Mapping")
- if df.empty or row >= len(df): return "", row
- return _full_udise(df.iloc[row]["List_of_UDISEs"]), row
-
-def on_select_cs(state, last_row, analysis, evt: gr.SelectData):
- row = evt.index[0]
- if row == last_row: return "", -1
- def _fs(sheet):
- df = analysis.get(sheet, pd.DataFrame()).copy()
- if state and state != "All States": df = df[df["State"] == state]
- return df.reset_index(drop=True)
- df = _fs("Complex Splits")
- if df.empty or row >= len(df): return "", row
- return _full_udise(df.iloc[row]["List_of_UDISEs"]), row
-
-def on_select_bm(state, last_row, analysis, evt: gr.SelectData):
- row = evt.index[0]
- if row == last_row: return "", -1
- def _fs(sheet):
- df = analysis.get(sheet, pd.DataFrame()).copy()
- if state and state != "All States": df = df[df["State"] == state]
- return df.reset_index(drop=True)
- df = _fs("Block Mapping")
- if df.empty or row >= len(df): return "", row
- return _full_udise(df.iloc[row]["List_of_UDISEs"]), row
-
-def on_select_ct(state, last_row, analysis, evt: gr.SelectData):
- row = evt.index[0]
- if row == last_row: return "", -1
- def _fs(sheet):
- df = analysis.get(sheet, pd.DataFrame()).copy()
- if state and state != "All States": df = df[df["State"] == state]
- return df.reset_index(drop=True)
- df = _fs("Category Transitions")
- if df.empty or row >= len(df): return "", row
- return _full_udise(df.iloc[row]["List_of_UDISEs"]), row
-
-# ── Analysis engine (ported from national_analysis.py) ───────────────────────
-def _load_file(path):
- if path is None:
- raise ValueError("No file uploaded.")
- ext = os.path.splitext(path)[-1].lower()
- if ext == ".csv":
- try:
- return pd.read_csv(path, low_memory=False)
- except UnicodeDecodeError:
- return pd.read_csv(path, encoding="cp1252", low_memory=False)
- elif ext in (".xlsx", ".xls"):
- return pd.read_excel(path)
- else:
- raise ValueError(f"Unsupported file type: {ext}. Use .csv or .xlsx")
-
-def _clean_text(s):
- return s.astype(str).str.strip().str.upper()
-
-def _clean_udise(s):
- return s.astype(str).str.replace(r"\.0$", "", regex=True).str.zfill(11)
-
-def _fmt_udise_list(series):
- joined = ", ".join(series.astype(str))
- return joined[:30000] + " ... (+ more)" if len(joined) > 30000 else joined
-
-def run_analysis(old_path, new_path):
- """Run the full comparison. Returns a dict of DataFrames."""
- df_old = _load_file(old_path)
- df_new = _load_file(new_path)
-
- for df in [df_old, df_new]:
- df["School_State__c"] = _clean_text(df["School_State__c"])
- df["UDISE"] = _clean_udise(df["School_Udise_Code__c"])
-
- df_old = df_old.drop_duplicates(subset=["UDISE"]).set_index("UDISE")
- df_new = df_new.drop_duplicates(subset=["UDISE"]).set_index("UDISE")
-
- old_u = set(df_old.index)
- new_u = set(df_new.index)
- added_u = new_u - old_u
- deleted_u = old_u - new_u
- common_u = old_u & new_u
-
- df_added_raw = df_new.loc[list(added_u)].reset_index()[["UDISE","School_State__c","School_District__c","School_Block__c"]]
- df_deleted_raw = df_old.loc[list(deleted_u)].reset_index()[["UDISE","School_State__c","School_District__c","School_Block__c"]]
-
- all_states = sorted(set(df_old["School_State__c"].unique()) | set(df_new["School_State__c"].unique()))
- state_summary = []
- for st in all_states:
- so = set(df_old[df_old["School_State__c"] == st].index)
- sn = set(df_new[df_new["School_State__c"] == st].index)
- state_summary.append({"State": st, "Old Count": len(so), "New Count": len(sn),
- "Added": len(sn - so), "Deleted": len(so - sn)})
- df_state_summary = pd.DataFrame(state_summary)
-
- df_common = df_old.loc[list(common_u)].join(df_new.loc[list(common_u)], lsuffix="_OLD", rsuffix="_NEW").reset_index()
- df_common["State"] = _clean_text(df_common["School_State__c_NEW"])
- df_common["Dist_OLD"] = _clean_text(df_common["School_District__c_OLD"])
- df_common["Dist_NEW"] = _clean_text(df_common["School_District__c_NEW"])
- df_common["Block_OLD"]= _clean_text(df_common["School_Block__c_OLD"])
- df_common["Block_NEW"]= _clean_text(df_common["School_Block__c_NEW"])
- df_common["Cat_OLD"] = _clean_text(df_common["schCategoryType__c_OLD"])
- df_common["Cat_NEW"] = _clean_text(df_common["schCategoryType__c_NEW"])
-
- dist_chg = df_common[df_common["Dist_OLD"] != df_common["Dist_NEW"]]
- district_mapping = pd.DataFrame()
- if len(dist_chg):
- district_mapping = dist_chg.groupby(["State","Dist_OLD","Dist_NEW"]).agg(
- Affected_Schools=("UDISE","count"), List_of_UDISEs=("UDISE", _fmt_udise_list)
- ).reset_index()
-
- # 3. Block to District Mapping (The Exact Matcher for Splits)
- # If a new district was formed, we want to know what the old district was for a given block.
- # So we group by (State, Dist_NEW, Block_NEW) and find the most common Dist_OLD.
- block_to_district = pd.DataFrame()
- if not df_common.empty:
- # Group by new block and old district
- block_dist_counts = df_common.groupby(["State", "Dist_NEW", "Block_NEW", "Dist_OLD"]).agg(
- Schools=("UDISE", "count"),
- List_of_UDISEs=("UDISE", _fmt_udise_list)
- ).reset_index()
-
- # For each new block, find the old district that had the most schools
- # (This establishes the definitive historical mapping for that block)
- idx = block_dist_counts.groupby(["State", "Dist_NEW", "Block_NEW"])["Schools"].idxmax()
- block_to_district = block_dist_counts.loc[idx].reset_index(drop=True)
-
- blk_chg = df_common[df_common["Block_OLD"] != df_common["Block_NEW"]]
- block_mapping = pd.DataFrame()
- if len(blk_chg):
- block_mapping = blk_chg.groupby(["State","Dist_OLD","Block_OLD","Block_NEW"]).agg(
- Affected_Schools=("UDISE","count"), List_of_UDISEs=("UDISE", _fmt_udise_list)
- ).reset_index()
-
- cat_chg = df_common[df_common["Cat_OLD"] != df_common["Cat_NEW"]]
- cat_migrations = pd.DataFrame()
- if len(cat_chg):
- cat_migrations = cat_chg.groupby(["State","Cat_OLD","Cat_NEW"]).agg(
- Count=("UDISE","count"), List_of_UDISEs=("UDISE", _fmt_udise_list)
- ).reset_index().sort_values(["State","Count"], ascending=[True,False])
-
- ns = pd.DataFrame({"Metric": [
- "Total Old Schools","Total New Schools","Total Schools Added","Total Schools Deleted",
- "Common Schools (In Both)","Schools with District Changes","Schools with Block Changes","Schools with Category Changes"
- ], "Count": [
- len(df_old), len(df_new), len(added_u), len(deleted_u),
- len(common_u), len(dist_chg), len(blk_chg), len(cat_chg)
- ]})
-
- from district_mapper import build_udise_geo_lookup
- geo_lookup = build_udise_geo_lookup(df_new)
- rows = []
- for ss, name in geo_lookup["state"].items(): rows.append(["state", ss, name])
- for ssdd, name in geo_lookup["district"].items(): rows.append(["district", ssdd, name])
- for ssddbb, name in geo_lookup["block"].items(): rows.append(["block", ssddbb, name])
- geo_df = pd.DataFrame(rows, columns=["Type", "Code", "Name"])
-
- # Find unmapped completely new districts
- old_dist_names = set(df_old["School_District__c"].dropna().str.strip().str.upper().unique())
- new_dist_df = df_new[["School_State__c", "School_District__c"]].drop_duplicates().dropna()
- new_dist_df["State"] = _clean_text(new_dist_df["School_State__c"])
- new_dist_df["New District Name"] = _clean_text(new_dist_df["School_District__c"])
-
- unmapped_new = new_dist_df[~new_dist_df["New District Name"].isin(old_dist_names)].copy()
- unmapped_new["Rename To (Type here...)"] = ""
- if not unmapped_new.empty:
- unmapped_new = unmapped_new[["State", "New District Name", "Rename To (Type here...)"]].sort_values(["State", "New District Name"]).reset_index(drop=True)
- else:
- unmapped_new = pd.DataFrame(columns=["State", "New District Name", "Rename To (Type here...)"])
-
- return {
- "National Summary": ns,
- "State by State Breakdown": df_state_summary,
- "All Added Schools": df_added_raw,
- "All Deleted Schools": df_deleted_raw,
- "District Mapping": district_mapping,
- "Block to District": block_to_district,
- "Block Mapping": block_mapping,
- "Category Transitions": cat_migrations,
- "New Districts": unmapped_new,
- "UDISE Geo": geo_df,
- }
-
-# ── CSS ───────────────────────────────────────────────────────────────────────
-css = """
-@import url('https://fonts.googleapis.com/css2?family=Inter:wght@400;500;600;700;800&display=swap');
-
-* { font-family: 'Inter', sans-serif !important; }
-
-#title {
- text-align: center;
- font-size: 2.2em !important;
- font-weight: 800 !important;
- background: linear-gradient(135deg, #6366f1, #8b5cf6, #a855f7);
- -webkit-background-clip: text;
- -webkit-text-fill-color: transparent;
- background-clip: text;
- margin-bottom: 4px !important;
- letter-spacing: -0.5px;
-}
-#subtitle {
- text-align: center;
- color: var(--body-text-color-subdued);
- margin-top: 0;
- margin-bottom: 24px;
- font-size: .95em;
-}
-
-/* Tab styling */
-.tab-nav button {
- font-weight: 600 !important;
- font-size: 1em !important;
- padding: 10px 20px !important;
- border-radius: 10px 10px 0 0 !important;
- transition: all 0.2s !important;
-}
-.tab-nav button.selected {
- background: linear-gradient(135deg, #6366f1, #8b5cf6) !important;
- color: white !important;
-}
-
-/* Step card styling */
-.step-card {
- background: var(--background-fill-secondary);
- border-radius: 14px;
- padding: 20px;
- border: 1px solid var(--border-color-primary);
- margin-bottom: 12px;
- transition: box-shadow 0.2s;
-}
-.step-card:hover { box-shadow: 0 4px 20px rgba(99,102,241,0.12); }
-
-/* Button beautification */
-.btn-primary { background: linear-gradient(135deg, #6366f1, #8b5cf6) !important; border: none !important; }
-.btn-primary:hover { transform: translateY(-1px) !important; box-shadow: 0 6px 20px rgba(99,102,241,0.4) !important; }
-
-/* Download file widget */
-.download-highlight {
- border: 2px dashed var(--color-accent) !important;
- border-radius: 12px !important;
- padding: 0 !important;
- position: relative !important;
- background: transparent !important;
- transition: all 0.25s !important;
-}
-.download-highlight:hover {
- border-style: solid !important;
- border-color: var(--color-accent) !important;
- background: color-mix(in srgb, var(--color-accent) 5%, transparent) !important;
-}
-.download-highlight * { position: static !important; }
-.download-highlight a { display: flex !important; align-items: center; padding: 14px !important; width: 100%; height: 100%; }
-.download-highlight a::after { content: ""; position: absolute !important; inset: 0 !important; z-index: 50 !important; cursor: pointer !important; }
-.hide-file { display: none !important; }
-
-/* Status banners */
-.status-ok { color: #22c55e; font-weight: 700; }
-.status-err { color: #ef4444; font-weight: 700; }
-.status-warn{ color: #f59e0b; font-weight: 700; }
-
-/* Accordion styling */
-.gr-accordion { border-radius: 12px !important; border: 1px solid var(--border-color-primary) !important; margin-bottom: 10px !important; }
-
-/* Section divider label */
-.section-label {
- font-size: 0.75em;
- font-weight: 600;
- text-transform: uppercase;
- letter-spacing: 1px;
- color: var(--body-text-color-subdued);
- margin: 20px 0 8px 0;
-}
-
-/* Metric cards */
-.metric-card {
- background: transparent;
- border-radius: 14px;
- padding: 20px 28px;
- text-align: center;
- transition: transform 0.2s;
-}
-.metric-card:hover { transform: translateY(-2px); }
-
-/* Log box */
-.log-box textarea { font-size:.76em !important; font-family: 'JetBrains Mono', monospace !important; }
-"""
-
-custom_theme = gr.themes.Soft(
- primary_hue="violet",
- secondary_hue="indigo",
- neutral_hue="slate",
- font=[gr.themes.GoogleFont("Inter"), "ui-sans-serif", "system-ui", "sans-serif"]
-)
-
-# ── UI ────────────────────────────────────────────────────────────────────────
-with gr.Blocks(title="School Data Fetcher", css=css, theme=custom_theme) as app:
- gr.HTML("""
-
-
- 🏫 School Data Fetcher
-
-
- Scrape → Auto-push Raw → Map Districts → Build Mapped Master
-
-
-
👋 How it works
-
- Step 1: Choose a state & categories, then click Start Scraping — school data is collected and automatically saved to the cloud.
- Step 2: If anything was missed, click Fix Missing Data to retry those records only.
- Step 3: Repeat for all states. Then go to the 📋 Master Sheet tab to review any newly detected districts and map them if necessary.
- Step 4: Once all states are scraped and mapped, click Build Master Sheet at the bottom of the tab to generate the final mapped excel file.
-
-
📍 Location
")
- state_dd = gr.Dropdown(
- choices=sorted(STATE_LABEL_TO_ID.keys()),
- label="Select a State or School Type",
- info="Includes regular states and pan-India school organisations (KVS, NVS, NAVY, IAF)"
- )
- with gr.Column(scale=2):
- gr.HTML("📚 School Types
")
- categories_cg = gr.CheckboxGroup(
- choices=list(ALL_CATEGORIES.values()),
- value=list(ALL_CATEGORIES.values()),
- label="Select which school categories to collect"
- )
-
- gr.HTML("")
-
- with gr.Row():
- scrape_btn = gr.Button("▶ Start Scraping", variant="primary", scale=3)
- stop_btn_placeholder = gr.HTML("", visible=False) # placeholder
-
- with gr.Column(visible=False) as confirm_col:
- gr.HTML("⚠️ Warning: This state (and all selected categories) is already present in your dataset. Are you sure you want to scrape it again?
")
- with gr.Row():
- confirm_yes_btn = gr.Button("Yes, Scrape Again", variant="stop")
- confirm_no_btn = gr.Button("No, Cancel", variant="secondary")
-
- with gr.Row(visible=False) as stop_row:
- stop_btn = gr.Button("⏹ Stop", variant="stop", scale=1)
-
- should_scrape_state = gr.State(False)
-
- progress_html = gr.HTML(value=_idle_html())
- phase_html = gr.HTML(value=_phase_html("Select a state and school types above, then click Start Scraping."))
- result_html = gr.HTML(value="")
-
- with gr.Row(visible=False) as retry_row:
- retry_btn = gr.Button("↻ Fix Missing Data (some records failed — click to retry)", variant="secondary", scale=1)
-
- with gr.Row(visible=False) as downloads_row:
- with gr.Column(scale=1):
- download_btn = gr.DownloadButton(
- label="📄 Download Raw JSON",
- value=None, visible=True,
- variant="secondary"
- )
- with gr.Column(scale=1):
- excel_dl_btn = gr.DownloadButton(
- label="📊 Download Excel",
- value=None, visible=False,
- variant="secondary"
- )
-
- with gr.Accordion("⚙️ Advanced Settings", open=False):
- retries_slider = gr.Slider(
- minimum=1, maximum=10, value=5, step=1,
- label="Retry attempts per failed record (higher = more thorough but slower)"
- )
-
-
- # Auto-push status (shown after scrape succeeds)
- auto_push_html = gr.HTML(value="")
-
- stream_outputs = [
- progress_html, phase_html, result_html,
- download_btn, downloads_row,
- excel_dl_btn,
- retry_row, stop_row,
- ]
-
- def handle_scrape_click(state_name, categories):
- from hf_store import check_state_scraped
- if not state_name:
- return gr.update(), gr.update(), False
- if check_state_scraped(state_name, categories):
- return gr.update(visible=False), gr.update(visible=True), False
- return gr.update(visible=False), gr.update(visible=False), True
-
- def ui_main_scrape_conditional(should_scrape, state, cats, retries):
- if should_scrape:
- yield from ui_main_scrape(state, cats, retries)
- else:
- yield tuple(gr.update() for _ in range(8))
-
- def do_auto_push_after_scrape(state, cats):
- """Auto-push to dataset after successful scrape and update status."""
- yield "⏳ Pushing data to HuggingFace dataset... please wait."
- res = _auto_push_to_dataset(state, cats)
- yield res
-
- scrape_btn.click(
- fn=handle_scrape_click,
- inputs=[state_dd, categories_cg],
- outputs=[scrape_btn, confirm_col, should_scrape_state]
- ).then(
- fn=ui_main_scrape_conditional,
- inputs=[should_scrape_state, state_dd, categories_cg, retries_slider],
- outputs=stream_outputs
- ).then(
- fn=do_auto_push_after_scrape,
- inputs=[state_dd, categories_cg],
- outputs=[auto_push_html]
- ).then(
- fn=lambda should: gr.update(visible=True) if should else gr.update(),
- inputs=[should_scrape_state],
- outputs=[scrape_btn]
- )
-
- confirm_yes_btn.click(
- fn=lambda: (gr.update(visible=False), gr.update(visible=False)),
- outputs=[confirm_col, scrape_btn]
- ).then(
- fn=ui_main_scrape,
- inputs=[state_dd, categories_cg, retries_slider],
- outputs=stream_outputs
- ).then(
- fn=do_auto_push_after_scrape,
- inputs=[state_dd, categories_cg],
- outputs=[auto_push_html]
- ).then(
- fn=lambda: gr.update(visible=True),
- outputs=[scrape_btn]
- )
-
- confirm_no_btn.click(
- fn=lambda: (gr.update(visible=False), gr.update(visible=True)),
- outputs=[confirm_col, scrape_btn]
- )
- retry_btn.click( fn=ui_retry, inputs=[state_dd, retries_slider], outputs=stream_outputs)
- stop_btn.click( fn=ui_stop, inputs=[], outputs=[progress_html])
-
- # ── Tab 2: Master Sheet Builder ───────────────────────────────────────
- with gr.Tab("📋 Master Sheet"):
-
- # ── HF connection status ──────────────────────────────────────────
- def _env_status():
- from hf_store import get_hf_credentials
- token, repo = get_hf_credentials()
- if token and repo:
- repo_name = repo.split("/")[-1] if "/" in repo else repo
- return f"""
-
-
✅
-
-
Connected to Dataset
-
Saving to: {repo}
-
-
-
⚠️
-
-
Dataset not set up
-
- Add your HF_TOKEN and HF_REPO to the .env file to enable dataset features.
-
-
-
📡 Dataset Coverage
-
- Only states with all 7 required school categories scraped will be included in the master sheet.
-
- """)
-
- with gr.Row():
- refresh_btn = gr.Button("🔄 Refresh from Dataset", variant="secondary", scale=1)
-
- coverage_table = gr.Dataframe(
- label="State Coverage (Click '🗑️ Delete' in the Action column to remove a state)",
- interactive=False, wrap=True,
- headers=["State", "Categories Scraped", "Complete?", "Last Scraped", "Action"],
- )
- coverage_summary_html = gr.HTML(value="")
-
- # ── Flagged Districts & Mapping Rules ─────────────────────────
- with gr.Accordion("Review New Districts", open=False):
- gr.HTML("Review newly scraped districts and map them to older SF districts or rename.
")
-
- with gr.Row():
- refresh_flags_btn = gr.Button("🔄 Refresh Flagged Districts", variant="secondary", scale=1)
- flagged_table = gr.Dataframe(
- label="New districts detected (Type the Old SF name in the last column to rename it)",
- headers=["State", "District", "Reason", "Rename District (Optional)"],
- interactive=True, wrap=True
- )
- flagged_status_html = gr.HTML(value="")
-
-
-
- gr.Markdown("---")
- build_master_btn = gr.Button("📋 Build Master Sheet", variant="primary")
- build_status_html = gr.HTML(value="")
- master_dl_btn = gr.DownloadButton(
- label="⬇ Download Master Excel",
- visible=False,
- variant="secondary"
- )
-
- # ── Callback: Refresh coverage table ─────────────────────────────
- coverage_state = gr.State(pd.DataFrame())
-
- def ui_refresh_coverage():
- from hf_store import get_hf_credentials
- token, repo = get_hf_credentials()
- if not token or not repo:
- return (
- pd.DataFrame(columns=["State", "Categories Scraped", "Complete?", "Last Scraped", "Action"]),
- "⚠️ HF_TOKEN or HF_REPO not set in environment.
"
- )
- try:
- from hf_store import pull_all_complete_states
- _, partial_info, coverage_df = pull_all_complete_states(token, repo)
- if not coverage_df.empty:
- coverage_df["Action"] = "🗑️ Delete"
- complete_count = int((coverage_df["Complete?"] == "✅ Yes").sum()) if not coverage_df.empty else 0
- total = len(coverage_df)
- summary = (
- f""
- f"✅ {complete_count} states ready to build
"
- )
- return coverage_df, summary
- except Exception as ex:
- return (
- pd.DataFrame(columns=["State", "Categories Scraped", "Complete?", "Last Scraped", "Action"]),
- f"❌ Error: {ex}
"
- )
-
- refresh_btn.click(
- fn=ui_refresh_coverage,
- inputs=[],
- outputs=[coverage_table, coverage_summary_html],
- ).then(
- fn=lambda df: df,
- inputs=[coverage_table],
- outputs=[coverage_state]
- )
-
- def on_coverage_select(evt: gr.SelectData, current_df):
- if evt.value != "🗑️ Delete":
- return gr.update(), gr.update()
-
- row_idx = evt.index[0]
- if row_idx >= len(current_df):
- return gr.update(), gr.update()
-
- state_name = current_df.iloc[row_idx]["State"]
- from hf_store import get_hf_credentials, delete_state_file
- token, repo = get_hf_credentials()
- try:
- delete_state_file(state_name, token, repo)
- except Exception as ex:
- return gr.update(), f"❌ Error deleting: {ex}
"
-
- new_df, new_summary = ui_refresh_coverage()
- msg = f"✅ Successfully deleted {state_name}.
" + new_summary
- return new_df, msg
-
- coverage_table.select(
- fn=on_coverage_select,
- inputs=[coverage_state],
- outputs=[coverage_table, coverage_summary_html]
- ).then(
- fn=lambda df: df,
- inputs=[coverage_table],
- outputs=[coverage_state]
- )
-
- # ── Callbacks: Flagged Districts & Mapping Rules ──────────────────────
- def ui_refresh_flags():
- from hf_store import get_hf_credentials, get_flagged_districts
- token, repo = get_hf_credentials()
- if not token or not repo:
- return pd.DataFrame(columns=["State", "District", "Reason"]), "⚠️ HF credentials not set.
"
- flags = get_flagged_districts(token, repo)
- if not flags:
- return pd.DataFrame(columns=["State", "District", "Reason"]), "✅ No flagged districts — all clear!
"
- df = pd.DataFrame(flags)
- df = df[["state", "district", "reason"]].rename(columns={"state": "State", "district": "District", "reason": "Reason"})
- df["Rename District (Optional)"] = ""
- return df, f"⚠️ {len(flags)} district(s) need review.
"
-
- refresh_flags_btn.click(fn=ui_refresh_flags, outputs=[flagged_table, flagged_status_html])
-
- # ── Callback: Build master sheet ──────────────────────────────────
- def ui_build_master(flagged_df):
- from district_mapper import build_udise_geo_lookup, apply_geo_decode, build_mapper, apply_district_backmap
- from hf_store import get_hf_credentials, pull_all_complete_states, pull_mapping_rules, push_mapped_master, upsert_mapping_rule, rename_district_in_reference
- token, repo = get_hf_credentials()
- if not token or not repo:
- return (
- "⚠️ HF_TOKEN or HF_REPO not set in environment.
",
- gr.update(elem_classes=["download-highlight", "hide-file"]),
- )
- try:
- # 1. Process inline table mappings
- if flagged_df is not None and not flagged_df.empty:
- for _, row in flagged_df.iterrows():
- mapped = str(row.get("Rename District (Optional)", "")).strip()
- state = str(row.get("State", "")).strip().upper()
- dist = str(row.get("District", "")).strip().upper()
- if mapped and mapped.upper() != "NAN" and mapped != "None":
- upsert_mapping_rule(state, dist, mapped.upper(), 0, token, repo)
- rename_district_in_reference(state, dist, mapped.upper(), token, repo)
- complete_dfs, _, _ = pull_all_complete_states(token, repo)
-
- if not complete_dfs:
- return (
- "⚠️ No states with all 7 categories found in dataset.
",
- gr.update(elem_classes=["download-highlight", "hide-file"]),
- )
-
- from hf_store import push_national_analysis_report
- master_df = pd.concat(complete_dfs, ignore_index=True)
- geo_lookup = build_udise_geo_lookup(master_df)
- master_df, geo_changes_df = apply_geo_decode(master_df, geo_lookup)
-
- # Pull mapping rules automatically
- dist_rules, block_rules = pull_mapping_rules(token, repo)
-
- backmap_changes_df = pd.DataFrame()
- if not dist_rules.empty:
- analysis_dict = {
- "District Mapping": dist_rules,
- "Block to District": block_rules if not block_rules.empty else pd.DataFrame(),
- }
- mapper = build_mapper(analysis_dict)
- master_df, backmap_changes_df = apply_district_backmap(master_df, mapper)
-
- # Combine reports and push
- combined_changes = pd.concat([geo_changes_df, backmap_changes_df], ignore_index=True)
- if not combined_changes.empty:
- push_national_analysis_report(combined_changes, token, repo)
-
- udise_col = "School_Udise_Code__c" if "School_Udise_Code__c" in master_df.columns else "UDISE"
- if udise_col in master_df.columns and "Scraped_Date" in master_df.columns:
- master_df = (
- master_df
- .sort_values("Scraped_Date", ascending=False)
- .drop_duplicates(subset=[udise_col], keep="first")
- .reset_index(drop=True)
- )
-
- # Filter out out-of-scope states (e.g., from Geo-Decoded Navy/KVS schools)
- from hf_store import pull_district_reference
- ref_df = pull_district_reference(token, repo)
- ref_states = set(ref_df["State"].str.strip().str.upper()) if not ref_df.empty else set()
-
- allowed_states = {
- "ARUNACHAL PRADESH", "ASSAM", "BIHAR", "CHHATTISGARH", "JHARKHAND",
- "KARNATAKA", "MADHYA PRADESH", "MANIPUR", "MEGHALAYA", "MIZORAM",
- "NAGALAND", "ODISHA", "PUDUCHERRY", "RAJASTHAN", "SIKKIM",
- "TELANGANA", "TRIPURA", "UTTAR PRADESH", "UTTARAKHAND", "DELHI",
- "ANDHRA PRADESH"
- }
- allowed_states.update(ref_states)
-
- state_col = "School_State__c" if "School_State__c" in master_df.columns else "State"
- if state_col in master_df.columns:
- master_df = master_df[master_df[state_col].str.strip().str.upper().isin(allowed_states)]
- master_df = master_df.sort_values(state_col, ascending=True).reset_index(drop=True)
-
- from datetime import datetime, timedelta
- ist_now = datetime.utcnow() + timedelta(hours=5, minutes=30)
- date_str = ist_now.strftime("%Y_%b_%d_%I_%M_%p").lower()
- os.makedirs(EXCEL_DIR, exist_ok=True)
- master_path = os.path.join(EXCEL_DIR, f"mapped_master_{date_str}.xlsx")
- master_df.to_excel(master_path, index=False)
-
- push_mapped_master(master_df, token, repo)
-
- # Auto-sync aliases CSV immediately after building a new master!
- try:
- from alias_sync import sync_aliases
- sync_aliases(token, repo)
- except Exception as e:
- print(f"Failed to auto-sync aliases: {e}")
-
- n = len(master_df)
- msg = (
- f""
- f"✅ Mapped master built! {n:,} records from {len(complete_dfs)} complete states. "
- f"District mapping applied. Also pushed to dataset."
- f"
"
- )
- return msg, gr.update(value=master_path, visible=True)
- except Exception as ex:
- return (
- f"❌ Error: {ex}
",
- gr.update(visible=False),
- )
-
- build_master_btn.click(
- fn=ui_build_master,
- inputs=[flagged_table],
- outputs=[build_status_html, master_dl_btn],
- )
-
- # ── Tab 3: Mapping Manager ─────────────────────────────────────────────
- with gr.Tab("🗺️ Mapping Manager"):
- gr.HTML("""
-
-
🗺️ Mapping Manager
-
- Scholarship Application Reference: The mirror of what Scholarship Application currently knows. When you add a new district to Scholarship Application, find it here and change its status from new districts found to present.
-
-
📁 Scholarship Application District Reference
")
- with gr.Row():
- refresh_ref_btn = gr.Button("🔄 Load from Dataset", variant="secondary", scale=1)
-
- # Filters
- with gr.Row():
- ref_filter_state = gr.Dropdown(label="Filter by State", choices=["All"], value="All", scale=1, interactive=True)
- ref_filter_status = gr.Dropdown(label="Filter by Status", choices=["All", "present", "new districts found"], value="All", scale=1, interactive=True)
- ref_filter_dist = gr.Textbox(label="Search District", placeholder="Type to search...", scale=2, interactive=True)
-
- # The DataFrame
- ref_full_state = gr.State(pd.DataFrame())
- ref_table = gr.Dataframe(
- label="Scholarship Application District Reference (Editable)",
- interactive=False, wrap=True
- )
-
- with gr.Accordion("📦 Bulk Excel Import / Export", open=False):
- with gr.Row():
- ref_prep_dl_btn = gr.Button("📦 Prepare Current View for Download")
- ref_dl_btn = gr.DownloadButton("📥 Download Excel", visible=False)
-
- with gr.Row():
- ref_upload = gr.File(label="Upload Updated Excel", file_types=[".xlsx"])
-
- gr.HTML("Click any row in the table above to edit its status here:
")
- with gr.Row():
- ref_edit_state = gr.Textbox(label="State", interactive=False, scale=2)
- ref_edit_dist = gr.Textbox(label="District", interactive=False, scale=2)
- ref_edit_status = gr.Dropdown(choices=["present", "new districts found"], label="Status", scale=1)
- ref_edit_btn = gr.Button("💾 Update Row", variant="primary", scale=1)
-
- ref_status_html = gr.HTML(value="")
-
- gr.HTML("Add a District
")
- with gr.Row():
- ref_state_add = gr.Textbox(label="State", placeholder="e.g. ANDHRA PRADESH", scale=2)
- ref_dist_add = gr.Textbox(label="District", placeholder="e.g. ALLURI SITHARAMA RAJU", scale=2)
- ref_status_add = gr.Dropdown(choices=["present", "new districts found"], value="present", label="Status", scale=1)
- ref_add_btn = gr.Button("➕ Add District", variant="primary", scale=1)
- ref_add_status = gr.HTML(value="")
-
- gr.HTML("Rename a District (automatically updates mappings)
")
- with gr.Row():
- ref_ren_state = gr.Dropdown(label="State", choices=[], allow_custom_value=True, scale=2)
- ref_ren_old = gr.Dropdown(label="Old Name", choices=[], allow_custom_value=True, scale=2)
- ref_ren_new = gr.Textbox(label="New Name", placeholder="e.g. VISAKHAPATNAM (NEW)", scale=2)
- ref_rename_btn = gr.Button("✏️ Rename", variant="secondary", scale=1)
- ref_rename_status = gr.HTML(value="")
-
- gr.HTML("Delete Actions
")
- with gr.Row():
- ref_del_state = gr.Dropdown(label="State", choices=[], allow_custom_value=True, scale=2)
- ref_del_dist = gr.Dropdown(label="District", choices=[], allow_custom_value=True, scale=2)
- ref_del_dist_btn = gr.Button("🗑️ Delete District", variant="stop", scale=1)
- ref_del_state_btn = gr.Button("🚨 Delete Entire State", variant="stop", scale=1)
- ref_del_status = gr.HTML(value="")
-
- def on_ref_table_select(df, evt: gr.SelectData):
- row = evt.index[0]
- if df is None or df.empty or row >= len(df):
- return gr.update(), gr.update(), gr.update(), ""
-
- state = df.iloc[row]["State"]
- dist = df.iloc[row]["District"]
- status = df.iloc[row]["Status"]
- return gr.update(value=state), gr.update(value=dist), gr.update(value=status), ""
-
- def ui_ref_edit_row(state, district, new_status):
- from hf_store import get_hf_credentials, update_district_reference_add
- token, repo = get_hf_credentials()
- if not token or not repo or not state or not district:
- return "⚠️ Please select a row from the table first.
"
- try:
- update_district_reference_add(state, district, new_status, token, repo)
- return f"✅ Successfully updated {district} to '{new_status}'. Refresh the table to see changes.
"
- except Exception as ex:
- return f"❌ Error: {ex}
"
-
- def ui_refresh_ref():
- from hf_store import get_hf_credentials, pull_district_reference
- token, repo = get_hf_credentials()
- if not token or not repo:
- empty = pd.DataFrame()
- u = gr.update()
- return empty, empty, u, u, u, u, u, "⚠️ HF credentials not set.
"
- df = pull_district_reference(token, repo)
- if df.empty:
- empty = pd.DataFrame()
- u = gr.update()
- return empty, empty, u, u, u, u, u, "⚠️ Scholarship Application Reference empty.
"
- states = sorted(list(df["State"].unique()))
- dists = sorted(list(df["District"].unique()))
- st_up = gr.update(choices=["All"] + states, value="All")
- st_dd = gr.update(choices=states)
- dist_dd = gr.update(choices=dists)
- return df, df, st_up, st_dd, dist_dd, st_dd, dist_dd, f"✅ Loaded {len(df)} rows.
"
-
- def filter_ref_table(df, filter_state, filter_status, search_dist):
- if df is None or df.empty: return df
- res = df.copy()
- if filter_state and filter_state != "All":
- res = res[res["State"] == filter_state]
- if filter_status and filter_status != "All":
- res = res[res["Status"] == filter_status]
- if search_dist:
- res = res[res["District"].str.contains(search_dist.upper(), na=False)]
- return res
-
- def ui_save_table_edits(df, full_df, filter_state, filter_status, search_dist):
- from hf_store import get_hf_credentials, push_district_reference
- token, repo = get_hf_credentials()
- if df is None or df.empty:
- return "⚠️ Table is empty.
", full_df
- # The user edited 'df'. We need to merge it back into 'full_df'
- # This is a bit complex if they deleted rows or changed keys.
- # Actually, if they are filtering, df only contains a subset.
- # Let's just overwrite the subset in full_df based on index if we kept index,
- # but Gradio df doesn't keep original indices easily.
- # Simplest way: They should only save edits if viewing All/All without search.
- if filter_state != "All" or filter_status != "All" or search_dist:
- return "⚠️ Please clear all filters (set to 'All', clear search) before saving direct table edits.
", full_df
-
- try:
- df["State"] = df["State"].astype(str).str.strip().str.upper()
- df["District"] = df["District"].astype(str).str.strip().str.upper()
- df["Status"] = df["Status"].astype(str).str.strip()
-
- invalid_statuses = df[~df["Status"].isin(["present", "new districts found"])]
- if not invalid_statuses.empty:
- return "⚠️ Invalid Status found. You can only use 'present' or 'new districts found'.
", full_df
-
- df = df.drop_duplicates()
- push_district_reference(df, token, repo)
- return f"✅ Saved {len(df)} rows directly to cloud.
", df
- except Exception as ex:
- return f"❌ Error: {ex}
", full_df
-
- def ui_prep_excel_download(df):
- if df is None or df.empty:
- return gr.update(visible=False), "⚠️ No data to download.
"
- import tempfile
- import os
- fd, temp_excel = tempfile.mkstemp(suffix=".xlsx")
- os.close(fd)
- df.to_excel(temp_excel, index=False)
- return gr.update(value=temp_excel, visible=True), "✅ Ready to download!
"
-
- def ui_upload_excel(file):
- if not file: return pd.DataFrame(), "⚠️ No file uploaded.
"
- import pandas as pd
- from hf_store import get_hf_credentials, push_district_reference, pull_district_reference
- token, repo = get_hf_credentials()
- try:
- uploaded_df = pd.read_excel(file.name)
- if "State" not in uploaded_df.columns or "District" not in uploaded_df.columns or "Status" not in uploaded_df.columns:
- return pd.DataFrame(), "⚠️ Excel must have State, District, and Status columns.
"
-
- uploaded_df["State"] = uploaded_df["State"].astype(str).str.strip().str.upper()
- uploaded_df["District"] = uploaded_df["District"].astype(str).str.strip().str.upper()
- uploaded_df["Status"] = uploaded_df["Status"].astype(str).str.strip()
-
- invalid_statuses = uploaded_df[~uploaded_df["Status"].isin(["present", "new districts found"])]
- if not invalid_statuses.empty:
- return pd.DataFrame(), "⚠️ Invalid Status found in Excel. You can only use 'present' or 'new districts found'.
"
-
- uploaded_df = uploaded_df.drop_duplicates(subset=["State", "District"])
-
- full_df = pull_district_reference(token, repo)
- if not full_df.empty:
- full_df.set_index(["State", "District"], inplace=True)
- uploaded_df.set_index(["State", "District"], inplace=True)
- full_df.update(uploaded_df)
- new_rows = uploaded_df[~uploaded_df.index.isin(full_df.index)]
- full_df = pd.concat([full_df, new_rows])
- full_df.reset_index(inplace=True)
- df_to_push = full_df
- else:
- df_to_push = uploaded_df.copy()
-
- push_district_reference(df_to_push, token, repo)
- return df_to_push, f"✅ Successfully merged {len(uploaded_df)} uploaded rows into the dataset!
"
- except Exception as ex:
- return pd.DataFrame(), f"❌ Error: {ex}
"
-
- def ui_ref_add(state, district, sf_status):
- from hf_store import get_hf_credentials, update_district_reference_add
- token, repo = get_hf_credentials()
- if not token or not repo or not state or not district or not state.strip() or not district.strip():
- return "⚠️ Fill in all fields and check HF credentials.
"
- try:
- update_district_reference_add(state.strip().upper(), district.strip().upper(), sf_status, token, repo)
- return f"✅ Added/updated {district.upper()} in {state.upper()}. Refresh table to see.
"
- except Exception as ex:
- return f"❌ Error: {ex}
"
-
- def ui_ref_rename(state, old_name, new_name):
- from hf_store import get_hf_credentials, update_district_reference_rename
- token, repo = get_hf_credentials()
- if not token or not repo or not state or not old_name or not new_name or not state.strip() or not old_name.strip() or not new_name.strip():
- return "⚠️ Fill in all fields.
"
- try:
- update_district_reference_rename(state.strip().upper(), old_name.strip().upper(), new_name.strip().upper(), token, repo)
- return f"✅ Renamed {old_name.upper()} → {new_name.upper()} (cascaded to Dataset 2). Refresh table to see.
"
- except Exception as ex:
- return f"❌ Error: {ex}
"
-
- def ui_ref_del_dist(state, district):
- from hf_store import get_hf_credentials, delete_district_reference
- token, repo = get_hf_credentials()
- if not token or not repo or not state or not district or not state.strip() or not district.strip():
- return "⚠️ Fill in both State and District.
"
- try:
- delete_district_reference(state.strip().upper(), district.strip().upper(), token, repo)
- return f"✅ Deleted {district.upper()} from {state.upper()}. Refresh table to see.
"
- except Exception as ex:
- return f"❌ Error: {ex}
"
-
- def ui_ref_del_state(state):
- from hf_store import get_hf_credentials, delete_state_reference
- token, repo = get_hf_credentials()
- if not token or not repo or not state:
- return "⚠️ Fill in State.
"
- try:
- delete_state_reference(state.strip().upper(), token, repo)
- return f"✅ Deleted entire state {state.upper()}. Refresh table to see.
"
- except Exception as ex:
- return f"❌ Error: {ex}
"
-
- # Wirings
- refresh_ref_btn.click(
- fn=ui_refresh_ref,
- outputs=[ref_table, ref_full_state, ref_filter_state,
- ref_ren_state, ref_ren_old,
- ref_del_state, ref_del_dist,
- ref_status_html]
- )
- ref_filter_state.change(fn=filter_ref_table, inputs=[ref_full_state, ref_filter_state, ref_filter_status, ref_filter_dist], outputs=[ref_table])
- ref_filter_status.change(fn=filter_ref_table, inputs=[ref_full_state, ref_filter_state, ref_filter_status, ref_filter_dist], outputs=[ref_table])
- ref_filter_dist.change(fn=filter_ref_table, inputs=[ref_full_state, ref_filter_state, ref_filter_status, ref_filter_dist], outputs=[ref_table])
-
- ref_table.select(fn=on_ref_table_select, inputs=[ref_table], outputs=[ref_edit_state, ref_edit_dist, ref_edit_status, ref_status_html])
- ref_edit_btn.click(fn=ui_ref_edit_row, inputs=[ref_edit_state, ref_edit_dist, ref_edit_status], outputs=[ref_status_html])
- ref_prep_dl_btn.click(fn=ui_prep_excel_download, inputs=[ref_table], outputs=[ref_dl_btn, ref_status_html])
- ref_upload.upload(fn=ui_upload_excel, inputs=[ref_upload], outputs=[ref_table, ref_status_html])
-
- ref_add_btn.click(fn=ui_ref_add, inputs=[ref_state_add, ref_dist_add, ref_status_add], outputs=[ref_add_status])
- ref_rename_btn.click(fn=ui_ref_rename, inputs=[ref_ren_state, ref_ren_old, ref_ren_new], outputs=[ref_rename_status])
- ref_del_dist_btn.click(fn=ui_ref_del_dist, inputs=[ref_del_state, ref_del_dist], outputs=[ref_del_status])
- ref_del_state_btn.click(fn=ui_ref_del_state, inputs=[ref_del_state], outputs=[ref_del_status])
-
- # ── Tab 4: Download Master Sheets ──────────────────────────────────────
- with gr.Tab("📥 Download History"):
- gr.HTML("Download Previously Built Master Sheets
")
-
- with gr.Row():
- dl_refresh_btn = gr.Button("🔄 Refresh File List", scale=1)
- dl_dropdown = gr.Dropdown(choices=[], label="Select a Master Sheet", scale=3)
-
- dl_status = gr.HTML()
- dl_download_btn = gr.DownloadButton("📥 Download as Excel", visible=False)
-
- def ui_refresh_dl_list():
- from hf_store import get_hf_credentials, list_mapped_master_files
- token, repo = get_hf_credentials()
- if not token or not repo:
- return gr.update(choices=[]), "⚠️ HF credentials not set.
"
-
- files = list_mapped_master_files(token, repo)
- if not files:
- return gr.update(choices=[]), "⚠️ No master sheets found in dataset.
"
-
- filenames = [f.split("/")[-1] for f in files]
- filenames.sort(reverse=True) # newest first
- return gr.update(choices=filenames), f"✅ Found {len(filenames)} master sheets.
"
-
- dl_refresh_btn.click(fn=ui_refresh_dl_list, outputs=[dl_dropdown, dl_status])
-
- def ui_prep_download(filename):
- if not filename: return gr.update(visible=False), ""
- from hf_store import get_hf_credentials
- import tempfile
- import pandas as pd
- from huggingface_hub import hf_hub_download
- token, repo = get_hf_credentials()
-
- try:
- local_path = hf_hub_download(repo_id=repo, repo_type="dataset", filename=f"scraped_data/mapped/{filename}", token=token)
- df = pd.read_parquet(local_path)
-
- temp_dir = tempfile.gettempdir()
- # Keep the original filename but change extension to .xlsx
- excel_filename = filename.replace('.parquet', '.xlsx')
- temp_excel = os.path.join(temp_dir, excel_filename)
- df.to_excel(temp_excel, index=False)
-
- return gr.update(value=temp_excel, visible=True), "✅ Ready to download!
"
- except Exception as e:
- return gr.update(visible=False), f"❌ Error preparing download: {e}
"
-
- dl_dropdown.change(fn=ui_prep_download, inputs=[dl_dropdown], outputs=[dl_download_btn, dl_status])
-
-
- # ── Tab 5: Master Analytics ────────────────────────────────────────────
- with gr.Tab("📊 Master Analytics"):
- gr.HTML('''
-
-
📈 Multi-Year Master Analytics
-
- This tab automatically loads the 2025 master and every
- mapped master sheet ever built. It analyzes how government school footprints and overall school counts shift year over year.
-
-
📅 Master Timeline
")
- gr.HTML("Year-over-year changes across all master sheets compared to 2025.
")
- timeline_table = gr.Dataframe(label="Timeline Metrics", interactive=False, wrap=True)
-
- gr.HTML("⚠️ Missing Schools (vs 2025)
")
- missing_schools_html = gr.HTML()
- missing_schools_table = gr.Dataframe(label="Missing Schools Detailed", interactive=False, wrap=True)
-
- gr.HTML("🔄 Management Shifts
")
- mgmt_shifts_html = gr.HTML()
- mgmt_shifts_table = gr.Dataframe(label="Government → Non-Govt Shifts", interactive=False, wrap=True)
-
- BASE_CSV_PATH = os.path.join(os.path.dirname(os.path.abspath(__file__)), "master_all_states.csv")
-
- def _comp_detect_year_label(filepath: str) -> str:
- import calendar as _cal
- import re as _re
- import os as _os
- name = _os.path.basename(filepath or "")
- m = _re.search(r"(\d{4})[-_](\d{2})", name)
- if m:
- yr, mo = int(m.group(1)), int(m.group(2))
- if 1 <= mo <= 12:
- return f"{_cal.month_abbr[mo]} {yr}"
- m2 = _re.search(r"(jan|feb|mar|apr|may|jun|jul|aug|sep|oct|nov|dec)[_\-]?(\d{4})", name, _re.IGNORECASE)
- if m2:
- return f"{m2.group(1).capitalize()} {m2.group(2)}"
- m3 = _re.search(r"(\d{4})", name)
- return m3.group(1) if m3 else name
-
- def _comp_find_col(df, *keywords):
- for k in keywords:
- for c in df.columns:
- if k in c.lower():
- return c
- return None
-
- def ui_run_analytics():
- from hf_store import get_hf_credentials, list_mapped_master_files, pull_baseline_master
- from huggingface_hub import hf_hub_download
- import pandas as pd
- import os
-
- token, repo = get_hf_credentials()
- if not token or not repo:
- return "⚠️ HF credentials not set.
", pd.DataFrame(), "", pd.DataFrame(), "", pd.DataFrame()
-
- masters = []
- # 1. Load baseline
- try:
- base_df = pull_baseline_master(token, repo)
- if not base_df.empty:
- masters.append({"label": "2025", "df": base_df})
- except Exception:
- pass
-
- # Fallback to CSV if HF fails
- if not masters and os.path.exists(BASE_CSV_PATH):
- masters.append({"label": "2025", "df": pd.read_csv(BASE_CSV_PATH, low_memory=False)})
-
- # 2. Pull HF mapped masters
- hf_files = list_mapped_master_files(token, repo)
- for hf_path in sorted(hf_files):
- try:
- local = hf_hub_download(repo_id=repo, repo_type="dataset", filename=hf_path, token=token, force_download=True)
- label = _comp_detect_year_label(hf_path)
- masters.append({"label": label, "df": pd.read_parquet(local)})
- except Exception:
- pass
-
- if not masters:
- return "⚠️ No master datasets found.
", pd.DataFrame(), "", pd.DataFrame(), "", pd.DataFrame()
-
- def _norm_udise(s):
- return s.astype(str).str.strip().str.zfill(11)
-
- def is_govt(s):
- val = str(s).lower()
- if 'aided' in val or 'private' in val or 'unaided' in val or 'un-aided' in val:
- return False
- return 'gov' in val or 'dept of edu' in val or 'local body' in val or 'panchayat' in val or 'municipal' in val
-
- baseline = masters[0]
- b_udise = _comp_find_col(baseline["df"], "udise")
- b_mgmt = _comp_find_col(baseline["df"], "school_management_type", "management", "mgmt")
- if b_udise: baseline["df"]["_U"] = _norm_udise(baseline["df"][b_udise])
- if b_mgmt: baseline["df"]["is_govt"] = baseline["df"][b_mgmt].apply(is_govt)
-
- timeline_data = []
- baseline_udises = set(baseline["df"]["_U"].dropna()) if b_udise else set()
-
- missing_df = pd.DataFrame()
- changed_mgmt = pd.DataFrame()
-
- for m in masters:
- m_df = m["df"]
- m_udise = _comp_find_col(m_df, "udise")
- m_mgmt = _comp_find_col(m_df, "school_management_type", "management", "mgmt")
- m_name = _comp_find_col(m_df, "school_name", "schoolname")
- m_state = _comp_find_col(m_df, "state")
-
- if m_udise: m_df["_U"] = _norm_udise(m_df[m_udise])
- if m_mgmt: m_df["is_govt"] = m_df[m_mgmt].apply(is_govt)
-
- tot_schools = len(m_df)
- tot_govt = m_df["is_govt"].sum() if m_mgmt else 0
-
- missing_count = 0
- shifts_count = 0
-
- if m != baseline and b_udise and m_udise:
- m_udises = set(m_df["_U"].dropna())
- missing = baseline_udises - m_udises
- missing_count = len(missing)
-
- if m == masters[-1]:
- b_name = _comp_find_col(baseline["df"], "school_name", "schoolname")
- b_state = _comp_find_col(baseline["df"], "state")
- missing_df = baseline["df"][baseline["df"]["_U"].isin(missing)].drop_duplicates(subset=["_U"]).copy()
-
- cols_to_keep = []
- for c in ["_U", b_name, b_state, b_mgmt]:
- if c and c in missing_df.columns:
- cols_to_keep.append(c)
- missing_df = missing_df[cols_to_keep] if cols_to_keep else missing_df
- missing_df.insert(0, "Missing In Master", m["label"])
-
- if b_mgmt and m_mgmt:
- b_cols = [c for c in ["_U", b_name, b_mgmt, "is_govt"] if c]
- m_cols = [c for c in ["_U", m_name, m_mgmt, "is_govt"] if c]
-
- merged = pd.merge(
- baseline["df"][b_cols],
- m_df[m_cols],
- on="_U", suffixes=("_old", "_new")
- )
- changed = merged[(merged["is_govt_old"] == True) & (merged["is_govt_new"] == False)]
- shifts_count = len(changed)
-
- c_name_old = str(b_name) + '_old' if b_name == m_name else b_name
- c_mgmt_old = str(b_mgmt) + '_old' if b_mgmt == m_mgmt else b_mgmt
- c_mgmt_new = str(m_mgmt) + '_new' if b_mgmt == m_mgmt else m_mgmt
-
- c_cols = [c for c in ["_U", c_name_old, c_mgmt_old, c_mgmt_new] if c and c in changed.columns]
- changed_mgmt = changed[c_cols] if c_cols else changed
- changed_mgmt.insert(0, "Shifted In Master", m["label"])
-
- timeline_data.append({
- "Master Name": m["label"],
- "Total Schools": f"{tot_schools:,}",
- "Total Govt Schools": f"{tot_govt:,}",
- "Missing (vs 2025)": f"{missing_count:,}" if m != baseline else "-",
- "Govt->Non-Govt Shifts": f"{shifts_count:,}" if m != baseline else "-"
- })
-
- timeline_df = pd.DataFrame(timeline_data)
-
- status_html = f"✅ Loaded {len(masters)} master datasets successfully.
"
- m_html = f"{len(missing_df):,} schools from 2025 are completely missing in the newest master.
"
- c_html = f"{len(changed_mgmt):,} schools shifted from Govt to Non-Govt management since 2025.
"
-
- return status_html, timeline_df, m_html, missing_df, c_html, changed_mgmt
-
- analytics_refresh_btn.click(
- fn=ui_run_analytics,
- outputs=[analytics_status, timeline_table, missing_schools_html, missing_schools_table, mgmt_shifts_html, mgmt_shifts_table]
- )
-
-if __name__ == "__main__":
- if os.environ.get("SPACE_ID"):
- print("Starting on Dataset Spaces (0.0.0.0:7860) …")
- app.queue().launch(server_name="0.0.0.0", server_port=7860)
- else:
- print("Starting locally …")
- app.queue().launch(server_name="127.0.0.1", server_port=7861, inbrowser=True)
+"""
+app.py – School Data Fetcher
+Progressive disclosure UI: each step appears only when it is needed.
+"""
+
+import gradio as gr
+import subprocess
+import sys
+import os
+import re
+import json
+import tempfile
+import pandas as pd
+from district_mapper import (
+ build_mapper, build_udise_geo_lookup,
+ apply_geo_decode, apply_district_backmap,
+ NON_ACTUAL_STATES,
+)
+from hf_store import (
+ push_state_file, pull_all_complete_states,
+ ALL_REQUIRED_CATEGORIES,
+ pull_district_reference, pull_mapping_rules,
+ push_district_reference, push_mapping_rules,
+ update_district_reference_add, update_district_reference_rename,
+ delete_district_reference, upsert_mapping_rule, delete_mapping_rule,
+ get_flagged_districts,
+ get_hf_credentials, seed_district_reference_from_csv,
+ seed_mapping_rules_from_excel,
+)
+
+# ── Auto-push helper (called after successful scrape) ─────────────────────────
+def _auto_push_to_dataset(state_name: str, categories: list) -> str:
+ """Automatically push scraped Excel to HF dataset after scrape succeeds."""
+ token, repo = get_hf_credentials()
+ if not token or not repo:
+ return "⚠️ No HF credentials — raw data not pushed to dataset."
+ excel_path = get_excel_file(state_name) if state_name else None
+ if not excel_path or not os.path.exists(excel_path):
+ return "⚠️ Excel not found — push skipped."
+ try:
+ df = pd.read_excel(excel_path)
+ cat_nums = [c.split(" - ")[0].strip() for c in (categories or [])]
+ if len(cat_nums) < 7 and "All Categories (1 to 7)" not in categories:
+ return "⚠️ Partial scrape completed. Excel downloaded, but NOT pushed to dataset. You must select all 7 categories to push."
+ push_state_file(df=df, state_name=state_name, categories_scraped=cat_nums, token=token, repo=repo)
+ return f"✅ Auto-pushed raw data for {state_name} to the cloud. Check the Master Sheet tab for any new districts flagged."
+ except Exception as ex:
+ return f"❌ Auto-push failed: {ex}"
+
+# ── State list ────────────────────────────────────────────────────────────────
+STATE_LABEL_TO_ID = {
+ "ANDAMAN & NICOBAR ISLANDS": 135, "ANDHRA PRADESH": 128,
+ "ARUNACHAL PRADESH": 112, "ASSAM": 118, "BIHAR": 110,
+ "CHANDIGARH": 104, "CHHATTISGARH": 122,
+ "DADRA & NAGAR HAVELI AND DAMAN & DIU": 138, "DELHI": 107,
+ "GOA": 130, "GUJARAT": 124, "HARYANA": 106, "HIMACHAL PRADESH": 102,
+ "IAF EC SOCIETY": 163, "JAMMU & KASHMIR": 101, "JHARKHAND": 120,
+ "KARNATAKA": 129, "KENDRIYA VIDYALAYA SANGHATHAN": 192,
+ "KERALA": 132, "LADAKH": 137, "LAKSHADWEEP": 131,
+ "MADHYA PRADESH": 123, "MAHARASHTRA": 127, "MANIPUR": 114,
+ "MEGHALAYA": 117, "MIZORAM": 115, "MSRVVP": 161,
+ "NAGALAND": 113, "NAVODAYA VIDYALAYA SAMITI": 193,
+ "NAVY EDUCATION SOCIETY": 162, "ODISHA": 121, "PUDUCHERRY": 134,
+ "PUNJAB": 103, "RAJASTHAN": 108, "SIKKIM": 111, "TAMILNADU": 133,
+ "TELANGANA": 136, "TEST STATE": 199, "TRIPURA": 116,
+ "UTTAR PRADESH": 109, "UTTARAKHAND": 105, "WEST BENGAL": 119,
+}
+
+ALL_CATEGORIES = {
+ "3": "3 - Pr. with Up.Pr. sec. and H.Sec.",
+ "5": "5 - Up. Pr. Secondary and Higher Sec",
+ "6": "6 - Pr. Up Pr. and Secondary Only",
+ "7": "7 - Upper Pr. and Secondary",
+ "8": "8 - Secondary Only",
+ "10": "10 - Secondary with Higher Secondary",
+ "11": "11 - Higher Secondary only/Jr. College",
+}
+
+OUTPUT_DIR = os.path.join(os.path.dirname(os.path.abspath(__file__)), "output")
+EXCEL_DIR = os.path.join(os.path.dirname(os.path.abspath(__file__)), "output_excel")
+_proc: dict = {"current": None}
+_batch_abort: dict = {"flag": False} # set by ui_stop() to abort a multi-state batch run
+
+# Analysis results are computed on demand (empty until user uploads files)
+_EMPTY_ANALYSIS: dict = {}
+
+
+def _state_prefix(s: str) -> str:
+ return re.sub(r"[^a-z0-9]+", "_", s.lower()).strip("_")
+
+def get_output_file(state: str) -> str:
+ return os.path.join(OUTPUT_DIR, f"{_state_prefix(state)}_schools_by_category.json")
+
+def get_excel_file(state: str) -> str:
+ return os.path.join(EXCEL_DIR, f"{_state_prefix(state)}_Schools.xlsx")
+
+def check_missed_schools(json_path: str):
+ if not os.path.exists(json_path):
+ return None
+ try:
+ import json
+ with open(json_path, "r", encoding="utf-8") as f:
+ data = json.load(f)
+ if not isinstance(data, list):
+ return None
+ def is_invalid(r):
+ if r.get("status") == "captcha_failed_all_retries":
+ return True
+ details = (r.get("response") or {}).get("error", {}).get("errorDetails", {}).get("details", "")
+ return isinstance(details, str) and "invalid captcha" in details.lower()
+ return sum(1 for r in data if is_invalid(r))
+ except Exception:
+ return None
+
+
+# ── HTML helpers ──────────────────────────────────────────────────────────────
+
+def _progress_html(current: int, total: int) -> str:
+ pct = round(current / total * 100) if total > 0 else 0
+ return f"""
+
+
+ Progress
+ {current} / {total} ({pct}%)
+
+
+
+ {"" + str(pct) + "%" if pct > 8 else ""}
+
+
+
+ {safe}
+
"""
+
+def _banner_html(icon: str, title: str, body: str, kind: str) -> str:
+ """Large status banner shown after a run completes."""
+ colors = {
+ "success": "var(--color-green-500)",
+ "warning": "var(--color-yellow-500)",
+ "nodata": "var(--color-blue-500)",
+ "stopped": "var(--color-red-500)",
+ }
+ border_col = colors.get(kind, "var(--border-color-primary)")
+ return f"""
+
+
{icon} {title}
+
{body}
+
Ready.
"
+
+def _stopped_html() -> str:
+ return "⏹ Stopped by user.
"
+
+
+# ── Log filter ─────────────────────────────────────────────────────────────────
+NOISE = [
+ "test session starts", "platform ", "cachedir:", "rootdir:",
+ "configfile:", "plugins:", "collected ", "live log call",
+ "PASSED", "FAILED", "=== 1 passed", "=== 1 failed", "no tests ran",
+ "tests/test_", "====================", "short test summary",
+ "UserWarning", "warnings.warn",
+]
+def _is_noise(line: str) -> bool:
+ return any(m in line for m in NOISE)
+
+def _clean(line: str) -> str:
+ return re.sub(r"(INFO|WARNING|ERROR|DEBUG)\s+\S+:\S+:\d+\s+", "", line).strip()
+
+
+# ── Core streamer ─────────────────────────────────────────────────────────────
+# Outputs (9 values):
+# 0 progress_html
+# 1 phase_html
+# 2 result_html
+# 3 download_btn ← gr.update(value) — raw JSON
+# 4 downloads_row ← gr.update(visible)
+# 5 excel_dl_btn ← gr.update(value) — auto-generated Excel
+# 6 retry_row ← gr.update(visible)
+# 7 map_row ← gr.update(visible) — show Map Districts after success
+# 8 stop_row ← gr.update(visible)
+
+def _stream(pytest_args: list, state: str, max_retries: int, mode: str, target_categories: str = None):
+ out_file = get_output_file(state)
+ excel_file = get_excel_file(state)
+
+ if mode == "scrape" and os.path.exists(excel_file):
+ try: os.remove(excel_file)
+ except Exception: pass
+
+ env = os.environ.copy()
+ env["KYS_MAX_RETRIES"] = str(int(max_retries))
+ if target_categories:
+ env["KYS_TARGET_CATEGORIES"] = target_categories
+
+ try:
+ process = subprocess.Popen(
+ [sys.executable, "-m"] + pytest_args,
+ stdout=subprocess.PIPE, stderr=subprocess.STDOUT,
+ encoding="utf-8", errors="replace", text=True, bufsize=1, universal_newlines=True,
+ cwd=os.path.dirname(os.path.abspath(__file__)),
+ env=env,
+ )
+ except Exception as e:
+ import traceback
+ err_banner = _banner_html("❌", "Error starting process", str(e), "stopped")
+ yield (_idle_html(), _phase_html(f"❌ Error: {e}"), _result_html(err_banner, ""), traceback.format_exc(),
+ gr.update(visible=False), gr.update(visible=False),
+ gr.update(visible=False, value=None), gr.update(visible=False),
+ gr.update(visible=False), gr.update(visible=False), gr.update(visible=False))
+ return
+
+ _proc["current"] = process
+ current, total = 0, 0
+ phase_text = "Initialising…"
+ log_lines = []
+ banner_rendered = ""
+ stats_rendered = ""
+ is_done = False
+ has_failures = False
+ success_count = 0
+ no_data_count = 0
+ failed_count = 0
+
+ re_p1 = re.compile(r"Found (\d+) districts")
+ re_p2 = re.compile(r"\[(\d+)/(\d+)\] NOW SEARCHING")
+ re_retry = re.compile(r"\[(\d+)/(\d+)\] RETRY")
+ re_dc = re.compile(r"District: (.+?)\s*\|\s*Category: (.+?)(?:\s*\(id=\d+\))?$")
+ re_done_s = re.compile(r"DONE\.\s+Total:\s*(\d+)\s*\|.*?Success:\s*(\d+).*?No Data:\s*(\d+).*?Captcha Failed:\s*(\d+)")
+ re_done_r = re.compile(r"RETRY DONE\.\s+Attempted:\s*(\d+)\s*\|.*?Resolved:\s*(\d+).*?Still captcha-failed:\s*(\d+)")
+
+ excel_cache = [None]
+ def _auto_export_excel():
+ """Auto-generate Excel from JSON after scrape completes."""
+ excel = get_excel_file(state)
+ if excel_cache[0] == excel and os.path.exists(excel):
+ return gr.update(value=excel, visible=True)
+
+ try:
+ proc = subprocess.run(
+ [sys.executable, "export_to_excel.py", "--state", state],
+ capture_output=True, encoding="utf-8", errors="replace", text=True,
+ cwd=os.path.dirname(os.path.abspath(__file__))
+ )
+ if proc.returncode == 0 and os.path.exists(excel):
+ excel_cache[0] = excel
+ return gr.update(value=excel, visible=True)
+ except Exception:
+ pass
+ return gr.update(value=None, visible=False)
+
+ def _emit():
+ prog = _progress_html(current, total) if total > 0 else _idle_html()
+ phase = _phase_html(phase_text)
+ res = _result_html(banner_rendered, stats_rendered)
+ stop_vis = gr.update(visible=not is_done)
+
+ if is_done:
+ if success_count == 0 and no_data_count > 0 and failed_count == 0:
+ dl_row_vis, retry_vis = gr.update(visible=False), gr.update(visible=False)
+ elif not has_failures:
+ # All data complete — show downloads, hide retry
+ dl_row_vis, retry_vis = gr.update(visible=True), gr.update(visible=False)
+ else:
+ # Some failures — show retry, hide downloads
+ dl_row_vis, retry_vis = gr.update(visible=False), gr.update(visible=True)
+ else:
+ dl_row_vis, retry_vis = gr.update(visible=False), gr.update(visible=False)
+
+ dl_val = gr.update(value=out_file) if (is_done and not has_failures and success_count > 0 and os.path.exists(out_file)) else gr.update(value=None)
+
+ # Auto-export excel only when scrape succeeds
+ if is_done and not has_failures and success_count > 0:
+ excel_update = _auto_export_excel()
+ else:
+ excel_update = gr.update(value=None, visible=False)
+
+ return (prog, phase, res, dl_val, dl_row_vis, excel_update, retry_vis, stop_vis)
+
+ yield _emit()
+ for raw in iter(process.stdout.readline, ""):
+ line = raw.rstrip()
+ if not line: continue
+ clean = _clean(line)
+ m = re_p1.search(line)
+ if m: phase_text = f"📍 {m.group(1)} districts found. Collecting data…"; total = current = 0
+ m2 = re_p2.search(line) or re_retry.search(line)
+ if m2:
+ current, total = int(m2.group(1)), int(m2.group(2))
+ md = re_dc.search(clean)
+ phase_text = (f"🔍 [{current}/{total}] {md.group(1).strip()} · {md.group(2).strip()}" if md else f"🔍 [{current}/{total}] Processing…")
+ if "PASS Call" in line or "[PASS" in line: phase_text = phase_text.replace("🔍", "✅")
+ elif "NO DATA" in line: phase_text = phase_text.replace("🔍", "○")
+ elif "ALL ROUNDS FAILED" in line: phase_text = phase_text.replace("🔍", "✗")
+ ms = re_done_s.search(line)
+ if ms:
+ t = int(ms.group(1)); success_count = int(ms.group(2)); no_data_count = int(ms.group(3)); failed_count = int(ms.group(4))
+ current = total = t; is_done = True; has_failures = failed_count > 0
+ stats_rendered = _stats_html(success_count, no_data_count, failed_count, t)
+ if success_count == 0 and failed_count == 0:
+ phase_text = "○ No schools found in selected categories."; banner_rendered = _banner_html("○", "No Schools Found", "The selected categories have no schools.", "nodata")
+ elif not has_failures:
+ phase_text = "✅ All data retrieved!"; banner_rendered = _banner_html("✅", "All Data Retrieved!", "All school data fetched successfully.", "success")
+ else:
+ phase_text = f"⚠️ {failed_count} record(s) incomplete."; banner_rendered = _banner_html("⚠️", "Some Records Are Incomplete", f"{failed_count} record(s) could not be fetched. Please proceed to Step 2 to fetch missing data.", "warning")
+ mr = re_done_r.search(line)
+ if mr:
+ attempted = int(mr.group(1)); resolved = int(mr.group(2)); still_failed = int(mr.group(3))
+ success_count = resolved; failed_count = still_failed; no_data_count = 0; current = total = attempted; is_done = True; has_failures = still_failed > 0
+ stats_rendered = _stats_html(resolved, 0, still_failed, attempted)
+ if not has_failures:
+ phase_text = "✅ All data retrieved!"; banner_rendered = _banner_html("✅", "All Data Retrieved!", "All missing data fetched.", "success")
+ else:
+ phase_text = f"⚠️ {still_failed} record(s) still incomplete."; banner_rendered = _banner_html("⚠️", "Some Records Still Incomplete", f"{still_failed} record(s) failed. You can run Step 2 again or proceed to Step 3.", "warning")
+ yield _emit()
+
+ process.stdout.close(); process.wait(); _proc["current"] = None
+ if process.returncode not in (0, 1):
+ is_done = True; phase_text = "Stopped by user."; banner_rendered = _banner_html("⏹", "Process Stopped", "The process was stopped.", "stopped"); yield _emit()
+ else:
+ fc = check_missed_schools(out_file)
+ if fc is not None:
+ if fc == 0 and success_count > 0:
+ is_done, has_failures = True, False; phase_text = "✅ All data retrieved!"; banner_rendered = _banner_html("✅", "All Data Retrieved!", "All school data fetched successfully.", "success")
+ elif fc > 0:
+ is_done, has_failures = True, True; failed_count = fc; phase_text = f"⚠️ {fc} record(s) incomplete."; banner_rendered = _banner_html("⚠️", "Some Records Are Incomplete", f"{fc} record(s) failed. Please proceed to Step 2 to fetch missing data.", "warning")
+ yield _emit()
+
+
+# ── Button handlers ───────────────────────────────────────────────────────────
+
+def _err_yield(msg: str):
+ # 8 outputs matching stream_outputs
+ return (_idle_html(), _phase_html(msg), "",
+ gr.update(value=None), gr.update(visible=False),
+ gr.update(value=None, elem_classes=["download-highlight", "hide-file"]),
+ gr.update(visible=False), gr.update(visible=False))
+
+def ui_main_scrape(state, selected_cats, max_retries):
+ if isinstance(state, list):
+ state = state[-1] if state else None
+ if not state: yield _err_yield("⚠️ Please select a state first."); return
+ if not selected_cats: yield _err_yield("⚠️ Please select at least one category."); return
+ cat_ids = [c.split(" - ")[0] for c in selected_cats]
+ yield from _stream(["pytest", "tests/test_scrape_districts_by_category.py", "-v", "-s", "--state", state], state, max_retries, "scrape", ",".join(cat_ids))
+
+def ui_retry(state, max_retries):
+ if isinstance(state, list):
+ state = state[-1] if state else None
+ if not state: yield _err_yield("⚠️ Please select a state first."); return
+ out = get_output_file(state)
+ if not os.path.exists(out): yield _err_yield("⚠️ No data file found."); return
+ fc = check_missed_schools(out)
+ if fc == 0:
+ # Auto-export Excel for complete data
+ excel_path = get_excel_file(state)
+ excel_update = gr.update(value=None, elem_classes=["download-highlight", "hide-file"])
+ try:
+ proc = subprocess.run([sys.executable, "export_to_excel.py", "--state", state],
+ capture_output=True, encoding="utf-8", errors="replace", text=True,
+ cwd=os.path.dirname(os.path.abspath(__file__)))
+ if proc.returncode == 0 and os.path.exists(excel_path):
+ excel_update = gr.update(value=excel_path, visible=True)
+ except Exception: pass
+ banner = _banner_html("✅", "All Data Complete!", "Nothing missing — your Excel is ready to download below.", "success")
+ yield (_idle_html(), _phase_html("✅ All data is complete!"), _result_html(banner, ""),
+ gr.update(value=out), gr.update(visible=True),
+ excel_update,
+ gr.update(visible=False), gr.update(visible=False))
+ return
+ yield from _stream(["pytest", "tests/test_retry_districts_by_category.py", "-v", "-s", "--state", state], state, max_retries, "retry")
+
+def ui_stop():
+ proc = _proc.get("current")
+ _batch_abort["flag"] = True # also abort any in-progress multi-state batch loop
+ if proc and proc.poll() is None:
+ proc.terminate()
+ try: proc.wait(timeout=5)
+ except subprocess.TimeoutExpired: proc.kill()
+ _proc["current"] = None
+ return _stopped_html()
+ return _phase_html("ℹ️ No process is currently running.")
+
+
+# ── Multi-state batch scraping ────────────────────────────────────────────────
+AUTO_RETRY_ROUNDS = 3 # automatic "Fix Missing Data" rounds per state before moving on
+
+def _batch_status_html(idx: int, total: int, state: str, log_lines: list) -> str:
+ header = ""
+ if total > 1:
+ header = f"""
+ 📦 Batch — state {idx} of {total}: {state}
+
"""
+ log_html = "".join(
+ f"{line}
"
+ for line in log_lines
+ )
+ return header + log_html
+
+
+def ui_batch_scrape(states, cats, retries, force=False):
+ """
+ Runs Start Scraping -> auto Fix Missing Data -> auto push for each selected
+ state in turn, so multiple states can be scraped unattended in one click.
+ force=True skips the "already scraped" check (used after the user has
+ explicitly confirmed a re-scrape for a single state).
+ """
+ from hf_store import check_state_scraped
+
+ if isinstance(states, str):
+ states = [states]
+ states = [s for s in (states or []) if s]
+ n = len(states)
+ noop8 = tuple(gr.update() for _ in range(8))
+
+ if n == 0:
+ yield ("",) + _err_yield("⚠️ Please select at least one state first.") + (gr.update(),)
+ return
+
+ _batch_abort["flag"] = False
+ log_lines = []
+ last_state = states[0]
+
+ for idx, state in enumerate(states, start=1):
+ last_state = state
+ if _batch_abort["flag"]:
+ break
+
+ if not force and check_state_scraped(state, cats):
+ log_lines.append(f"⏭️ {state} — already fully scraped for these categories, skipped.")
+ yield (_batch_status_html(idx, n, state, log_lines),) + noop8 + (gr.update(),)
+ continue
+
+ stopped = False
+ for out8 in ui_main_scrape(state, cats, retries):
+ yield (_batch_status_html(idx, n, state, log_lines),) + out8 + (gr.update(),)
+ if _batch_abort["flag"]:
+ stopped = True
+ break
+ if stopped:
+ break
+
+ out_file = get_output_file(state)
+ for attempt in range(1, AUTO_RETRY_ROUNDS + 1):
+ fc = check_missed_schools(out_file)
+ if fc in (0, None):
+ break
+ log_lines.append(f"↻ {state} — auto-retry round {attempt}/{AUTO_RETRY_ROUNDS} ({fc} missing record(s))…")
+ yield (_batch_status_html(idx, n, state, log_lines),) + noop8 + (gr.update(),)
+ for out8 in ui_retry(state, retries):
+ yield (_batch_status_html(idx, n, state, log_lines),) + out8 + (gr.update(),)
+ if _batch_abort["flag"]:
+ stopped = True
+ break
+ if stopped:
+ break
+ if stopped:
+ break
+
+ fc_final = check_missed_schools(out_file)
+ if fc_final:
+ log_lines.append(
+ f"⚠️ {state} — {fc_final} record(s) still failing after "
+ f"{AUTO_RETRY_ROUNDS} auto-retry round(s). Push skipped — use Fix Missing Data manually."
+ )
+ push_update = gr.update()
+ else:
+ push_msg = _auto_push_to_dataset(state, cats)
+ log_lines.append(f"{state}: {push_msg}")
+ push_update = push_msg
+
+ yield (_batch_status_html(idx, n, state, log_lines),) + noop8 + (push_update,)
+
+ if _batch_abort["flag"]:
+ log_lines.append("⏹ Batch stopped by user.")
+ else:
+ log_lines.append(f"🎉 Batch complete — {n} state(s) processed.")
+ yield (_batch_status_html(n, n, last_state, log_lines),) + noop8 + (gr.update(),)
+ _batch_abort["flag"] = False
+
+
+def ui_main_scrape_forced(states, cats, retries):
+ """Used after the user explicitly confirms re-scraping an already-scraped single state."""
+ yield from ui_batch_scrape(states, cats, retries, force=True)
+
+
+# ── Dashboard helpers ─────────────────────────────────────────────────────────
+def _metric_card(label, value, color="#6366f1"):
+ return f"""
+ """
+
+def _dash_headline_html():
+ ns = _ANALYSIS.get("National Summary", pd.DataFrame())
+ if ns.empty:
+ return "Analysis file not loaded.
"
+ vals = dict(zip(ns["Metric"], ns["Count"]))
+ old = f"{vals.get('Total Old Schools', 0):,}"
+ new = f"{vals.get('Total New Schools', 0):,}"
+ added= f"{vals.get('Total Schools Added', 0):,}"
+ # compute deleted
+ deleted_df = _ANALYSIS.get("All Deleted Schools", pd.DataFrame())
+ deleted = f"{len(deleted_df):,}" if not deleted_df.empty else "0"
+ return f"""
+
+ {_metric_card('Total Schools (Last Year)', old, '#6366f1')}
+ {_metric_card('Total Schools (This Year)', new, '#6366f1')}
+ {_metric_card('✅ Schools Added', added, '#22c55e')}
+ {_metric_card('❌ Schools Removed', deleted, '#ef4444')}
+
"""
+
+def _truncate_udise(val, n=3):
+ """Truncate long UDISE lists to first n codes + count."""
+ if pd.isna(val) or val == "":
+ return ""
+ codes = [c.strip() for c in str(val).split(",") if c.strip()]
+ if len(codes) <= n:
+ return ", ".join(codes)
+ shown = ", ".join(codes[:n])
+ return f"{shown} ... (+{len(codes)-n} more)"
+
+def _full_udise(val):
+ if pd.isna(val) or val == "":
+ return ""
+ return str(val).strip()
+
+def _filter_state(sheet_name, state_col="State", state=None):
+ df = _ANALYSIS.get(sheet_name, pd.DataFrame()).copy()
+ if df.empty:
+ return df
+ if state and state != "All States":
+ df = df[df[state_col] == state]
+ return df.reset_index(drop=True)
+
+def _prep_with_udise(df, udise_col):
+ """Return two versions: one with truncated UDISEs for display, one full."""
+ if df.empty or udise_col not in df.columns:
+ return df
+ df = df.copy()
+ df[udise_col] = df[udise_col].apply(_truncate_udise)
+ return df
+
+def dash_update(state, analysis):
+ """Returns all table data when state filter changes."""
+ def _fs(sheet, sc="State"):
+ df = analysis.get(sheet, pd.DataFrame()).copy()
+ if df.empty: return df
+ if state and state != "All States":
+ df = df[df[sc] == state]
+ return df.reset_index(drop=True)
+
+ dm = _prep_with_udise(_fs("District Mapping"), "List_of_UDISEs")
+ cs = _prep_with_udise(_fs("Complex Splits"), "List_of_UDISEs")
+ bm = _prep_with_udise(_fs("Block Mapping"), "List_of_UDISEs")
+ ct = _prep_with_udise(_fs("Category Transitions"), "List_of_UDISEs")
+ ad = _fs("All Added Schools", sc="School_State__c")
+ dl = _fs("All Deleted Schools", sc="School_State__c")
+ return dm, cs, bm, ct, ad, dl
+
+
+def _get_udise_for_row(sheet_name, row_idx, analysis):
+ """Look up the full UDISE list for a given row index from live analysis."""
+ df = analysis.get(sheet_name, pd.DataFrame())
+ if df.empty or row_idx >= len(df) or "List_of_UDISEs" not in df.columns:
+ return ""
+ return _full_udise(df.iloc[row_idx]["List_of_UDISEs"])
+
+
+def on_select_dm(state, last_row, analysis, evt: gr.SelectData):
+ row = evt.index[0]
+ if row == last_row: return "", -1
+ # filter to current state then get that row's udise
+ def _fs(sheet):
+ df = analysis.get(sheet, pd.DataFrame()).copy()
+ if state and state != "All States": df = df[df["State"] == state]
+ return df.reset_index(drop=True)
+ df = _fs("District Mapping")
+ if df.empty or row >= len(df): return "", row
+ return _full_udise(df.iloc[row]["List_of_UDISEs"]), row
+
+def on_select_cs(state, last_row, analysis, evt: gr.SelectData):
+ row = evt.index[0]
+ if row == last_row: return "", -1
+ def _fs(sheet):
+ df = analysis.get(sheet, pd.DataFrame()).copy()
+ if state and state != "All States": df = df[df["State"] == state]
+ return df.reset_index(drop=True)
+ df = _fs("Complex Splits")
+ if df.empty or row >= len(df): return "", row
+ return _full_udise(df.iloc[row]["List_of_UDISEs"]), row
+
+def on_select_bm(state, last_row, analysis, evt: gr.SelectData):
+ row = evt.index[0]
+ if row == last_row: return "", -1
+ def _fs(sheet):
+ df = analysis.get(sheet, pd.DataFrame()).copy()
+ if state and state != "All States": df = df[df["State"] == state]
+ return df.reset_index(drop=True)
+ df = _fs("Block Mapping")
+ if df.empty or row >= len(df): return "", row
+ return _full_udise(df.iloc[row]["List_of_UDISEs"]), row
+
+def on_select_ct(state, last_row, analysis, evt: gr.SelectData):
+ row = evt.index[0]
+ if row == last_row: return "", -1
+ def _fs(sheet):
+ df = analysis.get(sheet, pd.DataFrame()).copy()
+ if state and state != "All States": df = df[df["State"] == state]
+ return df.reset_index(drop=True)
+ df = _fs("Category Transitions")
+ if df.empty or row >= len(df): return "", row
+ return _full_udise(df.iloc[row]["List_of_UDISEs"]), row
+
+# ── Analysis engine (ported from national_analysis.py) ───────────────────────
+def _load_file(path):
+ if path is None:
+ raise ValueError("No file uploaded.")
+ ext = os.path.splitext(path)[-1].lower()
+ if ext == ".csv":
+ try:
+ return pd.read_csv(path, low_memory=False)
+ except UnicodeDecodeError:
+ return pd.read_csv(path, encoding="cp1252", low_memory=False)
+ elif ext in (".xlsx", ".xls"):
+ return pd.read_excel(path)
+ else:
+ raise ValueError(f"Unsupported file type: {ext}. Use .csv or .xlsx")
+
+def _clean_text(s):
+ return s.astype(str).str.strip().str.upper()
+
+def _clean_udise(s):
+ return s.astype(str).str.replace(r"\.0$", "", regex=True).str.zfill(11)
+
+def _fmt_udise_list(series):
+ joined = ", ".join(series.astype(str))
+ return joined[:30000] + " ... (+ more)" if len(joined) > 30000 else joined
+
+def run_analysis(old_path, new_path):
+ """Run the full comparison. Returns a dict of DataFrames."""
+ df_old = _load_file(old_path)
+ df_new = _load_file(new_path)
+
+ for df in [df_old, df_new]:
+ df["School_State__c"] = _clean_text(df["School_State__c"])
+ df["UDISE"] = _clean_udise(df["School_Udise_Code__c"])
+
+ df_old = df_old.drop_duplicates(subset=["UDISE"]).set_index("UDISE")
+ df_new = df_new.drop_duplicates(subset=["UDISE"]).set_index("UDISE")
+
+ old_u = set(df_old.index)
+ new_u = set(df_new.index)
+ added_u = new_u - old_u
+ deleted_u = old_u - new_u
+ common_u = old_u & new_u
+
+ df_added_raw = df_new.loc[list(added_u)].reset_index()[["UDISE","School_State__c","School_District__c","School_Block__c"]]
+ df_deleted_raw = df_old.loc[list(deleted_u)].reset_index()[["UDISE","School_State__c","School_District__c","School_Block__c"]]
+
+ all_states = sorted(set(df_old["School_State__c"].unique()) | set(df_new["School_State__c"].unique()))
+ state_summary = []
+ for st in all_states:
+ so = set(df_old[df_old["School_State__c"] == st].index)
+ sn = set(df_new[df_new["School_State__c"] == st].index)
+ state_summary.append({"State": st, "Old Count": len(so), "New Count": len(sn),
+ "Added": len(sn - so), "Deleted": len(so - sn)})
+ df_state_summary = pd.DataFrame(state_summary)
+
+ df_common = df_old.loc[list(common_u)].join(df_new.loc[list(common_u)], lsuffix="_OLD", rsuffix="_NEW").reset_index()
+ df_common["State"] = _clean_text(df_common["School_State__c_NEW"])
+ df_common["Dist_OLD"] = _clean_text(df_common["School_District__c_OLD"])
+ df_common["Dist_NEW"] = _clean_text(df_common["School_District__c_NEW"])
+ df_common["Block_OLD"]= _clean_text(df_common["School_Block__c_OLD"])
+ df_common["Block_NEW"]= _clean_text(df_common["School_Block__c_NEW"])
+ df_common["Cat_OLD"] = _clean_text(df_common["schCategoryType__c_OLD"])
+ df_common["Cat_NEW"] = _clean_text(df_common["schCategoryType__c_NEW"])
+
+ dist_chg = df_common[df_common["Dist_OLD"] != df_common["Dist_NEW"]]
+ district_mapping = pd.DataFrame()
+ if len(dist_chg):
+ district_mapping = dist_chg.groupby(["State","Dist_OLD","Dist_NEW"]).agg(
+ Affected_Schools=("UDISE","count"), List_of_UDISEs=("UDISE", _fmt_udise_list)
+ ).reset_index()
+
+ # 3. Block to District Mapping (The Exact Matcher for Splits)
+ # If a new district was formed, we want to know what the old district was for a given block.
+ # So we group by (State, Dist_NEW, Block_NEW) and find the most common Dist_OLD.
+ block_to_district = pd.DataFrame()
+ if not df_common.empty:
+ # Group by new block and old district
+ block_dist_counts = df_common.groupby(["State", "Dist_NEW", "Block_NEW", "Dist_OLD"]).agg(
+ Schools=("UDISE", "count"),
+ List_of_UDISEs=("UDISE", _fmt_udise_list)
+ ).reset_index()
+
+ # For each new block, find the old district that had the most schools
+ # (This establishes the definitive historical mapping for that block)
+ idx = block_dist_counts.groupby(["State", "Dist_NEW", "Block_NEW"])["Schools"].idxmax()
+ block_to_district = block_dist_counts.loc[idx].reset_index(drop=True)
+
+ blk_chg = df_common[df_common["Block_OLD"] != df_common["Block_NEW"]]
+ block_mapping = pd.DataFrame()
+ if len(blk_chg):
+ block_mapping = blk_chg.groupby(["State","Dist_OLD","Block_OLD","Block_NEW"]).agg(
+ Affected_Schools=("UDISE","count"), List_of_UDISEs=("UDISE", _fmt_udise_list)
+ ).reset_index()
+
+ cat_chg = df_common[df_common["Cat_OLD"] != df_common["Cat_NEW"]]
+ cat_migrations = pd.DataFrame()
+ if len(cat_chg):
+ cat_migrations = cat_chg.groupby(["State","Cat_OLD","Cat_NEW"]).agg(
+ Count=("UDISE","count"), List_of_UDISEs=("UDISE", _fmt_udise_list)
+ ).reset_index().sort_values(["State","Count"], ascending=[True,False])
+
+ ns = pd.DataFrame({"Metric": [
+ "Total Old Schools","Total New Schools","Total Schools Added","Total Schools Deleted",
+ "Common Schools (In Both)","Schools with District Changes","Schools with Block Changes","Schools with Category Changes"
+ ], "Count": [
+ len(df_old), len(df_new), len(added_u), len(deleted_u),
+ len(common_u), len(dist_chg), len(blk_chg), len(cat_chg)
+ ]})
+
+ from district_mapper import build_udise_geo_lookup
+ geo_lookup = build_udise_geo_lookup(df_new)
+ rows = []
+ for ss, name in geo_lookup["state"].items(): rows.append(["state", ss, name])
+ for ssdd, name in geo_lookup["district"].items(): rows.append(["district", ssdd, name])
+ for ssddbb, name in geo_lookup["block"].items(): rows.append(["block", ssddbb, name])
+ geo_df = pd.DataFrame(rows, columns=["Type", "Code", "Name"])
+
+ # Find unmapped completely new districts
+ old_dist_names = set(df_old["School_District__c"].dropna().str.strip().str.upper().unique())
+ new_dist_df = df_new[["School_State__c", "School_District__c"]].drop_duplicates().dropna()
+ new_dist_df["State"] = _clean_text(new_dist_df["School_State__c"])
+ new_dist_df["New District Name"] = _clean_text(new_dist_df["School_District__c"])
+
+ unmapped_new = new_dist_df[~new_dist_df["New District Name"].isin(old_dist_names)].copy()
+ unmapped_new["Rename To (Type here...)"] = ""
+ if not unmapped_new.empty:
+ unmapped_new = unmapped_new[["State", "New District Name", "Rename To (Type here...)"]].sort_values(["State", "New District Name"]).reset_index(drop=True)
+ else:
+ unmapped_new = pd.DataFrame(columns=["State", "New District Name", "Rename To (Type here...)"])
+
+ return {
+ "National Summary": ns,
+ "State by State Breakdown": df_state_summary,
+ "All Added Schools": df_added_raw,
+ "All Deleted Schools": df_deleted_raw,
+ "District Mapping": district_mapping,
+ "Block to District": block_to_district,
+ "Block Mapping": block_mapping,
+ "Category Transitions": cat_migrations,
+ "New Districts": unmapped_new,
+ "UDISE Geo": geo_df,
+ }
+
+# ── CSS ───────────────────────────────────────────────────────────────────────
+css = """
+@import url('https://fonts.googleapis.com/css2?family=Inter:wght@400;500;600;700;800&display=swap');
+
+* { font-family: 'Inter', sans-serif !important; }
+
+#title {
+ text-align: center;
+ font-size: 2.2em !important;
+ font-weight: 800 !important;
+ background: linear-gradient(135deg, #6366f1, #8b5cf6, #a855f7);
+ -webkit-background-clip: text;
+ -webkit-text-fill-color: transparent;
+ background-clip: text;
+ margin-bottom: 4px !important;
+ letter-spacing: -0.5px;
+}
+#subtitle {
+ text-align: center;
+ color: var(--body-text-color-subdued);
+ margin-top: 0;
+ margin-bottom: 24px;
+ font-size: .95em;
+}
+
+/* Tab styling */
+.tab-nav button {
+ font-weight: 600 !important;
+ font-size: 1em !important;
+ padding: 10px 20px !important;
+ border-radius: 10px 10px 0 0 !important;
+ transition: all 0.2s !important;
+}
+.tab-nav button.selected {
+ background: linear-gradient(135deg, #6366f1, #8b5cf6) !important;
+ color: white !important;
+}
+
+/* Step card styling */
+.step-card {
+ background: var(--background-fill-secondary);
+ border-radius: 14px;
+ padding: 20px;
+ border: 1px solid var(--border-color-primary);
+ margin-bottom: 12px;
+ transition: box-shadow 0.2s;
+}
+.step-card:hover { box-shadow: 0 4px 20px rgba(99,102,241,0.12); }
+
+/* Button beautification */
+.btn-primary { background: linear-gradient(135deg, #6366f1, #8b5cf6) !important; border: none !important; }
+.btn-primary:hover { transform: translateY(-1px) !important; box-shadow: 0 6px 20px rgba(99,102,241,0.4) !important; }
+
+/* Download file widget */
+.download-highlight {
+ border: 2px dashed var(--color-accent) !important;
+ border-radius: 12px !important;
+ padding: 0 !important;
+ position: relative !important;
+ background: transparent !important;
+ transition: all 0.25s !important;
+}
+.download-highlight:hover {
+ border-style: solid !important;
+ border-color: var(--color-accent) !important;
+ background: color-mix(in srgb, var(--color-accent) 5%, transparent) !important;
+}
+.download-highlight * { position: static !important; }
+.download-highlight a { display: flex !important; align-items: center; padding: 14px !important; width: 100%; height: 100%; }
+.download-highlight a::after { content: ""; position: absolute !important; inset: 0 !important; z-index: 50 !important; cursor: pointer !important; }
+.hide-file { display: none !important; }
+
+/* Status banners */
+.status-ok { color: #22c55e; font-weight: 700; }
+.status-err { color: #ef4444; font-weight: 700; }
+.status-warn{ color: #f59e0b; font-weight: 700; }
+
+/* Accordion styling */
+.gr-accordion { border-radius: 12px !important; border: 1px solid var(--border-color-primary) !important; margin-bottom: 10px !important; }
+
+/* Section divider label */
+.section-label {
+ font-size: 0.75em;
+ font-weight: 600;
+ text-transform: uppercase;
+ letter-spacing: 1px;
+ color: var(--body-text-color-subdued);
+ margin: 20px 0 8px 0;
+}
+
+/* Metric cards */
+.metric-card {
+ background: transparent;
+ border-radius: 14px;
+ padding: 20px 28px;
+ text-align: center;
+ transition: transform 0.2s;
+}
+.metric-card:hover { transform: translateY(-2px); }
+
+/* Log box */
+.log-box textarea { font-size:.76em !important; font-family: 'JetBrains Mono', monospace !important; }
+"""
+
+custom_theme = gr.themes.Soft(
+ primary_hue="violet",
+ secondary_hue="indigo",
+ neutral_hue="slate",
+ font=[gr.themes.GoogleFont("Inter"), "ui-sans-serif", "system-ui", "sans-serif"]
+)
+
+# ── UI ────────────────────────────────────────────────────────────────────────
+with gr.Blocks(title="School Data Fetcher", css=css, theme=custom_theme) as app:
+ gr.HTML("""
+
+
+ 🏫 School Data Fetcher
+
+
+ Scrape → Auto-push Raw → Map Districts → Build Mapped Master
+
+
+
👋 How it works
+
+ Step 1: Choose one or more states & categories, then click Start Scraping — selected states are scraped one after another automatically, each one retried and pushed to the cloud before moving to the next.
+ Step 2: If a state still has failures after the automatic retries, click Fix Missing Data to retry those records again.
+ Step 3: Once every state is done, go to the 📋 Master Sheet tab to review any newly detected districts and map them if necessary.
+ Step 4: Once all states are scraped and mapped, click Build Master Sheet at the bottom of the tab to generate the final mapped excel file.
+
+
📍 Location
")
+ state_dd = gr.Dropdown(
+ choices=sorted(STATE_LABEL_TO_ID.keys()),
+ multiselect=True,
+ value=[],
+ label="Select State(s) or School Type(s)",
+ info="Select one or more. Multiple states are scraped one after another automatically, "
+ "each retried and pushed to the cloud before moving to the next."
+ )
+ with gr.Column(scale=2):
+ gr.HTML("📚 School Types
")
+ categories_cg = gr.CheckboxGroup(
+ choices=list(ALL_CATEGORIES.values()),
+ value=list(ALL_CATEGORIES.values()),
+ label="Select which school categories to collect"
+ )
+
+ gr.HTML("")
+
+ with gr.Row():
+ scrape_btn = gr.Button("▶ Start Scraping", variant="primary", scale=3)
+ stop_btn_placeholder = gr.HTML("", visible=False) # placeholder
+
+ with gr.Column(visible=False) as confirm_col:
+ gr.HTML("⚠️ Warning: This state (and all selected categories) is already present in your dataset. Are you sure you want to scrape it again?
")
+ with gr.Row():
+ confirm_yes_btn = gr.Button("Yes, Scrape Again", variant="stop")
+ confirm_no_btn = gr.Button("No, Cancel", variant="secondary")
+
+ with gr.Row(visible=False) as stop_row:
+ stop_btn = gr.Button("⏹ Stop", variant="stop", scale=1)
+
+ should_scrape_state = gr.State(False)
+
+ batch_status_html = gr.HTML(value="")
+ progress_html = gr.HTML(value=_idle_html())
+ phase_html = gr.HTML(value=_phase_html("Select a state and school types above, then click Start Scraping."))
+ result_html = gr.HTML(value="")
+
+ with gr.Row(visible=False) as retry_row:
+ retry_btn = gr.Button("↻ Fix Missing Data (some records failed — click to retry)", variant="secondary", scale=1)
+
+ with gr.Row(visible=False) as downloads_row:
+ with gr.Column(scale=1):
+ download_btn = gr.DownloadButton(
+ label="📄 Download Raw JSON",
+ value=None, visible=True,
+ variant="secondary"
+ )
+ with gr.Column(scale=1):
+ excel_dl_btn = gr.DownloadButton(
+ label="📊 Download Excel",
+ value=None, visible=False,
+ variant="secondary"
+ )
+
+ with gr.Accordion("⚙️ Advanced Settings", open=False):
+ retries_slider = gr.Slider(
+ minimum=1, maximum=10, value=5, step=1,
+ label="Retry attempts per failed record (higher = more thorough but slower)"
+ )
+
+
+ # Auto-push status (shown after scrape succeeds)
+ auto_push_html = gr.HTML(value="")
+
+ stream_outputs = [
+ progress_html, phase_html, result_html,
+ download_btn, downloads_row,
+ excel_dl_btn,
+ retry_row, stop_row,
+ ]
+
+ def handle_scrape_click(states, categories):
+ from hf_store import check_state_scraped
+ states = states or []
+ if not states:
+ return gr.update(), gr.update(), False
+ if len(states) == 1 and check_state_scraped(states[0], categories):
+ return gr.update(visible=False), gr.update(visible=True), False
+ return gr.update(visible=False), gr.update(visible=False), True
+
+ def ui_main_scrape_conditional(should_scrape, states, cats, retries):
+ if should_scrape:
+ yield from ui_batch_scrape(states, cats, retries, force=False)
+ else:
+ yield tuple(gr.update() for _ in range(10))
+
+ batch_outputs = [batch_status_html] + stream_outputs + [auto_push_html]
+
+ scrape_btn.click(
+ fn=handle_scrape_click,
+ inputs=[state_dd, categories_cg],
+ outputs=[scrape_btn, confirm_col, should_scrape_state]
+ ).then(
+ fn=ui_main_scrape_conditional,
+ inputs=[should_scrape_state, state_dd, categories_cg, retries_slider],
+ outputs=batch_outputs
+ ).then(
+ fn=lambda should: gr.update(visible=True) if should else gr.update(),
+ inputs=[should_scrape_state],
+ outputs=[scrape_btn]
+ )
+
+ confirm_yes_btn.click(
+ fn=lambda: (gr.update(visible=False), gr.update(visible=False)),
+ outputs=[confirm_col, scrape_btn]
+ ).then(
+ fn=ui_main_scrape_forced,
+ inputs=[state_dd, categories_cg, retries_slider],
+ outputs=batch_outputs
+ ).then(
+ fn=lambda: gr.update(visible=True),
+ outputs=[scrape_btn]
+ )
+
+ confirm_no_btn.click(
+ fn=lambda: (gr.update(visible=False), gr.update(visible=True)),
+ outputs=[confirm_col, scrape_btn]
+ )
+ retry_btn.click( fn=ui_retry, inputs=[state_dd, retries_slider], outputs=stream_outputs)
+ stop_btn.click( fn=ui_stop, inputs=[], outputs=[progress_html])
+
+ # ── Tab 2: Master Sheet Builder ───────────────────────────────────────
+ with gr.Tab("📋 Master Sheet"):
+
+ # ── HF connection status ──────────────────────────────────────────
+ def _env_status():
+ from hf_store import get_hf_credentials
+ token, repo = get_hf_credentials()
+ if token and repo:
+ repo_name = repo.split("/")[-1] if "/" in repo else repo
+ return f"""
+
+
✅
+
+
Connected to Dataset
+
Saving to: {repo}
+
+
+
⚠️
+
+
Dataset not set up
+
+ Add your HF_TOKEN and HF_REPO to the .env file to enable dataset features.
+
+
+
📡 Dataset Coverage
+
+ Only states with all 7 required school categories scraped will be included in the master sheet.
+
+ """)
+
+ with gr.Row():
+ refresh_btn = gr.Button("🔄 Refresh from Dataset", variant="secondary", scale=1)
+
+ coverage_table = gr.Dataframe(
+ label="State Coverage (Click '🗑️ Delete' in the Action column to remove a state)",
+ interactive=False, wrap=True,
+ headers=["State", "Categories Scraped", "Complete?", "Last Scraped", "Action"],
+ )
+ coverage_summary_html = gr.HTML(value="")
+
+ # ── Flagged Districts & Mapping Rules ─────────────────────────
+ with gr.Accordion("Review New Districts", open=False):
+ gr.HTML("Review newly scraped districts and map them to older SF districts or rename.
")
+
+ with gr.Row():
+ refresh_flags_btn = gr.Button("🔄 Refresh Flagged Districts", variant="secondary", scale=1)
+ flagged_table = gr.Dataframe(
+ label="New districts detected (Type the Old SF name in the last column to rename it)",
+ headers=["State", "District", "Reason", "Rename District (Optional)"],
+ interactive=True, wrap=True
+ )
+ flagged_status_html = gr.HTML(value="")
+
+
+
+ gr.Markdown("---")
+ build_master_btn = gr.Button("📋 Build Master Sheet", variant="primary")
+ build_status_html = gr.HTML(value="")
+ master_dl_btn = gr.DownloadButton(
+ label="⬇ Download Master Excel",
+ visible=False,
+ variant="secondary"
+ )
+
+ # ── Callback: Refresh coverage table ─────────────────────────────
+ coverage_state = gr.State(pd.DataFrame())
+
+ def ui_refresh_coverage():
+ from hf_store import get_hf_credentials
+ token, repo = get_hf_credentials()
+ if not token or not repo:
+ return (
+ pd.DataFrame(columns=["State", "Categories Scraped", "Complete?", "Last Scraped", "Action"]),
+ "⚠️ HF_TOKEN or HF_REPO not set in environment.
"
+ )
+ try:
+ from hf_store import pull_all_complete_states
+ _, partial_info, coverage_df = pull_all_complete_states(token, repo)
+ if not coverage_df.empty:
+ coverage_df["Action"] = "🗑️ Delete"
+ complete_count = int((coverage_df["Complete?"] == "✅ Yes").sum()) if not coverage_df.empty else 0
+ total = len(coverage_df)
+ summary = (
+ f""
+ f"✅ {complete_count} states ready to build
"
+ )
+ return coverage_df, summary
+ except Exception as ex:
+ return (
+ pd.DataFrame(columns=["State", "Categories Scraped", "Complete?", "Last Scraped", "Action"]),
+ f"❌ Error: {ex}
"
+ )
+
+ refresh_btn.click(
+ fn=ui_refresh_coverage,
+ inputs=[],
+ outputs=[coverage_table, coverage_summary_html],
+ ).then(
+ fn=lambda df: df,
+ inputs=[coverage_table],
+ outputs=[coverage_state]
+ )
+
+ def on_coverage_select(evt: gr.SelectData, current_df):
+ if evt.value != "🗑️ Delete":
+ return gr.update(), gr.update()
+
+ row_idx = evt.index[0]
+ if row_idx >= len(current_df):
+ return gr.update(), gr.update()
+
+ state_name = current_df.iloc[row_idx]["State"]
+ from hf_store import get_hf_credentials, delete_state_file
+ token, repo = get_hf_credentials()
+ try:
+ delete_state_file(state_name, token, repo)
+ except Exception as ex:
+ return gr.update(), f"❌ Error deleting: {ex}
"
+
+ new_df, new_summary = ui_refresh_coverage()
+ msg = f"✅ Successfully deleted {state_name}.
" + new_summary
+ return new_df, msg
+
+ coverage_table.select(
+ fn=on_coverage_select,
+ inputs=[coverage_state],
+ outputs=[coverage_table, coverage_summary_html]
+ ).then(
+ fn=lambda df: df,
+ inputs=[coverage_table],
+ outputs=[coverage_state]
+ )
+
+ # ── Callbacks: Flagged Districts & Mapping Rules ──────────────────────
+ def ui_refresh_flags():
+ from hf_store import get_hf_credentials, get_flagged_districts
+ token, repo = get_hf_credentials()
+ if not token or not repo:
+ return pd.DataFrame(columns=["State", "District", "Reason"]), "⚠️ HF credentials not set.
"
+ flags = get_flagged_districts(token, repo)
+ if not flags:
+ return pd.DataFrame(columns=["State", "District", "Reason"]), "✅ No flagged districts — all clear!
"
+ df = pd.DataFrame(flags)
+ df = df[["state", "district", "reason"]].rename(columns={"state": "State", "district": "District", "reason": "Reason"})
+ df["Rename District (Optional)"] = ""
+ return df, f"⚠️ {len(flags)} district(s) need review.
"
+
+ refresh_flags_btn.click(fn=ui_refresh_flags, outputs=[flagged_table, flagged_status_html])
+
+ # ── Callback: Build master sheet ──────────────────────────────────
+ def ui_build_master(flagged_df):
+ from district_mapper import build_udise_geo_lookup, apply_geo_decode, build_mapper, apply_district_backmap
+ from hf_store import get_hf_credentials, pull_all_complete_states, pull_mapping_rules, push_mapped_master, upsert_mapping_rule, rename_district_in_reference
+ token, repo = get_hf_credentials()
+ if not token or not repo:
+ return (
+ "⚠️ HF_TOKEN or HF_REPO not set in environment.
",
+ gr.update(elem_classes=["download-highlight", "hide-file"]),
+ )
+ try:
+ # 1. Process inline table mappings
+ if flagged_df is not None and not flagged_df.empty:
+ for _, row in flagged_df.iterrows():
+ mapped = str(row.get("Rename District (Optional)", "")).strip()
+ state = str(row.get("State", "")).strip().upper()
+ dist = str(row.get("District", "")).strip().upper()
+ if mapped and mapped.upper() != "NAN" and mapped != "None":
+ upsert_mapping_rule(state, dist, mapped.upper(), 0, token, repo)
+ rename_district_in_reference(state, dist, mapped.upper(), token, repo)
+ complete_dfs, _, _ = pull_all_complete_states(token, repo)
+
+ if not complete_dfs:
+ return (
+ "⚠️ No states with all 7 categories found in dataset.
",
+ gr.update(elem_classes=["download-highlight", "hide-file"]),
+ )
+
+ from hf_store import push_national_analysis_report
+ master_df = pd.concat(complete_dfs, ignore_index=True)
+ geo_lookup = build_udise_geo_lookup(master_df)
+ master_df, geo_changes_df = apply_geo_decode(master_df, geo_lookup)
+
+ # Pull mapping rules automatically
+ dist_rules, block_rules = pull_mapping_rules(token, repo)
+
+ backmap_changes_df = pd.DataFrame()
+ if not dist_rules.empty:
+ analysis_dict = {
+ "District Mapping": dist_rules,
+ "Block to District": block_rules if not block_rules.empty else pd.DataFrame(),
+ }
+ mapper = build_mapper(analysis_dict)
+ master_df, backmap_changes_df = apply_district_backmap(master_df, mapper)
+
+ # Combine reports and push
+ combined_changes = pd.concat([geo_changes_df, backmap_changes_df], ignore_index=True)
+ if not combined_changes.empty:
+ push_national_analysis_report(combined_changes, token, repo)
+
+ udise_col = "School_Udise_Code__c" if "School_Udise_Code__c" in master_df.columns else "UDISE"
+ if udise_col in master_df.columns and "Scraped_Date" in master_df.columns:
+ master_df = (
+ master_df
+ .sort_values("Scraped_Date", ascending=False)
+ .drop_duplicates(subset=[udise_col], keep="first")
+ .reset_index(drop=True)
+ )
+
+ # Filter out out-of-scope states (e.g., from Geo-Decoded Navy/KVS schools)
+ from hf_store import pull_district_reference
+ ref_df = pull_district_reference(token, repo)
+ ref_states = set(ref_df["State"].str.strip().str.upper()) if not ref_df.empty else set()
+
+ allowed_states = {
+ "ARUNACHAL PRADESH", "ASSAM", "BIHAR", "CHHATTISGARH", "JHARKHAND",
+ "KARNATAKA", "MADHYA PRADESH", "MANIPUR", "MEGHALAYA", "MIZORAM",
+ "NAGALAND", "ODISHA", "PUDUCHERRY", "RAJASTHAN", "SIKKIM",
+ "TELANGANA", "TRIPURA", "UTTAR PRADESH", "UTTARAKHAND", "DELHI",
+ "ANDHRA PRADESH"
+ }
+ allowed_states.update(ref_states)
+
+ state_col = "School_State__c" if "School_State__c" in master_df.columns else "State"
+ if state_col in master_df.columns:
+ master_df = master_df[master_df[state_col].str.strip().str.upper().isin(allowed_states)]
+ master_df = master_df.sort_values(state_col, ascending=True).reset_index(drop=True)
+
+ from datetime import datetime, timedelta
+ ist_now = datetime.utcnow() + timedelta(hours=5, minutes=30)
+ date_str = ist_now.strftime("%Y_%b_%d_%I_%M_%p").lower()
+ os.makedirs(EXCEL_DIR, exist_ok=True)
+ master_path = os.path.join(EXCEL_DIR, f"mapped_master_{date_str}.xlsx")
+ master_df.to_excel(master_path, index=False)
+
+ push_mapped_master(master_df, token, repo)
+
+ # Auto-sync aliases CSV immediately after building a new master!
+ try:
+ from alias_sync import sync_aliases
+ sync_aliases(token, repo)
+ except Exception as e:
+ print(f"Failed to auto-sync aliases: {e}")
+
+ n = len(master_df)
+ msg = (
+ f""
+ f"✅ Mapped master built! {n:,} records from {len(complete_dfs)} complete states. "
+ f"District mapping applied. Also pushed to dataset."
+ f"
"
+ )
+ return msg, gr.update(value=master_path, visible=True)
+ except Exception as ex:
+ return (
+ f"❌ Error: {ex}
",
+ gr.update(visible=False),
+ )
+
+ build_master_btn.click(
+ fn=ui_build_master,
+ inputs=[flagged_table],
+ outputs=[build_status_html, master_dl_btn],
+ )
+
+ # ── Tab 3: Mapping Manager ─────────────────────────────────────────────
+ with gr.Tab("🗺️ Mapping Manager"):
+ gr.HTML("""
+
+
🗺️ Mapping Manager
+
+ Scholarship Application Reference: The mirror of what Scholarship Application currently knows. When you add a new district to Scholarship Application, find it here and change its status from new districts found to present.
+
+
📁 Scholarship Application District Reference
")
+ with gr.Row():
+ refresh_ref_btn = gr.Button("🔄 Load from Dataset", variant="secondary", scale=1)
+
+ # Filters
+ with gr.Row():
+ ref_filter_state = gr.Dropdown(label="Filter by State", choices=["All"], value="All", scale=1, interactive=True)
+ ref_filter_status = gr.Dropdown(label="Filter by Status", choices=["All", "present", "new districts found"], value="All", scale=1, interactive=True)
+ ref_filter_dist = gr.Textbox(label="Search District", placeholder="Type to search...", scale=2, interactive=True)
+
+ # The DataFrame
+ ref_full_state = gr.State(pd.DataFrame())
+ ref_table = gr.Dataframe(
+ label="Scholarship Application District Reference (Editable)",
+ interactive=False, wrap=True
+ )
+
+ with gr.Accordion("📦 Bulk Excel Import / Export", open=False):
+ with gr.Row():
+ ref_prep_dl_btn = gr.Button("📦 Prepare Current View for Download")
+ ref_dl_btn = gr.DownloadButton("📥 Download Excel", visible=False)
+
+ with gr.Row():
+ ref_upload = gr.File(label="Upload Updated Excel", file_types=[".xlsx"])
+
+ gr.HTML("Click any row in the table above to edit its status here:
")
+ with gr.Row():
+ ref_edit_state = gr.Textbox(label="State", interactive=False, scale=2)
+ ref_edit_dist = gr.Textbox(label="District", interactive=False, scale=2)
+ ref_edit_status = gr.Dropdown(choices=["present", "new districts found"], label="Status", scale=1)
+ ref_edit_btn = gr.Button("💾 Update Row", variant="primary", scale=1)
+
+ ref_status_html = gr.HTML(value="")
+
+ gr.HTML("Add a District
")
+ with gr.Row():
+ ref_state_add = gr.Textbox(label="State", placeholder="e.g. ANDHRA PRADESH", scale=2)
+ ref_dist_add = gr.Textbox(label="District", placeholder="e.g. ALLURI SITHARAMA RAJU", scale=2)
+ ref_status_add = gr.Dropdown(choices=["present", "new districts found"], value="present", label="Status", scale=1)
+ ref_add_btn = gr.Button("➕ Add District", variant="primary", scale=1)
+ ref_add_status = gr.HTML(value="")
+
+ gr.HTML("Rename a District (automatically updates mappings)
")
+ with gr.Row():
+ ref_ren_state = gr.Dropdown(label="State", choices=[], allow_custom_value=True, scale=2)
+ ref_ren_old = gr.Dropdown(label="Old Name", choices=[], allow_custom_value=True, scale=2)
+ ref_ren_new = gr.Textbox(label="New Name", placeholder="e.g. VISAKHAPATNAM (NEW)", scale=2)
+ ref_rename_btn = gr.Button("✏️ Rename", variant="secondary", scale=1)
+ ref_rename_status = gr.HTML(value="")
+
+ gr.HTML("Delete Actions
")
+ with gr.Row():
+ ref_del_state = gr.Dropdown(label="State", choices=[], allow_custom_value=True, scale=2)
+ ref_del_dist = gr.Dropdown(label="District", choices=[], allow_custom_value=True, scale=2)
+ ref_del_dist_btn = gr.Button("🗑️ Delete District", variant="stop", scale=1)
+ ref_del_state_btn = gr.Button("🚨 Delete Entire State", variant="stop", scale=1)
+ ref_del_status = gr.HTML(value="")
+
+ def on_ref_table_select(df, evt: gr.SelectData):
+ row = evt.index[0]
+ if df is None or df.empty or row >= len(df):
+ return gr.update(), gr.update(), gr.update(), ""
+
+ state = df.iloc[row]["State"]
+ dist = df.iloc[row]["District"]
+ status = df.iloc[row]["Status"]
+ return gr.update(value=state), gr.update(value=dist), gr.update(value=status), ""
+
+ def ui_ref_edit_row(state, district, new_status):
+ from hf_store import get_hf_credentials, update_district_reference_add
+ token, repo = get_hf_credentials()
+ if not token or not repo or not state or not district:
+ return "⚠️ Please select a row from the table first.
"
+ try:
+ update_district_reference_add(state, district, new_status, token, repo)
+ return f"✅ Successfully updated {district} to '{new_status}'. Refresh the table to see changes.
"
+ except Exception as ex:
+ return f"❌ Error: {ex}
"
+
+ def ui_refresh_ref():
+ from hf_store import get_hf_credentials, pull_district_reference
+ token, repo = get_hf_credentials()
+ if not token or not repo:
+ empty = pd.DataFrame()
+ u = gr.update()
+ return empty, empty, u, u, u, u, u, "⚠️ HF credentials not set.
"
+ df = pull_district_reference(token, repo)
+ if df.empty:
+ empty = pd.DataFrame()
+ u = gr.update()
+ return empty, empty, u, u, u, u, u, "⚠️ Scholarship Application Reference empty.
"
+ states = sorted(list(df["State"].unique()))
+ dists = sorted(list(df["District"].unique()))
+ st_up = gr.update(choices=["All"] + states, value="All")
+ st_dd = gr.update(choices=states)
+ dist_dd = gr.update(choices=dists)
+ return df, df, st_up, st_dd, dist_dd, st_dd, dist_dd, f"✅ Loaded {len(df)} rows.
"
+
+ def filter_ref_table(df, filter_state, filter_status, search_dist):
+ if df is None or df.empty: return df
+ res = df.copy()
+ if filter_state and filter_state != "All":
+ res = res[res["State"] == filter_state]
+ if filter_status and filter_status != "All":
+ res = res[res["Status"] == filter_status]
+ if search_dist:
+ res = res[res["District"].str.contains(search_dist.upper(), na=False)]
+ return res
+
+ def ui_save_table_edits(df, full_df, filter_state, filter_status, search_dist):
+ from hf_store import get_hf_credentials, push_district_reference
+ token, repo = get_hf_credentials()
+ if df is None or df.empty:
+ return "⚠️ Table is empty.
", full_df
+ # The user edited 'df'. We need to merge it back into 'full_df'
+ # This is a bit complex if they deleted rows or changed keys.
+ # Actually, if they are filtering, df only contains a subset.
+ # Let's just overwrite the subset in full_df based on index if we kept index,
+ # but Gradio df doesn't keep original indices easily.
+ # Simplest way: They should only save edits if viewing All/All without search.
+ if filter_state != "All" or filter_status != "All" or search_dist:
+ return "⚠️ Please clear all filters (set to 'All', clear search) before saving direct table edits.
", full_df
+
+ try:
+ df["State"] = df["State"].astype(str).str.strip().str.upper()
+ df["District"] = df["District"].astype(str).str.strip().str.upper()
+ df["Status"] = df["Status"].astype(str).str.strip()
+
+ invalid_statuses = df[~df["Status"].isin(["present", "new districts found"])]
+ if not invalid_statuses.empty:
+ return "⚠️ Invalid Status found. You can only use 'present' or 'new districts found'.
", full_df
+
+ df = df.drop_duplicates()
+ push_district_reference(df, token, repo)
+ return f"✅ Saved {len(df)} rows directly to cloud.
", df
+ except Exception as ex:
+ return f"❌ Error: {ex}
", full_df
+
+ def ui_prep_excel_download(df):
+ if df is None or df.empty:
+ return gr.update(visible=False), "⚠️ No data to download.
"
+ import tempfile
+ import os
+ fd, temp_excel = tempfile.mkstemp(suffix=".xlsx")
+ os.close(fd)
+ df.to_excel(temp_excel, index=False)
+ return gr.update(value=temp_excel, visible=True), "✅ Ready to download!
"
+
+ def ui_upload_excel(file):
+ if not file: return pd.DataFrame(), "⚠️ No file uploaded.
"
+ import pandas as pd
+ from hf_store import get_hf_credentials, push_district_reference, pull_district_reference
+ token, repo = get_hf_credentials()
+ try:
+ uploaded_df = pd.read_excel(file.name)
+ if "State" not in uploaded_df.columns or "District" not in uploaded_df.columns or "Status" not in uploaded_df.columns:
+ return pd.DataFrame(), "⚠️ Excel must have State, District, and Status columns.
"
+
+ uploaded_df["State"] = uploaded_df["State"].astype(str).str.strip().str.upper()
+ uploaded_df["District"] = uploaded_df["District"].astype(str).str.strip().str.upper()
+ uploaded_df["Status"] = uploaded_df["Status"].astype(str).str.strip()
+
+ invalid_statuses = uploaded_df[~uploaded_df["Status"].isin(["present", "new districts found"])]
+ if not invalid_statuses.empty:
+ return pd.DataFrame(), "⚠️ Invalid Status found in Excel. You can only use 'present' or 'new districts found'.
"
+
+ uploaded_df = uploaded_df.drop_duplicates(subset=["State", "District"])
+
+ full_df = pull_district_reference(token, repo)
+ if not full_df.empty:
+ full_df.set_index(["State", "District"], inplace=True)
+ uploaded_df.set_index(["State", "District"], inplace=True)
+ full_df.update(uploaded_df)
+ new_rows = uploaded_df[~uploaded_df.index.isin(full_df.index)]
+ full_df = pd.concat([full_df, new_rows])
+ full_df.reset_index(inplace=True)
+ df_to_push = full_df
+ else:
+ df_to_push = uploaded_df.copy()
+
+ push_district_reference(df_to_push, token, repo)
+ return df_to_push, f"✅ Successfully merged {len(uploaded_df)} uploaded rows into the dataset!
"
+ except Exception as ex:
+ return pd.DataFrame(), f"❌ Error: {ex}
"
+
+ def ui_ref_add(state, district, sf_status):
+ from hf_store import get_hf_credentials, update_district_reference_add
+ token, repo = get_hf_credentials()
+ if not token or not repo or not state or not district or not state.strip() or not district.strip():
+ return "⚠️ Fill in all fields and check HF credentials.
"
+ try:
+ update_district_reference_add(state.strip().upper(), district.strip().upper(), sf_status, token, repo)
+ return f"✅ Added/updated {district.upper()} in {state.upper()}. Refresh table to see.
"
+ except Exception as ex:
+ return f"❌ Error: {ex}
"
+
+ def ui_ref_rename(state, old_name, new_name):
+ from hf_store import get_hf_credentials, update_district_reference_rename
+ token, repo = get_hf_credentials()
+ if not token or not repo or not state or not old_name or not new_name or not state.strip() or not old_name.strip() or not new_name.strip():
+ return "⚠️ Fill in all fields.
"
+ try:
+ update_district_reference_rename(state.strip().upper(), old_name.strip().upper(), new_name.strip().upper(), token, repo)
+ return f"✅ Renamed {old_name.upper()} → {new_name.upper()} (cascaded to Dataset 2). Refresh table to see.
"
+ except Exception as ex:
+ return f"❌ Error: {ex}
"
+
+ def ui_ref_del_dist(state, district):
+ from hf_store import get_hf_credentials, delete_district_reference
+ token, repo = get_hf_credentials()
+ if not token or not repo or not state or not district or not state.strip() or not district.strip():
+ return "⚠️ Fill in both State and District.
"
+ try:
+ delete_district_reference(state.strip().upper(), district.strip().upper(), token, repo)
+ return f"✅ Deleted {district.upper()} from {state.upper()}. Refresh table to see.
"
+ except Exception as ex:
+ return f"❌ Error: {ex}
"
+
+ def ui_ref_del_state(state):
+ from hf_store import get_hf_credentials, delete_state_reference
+ token, repo = get_hf_credentials()
+ if not token or not repo or not state:
+ return "⚠️ Fill in State.
"
+ try:
+ delete_state_reference(state.strip().upper(), token, repo)
+ return f"✅ Deleted entire state {state.upper()}. Refresh table to see.
"
+ except Exception as ex:
+ return f"❌ Error: {ex}
"
+
+ # Wirings
+ refresh_ref_btn.click(
+ fn=ui_refresh_ref,
+ outputs=[ref_table, ref_full_state, ref_filter_state,
+ ref_ren_state, ref_ren_old,
+ ref_del_state, ref_del_dist,
+ ref_status_html]
+ )
+ ref_filter_state.change(fn=filter_ref_table, inputs=[ref_full_state, ref_filter_state, ref_filter_status, ref_filter_dist], outputs=[ref_table])
+ ref_filter_status.change(fn=filter_ref_table, inputs=[ref_full_state, ref_filter_state, ref_filter_status, ref_filter_dist], outputs=[ref_table])
+ ref_filter_dist.change(fn=filter_ref_table, inputs=[ref_full_state, ref_filter_state, ref_filter_status, ref_filter_dist], outputs=[ref_table])
+
+ ref_table.select(fn=on_ref_table_select, inputs=[ref_table], outputs=[ref_edit_state, ref_edit_dist, ref_edit_status, ref_status_html])
+ ref_edit_btn.click(fn=ui_ref_edit_row, inputs=[ref_edit_state, ref_edit_dist, ref_edit_status], outputs=[ref_status_html])
+ ref_prep_dl_btn.click(fn=ui_prep_excel_download, inputs=[ref_table], outputs=[ref_dl_btn, ref_status_html])
+ ref_upload.upload(fn=ui_upload_excel, inputs=[ref_upload], outputs=[ref_table, ref_status_html])
+
+ ref_add_btn.click(fn=ui_ref_add, inputs=[ref_state_add, ref_dist_add, ref_status_add], outputs=[ref_add_status])
+ ref_rename_btn.click(fn=ui_ref_rename, inputs=[ref_ren_state, ref_ren_old, ref_ren_new], outputs=[ref_rename_status])
+ ref_del_dist_btn.click(fn=ui_ref_del_dist, inputs=[ref_del_state, ref_del_dist], outputs=[ref_del_status])
+ ref_del_state_btn.click(fn=ui_ref_del_state, inputs=[ref_del_state], outputs=[ref_del_status])
+
+ # ── Tab 4: Download Master Sheets ──────────────────────────────────────
+ with gr.Tab("📥 Download History"):
+ gr.HTML("Download Previously Built Master Sheets
")
+
+ with gr.Row():
+ dl_refresh_btn = gr.Button("🔄 Refresh File List", scale=1)
+ dl_dropdown = gr.Dropdown(choices=[], label="Select a Master Sheet", scale=3)
+
+ dl_status = gr.HTML()
+ dl_download_btn = gr.DownloadButton("📥 Download as Excel", visible=False)
+
+ def ui_refresh_dl_list():
+ from hf_store import get_hf_credentials, list_mapped_master_files
+ token, repo = get_hf_credentials()
+ if not token or not repo:
+ return gr.update(choices=[]), "⚠️ HF credentials not set.
"
+
+ files = list_mapped_master_files(token, repo)
+ if not files:
+ return gr.update(choices=[]), "⚠️ No master sheets found in dataset.
"
+
+ filenames = [f.split("/")[-1] for f in files]
+ filenames.sort(reverse=True) # newest first
+ return gr.update(choices=filenames), f"✅ Found {len(filenames)} master sheets.
"
+
+ dl_refresh_btn.click(fn=ui_refresh_dl_list, outputs=[dl_dropdown, dl_status])
+
+ def ui_prep_download(filename):
+ if not filename: return gr.update(visible=False), ""
+ from hf_store import get_hf_credentials
+ import tempfile
+ import pandas as pd
+ from huggingface_hub import hf_hub_download
+ token, repo = get_hf_credentials()
+
+ try:
+ local_path = hf_hub_download(repo_id=repo, repo_type="dataset", filename=f"scraped_data/mapped/{filename}", token=token)
+ df = pd.read_parquet(local_path)
+
+ temp_dir = tempfile.gettempdir()
+ # Keep the original filename but change extension to .xlsx
+ excel_filename = filename.replace('.parquet', '.xlsx')
+ temp_excel = os.path.join(temp_dir, excel_filename)
+ df.to_excel(temp_excel, index=False)
+
+ return gr.update(value=temp_excel, visible=True), "✅ Ready to download!
"
+ except Exception as e:
+ return gr.update(visible=False), f"❌ Error preparing download: {e}
"
+
+ dl_dropdown.change(fn=ui_prep_download, inputs=[dl_dropdown], outputs=[dl_download_btn, dl_status])
+
+
+ # ── Tab 5: Master Analytics ────────────────────────────────────────────
+ with gr.Tab("📊 Master Analytics"):
+ gr.HTML('''
+
+
📈 Multi-Year Master Analytics
+
+ This tab automatically loads the 2025 master and every
+ mapped master sheet ever built. It analyzes how government school footprints and overall school counts shift year over year.
+
+
📅 Master Timeline
")
+ gr.HTML("Year-over-year changes across all master sheets compared to 2025.
")
+ timeline_table = gr.Dataframe(label="Timeline Metrics", interactive=False, wrap=True)
+
+ gr.HTML("⚠️ Missing Schools (vs 2025)
")
+ missing_schools_html = gr.HTML()
+ missing_schools_table = gr.Dataframe(label="Missing Schools Detailed", interactive=False, wrap=True)
+
+ gr.HTML("🔄 Management Shifts
")
+ mgmt_shifts_html = gr.HTML()
+ mgmt_shifts_table = gr.Dataframe(label="Government → Non-Govt Shifts", interactive=False, wrap=True)
+
+ BASE_CSV_PATH = os.path.join(os.path.dirname(os.path.abspath(__file__)), "master_all_states.csv")
+
+ def _comp_detect_year_label(filepath: str) -> str:
+ import calendar as _cal
+ import re as _re
+ import os as _os
+ name = _os.path.basename(filepath or "")
+ m = _re.search(r"(\d{4})[-_](\d{2})", name)
+ if m:
+ yr, mo = int(m.group(1)), int(m.group(2))
+ if 1 <= mo <= 12:
+ return f"{_cal.month_abbr[mo]} {yr}"
+ m2 = _re.search(r"(jan|feb|mar|apr|may|jun|jul|aug|sep|oct|nov|dec)[_\-]?(\d{4})", name, _re.IGNORECASE)
+ if m2:
+ return f"{m2.group(1).capitalize()} {m2.group(2)}"
+ m3 = _re.search(r"(\d{4})", name)
+ return m3.group(1) if m3 else name
+
+ def _comp_find_col(df, *keywords):
+ for k in keywords:
+ for c in df.columns:
+ if k in c.lower():
+ return c
+ return None
+
+ def ui_run_analytics():
+ from hf_store import get_hf_credentials, list_mapped_master_files, pull_baseline_master
+ from huggingface_hub import hf_hub_download
+ import pandas as pd
+ import os
+
+ token, repo = get_hf_credentials()
+ if not token or not repo:
+ return "⚠️ HF credentials not set.
", pd.DataFrame(), "", pd.DataFrame(), "", pd.DataFrame()
+
+ masters = []
+ # 1. Load baseline
+ try:
+ base_df = pull_baseline_master(token, repo)
+ if not base_df.empty:
+ masters.append({"label": "2025", "df": base_df})
+ except Exception:
+ pass
+
+ # Fallback to CSV if HF fails
+ if not masters and os.path.exists(BASE_CSV_PATH):
+ masters.append({"label": "2025", "df": pd.read_csv(BASE_CSV_PATH, low_memory=False)})
+
+ # 2. Pull HF mapped masters
+ hf_files = list_mapped_master_files(token, repo)
+ for hf_path in sorted(hf_files):
+ try:
+ local = hf_hub_download(repo_id=repo, repo_type="dataset", filename=hf_path, token=token, force_download=True)
+ label = _comp_detect_year_label(hf_path)
+ masters.append({"label": label, "df": pd.read_parquet(local)})
+ except Exception:
+ pass
+
+ if not masters:
+ return "⚠️ No master datasets found.
", pd.DataFrame(), "", pd.DataFrame(), "", pd.DataFrame()
+
+ def _norm_udise(s):
+ return s.astype(str).str.strip().str.zfill(11)
+
+ def is_govt(s):
+ val = str(s).lower()
+ if 'aided' in val or 'private' in val or 'unaided' in val or 'un-aided' in val:
+ return False
+ return 'gov' in val or 'dept of edu' in val or 'local body' in val or 'panchayat' in val or 'municipal' in val
+
+ baseline = masters[0]
+ b_udise = _comp_find_col(baseline["df"], "udise")
+ b_mgmt = _comp_find_col(baseline["df"], "school_management_type", "management", "mgmt")
+ if b_udise: baseline["df"]["_U"] = _norm_udise(baseline["df"][b_udise])
+ if b_mgmt: baseline["df"]["is_govt"] = baseline["df"][b_mgmt].apply(is_govt)
+
+ timeline_data = []
+ baseline_udises = set(baseline["df"]["_U"].dropna()) if b_udise else set()
+
+ missing_df = pd.DataFrame()
+ changed_mgmt = pd.DataFrame()
+
+ for m in masters:
+ m_df = m["df"]
+ m_udise = _comp_find_col(m_df, "udise")
+ m_mgmt = _comp_find_col(m_df, "school_management_type", "management", "mgmt")
+ m_name = _comp_find_col(m_df, "school_name", "schoolname")
+ m_state = _comp_find_col(m_df, "state")
+
+ if m_udise: m_df["_U"] = _norm_udise(m_df[m_udise])
+ if m_mgmt: m_df["is_govt"] = m_df[m_mgmt].apply(is_govt)
+
+ tot_schools = len(m_df)
+ tot_govt = m_df["is_govt"].sum() if m_mgmt else 0
+
+ missing_count = 0
+ shifts_count = 0
+
+ if m != baseline and b_udise and m_udise:
+ m_udises = set(m_df["_U"].dropna())
+ missing = baseline_udises - m_udises
+ missing_count = len(missing)
+
+ if m == masters[-1]:
+ b_name = _comp_find_col(baseline["df"], "school_name", "schoolname")
+ b_state = _comp_find_col(baseline["df"], "state")
+ missing_df = baseline["df"][baseline["df"]["_U"].isin(missing)].drop_duplicates(subset=["_U"]).copy()
+
+ cols_to_keep = []
+ for c in ["_U", b_name, b_state, b_mgmt]:
+ if c and c in missing_df.columns:
+ cols_to_keep.append(c)
+ missing_df = missing_df[cols_to_keep] if cols_to_keep else missing_df
+ missing_df.insert(0, "Missing In Master", m["label"])
+
+ if b_mgmt and m_mgmt:
+ b_cols = [c for c in ["_U", b_name, b_mgmt, "is_govt"] if c]
+ m_cols = [c for c in ["_U", m_name, m_mgmt, "is_govt"] if c]
+
+ merged = pd.merge(
+ baseline["df"][b_cols],
+ m_df[m_cols],
+ on="_U", suffixes=("_old", "_new")
+ )
+ changed = merged[(merged["is_govt_old"] == True) & (merged["is_govt_new"] == False)]
+ shifts_count = len(changed)
+
+ c_name_old = str(b_name) + '_old' if b_name == m_name else b_name
+ c_mgmt_old = str(b_mgmt) + '_old' if b_mgmt == m_mgmt else b_mgmt
+ c_mgmt_new = str(m_mgmt) + '_new' if b_mgmt == m_mgmt else m_mgmt
+
+ c_cols = [c for c in ["_U", c_name_old, c_mgmt_old, c_mgmt_new] if c and c in changed.columns]
+ changed_mgmt = changed[c_cols] if c_cols else changed
+ changed_mgmt.insert(0, "Shifted In Master", m["label"])
+
+ timeline_data.append({
+ "Master Name": m["label"],
+ "Total Schools": f"{tot_schools:,}",
+ "Total Govt Schools": f"{tot_govt:,}",
+ "Missing (vs 2025)": f"{missing_count:,}" if m != baseline else "-",
+ "Govt->Non-Govt Shifts": f"{shifts_count:,}" if m != baseline else "-"
+ })
+
+ timeline_df = pd.DataFrame(timeline_data)
+
+ status_html = f"✅ Loaded {len(masters)} master datasets successfully.
"
+ m_html = f"{len(missing_df):,} schools from 2025 are completely missing in the newest master.
"
+ c_html = f"{len(changed_mgmt):,} schools shifted from Govt to Non-Govt management since 2025.
"
+
+ return status_html, timeline_df, m_html, missing_df, c_html, changed_mgmt
+
+ analytics_refresh_btn.click(
+ fn=ui_run_analytics,
+ outputs=[analytics_status, timeline_table, missing_schools_html, missing_schools_table, mgmt_shifts_html, mgmt_shifts_table]
+ )
+
+if __name__ == "__main__":
+ if os.environ.get("SPACE_ID"):
+ print("Starting on Dataset Spaces (0.0.0.0:7860) …")
+ app.queue().launch(server_name="0.0.0.0", server_port=7860)
+ else:
+ print("Starting locally …")
+ app.queue().launch(server_name="127.0.0.1", server_port=7861, inbrowser=True)