"""
Nafs Results Analytics - Aseer Region
=====================================
End-to-end pipeline: raw extract -> data-quality rules -> conformed dataset
-> analytical model -> executive dashboard feed (JSON) + static charts.

Author : Adel Asiri (CDMP, PMP)
Stack  : Python 3 - pandas - NumPy - SciPy - Matplotlib

    python nafs_pipeline.py            # full run
Outputs (./output):
    dashboard.json         feed for the web executive dashboard
    dq_report.csv          data-quality rule results
    quarantine.csv         records held back by validation rules
    priority_schools.csv   schools recommended for intervention
    charts/*.png           static charts for reports
"""
from __future__ import annotations

import json
import re
from datetime import datetime, timezone
from pathlib import Path

import numpy as np
import pandas as pd
from scipy import stats

import matplotlib
matplotlib.use("Agg")
import matplotlib.pyplot as plt  # noqa: E402

ROOT = Path(__file__).resolve().parent
RAW = ROOT / "data" / "nafs_aseer_raw.csv"
REF = ROOT / "data" / "reference"
OUT = ROOT / "output"
CHARTS = OUT / "charts"
CHARTS.mkdir(parents=True, exist_ok=True)

MIN_N = 15        # small-cell suppression: minimum tested students per school-year
PRIORITY_N = 25   # minimum cohort to be listed for intervention
MATERIAL = 5.0    # material gap / decline in percentage points

# ---------------------------------------------------------------- column map
COLS = {
    "العام الدراسي": "year", "الرقم الوزاري": "school_id", "المدينة/المحافظة": "gov",
    "السلطة": "authority", "جنس المدرسة": "gender", "المرحلة": "stage", "الصف": "grade",
    "المجال": "domain", "المجال الفرعي": "subdomain",
    "عدد الطلاب المستهدفين": "expected", "مجموع المختبرين": "tested", "عدد المتقنين": "proficient",
    "نسبة المتقنين": "prof_rate", "متوسط الدرجة (مقياس 100)": "score",
    "نسبة الطلاب في مستوى الأداء المرتفع": "lvl_high", "نسبة الطلاب في مستوى الأداء المتوسط": "lvl_mid",
    "نسبة الطلاب في مستوى الأداء المنخفض": "lvl_low", "نسبة الطلاب في مستوى الأداء المنخفض جدا": "lvl_vlow",
    "المتوسط على مستوى المملكة": "national",
}
KEY = ["year", "school_id", "grade", "domain", "subdomain"]


def normalize_ar(s: str) -> str:
    """Conform free-text Arabic values before matching reference data."""
    if not isinstance(s, str):
        return s
    s = s.strip()
    s = re.sub(r"\s+", " ", s)
    s = re.sub("[إأآ]", "ا", s)
    return s


# ================================================================ 1. LOAD
def load():
    raw = pd.read_csv(RAW, encoding="utf-8-sig").rename(columns=COLS)
    govs = pd.read_csv(REF / "governorates.csv", encoding="utf-8-sig")
    subjects = pd.read_csv(REF / "subjects.csv", encoding="utf-8-sig")
    grades = pd.read_csv(REF / "grades.csv", encoding="utf-8-sig")
    return raw, govs, subjects, grades


# ================================================================ 2. DATA QUALITY
def data_quality(raw, govs, subjects, grades):
    df = raw.copy()
    n0 = len(df)
    rules, quarantine = [], []

    def rule(rid, dim_ar, dim_en, rule_ar, rule_en, failed, action_ar, action_en, checked=None):
        checked = checked or n0
        rules.append(dict(id=rid, dim_ar=dim_ar, dim_en=dim_en, rule_ar=rule_ar, rule_en=rule_en,
                          checked=int(checked), failed=int(failed),
                          pass_rate=round(100 * (1 - failed / checked), 2),
                          action_ar=action_ar, action_en=action_en))

    # DQ-001 completeness - school identifier is mandatory
    m = df["school_id"].isna()
    rule("DQ-001", "الاكتمال", "Completeness", "الرقم الوزاري غير فارغ", "School ID is not null",
         m.sum(), "عزل السجل وإعادته للمصدر", "Quarantine & return to source")
    quarantine.append(df[m].assign(dq_rule="DQ-001"))
    df = df[~m].copy()
    df["school_id"] = df["school_id"].astype(int)

    # DQ-002 consistency - governorate conforms to reference list
    ref_map = {normalize_ar(g): g for g in govs["gov_ar"]}
    conformed = df["gov"].map(lambda v: ref_map.get(normalize_ar(v)))
    changed = (conformed != df["gov"]) & conformed.notna()
    unknown = conformed.isna()
    rule("DQ-002", "الاتساق", "Consistency", "اسم المحافظة مطابق للبيانات المرجعية",
         "Governorate matches reference data", changed.sum() + unknown.sum(),
         "توحيد القيمة آليًا من الجدول المرجعي", "Auto-conform from reference table")
    df["gov"] = conformed.fillna(df["gov"])

    # DQ-003 validity - grade/domain codes belong to the reference lists
    bad_code = ~df["grade"].isin(grades["grade"]) | ~df["domain"].isin(subjects["subject_ar"])
    rule("DQ-003", "الصحة", "Validity", "رمز الصف والمجال من القوائم المرجعية",
         "Grade & domain codes are in reference lists", bad_code.sum(),
         "عزل السجل وتصحيح الترميز", "Quarantine & fix coding")
    quarantine.append(df[bad_code].assign(dq_rule="DQ-003"))
    df = df[~bad_code].copy()

    # DQ-004 validity - proficiency between 0 and 100 and consistent with counts
    recomputed = 100 * df["proficient"] / df["tested"]
    bad_rate = (df["prof_rate"] < 0) | (df["prof_rate"] > 100) | ((recomputed - df["prof_rate"]).abs() > 0.5)
    rule("DQ-004", "الصحة", "Validity", "نسبة الإتقان بين 0 و100 ومطابقة للأعداد",
         "Proficiency within 0-100 and consistent with counts", bad_rate.sum(),
         "إعادة الاحتساب من الأعداد الأصلية", "Recompute from source counts")
    df["prof_rate"] = recomputed.round(2)

    # DQ-005 validity - tested cannot exceed expected
    over = df["tested"] > df["expected"]
    rule("DQ-005", "الصحة", "Validity", "المختبرون لا يتجاوزون المستهدفين",
         "Tested does not exceed expected", over.sum(),
         "عزل السجل للتحقق من ملف المصدر", "Quarantine for source verification")
    quarantine.append(df[over].assign(dq_rule="DQ-005"))
    df = df[~over].copy()

    # DQ-006 uniqueness - one row per year/school/grade/domain/sub-domain
    dup = df.duplicated(KEY, keep="first")
    rule("DQ-006", "التفرد", "Uniqueness", "عدم تكرار المدرسة والعام والصف والمجال",
         "No duplicate year/school/grade/domain", dup.sum(),
         "حذف التكرار بعد المراجعة", "Remove duplicates after review")
    df = df[~dup].copy()

    # DQ-007 timeliness - extract covers every expected reporting year
    expected_years = {1444, 1445, 1446, 1447}
    missing_years = expected_years - set(df["year"].unique())
    rule("DQ-007", "الحداثة", "Timeliness", "المستخرج يغطي جميع سنوات التقرير",
         "Extract covers all reporting years", len(missing_years),
         "تصعيد التأخير لمالك البيانات", "Escalate delay to data owner", checked=len(expected_years))

    q = pd.concat(quarantine, ignore_index=True)
    dq = pd.DataFrame(rules)
    dims = (dq.groupby(["dim_ar", "dim_en"], sort=False)
              .apply(lambda g: pd.Series({"score": round(100 * (1 - g["failed"].sum() / g["checked"].sum()), 2)}))
              .reset_index())
    targets = {"Completeness": 99.5, "Consistency": 99.5, "Validity": 99.0, "Uniqueness": 100.0, "Timeliness": 100.0}
    dims["target"] = dims["dim_en"].map(targets)
    summary = dict(raw_rows=int(n0), clean_rows=int(len(df)), quarantined=int(len(q)),
                   duplicates_removed=int(dup.sum()), conformed=int(changed.sum()),
                   overall=round(float(dims["score"].mean()), 2))
    return df, dq, dims, q, summary


# ================================================================ 3. MODEL
def build_model(df, subjects):
    code = dict(zip(subjects["subject_ar"], subjects["code"]))
    df["subject"] = df["domain"].map(code)
    main = df[df["subdomain"] == df["domain"]].copy()
    subs = df[df["subdomain"] != df["domain"]].copy()
    for lv in ["lvl_high", "lvl_mid", "lvl_low", "lvl_vlow"]:
        main[lv + "_n"] = main[lv] / 100 * main["tested"]
    main["score_x_n"] = main["score"] * main["tested"]
    return main, subs


def rate(g):
    return 100 * g["proficient"].sum() / g["tested"].sum()


# ================================================================ 4. ANALYSIS
def analyse(main, subs, govs):
    years = sorted(main["year"].unique())
    y1, y0 = years[-1], years[-2]
    gov_list = list(govs["gov_ar"])
    gov_idx = {g: i for i, g in enumerate(gov_list)}
    auth_codes = {"حكومي": "PUB", "أهلي": "PRV", "عالمي وأجنبي": "INT"}
    gen_codes = {"بنين": "B", "بنات": "G"}

    # --- aggregation cube for the interactive dashboard
    cube = (main.assign(g=main["gov"].map(gov_idx), a=main["authority"].map(auth_codes), s=main["gender"].map(gen_codes))
                .groupby(["year", "grade", "subject", "s", "a", "g"])
                .agg(tested=("tested", "sum"), expected=("expected", "sum"), prof=("proficient", "sum"),
                     score=("score_x_n", "sum"), hi=("lvl_high_n", "sum"), mid=("lvl_mid_n", "sum"),
                     low=("lvl_low_n", "sum"), vlow=("lvl_vlow_n", "sum"), schools=("school_id", "nunique"))
                .reset_index())
    num = ["score", "hi", "mid", "low", "vlow"]
    cube[num] = cube[num].round(1)

    # --- national benchmark (constant within year x grade x subject)
    nat = main.groupby(["year", "grade", "subject"])["national"].mean().round(2)
    national = {}
    for (y, g, s), v in nat.items():
        national.setdefault(str(y), {}).setdefault(str(g), {})[s] = float(v)

    # --- regional trend + national weighted by regional mix
    trend = []
    for y in years:
        d = main[main["year"] == y]
        w = d.groupby(["grade", "subject"])["tested"].sum()
        nat_w = sum(national[str(y)][str(g)][s] * n for (g, s), n in w.items()) / w.sum()
        trend.append(dict(year=int(y), region=round(rate(d), 2), national=round(nat_w, 2),
                          participation=round(100 * d["tested"].sum() / d["expected"].sum(), 2)))

    # --- school-level indicators (latest two years)
    sch = (main[main["year"].isin([y0, y1])]
           .groupby(["school_id", "year"])
           .agg(prof=("proficient", "sum"), tested=("tested", "sum"))
           .reset_index())
    sch["rate"] = 100 * sch["prof"] / sch["tested"]
    wide = sch.pivot(index="school_id", columns="year", values=["rate", "tested"])
    wide.columns = [f"{a}_{b}" for a, b in wide.columns]
    wide = wide.dropna()
    wide = wide[(wide[f"tested_{y1}"] >= MIN_N) & (wide[f"tested_{y0}"] >= MIN_N)]
    attrs = main.drop_duplicates("school_id").set_index("school_id")[["gov", "stage", "gender", "authority"]]
    wide = wide.join(attrs)
    wide["change"] = wide[f"rate_{y1}"] - wide[f"rate_{y0}"]
    region_now = next(t["region"] for t in trend if t["year"] == y1)

    def seg(r):
        up = r["change"] >= 0
        hi = r[f"rate_{y1}"] >= region_now
        return "lead_up" if hi and up else "lead_down" if hi else "low_up" if up else "low_down"
    wide["segment"] = wide.apply(seg, axis=1)
    # priority = materially below the regional average AND materially declining, with a stable cohort
    wide["priority"] = ((wide[f"rate_{y1}"] <= region_now - MATERIAL) & (wide["change"] <= -MATERIAL)
                        & (wide[f"tested_{y1}"] >= PRIORITY_N))
    wide["priority_score"] = (region_now - wide[f"rate_{y1}"]) + (-wide["change"])
    pri = wide[wide["priority"]].sort_values("priority_score", ascending=False)

    # --- statistics
    girls = wide.loc[wide["gender"] == "بنات", f"rate_{y1}"]
    boys = wide.loc[wide["gender"] == "بنين", f"rate_{y1}"]
    t_gender = stats.ttest_ind(girls, boys, equal_var=False)
    pub = wide.loc[wide["authority"] == "حكومي", f"rate_{y1}"]
    prv = wide.loc[wide["authority"] != "حكومي", f"rate_{y1}"]
    t_auth = stats.ttest_ind(prv, pub, equal_var=False)
    rho = stats.spearmanr(wide[f"tested_{y1}"], wide[f"rate_{y1}"])
    x = np.array([t["year"] for t in trend], float)
    yv = np.array([t["region"] for t in trend], float)
    lr = stats.linregress(x, yv)
    x_new = x[-1] + 1
    y_hat = lr.intercept + lr.slope * x_new
    resid = yv - (lr.intercept + lr.slope * x)
    s_err = np.sqrt((resid ** 2).sum() / (len(x) - 2))
    pi = stats.t.ppf(0.90, len(x) - 2) * s_err * np.sqrt(1 + 1 / len(x) + (x_new - x.mean()) ** 2 / ((x - x.mean()) ** 2).sum())
    statistics = dict(
        gender=dict(girls=round(girls.mean(), 2), boys=round(boys.mean(), 2),
                    diff=round(girls.mean() - boys.mean(), 2), t=round(float(t_gender.statistic), 2),
                    p=float(t_gender.pvalue), n_girls=int(len(girls)), n_boys=int(len(boys))),
        authority=dict(private=round(prv.mean(), 2), public=round(pub.mean(), 2),
                       diff=round(prv.mean() - pub.mean(), 2), t=round(float(t_auth.statistic), 2), p=float(t_auth.pvalue)),
        size_corr=dict(rho=round(float(rho.statistic), 3), p=float(rho.pvalue)),
        trend=dict(slope=round(float(lr.slope), 2), r2=round(float(lr.rvalue ** 2), 3)),
        forecast=dict(year=int(x_new), value=round(float(y_hat), 1), low=round(float(y_hat - pi), 1), high=round(float(y_hat + pi), 1)),
        schools_classified=int(len(wide)), min_n=MIN_N,
    )

    # --- sub-domain diagnosis
    sd = (subs.groupby(["year", "grade", "subject", "subdomain"])
              .agg(tested=("tested", "sum"), prof=("proficient", "sum")).reset_index())
    sd_en = {"العلوم الفيزيائية والكيميائية": "Physical & chemical sciences", "علم الأرض والفلك": "Earth & space science",
             "علوم الحياة": "Life sciences", "الأعداد والعمليات عليها": "Numbers & operations", "الجبر": "Algebra",
             "الهندسة والقياس": "Geometry & measurement", "البيانات والاحتمالات": "Data & probability"}
    sd["sub_en"] = sd["subdomain"].map(sd_en)

    # --- governorates (latest year) for insights
    gv = main[main["year"] == y1].groupby("gov").apply(rate).sort_values()
    gv0 = main[main["year"] == y0].groupby("gov").apply(rate)
    gov_change = (gv - gv0).sort_values()

    # --- subject/grade for insights
    subj = main[main["year"] == y1].groupby("subject").apply(rate)
    en = dict(zip(govs["gov_ar"], govs["gov_en"]))
    t0, t1 = trend[0], trend[-1]
    seg_counts = wide["segment"].value_counts().to_dict()
    weakest_sub = (sd[sd["year"] == y1].assign(r=lambda d: 100 * d["prof"] / d["tested"]).sort_values("r").iloc[0])
    insights = [
        dict(kind="up",
             ar=f"ارتفعت نسبة الإتقان في المنطقة من {t0['region']:.1f}% ({t0['year']}هـ) إلى {t1['region']:.1f}% ({t1['year']}هـ) بزيادة {t1['region']-t0['region']:.1f} نقطة، بمعدل {lr.slope:.1f} نقطة سنويًا.",
             en=f"Regional proficiency rose from {t0['region']:.1f}% ({t0['year']} AH) to {t1['region']:.1f}% ({t1['year']} AH), +{t1['region']-t0['region']:.1f} pts at {lr.slope:.1f} pts per year."),
        dict(kind="gap",
             ar=f"الفجوة مع متوسط المملكة {t1['region']-t1['national']:+.1f} نقطة في {t1['year']}هـ مقابل {t0['region']-t0['national']:+.1f} في {t0['year']}هـ.",
             en=f"Gap to the national average is {t1['region']-t1['national']:+.1f} pts in {t1['year']} AH vs {t0['region']-t0['national']:+.1f} in {t0['year']} AH."),
        dict(kind="subject",
             ar=f"العلوم الأعلى إتقانًا ({subj['SCI']:.1f}%) والرياضيات الأدنى ({subj['MATH']:.1f}%) بفارق {subj['SCI']-subj['MATH']:.1f} نقطة؛ وأضعف مجال فرعي «{weakest_sub['subdomain']}» ({weakest_sub['r']:.1f}%).",
             en=f"Science leads ({subj['SCI']:.1f}%) and Mathematics trails ({subj['MATH']:.1f}%), a {subj['SCI']-subj['MATH']:.1f}-pt spread; weakest sub-domain: {weakest_sub['sub_en']} ({weakest_sub['r']:.1f}%)."),
        dict(kind="gov",
             ar=f"أعلى المحافظات {gv.index[-1]} ({gv.iloc[-1]:.1f}%) وأدناها {gv.index[0]} ({gv.iloc[0]:.1f}%)، والأكثر تحسنًا {gov_change.index[-1]} ({gov_change.iloc[-1]:+.1f}).",
             en=f"Top governorate {en[gv.index[-1]]} ({gv.iloc[-1]:.1f}%), lowest {en[gv.index[0]]} ({gv.iloc[0]:.1f}%); most improved {en[gov_change.index[-1]]} ({gov_change.iloc[-1]:+.1f})."),
        dict(kind="gender",
             ar=f"مدارس البنات تتفوق بمتوسط {statistics['gender']['diff']:.1f} نقطة، والفرق دال إحصائيًا (اختبار Welch، p < 0.001).",
             en=f"Girls' schools outperform by {statistics['gender']['diff']:.1f} pts on average; the difference is statistically significant (Welch t-test, p < 0.001)."),
        dict(kind="priority",
             ar=f"{len(pri)} مدرسة مرشحة لأولوية التدخل: تقل عن متوسط المنطقة بـ{MATERIAL:.0f} نقاط فأكثر وتراجعت {MATERIAL:.0f} نقاط فأكثر (من أصل {len(wide):,} مدرسة مصنفة).",
             en=f"{len(pri)} schools flagged for priority intervention: ≥{MATERIAL:.0f} pts below the regional average and down ≥{MATERIAL:.0f} pts year-on-year (of {len(wide):,} classified)."),
        dict(kind="forecast",
             ar=f"بالاتجاه الحالي تُقدَّر نسبة الإتقان لعام {statistics['forecast']['year']}هـ بنحو {statistics['forecast']['value']}% (نطاق 80%: {statistics['forecast']['low']}–{statistics['forecast']['high']}%).",
             en=f"On the current trend, {statistics['forecast']['year']} AH proficiency is projected at ~{statistics['forecast']['value']}% (80% range: {statistics['forecast']['low']}–{statistics['forecast']['high']}%)."),
    ]

    segments = dict(
        threshold=region_now, year=int(y1), prev=int(y0), counts={k: int(v) for k, v in seg_counts.items()},
        priority=int(len(pri)), material=MATERIAL, min_n=MIN_N,
        points=[[round(r[f"rate_{y1}"], 1), round(r["change"], 1), "priority" if r["priority"] else r["segment"]] for _, r in wide.iterrows()],
    )
    priority = [dict(id=f"SCH-{int(i) - 900000:04d}", gov=r["gov"], gov_en=en[r["gov"]], stage=r["stage"], gender=r["gender"],
                     rate=round(r[f"rate_{y1}"], 1), prev=round(r[f"rate_{y0}"], 1), change=round(r["change"], 1),
                     tested=int(r[f"tested_{y1}"]))
                for i, r in pri.head(12).iterrows()]
    pri.to_csv(OUT / "priority_schools.csv", encoding="utf-8-sig")

    # --- school register for distinct counts on the dashboard: [gov, gender, authority, grade-bitmask]
    bits = {3: 1, 6: 2, 9: 4}
    reg = main.groupby("school_id").agg(gov=("gov", "first"), gender=("gender", "first"),
                                       authority=("authority", "first"), grades=("grade", lambda g: sum(bits[x] for x in set(g))))
    register = [[gov_idx[r.gov], gen_codes[r.gender], auth_codes[r.authority], int(r.grades)] for r in reg.itertuples()]

    return dict(cube=cube, national=national, trend=trend, segments=segments, priority=priority, register=register,
                statistics=statistics, subdomains=sd, insights=insights, gv=gv, subj=subj, wide=wide)


# ================================================================ 5. CHARTS (static)
INK, MUTED, GRID = "#0B2E3A", "#5B6B73", "#E3E8EA"
C_REG, C_NAT, C_ACC = "#0E7C7B", "#C8A24A", "#C2410C"


def style(ax, title):
    ax.set_title(title, loc="left", fontsize=12, color=INK, fontweight="bold", pad=10)
    for s in ["top", "right"]:
        ax.spines[s].set_visible(False)
    for s in ["left", "bottom"]:
        ax.spines[s].set_color(GRID)
    ax.tick_params(colors=MUTED, labelsize=9)
    ax.grid(axis="y", color=GRID, linewidth=0.8)
    ax.set_axisbelow(True)


def charts(res, dims, govs, subjects):
    en = dict(zip(govs["gov_ar"], govs["gov_en"]))
    tr = pd.DataFrame(res["trend"])

    # 1 trend
    fig, ax = plt.subplots(figsize=(7, 3.8), dpi=150)
    ax.plot(tr["year"], tr["region"], marker="o", color=C_REG, lw=2.4, label="Aseer region")
    ax.plot(tr["year"], tr["national"], marker="o", color=C_NAT, lw=2, ls="--", label="National")
    for _, r in tr.iterrows():
        ax.annotate(f"{r['region']:.1f}%", (r["year"], r["region"]), textcoords="offset points", xytext=(0, -16), ha="center", fontsize=8, color=C_REG)
    ax.set_xticks(tr["year"]); ax.set_xticklabels([f"{y} AH" for y in tr["year"]])
    ax.set_ylabel("Proficiency %", color=MUTED); ax.legend(frameon=False, fontsize=9)
    ax.set_ylim(tr["region"].min() - 2.5, tr["national"].max() + 1.5)
    style(ax, "Proficiency trend vs national benchmark"); fig.tight_layout(); fig.savefig(CHARTS / "01_trend.png"); plt.close(fig)

    # 2 heatmap subject x grade (latest)
    c = res["cube"]; y1 = c["year"].max()
    h = c[c["year"] == y1].groupby(["subject", "grade"]).apply(lambda d: 100 * d["prof"].sum() / d["tested"].sum()).unstack()
    h = h.reindex(["MATH", "SCI", "READ"])
    fig, ax = plt.subplots(figsize=(6.2, 3.4), dpi=150)
    im = ax.imshow(h.values, cmap="YlGnBu", vmin=15, vmax=55)
    ax.set_xticks(range(h.shape[1])); ax.set_xticklabels([f"Grade {g}" for g in h.columns])
    ax.set_yticks(range(h.shape[0])); ax.set_yticklabels(["Mathematics", "Science", "Reading"])
    for i in range(h.shape[0]):
        for j in range(h.shape[1]):
            v = h.values[i, j]
            ax.text(j, i, "—" if np.isnan(v) else f"{v:.1f}%", ha="center", va="center", fontsize=10,
                    color="white" if (not np.isnan(v) and v > 40) else INK)
    ax.set_title(f"Subject x grade · {y1} AH", loc="left", fontsize=12, color=INK, fontweight="bold")
    for s in ax.spines.values():
        s.set_visible(False)
    fig.colorbar(im, ax=ax, fraction=0.04); fig.tight_layout(); fig.savefig(CHARTS / "02_subject_grade.png"); plt.close(fig)

    # 3 governorates
    gv = res["gv"]
    fig, ax = plt.subplots(figsize=(7, 5), dpi=150)
    colors = [C_REG if v >= res["segments"]["threshold"] else "#9FB8BF" for v in gv.values]
    ax.barh([en[g] for g in gv.index], gv.values, color=colors)
    ax.axvline(res["segments"]["threshold"], color=C_ACC, ls="--", lw=1.2)
    ax.text(res["segments"]["threshold"] + 0.3, -0.9, "Region avg", color=C_ACC, fontsize=8)
    for i, v in enumerate(gv.values):
        ax.text(v + 0.3, i, f"{v:.1f}%", va="center", fontsize=8, color=MUTED)
    style(ax, f"Proficiency by governorate · {y1} AH"); ax.grid(axis="x", color=GRID); ax.grid(axis="y", visible=False)
    fig.tight_layout(); fig.savefig(CHARTS / "03_governorates.png"); plt.close(fig)

    # 4 segmentation
    w = res["wide"]; thr = res["segments"]["threshold"]
    pal = {"lead_up": C_REG, "lead_down": C_NAT, "low_up": "#6B8E9B", "low_down": "#B7C4C9", "priority": C_ACC}
    lab = {"lead_up": "Leading & improving", "lead_down": "Leading, declining", "low_up": "Below avg, improving",
           "low_down": "Below avg, declining (watch)", "priority": "Priority for intervention"}
    fig, ax = plt.subplots(figsize=(7, 4.6), dpi=150)
    w = w.assign(segment=np.where(w["priority"], "priority", w["segment"]))
    for k in ["low_down", "low_up", "lead_down", "lead_up", "priority"]:
        d = w[w["segment"] == k]
        ax.scatter(d[f"rate_{y1}"], d["change"], s=9, alpha=0.55, color=pal[k], label=f"{lab[k]} ({len(d)})", edgecolors="none")
    ax.axvline(thr, color=MUTED, lw=0.8); ax.axhline(0, color=MUTED, lw=0.8)
    ax.set_xlabel(f"School proficiency {y1} AH (%)", color=MUTED); ax.set_ylabel("Change vs previous year (pts)", color=MUTED)
    ax.legend(frameon=False, fontsize=8, loc="upper left")
    style(ax, "School segmentation"); fig.tight_layout(); fig.savefig(CHARTS / "04_segmentation.png"); plt.close(fig)

    # 5 DQ scorecard
    fig, ax = plt.subplots(figsize=(6.5, 3.2), dpi=150)
    ax.barh(dims["dim_en"], dims["score"], color=[C_REG if s >= t else C_ACC for s, t in zip(dims["score"], dims["target"])])
    ax.scatter(dims["target"], dims["dim_en"], marker="|", s=300, color=INK, label="Target")
    ax.set_xlim(98, 100.2)
    for i, v in enumerate(dims["score"]):
        ax.text(98.05, i, f"{v:.2f}%", va="center", fontsize=8, color="white", fontweight="bold")
    style(ax, "Data-quality scorecard (raw extract) · | = target"); ax.grid(axis="x", color=GRID); ax.grid(axis="y", visible=False)
    fig.tight_layout(); fig.savefig(CHARTS / "05_dq_scorecard.png"); plt.close(fig)


# ================================================================ 6. EXPORT
def export(res, dq, dims, summary, govs, subjects, grades):
    c = res["cube"]
    fields = ["year", "grade", "subject", "s", "a", "g", "tested", "expected", "prof", "score", "hi", "mid", "low", "vlow", "schools"]
    sd = res["subdomains"]
    feed = dict(
        meta=dict(title_ar="تحليل نتائج اختبارات نافس — منطقة عسير", title_en="Nafs Results Analytics — Aseer Region",
                  generated=datetime.now(timezone.utc).strftime("%Y-%m-%d"), years=sorted(int(y) for y in c["year"].unique()),
                  schools=int(res["wide"].shape[0]), pipeline="Python · pandas · NumPy · SciPy · Matplotlib"),
        labels=dict(
            gov=[dict(ar=a, en=e) for a, e in zip(govs["gov_ar"], govs["gov_en"])],
            subject={r.code: dict(ar=r.subject_ar, en=r.subject_en) for r in subjects.itertuples()},
            grade={str(r.grade): dict(ar=r.grade_ar, en=r.grade_en) for r in grades.itertuples()},
        ),
        dq=dict(rules=dq.to_dict("records"), dims=dims.to_dict("records"), **summary),
        cube=dict(fields=fields, rows=c[fields].values.tolist()),
        national=res["national"], trend=res["trend"], segments=res["segments"], priority=res["priority"],
        statistics=res["statistics"], insights=res["insights"], register=res["register"],
        subdomains=[[int(r.year), int(r.grade), r.subject, r.subdomain, r.sub_en, int(r.tested), int(r.prof)] for r in sd.itertuples()],
    )
    (OUT / "dashboard.json").write_text(json.dumps(feed, ensure_ascii=False, separators=(",", ":"), default=float), encoding="utf-8")
    dq.to_csv(OUT / "dq_report.csv", index=False, encoding="utf-8-sig")
    return feed


def main():
    raw, govs, subjects, grades = load()
    clean, dq, dims, quarantine, summary = data_quality(raw, govs, subjects, grades)
    quarantine.to_csv(OUT / "quarantine.csv", index=False, encoding="utf-8-sig")
    main_df, subs = build_model(clean, subjects)
    res = analyse(main_df, subs, govs)
    charts(res, dims, govs, subjects)
    feed = export(res, dq, dims, summary, govs, subjects, grades)
    print(f"raw={summary['raw_rows']:,} clean={summary['clean_rows']:,} quarantined={summary['quarantined']} "
          f"dq={summary['overall']}% cube_rows={len(feed['cube']['rows']):,}")
    for i in feed["insights"]:
        print(" •", i["en"])


if __name__ == "__main__":
    main()