"""
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 ""}
"""
def _phase_html(text: str) -> str:
safe = re.sub(r"<[^>]+>", "", re.sub(r"\s*\(id=\d+\)", "", text.split("\n")[0]))[:200]
return f"""
{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"""
"""
def _stats_html(success: int, no_data: int, failed: int, total: int) -> str:
return f"""
"""
def _result_html(banner: str, stats: str) -> str:
if not banner and not stats:
return ""
return banner + stats
def _idle_html() -> str:
return "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
""")
# Holds the live analysis results dict across tabs
analysis_state = gr.State({})
with gr.Tabs():
# ── Tab 1: Scraper ────────────────────────────────────────────────────
with gr.Tab("🔍 Scraper"):
gr.HTML("""
👋 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.
""")
with gr.Row(equal_height=True):
with gr.Column(scale=1):
gr.HTML("📍 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}
"""
return """
⚠️
Dataset not set up
Add your HF_TOKEN and HF_REPO to the .env file to enable dataset features.
"""
hf_env_status = gr.HTML(value=_env_status())
gr.HTML("")
gr.HTML("""
📡 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.
""")
gr.HTML("📁 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])
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)