267 lines
8.7 KiB
Python
267 lines
8.7 KiB
Python
#!/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()
|