"""Turn the CBS 'Local Authorities in Israel' workbooks (2021-2024) into tidy parquet tables.

  data/x_budget.parquet  one row per authority x year: CBS budget-execution figures (₪ thousands)
  data/x_phys24.parquet  one row per authority: 2024 population / socio-economic context

Columns are picked by header text (leaf label + occurrence), never by position, because the
layout shifts between editions. Every selector must match exactly once per workbook or we stop.
"""
import sys
from pathlib import Path

import pandas as pd

sys.path.insert(0, str(Path(__file__).parent))
from cbs_parse import find_sheet, load, norm, sheet_table

ROOT = Path(__file__).resolve().parent.parent
FILES = {2021: "p_libud_2021.xlsx", 2022: "p_libud_22.xlsx", 2023: "p_libud_23.xlsx", 2024: "p_libud_24.xlsx"}

# key -> (leaf label or prefix, occurrence among columns with that leaf, prefix match?)
BUDGET = {
    "name": ("שם הרשות", 0, False),
    "code": ("סמל הרשות", 0, False),
    "district": ("מחוז", 0, False),
    "status": ("מעמד מוניציפ", 0, True),
    "pop_budget": ('סה"כ אוכלוסייה בסוף השנה', 0, False),
    "se_cluster_budget": ("אשכול מדד חברתי כלכלי", 0, True),
    "rev_regular": ('סה"כ הכנסות בתקציב הרגיל', 0, False),
    "own_income": ('סה"כ הכנסות עצמיות', 0, False),
    "arnona_total": ("הכנסות עצמיות: מארנונה", 0, False),
    "arnona_res_collected": ('ארנונה למגורים: סה"כ גביות', 0, False),
    "arnona_res_net_charge": ('ארנונה למגורים: סה"כ חיובים נטו', 0, False),
    "arnona_res_rate_all": ("ארנונה למגורים: יחס גבייה ב-% לכלל החיובים", 0, False),
    "arnona_res_rate_year": ("ארנונה למגורים: יחס גבייה ב-% לחיוב השנה", 0, False),
    "arnona_res_discounts": ('ארנונה למגורים: סה"כ פטורים, שחרורים והנחות', 0, False),
    "arnona_nonres_collected": ("ארנונה לא למגורים: גבייה", 0, False),
    "own_education": ("הכנסות עצמיות: מחינוך", 0, False),
    "own_welfare": ("הכנסות עצמיות: מרווחה", 0, False),
    "own_other": ("יתר הכנסות עצמיות", 0, False),
    "gov_total": ('סה"כ הכנסות מהממשלה', 0, False),
    "gov_education": ("הכנסות מהממשלה: ממשרד החינוך", 0, False),
    "gov_welfare": ("הכנסות מהממשלה: ממשרד הרווחה", 0, False),
    "gov_general_grant": ("הכנסות מהממשלה: מענק כללי", 0, False),
    "gov_special_grants": ("הכנסות מהממשלה: מענקים מיוחדים", 0, True),
    "gov_deficit_grants": ("הכנסות מהממשלה: מענקים להקטנת גירעון", 0, True),
    "exp_regular": ('סה"כ הוצאות בתקציב הרגיל', 0, False),
    "exp_pc_cbs": ('הוצאה לנפש בתקציב הרגיל (ש"ח)', 0, False),
    "exp_admin": ("הנהלה וכלליות", 0, False),
    "exp_local": ("שירותים מקומיים", 1, False),
    "exp_state": ('סה"כ שירותים ממלכתיים', 1, False),
    "exp_education": ("שירותים ממלכתיים: חינוך", 0, False),
    "exp_welfare": ("שירותים ממלכתיים: רווחה", 0, False),
    "exp_culture": ("שירותים ממלכתיים: תרבות", 0, False),
    "exp_enterprises": ("מפעלים", 0, False),
    "exp_extraordinary": ("תשלומים בלתי רגילים", 0, False),
    "exp_labor": ('סה"כ הוצאות לעלות עבודה', 0, False),
    "exp_loan_repay": ("הוצאות לפירעון מלוות", 0, False),
    "dev_exp": ('סה"כ הוצאות בתקציב בלתי רגיל', 0, False),
    "dev_gov": ("השתתפות הממשלה", 0, False),
    "surplus_year": ("עודף/גירעון בתקציב הרגיל השנה", 0, False),
    "surplus_acc": ("עודף/גירעון מצטבר בתקציב הרגיל", 0, True),
    "loan_burden": ("עומס מלוות לסוף השנה", 0, True),
    "charge_res_area": ("למגורים", 1, False),  # occurrence 0 = ₪ charge, 1 = area (thousand m²)
    "charge_office_area": ("למשרדים, שירותים ומסחר", 1, False),
    "charge_industry_area": ("לתעשייה", 1, False),
}
OPTIONAL = {"pop_budget", "se_cluster_budget"}  # missing from the 2021 edition
EMERGENCY = ("פעולות חירום", 1)  # added to the expenditure side in 2023 (Iron Swords war)

PHYS = {
    "name": ("שם הרשות", 0, False),
    "code": ("סמל הרשות", 0, False),
    "status": ("מעמד מוניציפ", 0, True),
    "pop": ('סה"כ אוכלוסייה בסוף השנה', 0, False),
    "area_km2": ('שטח (קמ"ר)', 0, False),
    "pct_jews_others": ("יהודים ואחרים (אחוזים מתוך כלל אוכלוסיית הישראלים)", 0, False),
    "pct_arabs": ("ערבים (אחוזים מתוך כלל אוכלוסיית הישראלים)", 0, False),
    "pct_0_17": ("בני 17-0", 0, False),
    "pct_65p": ("בני 65 ומעלה", 0, False),
    "dependency_ratio": ("יחס תלות (ל-1,000 תושבים בלתי תלויים)", 0, False),
    "pop_growth_pct": ("אחוז גידול האוכלוסייה לעומת השנה הקודמת", 0, False),
    "income_support_recipients": ("מקבלי גמלת הבטחת הכנסה (נפשות, בסוף השנה)", 0, False),
    "kids_kindergarten": ('ילדים בגנים של משרד החינוך | סה"כ', 0, False),
    "pupils_total": ('תלמידים | סה"כ', 0, False),
    "bagrut_pct": ("אחוז זכאים לתעודת בגרות מבין תלמידי כיתות יב", 0, False),
    "wage_avg_2023": ("שכר ממוצע", 0, False),
    "se_cluster": ("אשכול (מ-1 עד 10, 1 הנמוך ביותר)", 0, False),
    "se_index": ("ערך מדד", 0, False),
    "se_rank": ("דירוג (מ-1 עד 255, 1 הנמוך ביותר)", 0, False),
    "periph_cluster": ("אשכול (מ-1 עד 10, 1 הפריפריאלי ביותר)", 0, False),
    "periph_index": ("ערך מדד", 2, False),
    "pct_rent": ("דירות בשכירות | אחוז", 0, False),
    "built_density": ('צפיפות אוכלוסייה לשטח בנוי למגורים | נפשות לקמ"ר', 0, False),
}


def pick(heads, spec, label):
    """Map key -> column index. Leaf = last header part; 'a | b' labels match the last two parts."""
    out = {}
    for key, (want, occ, prefix) in spec.items():
        want = norm(want)
        depth = want.count(" | ") + 1
        hits = []
        for i, parts in enumerate(heads):
            leaf = " | ".join(parts[-depth:])
            if (leaf.startswith(want) if prefix else leaf == want):
                hits.append(i)
        if len(hits) <= occ:
            out[key] = None
            continue
        out[key] = hits[occ]
    missing = [k for k, v in out.items() if v is None and k not in OPTIONAL]
    if missing:
        raise SystemExit(f"{label}: no column for {missing}")
    return out


def num(v):
    if v is None:
        return None
    if isinstance(v, (int, float)):
        return float(v)
    s = str(v).strip().replace(",", "")
    try:
        return float(s)
    except ValueError:
        return None  # CBS uses '..', '-', '' for suppressed / not applicable


def budget_year(year, wb):
    heads, rows = sheet_table(find_sheet(wb, "תקציב"))
    cols = pick(heads, BUDGET, f"budget {year}")
    leaf = [p[-1] if p else "" for p in heads]
    em = [i for i, h in enumerate(leaf) if h == EMERGENCY[0]]
    cols["exp_emergency"] = em[EMERGENCY[1]] if len(em) > EMERGENCY[1] else None
    recs = []
    for r in rows:
        rec = {"year": year}
        for k, i in cols.items():
            v = r[i] if i is not None else None
            rec[k] = (norm(v) if v is not None else None) if k in ("name", "district", "status") else num(v)
        recs.append(rec)
    df = pd.DataFrame(recs)
    df["code"] = df["code"].astype(int)
    # Authorities that did not report that year appear with every amount = 0; treat as missing.
    money = [k for k in cols if k not in ("name", "code", "district", "status", "pop_budget", "se_cluster_budget")]
    unreported = (df["exp_regular"].fillna(0) == 0) & (df["rev_regular"].fillna(0) == 0)
    df.loc[unreported, money] = None
    df["reported"] = ~unreported
    # 2022 published collection ratios as fractions, later editions as percent
    for c in ("arnona_res_rate_all", "arnona_res_rate_year"):
        if df[c].median() < 2:
            df[c] = df[c] * 100
    return df


def phys_year(year, wb):
    heads, rows = sheet_table(find_sheet(wb, "פיזיים"))
    spec = PHYS if year == 2024 else {k: PHYS[k] for k in ("name", "code", "status", "pop")}
    cols = pick(heads, spec, f"phys {year}")
    text = ("name", "status")
    df = pd.DataFrame([{k: norm(r[i]) if k in text else num(r[i]) for k, i in cols.items()} for r in rows])
    df["code"] = df["code"].astype(int)
    df["year"] = year
    return df


def main():
    raw = ROOT / "data" / "cbs_raw"
    budgets, physs = [], []
    for year, fn in FILES.items():
        wb = load(raw / fn)
        b, p = budget_year(year, wb), phys_year(year, wb)
        print(year, "budget rows", len(b), "phys rows", len(p), "emergency col", b["exp_emergency"].notna().sum())
        budgets.append(b)
        physs.append(p)
    budget = pd.concat(budgets, ignore_index=True)
    # Before 2024 regional-council codes collide with small local-council codes, so (code, status)
    # is the key; status spelling also drifts (מוניציפאלי/מוניציפלי) but its values do not.
    pops = pd.concat([p[["year", "code", "status", "pop"]] for p in physs], ignore_index=True)
    assert not pops.duplicated(["year", "code", "status"]).any()
    assert not budget.duplicated(["year", "code", "status"]).any()
    n = len(budget)
    budget = budget.merge(pops, on=["year", "code", "status"], how="left", validate="one_to_one")
    assert len(budget) == n and budget["pop"].notna().all(), "population join failed"
    # From 2024 CBS codes regional councils as 55XX; older editions use the bare council number.
    rc = budget["status"].str.contains("אזורית") & (budget["code"] < 100)
    budget["code24"] = budget["code"].where(~rc, 5500 + budget["code"])
    budget.to_parquet(ROOT / "data" / "x_budget.parquet", index=False)
    physs[-1].drop(columns=["year"]).to_parquet(ROOT / "data" / "x_phys24.parquet", index=False)
    print("wrote", len(budget), "budget rows")


if __name__ == "__main__":
    main()