""" national_analysis.py ───────────────────── Background analysis engine for KYS school scraper. This module is NOT meant to be run as a standalone script. It exposes reusable functions that are called automatically in the background after a scrape is pushed to the dataset. Key function: run_analysis(old_df, new_df) -> AnalysisResult """ import pandas as pd def clean_text(series: pd.Series) -> pd.Series: return series.astype(str).str.strip().str.upper() def clean_udise(series: pd.Series) -> pd.Series: return series.astype(str).str.replace(r'\.0$', '', regex=True).str.zfill(11) def format_udise_list(udise_series: pd.Series) -> str: """Joins UDISEs with commas. Safely truncates to prevent Excel crash.""" joined = ", ".join(udise_series.astype(str)) if len(joined) > 30000: return joined[:30000] + " ... (+ more)" return joined class AnalysisResult: """Container for all analysis outputs.""" def __init__(self): self.national_summary: pd.DataFrame = pd.DataFrame() self.state_summary: pd.DataFrame = pd.DataFrame() self.district_mapping: pd.DataFrame = pd.DataFrame() self.block_to_district: pd.DataFrame = pd.DataFrame() self.block_mapping: pd.DataFrame = pd.DataFrame() self.added_schools: pd.DataFrame = pd.DataFrame() self.deleted_schools: pd.DataFrame = pd.DataFrame() self.new_districts: pd.DataFrame = pd.DataFrame() self.category_transitions: pd.DataFrame = pd.DataFrame() self.flagged_new_districts: list = [] # [{state, district, reason}] def run_analysis(old_df: pd.DataFrame, new_df: pd.DataFrame) -> AnalysisResult: """ Compare old master (SF/2025 data) and new scraped master data. Returns AnalysisResult with all comparison DataFrames. Parameters ---------- old_df : pd.DataFrame The old/SF master data (2025). new_df : pd.DataFrame The newly scraped master data. """ result = AnalysisResult() # ── Clean inputs ────────────────────────────────────────────────────────── for df in [old_df, new_df]: df["School_State__c"] = clean_text(df["School_State__c"]) df["UDISE"] = clean_udise(df["School_Udise_Code__c"]) old_df = old_df.drop_duplicates(subset=["UDISE"]).set_index("UDISE") new_df = new_df.drop_duplicates(subset=["UDISE"]).set_index("UDISE") old_udises = set(old_df.index) new_udises = set(new_df.index) added_udises = new_udises - old_udises deleted_udises = old_udises - new_udises common_udises = old_udises & new_udises # ── Added / Deleted schools ─────────────────────────────────────────────── _school_cols = ["UDISE", "School_State__c", "School_District__c", "School_Block__c"] result.added_schools = ( new_df.loc[list(added_udises)].reset_index()[_school_cols] if added_udises else pd.DataFrame(columns=_school_cols) ) result.deleted_schools = ( old_df.loc[list(deleted_udises)].reset_index()[_school_cols] if deleted_udises else pd.DataFrame(columns=_school_cols) ) # ── State-by-state summary ──────────────────────────────────────────────── all_states = sorted( set(old_df["School_State__c"].unique()) | set(new_df["School_State__c"].unique()) ) state_rows = [] for state in all_states: s_old = set(old_df[old_df["School_State__c"] == state].index) s_new = set(new_df[new_df["School_State__c"] == state].index) state_rows.append({ "State": state, "Old Count": len(s_old), "New Count": len(s_new), "Added": len(s_new - s_old), "Deleted": len(s_old - s_new), }) result.state_summary = pd.DataFrame(state_rows) # ── Inner join for deep comparison ──────────────────────────────────────── if not common_udises: return result df_common = old_df.loc[list(common_udises)].join( new_df.loc[list(common_udises)], lsuffix="_OLD", rsuffix="_NEW" ).reset_index() # Output Schema: OLD = Scraped Data (needs to be fixed), NEW = SF Baseline (the correct target) df_common["State"] = clean_text(df_common["School_State__c_NEW"]) df_common["Dist_OLD"] = clean_text(df_common["School_District__c_NEW"]) # Scraped df_common["Dist_NEW"] = clean_text(df_common["School_District__c_OLD"]) # SF df_common["Block_OLD"] = clean_text(df_common["School_Block__c_NEW"]) # Scraped df_common["Block_NEW"] = clean_text(df_common["School_Block__c_OLD"]) # SF if "schCategoryType__c_OLD" in df_common.columns: df_common["Cat_OLD"] = clean_text(df_common["schCategoryType__c_OLD"]) df_common["Cat_NEW"] = clean_text(df_common["schCategoryType__c_NEW"]) # ── District mapping (includes unchanged for dominant count math) ───────── result.district_mapping = df_common.groupby(["State", "Dist_OLD", "Dist_NEW"]).agg( Affected_Schools=("UDISE", "count"), List_of_UDISEs=("UDISE", format_udise_list) ).reset_index() # ── Block to District mapping ───────────────────────────────────────────── result.block_to_district = df_common.groupby( ["State", "Dist_NEW", "Block_NEW", "Dist_OLD"] ).agg( Schools=("UDISE", "count"), List_of_UDISEs=("UDISE", format_udise_list) ).reset_index().sort_values( ["State", "Dist_NEW", "Block_NEW", "Schools"], ascending=[True, True, True, False] ) # ── Block mapping ───────────────────────────────────────────────────────── result.block_mapping = df_common.groupby( ["State", "Dist_OLD", "Block_OLD", "Block_NEW"] ).agg( Affected_Schools=("UDISE", "count"), List_of_UDISEs=("UDISE", format_udise_list) ).reset_index() # ── Category transitions ────────────────────────────────────────────────── if "Cat_OLD" in df_common.columns: cat_changes = df_common[df_common["Cat_OLD"] != df_common["Cat_NEW"]] if not cat_changes.empty: result.category_transitions = cat_changes.groupby( ["State", "Cat_OLD", "Cat_NEW"] ).agg( Count=("UDISE", "count"), List_of_UDISEs=("UDISE", format_udise_list) ).reset_index().sort_values(["State", "Count"], ascending=[True, False]) # ── National summary ────────────────────────────────────────────────────── result.national_summary = pd.DataFrame({ "Metric": [ "Total Old Schools", "Total New Schools", "Total Schools Added", "Total Schools Deleted", "Common Schools", "Schools with District Changes", "Schools with Block Changes", ], "Count": [ len(old_df), len(new_df), len(added_udises), len(deleted_udises), len(common_udises), int((df_common["Dist_OLD"] != df_common["Dist_NEW"]).sum()), int((df_common["Block_OLD"] != df_common["Block_NEW"]).sum()), ] }) # ── New districts (districts in new data but not in old data at all) ────── old_dist_names = set( old_df["School_District__c"].dropna().astype(str).str.strip().str.upper().unique() ) new_dist_df = new_df[["School_State__c", "School_District__c"]].drop_duplicates().dropna().copy() 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"]) new_dist_df = new_dist_df[~new_dist_df["New District Name"].isin(old_dist_names)] if not new_dist_df.empty: result.new_districts = new_dist_df[["State", "New District Name"]].sort_values( ["State", "New District Name"] ).reset_index(drop=True) result.flagged_new_districts = [ {"state": row["State"], "district": row["New District Name"], "reason": "new_in_udise"} for _, row in result.new_districts.iterrows() ] return result def detect_new_districts_against_reference(new_df: pd.DataFrame, ref_df: pd.DataFrame) -> list: """ Compare districts in newly scraped data against Dataset 1 (district_reference). Returns list of dicts: [{state, district, reason}] for districts not in reference. Parameters ---------- new_df : scraped DataFrame with School_State__c and School_District__c columns ref_df : district_reference DataFrame with State and District columns """ if ref_df.empty or "District" not in ref_df.columns: return [] known = set(zip( ref_df["State"].astype(str).str.strip().str.upper(), ref_df["District"].astype(str).str.strip().str.upper() )) scraped = ( new_df[["School_State__c", "School_District__c"]] .dropna() .drop_duplicates() ) scraped["State"] = clean_text(scraped["School_State__c"]) scraped["District"] = clean_text(scraped["School_District__c"]) flagged = [] for _, row in scraped.iterrows(): pair = (row["State"], row["District"]) if pair not in known: flagged.append({ "state": row["State"], "district": row["District"], "reason": "not_in_district_reference" }) return flagged