150 lines
4.2 KiB
Python
150 lines
4.2 KiB
Python
#!/usr/bin/env python3
|
|
"""Build a BARON-SSO identity mapping template from the EGBIM QA dump.
|
|
|
|
The output is a review template. It never invents an SSO subject or tenant and
|
|
it never assigns administrator permissions.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import csv
|
|
from pathlib import Path
|
|
|
|
import migrate_egbim_dump
|
|
|
|
|
|
migrate_egbim_dump.TABLES["qa_comments"] = (
|
|
"comment_id",
|
|
"post_id",
|
|
"commenter",
|
|
"content",
|
|
"created_at",
|
|
"user_name",
|
|
"updated_at",
|
|
)
|
|
|
|
ADMIN_CANDIDATE_EMAILS = {
|
|
"cjy627@hanmaceng.co.kr",
|
|
"b23072@hanmaceng.co.kr",
|
|
"kjy0426@hanmaceng.co.kr",
|
|
"b21367@hanmaceng.co.kr",
|
|
"rmsgud1202@hanmaceng.co.kr",
|
|
}
|
|
|
|
|
|
def add_identity(
|
|
identities: dict[str, dict[str, object]],
|
|
identifier: str | None,
|
|
*,
|
|
legacy_user_id: str | None = None,
|
|
legacy_name: str | None = None,
|
|
post_count: bool = False,
|
|
comment_count: bool = False,
|
|
) -> None:
|
|
key = (identifier or "").strip().lower()
|
|
if not key:
|
|
return
|
|
item = identities.setdefault(
|
|
key,
|
|
{
|
|
"legacy_user_ids": set(),
|
|
"legacy_names": set(),
|
|
"post_count": 0,
|
|
"comment_count": 0,
|
|
"admin_candidate": key in ADMIN_CANDIDATE_EMAILS,
|
|
},
|
|
)
|
|
if legacy_user_id:
|
|
item["legacy_user_ids"].add(str(legacy_user_id))
|
|
if legacy_name:
|
|
item["legacy_names"].add(str(legacy_name))
|
|
if post_count:
|
|
item["post_count"] += 1
|
|
if comment_count:
|
|
item["comment_count"] += 1
|
|
|
|
|
|
def write_template(posts: list[dict], comments: list[dict], output: Path) -> int:
|
|
identities: dict[str, dict[str, object]] = {}
|
|
for row in posts:
|
|
add_identity(
|
|
identities,
|
|
row.get("login_id"),
|
|
legacy_user_id=row.get("user_id"),
|
|
legacy_name=row.get("user_name"),
|
|
post_count=True,
|
|
)
|
|
for row in comments:
|
|
add_identity(
|
|
identities,
|
|
row.get("commenter"),
|
|
legacy_name=row.get("user_name"),
|
|
comment_count=True,
|
|
)
|
|
|
|
output.parent.mkdir(parents=True, exist_ok=True)
|
|
with output.open("w", newline="", encoding="utf-8") as stream:
|
|
writer = csv.writer(stream)
|
|
writer.writerow(
|
|
[
|
|
"legacy_identifier",
|
|
"legacy_user_ids",
|
|
"legacy_names",
|
|
"post_count",
|
|
"comment_count",
|
|
"admin_candidate",
|
|
"sso_subject",
|
|
"tenant_id",
|
|
"email",
|
|
"name",
|
|
"department",
|
|
"phone",
|
|
"resolution_status",
|
|
"notes",
|
|
]
|
|
)
|
|
for identifier in sorted(identities):
|
|
item = identities[identifier]
|
|
writer.writerow(
|
|
[
|
|
identifier,
|
|
";".join(sorted(item["legacy_user_ids"])),
|
|
";".join(sorted(item["legacy_names"])),
|
|
item["post_count"],
|
|
item["comment_count"],
|
|
"Y" if item["admin_candidate"] else "N",
|
|
"",
|
|
"",
|
|
"",
|
|
"",
|
|
"",
|
|
"",
|
|
"UNRESOLVED",
|
|
"BARON-SSO profile/secondary_emails 확인 후 입력",
|
|
]
|
|
)
|
|
return len(identities)
|
|
|
|
|
|
def main() -> None:
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument("--source-sql", type=Path, default=Path("scripts/egbim_qa.sql"))
|
|
parser.add_argument(
|
|
"--output",
|
|
type=Path,
|
|
default=Path("scripts/generated/egbim_migration/identity_mapping_template.csv"),
|
|
)
|
|
args = parser.parse_args()
|
|
|
|
text = args.source_sql.read_text(encoding="utf-8")
|
|
posts = migrate_egbim_dump.load_table(text, "qa_posts")
|
|
comments = migrate_egbim_dump.load_table(text, "qa_comments")
|
|
identities = write_template(posts, comments, args.output)
|
|
print(f"posts={len(posts)} comments={len(comments)} identities={identities}")
|
|
print(f"output={args.output}")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|