This commit is contained in:
@@ -0,0 +1,93 @@
|
||||
#!/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()
|
||||
Reference in New Issue
Block a user