#!/usr/bin/env python3 import argparse import csv import json from collections import defaultdict from pathlib import Path ROOT_SLUG = "hanmac-family" def parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser( description="Export Baron org-context JSON to CSV files for photo mapping." ) parser.add_argument("--input", required=True, help="Path to org-context JSON file") parser.add_argument( "--output-dir", required=True, help="Directory where CSV files will be written" ) return parser.parse_args() def normalize_phone(phone: str) -> str: digits = "".join(ch for ch in (phone or "") if ch.isdigit()) if digits.startswith("82") and len(digits) >= 11: digits = "0" + digits[2:] return digits def format_phone(phone: str) -> str: digits = normalize_phone(phone) if len(digits) == 11: return f"{digits[:3]}-{digits[3:7]}-{digits[7:]}" if len(digits) == 10: return f"{digits[:3]}-{digits[3:6]}-{digits[6:]}" return phone or "" def email_local_part(email: str) -> str: if "@" not in (email or ""): return "" return email.split("@", 1)[0].strip() def load_json(path: Path) -> dict: with path.open("r", encoding="utf-8") as handle: return json.load(handle) def build_tree_indexes(tree: dict) -> tuple[dict, dict, dict]: tenant_by_id: dict[str, dict] = {} parent_by_id: dict[str, str | None] = {} children_by_id: dict[str, list[str]] = defaultdict(list) def visit(node: dict, parent_id: str | None) -> None: tenant_id = node.get("id", "") if not tenant_id: return tenant_by_id[tenant_id] = node parent_by_id[tenant_id] = parent_id if parent_id: children_by_id[parent_id].append(tenant_id) for child in node.get("children", []) or []: if isinstance(child, dict): visit(child, tenant_id) visit(tree, None) return tenant_by_id, parent_by_id, children_by_id def build_flat_tenant_index(tenants: list[dict]) -> dict[str, dict]: return { tenant.get("id", ""): tenant for tenant in tenants if isinstance(tenant, dict) and tenant.get("id") } def tenant_path( tenant_id: str, tenant_by_id: dict[str, dict], parent_by_id: dict[str, str | None] ) -> list[dict]: path: list[dict] = [] current_id = tenant_id while current_id: node = tenant_by_id.get(current_id) if node is None: break path.append(node) current_id = parent_by_id.get(current_id) or "" path.reverse() return path def resolve_company(path: list[dict], tenant: dict) -> tuple[str, str]: if not path: return tenant.get("id", ""), tenant.get("name", "") if len(path) == 1: return path[0].get("id", ""), path[0].get("name", "") for node in path[1:]: node_type = (node.get("type") or "").upper() if node_type in {"COMPANY", "ORGANIZATION"}: return node.get("id", ""), node.get("name", "") first_child = path[1] return first_child.get("id", ""), first_child.get("name", "") def row_sort_key(row: dict) -> tuple: phone_key = row["phone_normalized"] or "99999999999" return ( row["company_name"], phone_key, row["department_name"], row["name"], row["email"], ) def primary_preference(row: dict) -> tuple: return ( 0 if row["is_primary"] == "true" else 1, 0 if row["phone_normalized"] else 1, 0 if row["grade"] else 1, 0 if row["position"] else 1, row["company_name"], row["department_name"], row["email"], ) def build_rows(payload: dict) -> tuple[list[dict], list[dict], list[dict]]: tree = payload.get("tree") or {} tenants = payload.get("tenants") or [] tree_tenant_by_id, parent_by_id, _children_by_id = build_tree_indexes(tree) flat_tenant_by_id = build_flat_tenant_index(tenants) for tenant_id, tenant in flat_tenant_by_id.items(): tree_tenant_by_id.setdefault(tenant_id, tenant) membership_rows: list[dict] = [] for tenant in tenants: if not isinstance(tenant, dict): continue tenant_id = tenant.get("id", "") tenant_name = tenant.get("name", "") path = tenant_path(tenant_id, tree_tenant_by_id, parent_by_id) company_tenant_id, company_name = resolve_company(path, tenant) path_names = " > ".join( node.get("name", "") for node in path if node.get("name") and node.get("slug") != ROOT_SLUG ) for member in tenant.get("members", []) or []: if not isinstance(member, dict): continue raw_phone = (member.get("phone") or member.get("phoneNumber") or "").strip() row = { "company_tenant_id": company_tenant_id, "company_name": company_name, "department_tenant_id": tenant_id, "department_name": tenant_name, "department_path": path_names, "email": (member.get("email") or "").strip(), "email_id": email_local_part(member.get("email") or ""), "phone_raw": raw_phone, "phone_normalized": normalize_phone(raw_phone), "phone_display": format_phone(raw_phone), "name": (member.get("name") or "").strip(), "grade": (member.get("grade") or "").strip(), "position": (member.get("position") or "").strip(), "member_department": (member.get("department") or "").strip(), "is_primary": "true" if member.get("isPrimary") else "false", "member_id": (member.get("id") or "").strip(), } membership_rows.append(row) membership_rows.sort(key=row_sort_key) primary_rows = [row for row in membership_rows if row["is_primary"] == "true"] phone_index: dict[str, list[dict]] = defaultdict(list) for row in membership_rows: if row["phone_normalized"]: phone_index[row["phone_normalized"]].append(row) deduped_rows: list[dict] = [] for phone in sorted(phone_index): best = sorted(phone_index[phone], key=primary_preference)[0] duplicate_count = len(phone_index[phone]) selected = dict(best) selected["duplicate_row_count_for_phone"] = str(duplicate_count) deduped_rows.append(selected) deduped_rows.sort(key=row_sort_key) return membership_rows, primary_rows, deduped_rows def write_csv(path: Path, rows: list[dict], fieldnames: list[str]) -> None: path.parent.mkdir(parents=True, exist_ok=True) with path.open("w", encoding="utf-8-sig", newline="") as handle: writer = csv.DictWriter(handle, fieldnames=fieldnames) writer.writeheader() for row in rows: writer.writerow({field: row.get(field, "") for field in fieldnames}) def write_summary(path: Path, membership_rows: list[dict], primary_rows: list[dict], deduped_rows: list[dict]) -> None: company_counts: dict[str, int] = defaultdict(int) for row in deduped_rows: company_counts[row["company_name"]] += 1 lines = [ "Baron org-context CSV export summary", f"membership_rows={len(membership_rows)}", f"primary_rows={len(primary_rows)}", f"deduped_phone_rows={len(deduped_rows)}", "company_counts:", ] for company_name in sorted(company_counts): lines.append(f"- {company_name}: {company_counts[company_name]}") path.write_text("\n".join(lines) + "\n", encoding="utf-8") def main() -> None: args = parse_args() payload = load_json(Path(args.input)) membership_rows, primary_rows, deduped_rows = build_rows(payload) output_dir = Path(args.output_dir) fieldnames = [ "company_tenant_id", "company_name", "department_tenant_id", "department_name", "department_path", "email", "email_id", "phone_raw", "phone_normalized", "phone_display", "name", "grade", "position", "member_department", "is_primary", "member_id", ] deduped_fieldnames = fieldnames + ["duplicate_row_count_for_phone"] write_csv(output_dir / "baron_org_context_membership_rows.csv", membership_rows, fieldnames) write_csv(output_dir / "baron_org_context_primary_rows.csv", primary_rows, fieldnames) write_csv(output_dir / "baron_org_context_phone_mapping.csv", deduped_rows, deduped_fieldnames) write_summary( output_dir / "baron_org_context_export_summary.txt", membership_rows, primary_rows, deduped_rows, ) if __name__ == "__main__": main()