""" ATIME CDR analysis pipeline. Reads CDRs from SQLite (or an in-memory DataFrame), classifies each call by division / route / voicemail / caller-geography, and emits a single JSON-serializable metrics dict consumed by render_html and render_xlsx. Run standalone for a console validation dump: python3 analyze.py [--start YYYY-MM-DD] [--end YYYY-MM-DD] """ import argparse import json import sqlite3 import numpy as np import pandas as pd import config from mapping import load_maps, load_live_departments # --------------------------------------------------------------------------- # Load + classify # --------------------------------------------------------------------------- def load_df(db_path=None, start=None, end=None): conn = sqlite3.connect(db_path or config.DB_PATH) q = 'SELECT * FROM cdrs' clauses = [] if start: clauses.append(f'"call-start-datetime" >= "{start}"') if end: # inclusive end-of-day clauses.append(f'"call-start-datetime" <= "{end}T23:59:59-05:00"') if clauses: q += " WHERE " + " AND ".join(clauses) df = pd.read_sql_query(q, conn) conn.close() return df def classify(df, queue_map, direct_map, live_map=None, live_queues=None): """Add derived columns: division, route_type, is_voicemail, caller, area_code, dt.""" df = df.copy() # --- timestamps (already Eastern wall-clock with -05:00 offset) --- dt = pd.to_datetime(df["call-start-datetime"], errors="coerce", utc=True) # shift to fixed Eastern offset for wall-clock hour/day dt_local = dt.dt.tz_convert("Etc/GMT+5") df["dt"] = dt_local df["month"] = dt_local.dt.strftime("%Y-%m") df["hour"] = dt_local.dt.hour df["dow"] = dt_local.dt.dayofweek # Mon=0 # --- direction --- df["dir_label"] = df["call-direction"].map(config.DIRECTION_LABELS).fillna("other") # --- voicemail --- to_uri = df["call-term-to-uri"].fillna("").str.lower() match_uri = df["call-term-match-uri"].fillna("").str.lower() df["is_voicemail"] = (to_uri.str.contains(config.VMAIL_TO_URI_TOKEN, regex=False) | match_uri.str.contains(config.VMAIL_MATCH_URI_TOKEN, regex=False)) # --- caller number + area code (from call-orig-from-uri) --- df["caller"] = df["call-orig-from-uri"].fillna("").str.extract(config.PHONE_RE)[0] df["area_code"] = df["caller"].str[:3] # --- talk seconds numeric --- df["talk_s"] = pd.to_numeric(df["call-talking-duration-seconds"], errors="coerce").fillna(0) # --- division + route_type (priority chain) --- tu = df["call-through-user"].fillna("").str.strip() mu = df["call-term-user"].fillna("").str.strip() def dept_div(series): return series.fillna("").map(config.normalize_dept).map(config.DEPT_CROSSWALK) through_dept = dept_div(df["call-through-department"]) term_dept = dept_div(df["call-term-department"]) orig_dept = dept_div(df["call-orig-department"]) n = len(df) division = np.array([None] * n, dtype=object) route = np.array([None] * n, dtype=object) def fill(mask, div_series, route_label): m = mask & pd.isna(pd.Series(division, index=df.index)) idx = np.where(m.values)[0] for i in idx: division[i] = div_series.iloc[i] route[i] = route_label # 1. through-user is a queue/AA/bridge fill(tu.isin(queue_map), tu.map(queue_map), "queue") # 2. through-user is a direct user extension fill(tu.isin(direct_map), tu.map(direct_map), "direct") # 3. term-user is a queue fill(mu.isin(queue_map), mu.map(queue_map), "queue") # 4. term-user is a direct user fill(mu.isin(direct_map), mu.map(direct_map), "direct") # 5. live NS department cache (covers users/queues the xlsx doesn't know yet) if live_map: lq = live_queues or set() lm_q = {e: d for e, d in live_map.items() if e in lq} lm_d = {e: d for e, d in live_map.items() if e not in lq} fill(tu.isin(lm_q), tu.map(lm_q), "queue") fill(tu.isin(lm_d), tu.map(lm_d), "direct") fill(mu.isin(lm_q), mu.map(lm_q), "queue") fill(mu.isin(lm_d), mu.map(lm_d), "direct") # 6. raw department text fields fill(through_dept.notna(), through_dept, "department") fill(term_dept.notna(), term_dept, "department") fill(orig_dept.notna(), orig_dept, "department") df["division"] = division df["route_type"] = route # Force the main Auto-Attendant IVR leg to UNMAPPED (avoid double-counting greeting). aa = tu.eq(config.MAIN_AA_EXTENSION) df.loc[aa, "division"] = None df.loc[aa, "route_type"] = None df["is_aa998"] = aa.values # flag the IVR greeting leg for geography de-duplication df["division"] = df["division"].where(df["division"].notna(), None) # In-scope flag: the call belongs to a Reporting Division (not Admin, not unmapped) df["is_helpline"] = df["division"].isin(config.REPORTING_DIVISIONS) return df # --------------------------------------------------------------------------- # Metric helpers # --------------------------------------------------------------------------- def _pct(num, den): return round(100.0 * num / den, 1) if den else 0.0 def _hours(seconds): return round(float(seconds) / 3600.0, 1) def _summary(hl): """Headline summary over the in-scope frame (dir 0/1/2, all reporting divisions). Inbound and outbound are reported separately: voicemail, routing (queue/ direct/dept) and per-direction talk time are inbound-scoped, so queue + direct + dept == inbound attempts. Totals are kept alongside. """ answered = int((hl["call-direction"] == "1").sum()) missed = int((hl["call-direction"] == "2").sum()) attempts = answered + missed inbound = hl[hl["call-direction"].isin(["1", "2"])] outbound = hl[hl["call-direction"] == "0"] callers = inbound["caller"].dropna() callers = callers[callers != ""] unique_callers = callers.nunique() reached = inbound[inbound["call-direction"] == "1"]["caller"] reached = reached[reached.notna() & (reached != "")].nunique() vm = int(inbound["is_voicemail"].sum()) talk_in = inbound["talk_s"].sum() talk_out = outbound["talk_s"].sum() after_hours = inbound[(inbound["hour"] < config.BUSINESS_START_HOUR) | (inbound["hour"] >= config.BUSINESS_END_HOUR)] overnight = inbound[(inbound["hour"] >= config.OVERNIGHT_START_HOUR) | (inbound["hour"] < config.OVERNIGHT_END_HOUR)] return { "inbound_answered": answered, "inbound_missed": missed, "outbound": int(len(outbound)), "inbound_attempts": attempts, "total_calls": answered + missed + int(len(outbound)), "answer_rate": _pct(answered, attempts), "talk_seconds": int(hl["talk_s"].sum()), "talk_hours": _hours(hl["talk_s"].sum()), "talk_minutes": int(round(hl["talk_s"].sum() / 60)), "talk_hours_inbound": _hours(talk_in), "talk_minutes_inbound": int(round(talk_in / 60)), "talk_hours_outbound": _hours(talk_out), "talk_minutes_outbound": int(round(talk_out / 60)), "unique_callers": int(unique_callers), "callers_reached": int(reached), "reach_rate": _pct(reached, unique_callers), "voicemail_count": vm, "voicemail_pct_inbound": _pct(vm, attempts), "queue_count": int((inbound["route_type"] == "queue").sum()), "direct_count": int((inbound["route_type"] == "direct").sum()), "department_count": int((inbound["route_type"] == "department").sum()), "after_hours_pct": _pct(len(after_hours), len(inbound)), "overnight_pct": _pct(len(overnight), len(inbound)), "overnight_count": int(len(overnight)), } def _division_table(hl): """Per-division rows with inbound and outbound reported separately. Voicemail and routing (queue/direct/dept) are inbound-scoped, so per row queue + direct + dept == inbound attempts (answered + missed). """ rows = [] for div in config.REPORTING_DIVISIONS: d = hl[hl["division"] == div] if len(d) == 0: continue answered = int((d["call-direction"] == "1").sum()) missed = int((d["call-direction"] == "2").sum()) inbound = d[d["call-direction"].isin(["1", "2"])] outb = d[d["call-direction"] == "0"] callers = inbound["caller"] callers = callers[callers.notna() & (callers != "")] rows.append({ "division": div, "inbound_answered": answered, "inbound_missed": missed, "outbound": int(len(outb)), "inbound_attempts": answered + missed, "total": answered + missed + int(len(outb)), "answer_rate": _pct(answered, answered + missed), "talk_hours": _hours(d["talk_s"].sum()), "talk_hours_in": _hours(inbound["talk_s"].sum()), "talk_hours_out": _hours(outb["talk_s"].sum()), "unique_callers": int(callers.nunique()), "voicemail": int(inbound["is_voicemail"].sum()), "queue": int((inbound["route_type"] == "queue").sum()), "direct": int((inbound["route_type"] == "direct").sum()), "dept": int((inbound["route_type"] == "department").sum()), }) rows.sort(key=lambda r: r["inbound_attempts"], reverse=True) return rows def _monthly(hl, divisions=None): d = hl if divisions is None else hl[hl["division"].isin(divisions)] out = [] for m, g in d.groupby("month"): if not m: continue answered = int((g["call-direction"] == "1").sum()) missed = int((g["call-direction"] == "2").sum()) outbound = int((g["call-direction"] == "0").sum()) g_in = g[g["call-direction"].isin(["1", "2"])] g_out = g[g["call-direction"] == "0"] out.append({ "month": m, "inbound_answered": answered, "inbound_missed": missed, "outbound": outbound, "total": answered + missed + outbound, "answer_rate": _pct(answered, answered + missed), "talk_hours": _hours(g["talk_s"].sum()), "talk_hours_in": _hours(g_in["talk_s"].sum()), "talk_hours_out": _hours(g_out["talk_s"].sum()), }) out.sort(key=lambda r: r["month"]) return out def _monthly_by_division(hl): res = {} for div in config.REPORTING_DIVISIONS: rows = _monthly(hl, [div]) if rows: res[div] = rows return res def _handler_ext(row): tu = (row.get("call-through-user") or "").strip() return tu if tu else (row.get("call-term-user") or "").strip() def _mhl_case_managers(hl, user_names): med = hl[(hl["division"] == "Medical")].copy() med["ext"] = med["call-through-user"].fillna("").str.strip() med = med[med["ext"] != ""] rows = [] for ext, g in med.groupby("ext"): answered = int((g["call-direction"] == "1").sum()) missed = int((g["call-direction"] == "2").sum()) outbound = int((g["call-direction"] == "0").sum()) if answered + missed + outbound == 0: continue rows.append({ "extension": ext, "name": user_names.get(ext, ""), "inbound_answered": answered, "inbound_missed": missed, "outbound": outbound, "total": answered + missed + outbound, "answer_rate": _pct(answered, answered + missed), "talk_hours": _hours(g["talk_s"].sum()), "talk_hours_in": _hours(g[g["call-direction"].isin(["1", "2"])]["talk_s"].sum()), "talk_hours_out": _hours(g[g["call-direction"] == "0"]["talk_s"].sum()), }) rows.sort(key=lambda r: r["total"], reverse=True) return rows def _mhl_by_person(case_managers): """Aggregate the by-extension MHL table to one row per named person. Hunt pilots (9256) and direct extensions (256) for the same case manager are summed; unnamed extensions (e.g. queue pilot 501) stay as their own row. """ buckets = {} for r in case_managers: key = r["name"] if r["name"] else f"ext {r['extension']}" b = buckets.setdefault(key, {"name": r["name"] or f"Ext {r['extension']}", "extensions": [], "inbound_answered": 0, "inbound_missed": 0, "outbound": 0, "total": 0, "talk_seconds": 0.0, "talk_seconds_in": 0.0, "talk_seconds_out": 0.0}) b["extensions"].append(r["extension"]) b["inbound_answered"] += r["inbound_answered"] b["inbound_missed"] += r["inbound_missed"] b["outbound"] += r["outbound"] b["total"] += r["total"] b["talk_seconds"] += r["talk_hours"] * 3600.0 b["talk_seconds_in"] += r.get("talk_hours_in", 0.0) * 3600.0 b["talk_seconds_out"] += r.get("talk_hours_out", 0.0) * 3600.0 rows = [] for b in buckets.values(): att = b["inbound_answered"] + b["inbound_missed"] rows.append({ "name": b["name"], "extensions": ", ".join(sorted(b["extensions"])), "inbound_answered": b["inbound_answered"], "inbound_missed": b["inbound_missed"], "outbound": b["outbound"], "total": b["total"], "answer_rate": _pct(b["inbound_answered"], att), "talk_hours": round(b["talk_seconds"] / 3600.0, 1), "talk_hours_in": round(b["talk_seconds_in"] / 3600.0, 1), "talk_hours_out": round(b["talk_seconds_out"] / 3600.0, 1), }) rows.sort(key=lambda r: r["total"], reverse=True) return rows def _missed_by_extension(hl, user_names): res = {} missed = hl[hl["call-direction"] == "2"].copy() missed["ext"] = missed.apply(_handler_ext, axis=1) missed = missed[missed["ext"] != ""] for div in config.REPORTING_DIVISIONS: d = missed[missed["division"] == div] if len(d) == 0: continue counts = d.groupby("ext").size().sort_values(ascending=False) rows = [{"extension": ext, "name": user_names.get(ext, ""), "missed": int(c)} for ext, c in counts.items()] res[div] = rows return res def _hourly_dow_heatmap(hl): inbound = hl[hl["call-direction"].isin(["1", "2"])] hourly = [int((inbound["hour"] == h).sum()) for h in range(24)] dow = [int((inbound["dow"] == d).sum()) for d in range(7)] heat = [[int(((inbound["dow"] == d) & (inbound["hour"] == h)).sum()) for h in range(24)] for d in range(7)] return hourly, dow, heat def _geo(hl): inbound = hl[hl["call-direction"].isin(["1", "2"])].copy() inbound = inbound[inbound["area_code"].notna() & (inbound["area_code"] != "")] # area code call counts + unique callers ac_rows = [] for ac, g in inbound.groupby("area_code"): callers = g["caller"] ac_rows.append({ "area_code": ac, "calls": int(len(g)), "unique_callers": int(callers.nunique()), "answered": int((g["call-direction"] == "1").sum()), "region": config.REGION_BY_AREA_CODE.get(ac, ""), }) ac_rows.sort(key=lambda r: r["calls"], reverse=True) # regional rollup inbound["region"] = inbound["area_code"].map(config.REGION_BY_AREA_CODE).fillna("Other / Outside listed regions") reg_rows = [] for region, g in inbound.groupby("region"): reg_rows.append({ "region": region, "calls": int(len(g)), "unique_callers": int(g["caller"].nunique()), }) reg_rows.sort(key=lambda r: r["calls"], reverse=True) return {"area_codes": ac_rows, "regional": reg_rows, "total_geocoded_calls": int(len(inbound))} def _region(inbound_df, region): """Regional deep-dive over the Scope-A inbound frame (dir 1/2, IVR-998 excluded). `inbound_df` is already all real inbound calls; `region` is a config.REGIONS entry. Per-division figures cover every Reporting Division (loop is over REPORTING_DIVISIONS). """ codes = list(region["area_codes"].keys()) sub = inbound_df[inbound_df["area_code"].isin(codes)].copy() answered = int((sub["call-direction"] == "1").sum()) missed = int((sub["call-direction"] == "2").sum()) callers = sub["caller"] callers = callers[callers.notna() & (callers != "")] reached = sub[sub["call-direction"] == "1"]["caller"] reached = reached[reached.notna() & (reached != "")].nunique() summary = { "inbound_answered": answered, "inbound_missed": missed, "inbound_attempts": answered + missed, "answer_rate": _pct(answered, answered + missed), "talk_hours": _hours(sub["talk_s"].sum()), "talk_minutes": int(round(sub["talk_s"].sum() / 60)), "unique_callers": int(callers.nunique()), "callers_reached": int(reached), "reach_rate": _pct(reached, callers.nunique()), "voicemail_count": int(sub["is_voicemail"].sum()), "voicemail_pct_inbound": _pct(int(sub["is_voicemail"].sum()), answered + missed), "pct_of_all_inbound": _pct(len(sub), len(inbound_df)), } ac = [] for code in codes: g = sub[sub["area_code"] == code] ac.append({ "area_code": code, "label": region["area_codes"][code], "calls": int(len(g)), "answered": int((g["call-direction"] == "1").sum()), "missed": int((g["call-direction"] == "2").sum()), "unique_callers": int(g["caller"][g["caller"].notna() & (g["caller"] != "")].nunique()), }) ac.sort(key=lambda r: r["calls"], reverse=True) by_div = [] for div in config.REPORTING_DIVISIONS: g = sub[sub["division"] == div] if len(g) == 0: continue by_div.append({ "division": div, "inbound_answered": int((g["call-direction"] == "1").sum()), "inbound_missed": int((g["call-direction"] == "2").sum()), "inbound_attempts": int(len(g)), "talk_hours": _hours(g["talk_s"].sum()), "unique_callers": int(g["caller"][g["caller"].notna() & (g["caller"] != "")].nunique()), }) by_div.sort(key=lambda r: r["inbound_attempts"], reverse=True) monthly = [] for m, g in sub.groupby("month"): if not m: continue monthly.append({ "month": m, "inbound_answered": int((g["call-direction"] == "1").sum()), "inbound_missed": int((g["call-direction"] == "2").sum()), "inbound_attempts": int(len(g)), "talk_hours": _hours(g["talk_s"].sum()), }) monthly.sort(key=lambda r: r["month"]) meta = {k: region.get(k, "") for k in ("key", "name", "display", "msa_name", "accent", "counties", "excluded_note")} return {"meta": meta, "summary": summary, "area_codes": ac, "by_division": by_div, "monthly": monthly} def _validation(df, hl): """Cross-check user-based classification against CDR department text fields.""" user_routed = df[df["route_type"].isin(["queue", "direct"])].copy() # crosswalk the through-department for an independent opinion cdr_div = user_routed["call-through-department"].fillna("").map(config.normalize_dept).map(config.DEPT_CROSSWALK) comparable = user_routed[cdr_div.notna() & (user_routed["division"].notna())] cdr_div_c = cdr_div[comparable.index] agree = (comparable["division"] == cdr_div_c).sum() total = len(comparable) disagreements = comparable[comparable["division"] != cdr_div_c] dis_ext = disagreements["call-through-user"].value_counts().head(10).to_dict() total_records = int(len(df)) internal = int((df["call-direction"] == "3").sum()) admin = int((df["division"] == config.ADMIN).sum()) unmapped = int(df["division"].isna().sum()) aa_998 = int((df["call-through-user"].fillna("").str.strip() == config.MAIN_AA_EXTENSION).sum()) extracted = int((df["caller"].notna() & (df["caller"] != "")).sum()) inbound_all = df[df["call-direction"].isin(["1", "2"])] inbound_extracted = int((inbound_all["caller"].notna() & (inbound_all["caller"] != "")).sum()) return { "classification_agreement_pct": _pct(agree, total), "classification_compared": total, "classification_disagreements": int(total - agree), "disagreement_top_extensions": {str(k): int(v) for k, v in dis_ext.items()}, "total_records": total_records, "internal_excluded": internal, "admin_excluded": admin, "unmapped_excluded": unmapped, "unmapped_pct": _pct(unmapped, total_records), "main_aa_998_legs": aa_998, "phone_extracted_pct_all": _pct(extracted, total_records), "phone_extracted_pct_inbound": _pct(inbound_extracted, len(inbound_all)), "helpline_records": int(len(hl)), } # --------------------------------------------------------------------------- # Top-level # --------------------------------------------------------------------------- def run(db_path=None, start=None, end=None, generated_at=None): queue_map, direct_map, mp = load_maps() live_map, live_queues, live_names = load_live_departments() df = load_df(db_path, start, end) df = classify(df, queue_map, direct_map, live_map, live_queues) # Display names: spreadsheet names win; live pull fills the gaps. ext_names = dict(live_names) ext_names.update({e: n for e, n in mp["ext_names"].items() if n}) mp["ext_names"] = ext_names hl = df[df["is_helpline"] & df["call-direction"].isin(["0", "1", "2"])].copy() # Scope A geography frame: every real inbound call (dir 1/2) EXCEPT the main-AA # IVR greeting leg (ext 998), which is excluded to avoid double-counting. This # includes unmapped callers — they are still real people reaching out from a # region. Per-division tables remain division-scoped. geo_inbound = df[df["call-direction"].isin(["1", "2"]) & ~df["is_aa998"]].copy() dts = pd.to_datetime(df["call-start-datetime"], errors="coerce", utc=True) metrics = { "meta": { "generated_at": generated_at or "", "domain": config.DOMAIN, "period_start": (start or (str(dts.min())[:10] if dts.notna().any() else "")), "period_end": (end or (str(dts.max())[:10] if dts.notna().any() else "")), "org_name": config.ORG_NAME, "org_tagline": config.ORG_TAGLINE, }, "summary": _summary(hl), "by_division": _division_table(hl), "monthly": _monthly(hl), "monthly_by_division": _monthly_by_division(hl), "mhl_case_managers": _mhl_case_managers(hl, mp["ext_names"]), "mhl_by_person": _mhl_by_person(_mhl_case_managers(hl, mp["ext_names"])), "missed_by_extension": _missed_by_extension(hl, mp["ext_names"]), "geo": _geo(geo_inbound), "regions": {key: _region(geo_inbound, cfg) for key, cfg in config.REGIONS.items()}, "validation": _validation(df, hl), "brand": config.BRAND, } hourly, dow, heat = _hourly_dow_heatmap(hl) metrics["hourly"] = hourly metrics["dow"] = dow metrics["heatmap"] = heat return metrics, df def _print_validation(m): s, v = m["summary"], m["validation"] print("\n" + "=" * 64) print(f"ATIME CDR — {m['meta']['period_start']} .. {m['meta']['period_end']}") print("=" * 64) print(f"Total records in DB ............ {v['total_records']:,}") print(f" internal (dir 3) excluded .... {v['internal_excluded']:,}") print(f" admin/back-office excluded ... {v['admin_excluded']:,}") print(f" unmapped excluded ............ {v['unmapped_excluded']:,} ({v['unmapped_pct']}%)") print(f" main AA ext 998 legs ......... {v['main_aa_998_legs']:,}") print(f" in-scope records ............. {v['helpline_records']:,}") print("\n-- HEADLINE (all reporting divisions) --") print(f" inbound answered ............. {s['inbound_answered']:,}") print(f" inbound missed ............... {s['inbound_missed']:,}") print(f" outbound ..................... {s['outbound']:,}") print(f" answer rate .................. {s['answer_rate']}%") print(f" talk time ................... {s['talk_hours']:,} hrs total " f"(inbound {s['talk_hours_inbound']:,} / outbound {s['talk_hours_outbound']:,})") print(f" unique callers .............. {s['unique_callers']:,}") print(f" reach rate .................. {s['reach_rate']}%") print(f" voicemail ................... {s['voicemail_count']:,} ({s['voicemail_pct_inbound']}% of inbound)") print(f" inbound routed q/d/dept ..... {s['queue_count']:,} / {s['direct_count']:,} / {s['department_count']:,}" f" (= attempts {s['inbound_attempts']:,})") print(f" after-hours / overnight ..... {s['after_hours_pct']}% / {s['overnight_pct']}%") print("\n-- VALIDATION --") print(f" classification agreement .... {v['classification_agreement_pct']}% " f"({v['classification_disagreements']}/{v['classification_compared']} disagree)") print(f" top disagree extensions ..... {v['disagreement_top_extensions']}") print(f" phone extracted (inbound) ... {v['phone_extracted_pct_inbound']}%") print("\n-- BY DIVISION (inbound ans/miss/answer% | talk in+out | callers | routing q/d/dept | outbound) --") for r in m["by_division"]: print(f" {r['division']:<20} {r['inbound_answered']:>6,} / {r['inbound_missed']:>6,} " f"{r['answer_rate']:>5}% {r['talk_hours_in']:>7,}h+{r['talk_hours_out']:,}h " f"{r['unique_callers']:>5,} [q{r['queue']:,}/d{r['direct']:,}/dp{r['dept']:,}] " f"out {r['outbound']:,}") for key, reg in m["regions"].items(): c = reg["summary"] codes = " + ".join(a["area_code"] for a in reg["area_codes"]) print(f"\n-- {reg['meta']['display'].upper()} ({codes}) --") print(f" inbound attempts ............ {c['inbound_attempts']:,} ({c['pct_of_all_inbound']}% of all inbound)") print(f" answered / missed ........... {c['inbound_answered']:,} / {c['inbound_missed']:,} ({c['answer_rate']}% answer)") print(f" unique callers .............. {c['unique_callers']:,} reach {c['reach_rate']}%") print(f" talk time ................... {c['talk_hours']:,} hrs voicemail {c['voicemail_count']:,}") for a in reg["area_codes"][:6]: print(f" {a['area_code']}: {a['calls']:,} inbound, {a['unique_callers']:,} callers — {a['label']}") print("\n-- TOP AREA CODES (inbound) --") for a in m["geo"]["area_codes"][:12]: print(f" {a['area_code']}: {a['calls']:,} calls, {a['unique_callers']:,} callers {a['region']}") if __name__ == "__main__": ap = argparse.ArgumentParser() ap.add_argument("--start") ap.add_argument("--end") ap.add_argument("--json", help="write metrics JSON to this path") args = ap.parse_args() m, _ = run(start=args.start, end=args.end) _print_validation(m) if args.json: with open(args.json, "w") as f: json.dump(m, f, indent=2) print(f"\nwrote {args.json}")