"""Builds the Looker Studio data source from the Jobber and Xero exports: tidy fact tables, the Jobber-Xero tie-out,
the dashboard figures (results/kpi.json) and a test against the answer key."""
import csv, json, os, datetime as dt
from collections import defaultdict, Counter
HERE = os.path.dirname(os.path.abspath(__file__)); D = dt.date.fromisoformat
rd = lambda p: list(csv.DictReader(open(os.path.join(HERE, p))))
def w(path, rows):
    with open(os.path.join(HERE, path), "w", newline="") as f: c = csv.DictWriter(f, list(rows[0].keys())); c.writeheader(); c.writerows(rows)
ASAT = dt.date(2026, 9, 30); MONTHS = [f"{y}-{m:02d}" for y, m in [(2024, 10), (2024, 11), (2024, 12)] + [(2025, k) for k in range(1, 13)] + [(2026, k) for k in range(1, 10)]]
CL = {c["Client ID"]: c for c in rd("data/jobber_clients.csv")}; BYNAME = {c["Name"]: c for c in CL.values()}
Q = rd("data/jobber_quotes.csv"); J = rd("data/jobber_jobs.csv"); V = rd("data/jobber_visits.csv"); JI = rd("data/jobber_invoices.csv")
XI = rd("data/xero_invoices.csv"); XP = rd("data/xero_payments.csv"); ST = {s["Technician"]: s for s in rd("data/staff.csv")}
f = lambda v: round(float(v or 0), 2)
# ------------------------------------------------ tie-out: Jobber invoices against Xero
XV = [x for x in XI if x["Status"] != "VOIDED"]; XBY = {x["InvoiceNumber"]: x for x in XV}
VOID = [x for x in XI if x["Status"] == "VOIDED"]; JBY = {i["Invoice #"]: i for i in JI}
tie = []
for i in JI:
    x = XBY.get(i["Invoice #"])
    st = "Not in Xero" if not x else ("Matched" if abs(f(x["SubTotal"]) - f(i["Subtotal"])) < .005 else "Amount differs")
    tie.append({"Invoice #": i["Invoice #"], "Job #": i["Job #"], "Client": i["Client name"], "Jobber subtotal": f(i["Subtotal"]), "Xero subtotal": f(x["SubTotal"]) if x else 0,
                "Difference": round((f(x["SubTotal"]) if x else 0) - f(i["Subtotal"]), 2), "Status": st})
for x in XV:
    if x["InvoiceNumber"] not in JBY:
        tie.append({"Invoice #": x["InvoiceNumber"], "Job #": "", "Client": x["ContactName"], "Jobber subtotal": 0, "Xero subtotal": f(x["SubTotal"]), "Difference": f(x["SubTotal"]), "Status": "Xero only"})
for x in VOID:
    tie.append({"Invoice #": x["InvoiceNumber"], "Job #": x["Reference"], "Client": x["ContactName"], "Jobber subtotal": 0, "Xero subtotal": 0, "Difference": 0, "Status": "Voided duplicate (excluded)"})
w("results/tieout.csv", tie); TS = Counter(t["Status"] for t in tie)
# ------------------------------------------------ invoices fact (Xero is the revenue source)
PAY = {p["InvoiceNumber"]: p for p in XP}
def bucket(d):
    if d <= 0: return "Not yet due"
    return "1-30 days" if d <= 30 else "31-60 days" if d <= 60 else "61-90 days" if d <= 90 else "90+ days"
FI = []
for x in XV:
    c = BYNAME.get(x["ContactName"], {}); due = f(x["AmountDue"]); od = (ASAT - D(x["DueDate"])).days if due else 0; p = PAY.get(x["InvoiceNumber"])
    FI.append({"Invoice #": x["InvoiceNumber"], "Contact": x["ContactName"], "Client type": c.get("Client type", ""), "Invoice date": x["InvoiceDate"], "Month": x["InvoiceDate"][:7], "Due date": x["DueDate"],
               "Reference": x["Reference"], "Source": "Jobber sync" if x["InvoiceNumber"] in JBY else "Xero only", "Ex GST": f(x["SubTotal"]), "GST": f(x["TotalTax"]), "Total": f(x["Total"]),
               "Paid": f(x["AmountPaid"]), "Amount due": due, "Paid date": p["Date"] if p else "", "Days to pay": (D(p["Date"]) - D(x["InvoiceDate"])).days if p else "",
               "Days overdue": max(od, 0) if due else "", "Ageing": bucket(od) if due else "Paid"})
w("results/fact_invoices.csv", FI)
# ------------------------------------------------ jobs fact
VH = defaultdict(float)
for v in V:
    if v["Completed"] == "Yes": VH[v["Job #"]] += float(v["Duration (h)"])
INV_BY_JOB = {}
for i in JI:
    if i["Invoice #"] in XBY: INV_BY_JOB[i["Job #"]] = XBY[i["Invoice #"]]
contract_inv = defaultdict(list)
for x in XV:
    if x["Reference"] == "Maintenance contract": contract_inv[x["ContactName"]].append(x)
FJ = []
for j in J:
    if j["Status"] == "Upcoming": continue
    c = CL[j["Client ID"]]; hrs = round(VH[j["Job #"]], 2); lab = round(hrs * float(ST[j["Assigned to"]]["Hourly cost incl. on-costs"]), 2); mat = f(j["Line item costs"])
    x = INV_BY_JOB.get(j["Job #"]); rev = f(x["SubTotal"]) if x else 0.0; invn = x["InvoiceNumber"] if x else ""; idate = x["InvoiceDate"] if x else ""
    if j["Billing"]:  # contract work billed in Xero: the quarter's contract invoice for that client
        cand = sorted(contract_inv[j["Client name"]], key=lambda y: abs((D(y["InvoiceDate"]) - D(j["End date"])).days))
        if cand: x = cand[0]; contract_inv[j["Client name"]].remove(x); rev = f(x["SubTotal"]); invn = x["InvoiceNumber"]; idate = x["InvoiceDate"]
    cb = j["Title"].startswith("Callback")
    p = PAY.get(invn); done = j["Status"] in ("Completed", "Requires invoicing")
    FJ.append({"Job #": j["Job #"], "Client ID": j["Client ID"], "Client": j["Client name"], "Client type": c["Client type"], "Lead source": c["Lead source"], "Service line": j["Service line"], "Technician": j["Assigned to"],
               "Start date": j["Start date"], "End date": j["End date"], "Month": j["End date"][:7], "Status": j["Status"], "Callback": "Yes" if cb else "No", "Callback of": j["Title"][11:] if cb else "",
               "Quote #": j["Quote #"], "Jobber total": f(j["Job total"]), "Visit hours": hrs, "Labour cost": lab, "Materials": mat, "Invoice #": invn, "Revenue ex GST": rev,
               "Gross margin": round(rev - lab - mat, 2), "Invoice date": idate, "Days to invoice": (D(idate) - D(j["End date"])).days if idate and not j["Billing"] else "",
               "Paid date": p["Date"] if p else "", "Billing": "Xero contract" if j["Billing"] else "Jobber", "Done": "Yes" if done else "No"})
w("results/fact_jobs.csv", FJ)
CB_TRUTH = rd("truth/callbacks.csv")
# not invoiced: completed in Jobber more than 14 days ago with no invoice
NI = [r for r in FJ if r["Status"] == "Requires invoicing" and (ASAT - D(r["End date"])).days > 14]
w("results/not_invoiced.csv", [{k: r[k] for k in ("Job #", "Client", "Service line", "Technician", "End date", "Jobber total")} | {"Days since completed": (ASAT - D(r["End date"])).days} for r in NI])
# ------------------------------------------------ quotes fact
FQ = []
for q in Q:
    c = CL[q["Client ID"]]; dec = q["Status"] in ("Approved", "Declined", "Expired")
    FQ.append({"Quote #": q["Quote #"], "Client ID": q["Client ID"], "Client type": c["Client type"], "Lead source": c["Lead source"], "Service line": q["Title"], "Created": q["Created"], "Sent": q["Sent"],
               "Month sent": q["Sent"][:7], "Status": q["Status"], "Decided": "Yes" if dec else "No", "Approved": 1 if q["Status"] == "Approved" else 0,
               "Days to approve": (D(q["Approved"]) - D(q["Sent"])).days if q["Approved"] else "", "Subtotal": f(q["Subtotal"])})
w("results/fact_quotes.csv", FQ)
# ------------------------------------------------ technician by month
def weekdays(m):
    y, mm = map(int, m.split("-")); d = dt.date(y, mm, 1); n = 0
    while d.month == mm: n += d.weekday() < 5; d += dt.timedelta(1)
    return n
TM = []
JOBT = {j["Job #"]: j for j in FJ}
vh = defaultdict(float)
for v in V:
    if v["Completed"] == "Yes": vh[(v["Assigned to"], v["Date"][:7])] += float(v["Duration (h)"])
for t in ST:
    for m in MONTHS:
        jm = [r for r in FJ if r["Technician"] == t and r["Month"] == m and r["Done"] == "Yes"]
        avail = round(weekdays(m) * float(ST[t]["Contract hours per week"]) / 5, 1)
        TM.append({"Technician": t, "Month": m, "Contract hours": avail, "Visit hours": round(vh[(t, m)], 2), "Utilisation": round(vh[(t, m)] / avail, 4),
                   "Jobs done": sum(1 for r in jm if r["Callback"] == "No"), "Callbacks": sum(1 for r in jm if r["Callback"] == "Yes"), "Revenue ex GST": round(sum(r["Revenue ex GST"] for r in jm), 2)})
w("results/tech_month.csv", TM)
# ------------------------------------------------ dashboard figures (the same definitions as the Looker Studio calculated fields)
done = [r for r in FJ if r["Done"] == "Yes" and r["Callback"] == "No"]; rev_jobs = [r for r in done if r["Revenue ex GST"] > 0]
S = lambda rs, k: round(sum(r[k] for r in rs), 2)
K = {}
K["revenue_ex"] = S(FI, "Ex GST"); K["revenue_jobber_only"] = S([r for r in FI if r["Source"] == "Jobber sync"], "Ex GST"); K["revenue_xero_only"] = S([r for r in FI if r["Source"] == "Xero only"], "Ex GST")
K["jobs_done"] = len(done); K["callbacks"] = sum(1 for r in FJ if r["Done"] == "Yes" and r["Callback"] == "Yes")
K["callback_rate"] = round(K["callbacks"] / K["jobs_done"], 4)
K["avg_job_value"] = round(S(rev_jobs, "Revenue ex GST") / len(rev_jobs), 2)
K["gm"] = round(S(rev_jobs, "Gross margin"), 2); K["gm_pct"] = round(K["gm"] / S(rev_jobs, "Revenue ex GST"), 4)
qs = [q for q in FQ if q["Sent"]]; qd = [q for q in FQ if q["Decided"] == "Yes"]
K["quotes_sent"] = len(qs); K["quotes_decided"] = len(qd); K["quotes_approved"] = sum(q["Approved"] for q in FQ); K["conversion"] = round(K["quotes_approved"] / len(qd), 4)
K["approved_value"] = round(sum(q["Subtotal"] for q in FQ if q["Approved"]), 2); K["open_quotes"] = sum(1 for q in FQ if q["Status"] == "Awaiting response"); K["open_quote_value"] = round(sum(q["Subtotal"] for q in FQ if q["Status"] == "Awaiting response"), 2)
dta = [q["Days to approve"] for q in FQ if q["Days to approve"] != ""]; K["days_to_approve"] = round(sum(dta) / len(dta), 1)
K["ar"] = S(FI, "Amount due"); K["ar_overdue"] = round(sum(r["Amount due"] for r in FI if r["Ageing"] not in ("Paid", "Not yet due")), 2)
K["ageing"] = {b: round(sum(r["Amount due"] for r in FI if r["Ageing"] == b), 2) for b in ["Not yet due", "1-30 days", "31-60 days", "61-90 days", "90+ days"]}
dtp = defaultdict(list)
for r in FI:
    if r["Days to pay"] != "": dtp[r["Client type"]].append(r["Days to pay"])
K["days_to_pay"] = {k: round(sum(v) / len(v), 1) for k, v in sorted(dtp.items())}
dti = defaultdict(list)
for r in done:
    if r["Days to invoice"] != "": dti[r["Client type"]].append(r["Days to invoice"])
K["days_to_invoice"] = {k: round(sum(v) / len(v), 1) for k, v in sorted(dti.items())}
K["rev_month_line"] = {m: {} for m in MONTHS}
for r in FI:
    j = JOBT.get(r["Reference"]); line = j["Service line"] if j else ("Commercial maintenance" if r["Reference"] == "Maintenance contract" else "Other (Xero only)")
    K["rev_month_line"][r["Month"]][line] = round(K["rev_month_line"][r["Month"]].get(line, 0) + r["Ex GST"], 2)
K["paid_month"] = {m: round(sum(float(p["Amount"]) for p in XP if p["Date"][:7] == m), 2) for m in MONTHS}
K["invoiced_month"] = {m: round(sum(r["Total"] for r in FI if r["Month"] == m), 2) for m in MONTHS}
lines = sorted({r["Service line"] for r in done})
K["by_line"] = {l: dict(jobs=sum(1 for r in done if r["Service line"] == l), revenue=S([r for r in rev_jobs if r["Service line"] == l], "Revenue ex GST"), gm=S([r for r in rev_jobs if r["Service line"] == l], "Gross margin")) for l in lines}
for l, v in K["by_line"].items(): v["gm_pct"] = round(v["gm"] / v["revenue"], 4) if v["revenue"] else 0
K["by_source"] = {}
for s in sorted({q["Lead source"] for q in FQ}):
    d_ = [q for q in qd if q["Lead source"] == s]; K["by_source"][s] = dict(decided=len(d_), approved=sum(q["Approved"] for q in d_), conversion=round(sum(q["Approved"] for q in d_) / len(d_), 4) if d_ else 0,
                                                                           value=round(sum(q["Subtotal"] for q in d_ if q["Approved"]), 2))
K["quotes_month"] = {m: dict(sent=sum(1 for q in qs if q["Month sent"] == m), approved=sum(1 for q in qs if q["Month sent"] == m and q["Approved"])) for m in MONTHS}
K["by_tech"] = {}
for t in ST:
    tm = [r for r in TM if r["Technician"] == t]; jd = [r for r in done if r["Technician"] == t]; cb = sum(r["Callbacks"] for r in tm)
    hrs = sum(r["Visit hours"] for r in tm); av = sum(r["Contract hours"] for r in tm)
    K["by_tech"][t] = dict(jobs=len(jd), callbacks=cb, callback_rate=round(cb / len(jd), 4), hours=round(hrs, 1), utilisation=round(hrs / av, 4), revenue=S(jd, "Revenue ex GST"), rev_per_hour=round(S(jd, "Revenue ex GST") / hrs, 2))
K["visits_week"] = {}
for v in V:
    if v["Completed"] == "Yes":
        d = D(v["Date"]); wk = (d - dt.timedelta(d.weekday())).isoformat(); K["visits_week"][wk] = K["visits_week"].get(wk, 0) + 1
K["tie"] = dict(TS); K["tie_diff_value"] = round(sum(t["Difference"] for t in tie if t["Status"] == "Amount differs"), 2); K["not_in_xero_value"] = round(sum(t["Jobber subtotal"] for t in tie if t["Status"] == "Not in Xero"), 2)
K["not_invoiced"] = len(NI); K["not_invoiced_value"] = round(sum(r["Jobber total"] for r in NI), 2)
K["top_overdue"] = sorted([(c, round(sum(r["Amount due"] for r in FI if r["Contact"] == c and r["Ageing"] not in ("Paid", "Not yet due")), 2), max([r["Days overdue"] for r in FI if r["Contact"] == c and r["Days overdue"] != ""] or [0])) for c in {r["Contact"] for r in FI}], key=lambda x: -x[1])[:8]
K["drafts"] = sum(1 for q in FQ if q["Status"] == "Draft"); K["xero_invoices"] = len(FI); K["voided"] = len(VOID); K["jobs_rows"] = len(FJ)
json.dump(K, open(os.path.join(HERE, "results/kpi.json"), "w"), indent=1)
# ------------------------------------------------ test against the answer key
KEY = json.load(open(os.path.join(HERE, "truth/key.json"))); TR = {t["Invoice #"]: t["issue"] for t in rd("truth/sync_issues.csv")}
found = {t["Invoice #"] if t["Status"] != "Voided duplicate (excluded)" else t["Invoice #"][:-2]: {"Not in Xero": "Not in Xero", "Amount differs": "Amount differs", "Xero only": "Xero only", "Voided duplicate (excluded)": "Voided duplicate"}[t["Status"]] for t in tie if t["Status"] != "Matched"}
TNI = {t["Job #"] for t in rd("truth/not_invoiced.csv")}
T = dict(revenue=[KEY["revenue_ex"], K["revenue_ex"]], ar=[KEY["ar"], K["ar"]], invoices=[KEY["invoices"], K["xero_invoices"]], jobs_done=[KEY["completed_jobs"], K["jobs_done"]], callbacks=[KEY["callbacks"], K["callbacks"]],
         quotes_sent=[KEY["quotes_sent"], K["quotes_sent"]], approved=[KEY["quotes_approved"], K["quotes_approved"]], decided=[KEY["quotes_decided"], K["quotes_decided"]],
         sync_issues=[len(TR), sum(1 for k, v in found.items() if TR.get(k) == v)], sync_false=[0, sum(1 for k, v in found.items() if TR.get(k) != v)],
         not_invoiced=[len(TNI), len(TNI & {r["Job #"] for r in NI})], not_invoiced_false=[0, len({r["Job #"] for r in NI} - TNI)], not_invoiced_value=[KEY["not_invoiced_value"], K["not_invoiced_value"]],
         callbacks_linked=[len(CB_TRUTH), sum(1 for c in CB_TRUTH if JOBT.get(c["Job #"], {}).get("Callback of") == c["original"])])
json.dump(T, open(os.path.join(HERE, "results/engine_test.json"), "w"), indent=1)
print({k: v for k, v in T.items()}); print("tie", dict(TS)); print({k: K[k] for k in ("revenue_ex", "revenue_xero_only", "gm_pct", "avg_job_value", "conversion", "ar", "ar_overdue", "callback_rate", "not_invoiced", "not_invoiced_value", "days_to_pay", "days_to_invoice")})
print({t: (v["utilisation"], v["callback_rate"], v["rev_per_hour"]) for t, v in K["by_tech"].items()}); print({l: (v["jobs"], v["gm_pct"]) for l, v in K["by_line"].items()}); print({s: v["conversion"] for s, v in K["by_source"].items()}); print(K["ageing"]); print(K["top_overdue"][:4])
bad = [k for k, (a, b) in T.items() if a != b]; print("FAIL" if bad else "ALL PASS", bad)
# ------------------------------------------------ steer22: 24-month series and year on year (12 months to Sep 2025 vs 12 months to Sep 2026)
YR = lambda m: "FY25" if m <= "2025-09" else "FY26"  # labels used on the dashboard: 12 months to 30 Sep
KM = []
for m in MONTHS:
    dm = [r for r in done if r["Month"] == m]; rj = [r for r in dm if r["Revenue ex GST"] > 0]
    KM.append({"Month": m, "Period": "Oct 2024 - Sep 2025" if YR(m) == "FY25" else "Oct 2025 - Sep 2026", "Revenue ex GST": round(sum(r["Ex GST"] for r in FI if r["Month"] == m), 2),
               "Jobs done": len(dm), "Callbacks": sum(1 for r in FJ if r["Done"] == "Yes" and r["Callback"] == "Yes" and r["Month"] == m),
               "Job revenue ex GST": S(rj, "Revenue ex GST"), "Gross margin": S(rj, "Gross margin"), "Quotes sent": K["quotes_month"][m]["sent"], "Quotes approved (of sent that month)": K["quotes_month"][m]["approved"],
               "Invoiced incl GST": K["invoiced_month"][m], "Payments received": K["paid_month"][m], "Visit hours": round(sum(r["Visit hours"] for r in TM if r["Month"] == m), 2)})
w("results/kpi_month.csv", KM)
YOY = {}
for y, lab in (("FY25", "12 months to Sep 2025"), ("FY26", "12 months to Sep 2026")):
    km = [r for r in KM if YR(r["Month"]) == y]; dq = [q for q in qd if q["Sent"] and YR(q["Month sent"]) == y]
    jr = sum(r["Job revenue ex GST"] for r in km); nrj = sum(1 for r in done if YR(r["Month"]) == y and r["Revenue ex GST"] > 0)
    dp = [r["Days to pay"] for r in FI if r["Days to pay"] != "" and YR(r["Month"]) == y and r["Client type"] == "Residential"]
    YOY[y] = dict(label=lab, revenue=round(sum(r["Revenue ex GST"] for r in km), 2), jobs=sum(r["Jobs done"] for r in km), callbacks=sum(r["Callbacks"] for r in km),
                  gm_pct=round(sum(r["Gross margin"] for r in km) / jr, 4), avg_job=round(jr / nrj, 2), decided=len(dq), approved=sum(q["Approved"] for q in dq),
                  conversion=round(sum(q["Approved"] for q in dq) / len(dq), 4), hours=round(sum(r["Visit hours"] for r in km), 1))
    YOY[y]["callback_rate"] = round(YOY[y]["callbacks"] / YOY[y]["jobs"], 4)
YOY["growth"] = round(YOY["FY26"]["revenue"] / YOY["FY25"]["revenue"] - 1, 4)
K["yoy"] = YOY; K["months"] = len(MONTHS); K["clients"] = len(CL); K["visits"] = sum(1 for v in V if v["Completed"] == "Yes"); K["quotes_rows"] = len(FQ)
K["series_ok"] = dict(revenue=round(sum(r["Revenue ex GST"] for r in KM) - K["revenue_ex"], 2), jobs=sum(r["Jobs done"] for r in KM) - K["jobs_done"])
json.dump(K, open(os.path.join(HERE, "results/kpi.json"), "w"), indent=1)
print("yoy", json.dumps(YOY), "series", K["series_ok"])