94 lines
3.2 KiB
Python
94 lines
3.2 KiB
Python
#!/usr/bin/env python3
|
|
"""Apply exact email/alias matches from a BARON-SSO export to the template."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import csv
|
|
import re
|
|
from collections import defaultdict
|
|
from pathlib import Path
|
|
|
|
|
|
EMAIL_RE = re.compile(r"[A-Z0-9._%+-]+@[A-Z0-9.-]+\.[A-Z]{2,}", re.IGNORECASE)
|
|
MATCH_COLUMNS = (
|
|
"Email",
|
|
"Meta:secondary_emails",
|
|
"Meta:aliasEmails",
|
|
"Meta:worksmobileAliasEmails",
|
|
"Meta:sub_email",
|
|
"Meta:external_sub_email",
|
|
"Meta:source_login_id",
|
|
)
|
|
|
|
|
|
def emails(value: str | None) -> set[str]:
|
|
return {match.lower() for match in EMAIL_RE.findall(value or "")}
|
|
|
|
|
|
def main() -> None:
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument("--users", type=Path, default=Path("users_export_20260805.csv"))
|
|
parser.add_argument(
|
|
"--template",
|
|
type=Path,
|
|
default=Path("scripts/generated/egbim_migration/identity_mapping_template.csv"),
|
|
)
|
|
parser.add_argument("--output", type=Path)
|
|
args = parser.parse_args()
|
|
output = args.output or args.template
|
|
|
|
with args.users.open(encoding="utf-8-sig", newline="") as stream:
|
|
users = list(csv.DictReader(stream))
|
|
with args.template.open(encoding="utf-8", newline="") as stream:
|
|
template = list(csv.DictReader(stream))
|
|
|
|
index: dict[str, list[tuple[dict[str, str], str]]] = defaultdict(list)
|
|
for user in users:
|
|
for column in MATCH_COLUMNS:
|
|
for identifier in emails(user.get(column)):
|
|
index[identifier].append((user, column))
|
|
|
|
resolved = 0
|
|
ambiguous = 0
|
|
unmatched = 0
|
|
for row in template:
|
|
identifier = row["legacy_identifier"].strip().lower()
|
|
candidates = index.get(identifier, [])
|
|
users_by_id = {user["user_id"]: (user, column) for user, column in candidates}
|
|
if len(users_by_id) != 1:
|
|
row["resolution_status"] = "AMBIGUOUS" if len(users_by_id) > 1 else "UNRESOLVED"
|
|
row["notes"] = (
|
|
"BARON-SSO export에서 동일 alias가 여러 사용자와 일치함"
|
|
if len(users_by_id) > 1
|
|
else "정확한 이메일/보조 이메일 일치 없음; 이름만으로 자동 매핑하지 않음"
|
|
)
|
|
if len(users_by_id) > 1:
|
|
ambiguous += 1
|
|
else:
|
|
unmatched += 1
|
|
continue
|
|
|
|
user, matched_column = next(iter(users_by_id.values()))
|
|
row["sso_subject"] = user.get("user_id", "")
|
|
row["tenant_id"] = user.get("tenant_id", "")
|
|
row["email"] = user.get("Email", "")
|
|
row["name"] = user.get("Name", "")
|
|
row["department"] = user.get("Meta:department") or user.get("JobTitle") or user.get("Position") or ""
|
|
row["phone"] = user.get("Phone", "")
|
|
row["resolution_status"] = "RESOLVED"
|
|
row["notes"] = f"exact match: {matched_column}"
|
|
resolved += 1
|
|
|
|
with output.open("w", encoding="utf-8", newline="") as stream:
|
|
writer = csv.DictWriter(stream, fieldnames=template[0].keys())
|
|
writer.writeheader()
|
|
writer.writerows(template)
|
|
|
|
print(f"template_rows={len(template)} resolved={resolved} ambiguous={ambiguous} unmatched={unmatched}")
|
|
print(f"output={output}")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|