"""
Workforce Planning Analytics — Sunny Valley International
=========================================================
Employee register -> HR data-quality rules -> current-state workforce analytics
-> 3-year workforce plan (demand, attrition, hiring, Saudization, payroll index)
-> JSON feed for the bilingual executive dashboard.

Author : Adel Asiri (CDMP, PMP)
Stack  : Python 3 · pandas · NumPy

    python hr_pipeline.py      ->  output/hr_dashboard.json, output/dq_report.csv
Privacy: pseudonymised IDs only; pay is published as shares and a relative index,
never as amounts; cost groups with fewer than 3 employees are merged.
"""
from __future__ import annotations

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

import numpy as np
import pandas as pd

ROOT = Path(__file__).resolve().parent
DATA, OUT = ROOT / "data", ROOT / "output"
OUT.mkdir(exist_ok=True)
TODAY = date(2026, 9, 27)
YEARS = [2027, 2028, 2029]          # plan horizon: end of academic years 2026-27 … 2028-29
MIN_CELL = 3                        # small-cell suppression for pay analytics

FAMILIES = {
    "TCH": ("التعليم", "Teaching"), "TRN": ("النقل المدرسي", "School transport"),
    "SUP": ("الخدمات المساندة", "Support services"), "ADM": ("الإدارة", "Administration"),
    "FIN": ("المالية", "Finance"), "HRM": ("الموارد البشرية", "Human resources"),
}
NATS = {"SA": ("سعودي", "Saudi"), "EG": ("مصري", "Egyptian"), "IN": ("هندي", "Indian")}
LOCALIZABLE = {"TCH", "TRN"}             # hiring mix follows the Saudi-share policy
SAUDI_ONLY = {"ADM", "FIN", "HRM"}        # already fully Saudi today; replacements stay Saudi
EXPAT_OF = {"TCH": "EG", "TRN": "EG", "SUP": "IN", "ADM": "EG", "FIN": "EG", "HRM": "EG"}
GROWTH_FACTOR = {"TCH": 1.0, "TRN": 1.0, "ADM": 0.5, "FIN": 0.0, "HRM": 0.0, "SUP": 0.0}


def load():
    emp = pd.read_csv(DATA / "employees.csv", encoding="utf-8-sig", keep_default_na=False)
    struct = pd.read_csv(DATA / "workforce_structure.csv")
    a = pd.read_csv(DATA / "planning_assumptions.csv").set_index("parameter")["value"].astype(float).to_dict()
    for c in ["hire_date", "contract_end"]:
        emp[c] = pd.to_datetime(emp[c]).dt.date
    emp["iqama_expiry"] = pd.to_datetime(emp["iqama_expiry"].replace("", None)).dt.date
    return emp, struct, a


# ---------------------------------------------------------------- data quality
def data_quality(emp, struct):
    n = len(emp); rules = []
    def rule(rid, dim_ar, dim_en, ar, en, failed, act_ar, act_en, checked=None):
        checked = checked or n
        rules.append(dict(id=rid, dim_ar=dim_ar, dim_en=dim_en, rule_ar=ar, rule_en=en, checked=int(checked),
                          failed=int(failed), pass_rate=round(100 * (1 - failed / checked), 1), action_ar=act_ar, action_en=act_en))
    expat = emp["nationality"] != "SA"
    rule("HR-01", "التفرد", "Uniqueness", "رقم الموظف المستعار فريد", "Pseudonymised employee ID is unique",
         emp["employee_id"].duplicated().sum(), "دمج السجلات المكررة", "Merge duplicate records")
    rule("HR-02", "الصحة", "Validity", "الجنسية من القائمة المرجعية", "Nationality from reference list",
         (~emp["nationality"].isin(NATS)).sum(), "تصحيح الترميز", "Fix coding")
    rule("HR-03", "الصحة", "Validity", "الأسرة الوظيفية من الدليل المعتمد", "Job family from approved catalogue",
         (~emp["job_family"].isin(FAMILIES)).sum(), "تصحيح الترميز", "Fix coding")
    rule("HR-04", "الاكتمال", "Completeness", "تاريخ انتهاء الإقامة موجود لغير السعوديين", "Iqama expiry present for expatriates",
         (expat & emp["iqama_expiry"].isna()).sum(), "استكمال التاريخ من منصة مقيم", "Complete from Muqeem", checked=int(expat.sum()))
    rule("HR-05", "الاتساق", "Consistency", "لا يوجد تاريخ إقامة لموظف سعودي", "No iqama date on Saudi records",
         ((~expat) & emp["iqama_expiry"].notna()).sum(), "حذف القيمة غير المنطقية", "Clear invalid value", checked=int((~expat).sum()))
    sec_ok = ~(((emp["section"] == "Boys section") & (emp["gender"] != "M")) |
               (emp["section"].isin(["Girls section", "Kindergarten"]) & (emp["gender"] != "F")))
    rule("HR-06", "الاتساق", "Consistency", "جنس المعلم متوافق مع القسم", "Teacher gender matches section",
         (~sec_ok).sum(), "مراجعة التكليف", "Review assignment")
    rule("HR-07", "الصحة", "Validity", "العمر 18–65 وتاريخ التعيين غير مستقبلي", "Age 18–65 and hire date not in future",
         ((emp["age"] < 18) | (emp["age"] > 65) | (emp["hire_date"] > TODAY)).sum(), "التحقق من ملف الموظف", "Verify employee file")
    approved = struct.groupby("nationality")["count"].sum()
    actual = emp.groupby("nationality").size().reindex(approved.index, fill_value=0)
    rule("HR-08", "الاتساق", "Consistency", "العدد الفعلي مطابق للهيكل المعتمد لكل جنسية", "Headcount reconciles to approved structure",
         int((actual != approved).sum()), "مطابقة مع مسيّر الرواتب ومدد", "Reconcile with payroll & Mudad", checked=len(approved))
    return pd.DataFrame(rules)


# ---------------------------------------------------------------- helpers
def band_age(a):
    return "<30" if a < 30 else "30–39" if a < 40 else "40–49" if a < 50 else "50–59" if a < 60 else "60+"


def band_tenure(t):
    return "<1" if t < 1 else "1–3" if t < 3 else "3–5" if t < 5 else "5–10" if t < 10 else "10+"


def month_idx(d):
    """0..11 = months ahead (Oct 2026 … Sep 2027); -1 outside the 12-month window."""
    if d is None or pd.isna(d):
        return -1
    m = (d.year - TODAY.year) * 12 + (d.month - TODAY.month)
    return m - 1 if 1 <= m <= 12 else (0 if m == 0 and d >= TODAY else -1)


# ---------------------------------------------------------------- 3-year plan model
def project(base, retire, a, share=None):
    """Deterministic workforce projection. base[f][g] = headcount (g in SA / EX)."""
    share = a["saudi_share_new_hires"] if share is None else share
    e, out = a["enrollment_growth"], []
    H = {f: dict(v) for f, v in base.items()}
    D0 = {f: v["SA"] + v["EX"] for f, v in base.items()}
    for y, yr in enumerate(YEARS, start=1):
        row = dict(year=yr, fam={}, hires=0.0, saudi_hires=0.0, leavers=0.0)
        for f in H:
            if f == "SUP" and a["outsource_support"] >= 1:
                lv = H[f]["SA"] + H[f]["EX"]
                row["fam"][f] = dict(demand=0, leavers=lv, hires=0, saudi_hires=0, SA=0, EX=0)
                H[f] = {"SA": 0.0, "EX": 0.0}; row["leavers"] += lv
                continue
            demand = D0[f] * (1 + e * GROWTH_FACTOR[f]) ** y
            lv = {g: min(H[f][g], H[f][g] * a["attrition_saudi" if g == "SA" else "attrition_expat"] + retire[f][g][y - 1]) for g in ("SA", "EX")}
            rem = {g: H[f][g] - lv[g] for g in lv}
            need = max(0.0, demand - rem["SA"] - rem["EX"])
            s = need * (1.0 if f in SAUDI_ONLY else share if f in LOCALIZABLE else 0.0)
            H[f] = {"SA": rem["SA"] + s, "EX": rem["EX"] + need - s}
            row["fam"][f] = dict(demand=demand, leavers=lv["SA"] + lv["EX"], hires=need, saudi_hires=s, **H[f])
            row["hires"] += need; row["saudi_hires"] += s; row["leavers"] += lv["SA"] + lv["EX"]
        tot = sum(H[f]["SA"] + H[f]["EX"] for f in H)
        row["total"] = tot
        row["saudi"] = sum(H[f]["SA"] for f in H)
        row["saudization"] = 100 * row["saudi"] / tot if tot else 0
        out.append(row)
    return out


def min_share_for_target(base, retire, a):
    lo, hi = 0.0, 1.0
    if project(base, retire, a, 1.0)[-1]["saudization"] < 100 * a["saudization_target"]:
        return None
    for _ in range(30):
        mid = (lo + hi) / 2
        if project(base, retire, a, mid)[-1]["saudization"] >= 100 * a["saudization_target"]:
            hi = mid
        else:
            lo = mid
    return hi


# ---------------------------------------------------------------- main analytics
def analyse(emp, a):
    emp = emp.copy()
    emp["tenure"] = emp["hire_date"].map(lambda d: (TODAY - d).days / 365.25)
    emp["age_band"] = emp["age"].map(band_age)
    emp["tenure_band"] = emp["tenure"].map(band_tenure)
    emp["grp"] = np.where(emp["nationality"] == "SA", "SA", "EX")
    emp["c_m"] = emp["contract_end"].map(month_idx)
    emp["i_m"] = emp["iqama_expiry"].map(month_idx)

    # register for client-side filtering (no pay, no identifiers)
    register = emp[["nationality", "job_family", "gender", "section", "age_band", "tenure_band", "c_m", "i_m"]].values.tolist()

    # baseline for the plan model
    base = {f: {"SA": 0.0, "EX": 0.0} for f in FAMILIES}
    for r in emp.itertuples():
        base[r.job_family][r.grp] += 1
    retire = {f: {g: [0, 0, 0] for g in ("SA", "EX")} for f in FAMILIES}
    ra = a["retirement_age"]
    for r in emp.itertuples():
        for y in range(1, 4):
            if r.age + y >= ra > r.age + y - 1:
                retire[r.job_family][r.grp][y - 1] += 1

    # pay analytics with small-cell suppression (FIN + HRM merged)
    emp["cost_group"] = emp["job_family"] + "_" + emp["grp"]
    counts = emp["cost_group"].value_counts()
    small = counts[counts < MIN_CELL].index
    emp.loc[emp["cost_group"].isin(small) & emp["job_family"].isin(["FIN", "HRM"]), "cost_group"] = "FINHR_SA"
    total_pay = emp["monthly_package"].sum()
    share = (emp.groupby("cost_group")["monthly_package"].sum() / total_pay * 100).round(1).to_dict()
    share_nat = (emp.groupby("nationality")["monthly_package"].sum() / total_pay * 100).round(1).to_dict()
    per_head_index = (emp.groupby("cost_group")["monthly_package"].mean() / (total_pay / 100)).to_dict()
    weights = {}
    for f in FAMILIES:
        for g in ("SA", "EX"):
            key = f"{f}_{g}" if f"{f}_{g}" in per_head_index else ("FINHR_SA" if f in ("FIN", "HRM") and g == "SA" else None)
            if key is None:  # families with no expatriates today: assume the family average
                fam = [k for k in per_head_index if k.startswith(f + "_")] or ["FINHR_SA"]
                key = fam[0]
            weights[f"{f}_{g}"] = round(per_head_index[key], 4)

    plan = project(base, retire, a)
    need_share = min_share_for_target(base, retire, a)

    # ---- insights
    n = len(emp); sa = int((emp["nationality"] == "SA").sum()); eg = int((emp["nationality"] == "EG").sum())
    tch = int((emp["job_family"] == "TCH").sum()); tch_eg = int(((emp["job_family"] == "TCH") & (emp["nationality"] == "EG")).sum())
    renew90 = int(emp["c_m"].between(0, 2).sum() + emp["i_m"].between(0, 2).sum())   # next three calendar months
    ev = pd.concat([emp.loc[emp["c_m"] >= 0, "c_m"], emp.loc[emp["i_m"] >= 0, "i_m"]])
    peak_m = int(ev.value_counts().idxmax()); peak_n = int(ev.value_counts().max())
    months_ar = ["أكتوبر 2026", "نوفمبر 2026", "ديسمبر 2026", "يناير 2027", "فبراير 2027", "مارس 2027", "أبريل 2027", "مايو 2027", "يونيو 2027", "يوليو 2027", "أغسطس 2027", "سبتمبر 2027"]
    months_en = ["Oct 2026", "Nov 2026", "Dec 2026", "Jan 2027", "Feb 2027", "Mar 2027", "Apr 2027", "May 2027", "Jun 2027", "Jul 2027", "Aug 2027", "Sep 2027"]
    y3 = plan[-1]; hires3 = sum(p["hires"] for p in plan); sh3 = sum(p["saudi_hires"] for p in plan)
    target = 100 * a["saudization_target"]
    insights = [
        dict(kind="gap", ar=f"نسبة التوطين الحالية {100*sa/n:.1f}% ({sa} من {n}) — أولوية تخطيطية للسنوات الثلاث القادمة.",
             en=f"Saudization currently stands at {100*sa/n:.1f}% ({sa} of {n}) — the planning priority for the next three years."),
        dict(kind="subject", ar=f"المصريون {100*eg/n:.1f}% من القوى العاملة، ويشغلون {tch_eg} من {tch} وظيفة تعليمية — تركّز يتطلب خطة إحلال وتعاقب.",
             en=f"Egyptian nationals are {100*eg/n:.1f}% of the workforce and hold {tch_eg} of {tch} teaching posts — a concentration that calls for succession and localisation planning."),
        dict(kind="priority", ar=f"{renew90} تجديدًا (عقود وإقامات) يستحق خلال الأشهر الثلاثة القادمة، وذروة التجديدات في {months_ar[peak_m]} بعدد {peak_n}.",
             en=f"{renew90} renewals (contracts and iqamas) fall due in the next three months; the peak is {months_en[peak_m]} with {peak_n}."),
        dict(kind="up", ar=f"وفق السيناريو الأساسي (نمو الطلاب {a['enrollment_growth']*100:.0f}% سنويًا) يصل العدد إلى {y3['total']:.0f} موظفًا بنهاية {YEARS[-1]} مع {hires3:.0f} تعيينًا خلال ثلاث سنوات، منها {sh3:.0f} سعوديًا.",
             en=f"Under the base scenario ({a['enrollment_growth']*100:.0f}% annual enrolment growth) headcount reaches {y3['total']:.0f} by {YEARS[-1]}, with {hires3:.0f} hires over three years — {sh3:.0f} of them Saudi."),
        dict(kind="forecast", ar=(f"التوطين المتوقع بنهاية السنة الثالثة {y3['saudization']:.1f}% مقابل مستهدف {target:.0f}%؛ ويتطلب بلوغه ألا تقل نسبة السعوديين من تعيينات التعليم والنقل عن {need_share*100:.0f}%." if need_share is not None else f"المستهدف {target:.0f}% لا يتحقق بالتعيين الجديد وحده ويتطلب إحلالًا مباشرًا."),
             en=(f"Projected Saudization at end of Year 3 is {y3['saudization']:.1f}% against a {target:.0f}% target; reaching it requires at least {need_share*100:.0f}% of new teaching and transport hires to be Saudi." if need_share is not None else f"The {target:.0f}% target cannot be reached through new hiring alone and needs direct replacement.")),
        dict(kind="subject", ar="سائقو الحافلات (14) وعمال النظافة (3) جميعهم غير سعوديين: الأولى مرشحة للتوطين التدريجي، والثانية للإسناد لمقاول خدمات.",
             en="All 14 bus drivers and 3 cleaners are expatriates: drivers are a phased-localisation candidate; cleaning is a candidate for outsourcing."),
    ]

    fam_counts = emp.groupby(["job_family", "nationality"]).size().unstack(fill_value=0)
    return dict(
        register=register, base=base, retire=retire, weights=weights, plan=plan, need_share=need_share,
        pay_share=share, pay_share_nat=share_nat, insights=insights,
        stats=dict(headcount=n, saudi=sa, avg_age=round(emp["age"].mean(), 1), avg_tenure=round(emp["tenure"].mean(), 1),
                   under1=int((emp["tenure"] < 1).sum()), renew90=renew90,
                   fam_nat={f: {k: int(v) for k, v in fam_counts.loc[f].items()} for f in fam_counts.index}),
    )


def main():
    emp, struct, a = load()
    dq = data_quality(emp, struct)
    res = analyse(emp, a)
    feed = dict(
        meta=dict(title_ar="تحليلات تخطيط القوى العاملة — شركة صني فالي العالمية", title_en="Workforce Planning Analytics — Sunny Valley International",
                  as_of=TODAY.isoformat(), years=YEARS, generated=datetime.now(timezone.utc).strftime("%Y-%m-%d")),
        labels=dict(fam={k: dict(ar=v[0], en=v[1]) for k, v in FAMILIES.items()}, nat={k: dict(ar=v[0], en=v[1]) for k, v in NATS.items()}),
        assumptions=a, localizable=sorted(LOCALIZABLE), saudi_only=sorted(SAUDI_ONLY), growth_factor=GROWTH_FACTOR,
        dq=dict(rules=dq.to_dict("records"), overall=round(float(dq["pass_rate"].mean()), 1)),
        **{k: res[k] for k in ["register", "base", "retire", "weights", "plan", "need_share", "pay_share", "pay_share_nat", "insights", "stats"]},
    )
    (OUT / "hr_dashboard.json").write_text(json.dumps(feed, ensure_ascii=False, default=float, separators=(",", ":")), encoding="utf-8")
    dq.to_csv(OUT / "dq_report.csv", index=False, encoding="utf-8-sig")
    print("DQ:", dq[["id", "failed"]].values.tolist())
    for p in res["plan"]:
        print(p["year"], round(p["total"], 1), f"{p['saudization']:.1f}%", "hires", round(p["hires"], 1), "saudi", round(p["saudi_hires"], 1))
    print("need share:", res["need_share"]); print("pay share:", res["pay_share"])
    for i in res["insights"]:
        print(" •", i["en"])


if __name__ == "__main__":
    main()