national_analysis.py

"""
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