660 lines
27 KiB
Python
660 lines
27 KiB
Python
#!/usr/bin/env python3
|
|
"""Convert the EGBIM QA dump into staging migration artifacts.
|
|
|
|
This tool deliberately does not execute SQL against a database. It reads the
|
|
legacy dump, optionally copies referenced files into the staging uploads
|
|
volume, and writes:
|
|
|
|
* userfeedback_feedbacks.sql: idempotent ABC feedback inserts
|
|
* support_attachments.sql: attachment inserts resolved through the planned
|
|
EGBIM_QA migration mappings in baron_support
|
|
* uploads_manifest.csv: source/target file and checksum audit data
|
|
|
|
The generated support SQL must run after support_tickets and ticket_comments
|
|
have been migrated. Comment migration must store abc_comment_id as
|
|
EGBIM_QA:<legacy_comment_id> so comment image rows can resolve their target.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import csv
|
|
import hashlib
|
|
import json
|
|
import mimetypes
|
|
import re
|
|
import shutil
|
|
from dataclasses import dataclass
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
|
|
TABLES = {
|
|
"qa_posts": (
|
|
"post_id",
|
|
"login_id",
|
|
"phone",
|
|
"user_id",
|
|
"user_name",
|
|
"category",
|
|
"is_internal",
|
|
"company",
|
|
"family_company",
|
|
"department",
|
|
"position",
|
|
"title",
|
|
"content",
|
|
"attachment",
|
|
"status",
|
|
"is_private",
|
|
"complete_form",
|
|
"created_at",
|
|
"updated_at",
|
|
"is_secret",
|
|
"is_read_admin",
|
|
),
|
|
"qa_attachments": (
|
|
"id",
|
|
"post_id",
|
|
"ori_name",
|
|
"save_path",
|
|
"file_size",
|
|
"uploaded_at",
|
|
),
|
|
"qa_comment_images": (
|
|
"id",
|
|
"comment_id",
|
|
"file_name",
|
|
"file_path",
|
|
"thumb_path",
|
|
"file_size",
|
|
"uploaded_at",
|
|
),
|
|
"qa_comments": (
|
|
"comment_id",
|
|
"post_id",
|
|
"commenter",
|
|
"content",
|
|
"created_at",
|
|
"user_name",
|
|
"updated_at",
|
|
),
|
|
}
|
|
|
|
|
|
@dataclass
|
|
class FileRecord:
|
|
entity_type: str
|
|
source_id: str
|
|
parent_id: str
|
|
original_name: str
|
|
source_value: str
|
|
source_path: Path
|
|
storage_key: str
|
|
uploaded_at: str | None
|
|
file_size: int | None
|
|
checksum: str | None = None
|
|
status: str = "MISSING"
|
|
|
|
|
|
def find_insert_block(text: str, table: str) -> tuple[list[str], str]:
|
|
marker = f"INSERT INTO `{table}`"
|
|
start = text.find(marker)
|
|
if start < 0:
|
|
return [], ""
|
|
columns_start = text.find("(", start)
|
|
columns_end = text.find(")", columns_start)
|
|
values_start = text.find("VALUES", columns_end)
|
|
if columns_start < 0 or columns_end < 0 or values_start < 0:
|
|
return [], ""
|
|
columns = [part.strip().strip("`") for part in text[columns_start + 1 : columns_end].split(",")]
|
|
body_start = values_start + len("VALUES")
|
|
in_quote = False
|
|
escaped = False
|
|
quote = ""
|
|
for index in range(body_start, len(text)):
|
|
char = text[index]
|
|
if in_quote:
|
|
if escaped:
|
|
escaped = False
|
|
elif char == "\\":
|
|
escaped = True
|
|
elif char == quote:
|
|
if index + 1 < len(text) and text[index + 1] == quote:
|
|
continue
|
|
in_quote = False
|
|
elif char in "'\"":
|
|
in_quote = True
|
|
quote = char
|
|
elif char == ";":
|
|
return columns, text[body_start:index]
|
|
raise ValueError(f"INSERT block for {table} has no terminating semicolon")
|
|
|
|
|
|
def parse_rows(body: str) -> list[list[Any]]:
|
|
rows: list[list[Any]] = []
|
|
index = 0
|
|
length = len(body)
|
|
while index < length:
|
|
while index < length and body[index] in " \t\r\n,":
|
|
index += 1
|
|
if index >= length:
|
|
break
|
|
if body[index] != "(":
|
|
raise ValueError(f"Expected row at offset {index}")
|
|
index += 1
|
|
row: list[Any] = []
|
|
while index < length:
|
|
while index < length and body[index] in " \t\r\n":
|
|
index += 1
|
|
if index >= length:
|
|
raise ValueError("Unterminated row")
|
|
if body[index] in "'\"":
|
|
quote = body[index]
|
|
index += 1
|
|
value: list[str] = []
|
|
while index < length:
|
|
char = body[index]
|
|
if char == "\\" and index + 1 < length:
|
|
escapes = {"n": "\n", "r": "\r", "t": "\t", "0": "\0"}
|
|
value.append(escapes.get(body[index + 1], body[index + 1]))
|
|
index += 2
|
|
elif char == quote:
|
|
if index + 1 < length and body[index + 1] == quote:
|
|
value.append(quote)
|
|
index += 2
|
|
else:
|
|
index += 1
|
|
break
|
|
else:
|
|
value.append(char)
|
|
index += 1
|
|
row.append("".join(value))
|
|
else:
|
|
start = index
|
|
while index < length and body[index] not in ",)\r\n":
|
|
index += 1
|
|
token = body[start:index].strip()
|
|
if token.upper() == "NULL":
|
|
row.append(None)
|
|
else:
|
|
row.append(token)
|
|
while index < length and body[index] in " \t\r\n":
|
|
index += 1
|
|
if index >= length:
|
|
raise ValueError("Unterminated row")
|
|
if body[index] == ",":
|
|
index += 1
|
|
continue
|
|
if body[index] == ")":
|
|
index += 1
|
|
rows.append(row)
|
|
break
|
|
raise ValueError(f"Expected comma or closing parenthesis at offset {index}")
|
|
return rows
|
|
|
|
|
|
def load_table(text: str, table: str) -> list[dict[str, Any]]:
|
|
columns, body = find_insert_block(text, table)
|
|
if not body:
|
|
return []
|
|
rows = parse_rows(body)
|
|
expected = TABLES[table]
|
|
if tuple(columns) != expected:
|
|
raise ValueError(f"Unexpected columns for {table}: {columns}")
|
|
return [dict(zip(columns, row, strict=True)) for row in rows]
|
|
|
|
|
|
def sql_string(value: str | None) -> str:
|
|
if value is None:
|
|
return "NULL"
|
|
return "'" + value.replace("\\", "\\\\").replace("'", "''") + "'"
|
|
|
|
|
|
def sql_json(value: dict[str, Any]) -> str:
|
|
encoded = json.dumps(value, ensure_ascii=False, separators=(",", ":"))
|
|
return "CAST(" + sql_string(encoded) + " AS JSON)"
|
|
|
|
|
|
def safe_extension(name: str) -> str:
|
|
extension = Path(name).suffix.lower()
|
|
return extension if re.fullmatch(r"\.[a-z0-9]{1,10}", extension) else ".bin"
|
|
|
|
|
|
def source_relative(path_value: str) -> Path:
|
|
normalized = path_value.replace("\\", "/")
|
|
normalized = re.sub(r"^/egbim/uploads/", "", normalized)
|
|
normalized = re.sub(r"^/uploads/", "", normalized)
|
|
return Path(normalized.lstrip("/"))
|
|
|
|
|
|
def checksum(path: Path) -> str:
|
|
digest = hashlib.sha256()
|
|
with path.open("rb") as stream:
|
|
for chunk in iter(lambda: stream.read(1024 * 1024), b""):
|
|
digest.update(chunk)
|
|
return digest.hexdigest()
|
|
|
|
|
|
def build_file_records(
|
|
attachments: list[dict[str, Any]],
|
|
comment_images: list[dict[str, Any]],
|
|
source_root: Path,
|
|
target_root: Path,
|
|
copy_files: bool,
|
|
) -> list[FileRecord]:
|
|
records: list[FileRecord] = []
|
|
for row in attachments:
|
|
source_rel = source_relative(str(row["save_path"]))
|
|
source_path = source_root / source_rel
|
|
target_key = f"EGBIM_QA/attachments/{row['id']}{safe_extension(str(row['ori_name']))}"
|
|
record = FileRecord(
|
|
entity_type="POST_ATTACHMENT",
|
|
source_id=str(row["id"]),
|
|
parent_id=str(row["post_id"]),
|
|
original_name=str(row["ori_name"]),
|
|
source_value=str(row["save_path"]),
|
|
source_path=source_path,
|
|
storage_key=target_key,
|
|
uploaded_at=row["uploaded_at"],
|
|
file_size=int(row["file_size"]) if row["file_size"] else None,
|
|
)
|
|
records.append(record)
|
|
|
|
for row in comment_images:
|
|
source_rel = source_relative(str(row["file_path"]))
|
|
source_path = source_root / source_rel
|
|
target_key = f"EGBIM_QA/comment-images/{row['id']}{safe_extension(str(row['file_name']))}"
|
|
record = FileRecord(
|
|
entity_type="COMMENT_IMAGE",
|
|
source_id=str(row["id"]),
|
|
parent_id=str(row["comment_id"]),
|
|
original_name=str(row["file_name"]),
|
|
source_value=str(row["file_path"]),
|
|
source_path=source_path,
|
|
storage_key=target_key,
|
|
uploaded_at=row["uploaded_at"],
|
|
file_size=int(row["file_size"]) if row["file_size"] else None,
|
|
)
|
|
records.append(record)
|
|
|
|
thumb_value = row.get("thumb_path")
|
|
if thumb_value:
|
|
thumb_source = source_root / source_relative(str(thumb_value))
|
|
thumb_key = f"EGBIM_QA/comment-thumbnails/{row['id']}{safe_extension(str(row['file_name']))}"
|
|
records.append(
|
|
FileRecord(
|
|
entity_type="COMMENT_THUMBNAIL",
|
|
source_id=str(row["id"]),
|
|
parent_id=str(row["comment_id"]),
|
|
original_name=str(row["file_name"]),
|
|
source_value=str(thumb_value),
|
|
source_path=thumb_source,
|
|
storage_key=thumb_key,
|
|
uploaded_at=row["uploaded_at"],
|
|
file_size=None,
|
|
)
|
|
)
|
|
|
|
for record in records:
|
|
if not record.source_path.is_file():
|
|
continue
|
|
record.checksum = checksum(record.source_path)
|
|
record.file_size = record.source_path.stat().st_size
|
|
record.status = "READY"
|
|
if copy_files:
|
|
destination = target_root / record.storage_key
|
|
destination.parent.mkdir(parents=True, exist_ok=True)
|
|
if destination.exists() and checksum(destination) != record.checksum:
|
|
raise RuntimeError(f"Target exists with different checksum: {destination}")
|
|
if not destination.exists():
|
|
shutil.copy2(record.source_path, destination)
|
|
return records
|
|
|
|
|
|
def write_manifest(records: list[FileRecord], output: Path) -> None:
|
|
with output.open("w", newline="", encoding="utf-8") as stream:
|
|
writer = csv.writer(stream)
|
|
writer.writerow(
|
|
[
|
|
"entity_type",
|
|
"source_id",
|
|
"parent_id",
|
|
"original_name",
|
|
"source_value",
|
|
"source_path",
|
|
"storage_key",
|
|
"uploaded_at",
|
|
"file_size",
|
|
"checksum_sha256",
|
|
"status",
|
|
]
|
|
)
|
|
for record in records:
|
|
writer.writerow(
|
|
[
|
|
record.entity_type,
|
|
record.source_id,
|
|
record.parent_id,
|
|
record.original_name,
|
|
record.source_value,
|
|
record.source_path,
|
|
record.storage_key,
|
|
record.uploaded_at or "",
|
|
record.file_size or "",
|
|
record.checksum or "",
|
|
record.status,
|
|
]
|
|
)
|
|
|
|
|
|
def write_feedback_sql(posts: list[dict[str, Any]], output: Path, channel_id: int) -> None:
|
|
lines = [
|
|
"-- Generated from EGBIM qa_posts. Run against ABC userfeedback only.",
|
|
"-- This file inserts feedback metadata. Comments and local attachments belong to baron_support.",
|
|
"SET NAMES utf8mb4;",
|
|
"USE `userfeedback`;",
|
|
"START TRANSACTION;",
|
|
]
|
|
category_map = {"error": "ERROR_QNA", "improvement": "IMPROVEMENT_QNA", "general": "GENERAL_QNA", "notice": "GENERAL_QNA"}
|
|
for row in posts:
|
|
source_id = str(row["post_id"])
|
|
category = str(row.get("category") or "general")
|
|
data = {
|
|
"title": row.get("title") or f"EGBIM 문의 #{source_id}",
|
|
"contents": row.get("content") or "",
|
|
"category": category_map.get(category, "GENERAL_QNA"),
|
|
"source_system": "EGBIM_QA",
|
|
"source_post_id": source_id,
|
|
"requester": {
|
|
"login_id": row.get("login_id"),
|
|
"legacy_user_id": row.get("user_id"),
|
|
"name": row.get("user_name"),
|
|
"phone": row.get("phone"),
|
|
"company": row.get("company"),
|
|
"department": row.get("department"),
|
|
},
|
|
}
|
|
source_is_secret = bool(int(row.get("is_secret") or 0))
|
|
source_is_private = bool(int(row.get("is_private") or 0))
|
|
additional = {
|
|
"migration_key": f"EGBIM_QA:POST:{source_id}",
|
|
"source_status": row.get("status"),
|
|
"is_internal": bool(int(row.get("is_internal") or 0)),
|
|
"is_secret": source_is_secret or source_is_private,
|
|
"source_is_secret": source_is_secret,
|
|
"source_is_private": source_is_private,
|
|
}
|
|
# The staging ABC schema stores feedback JSON in `data` and does not
|
|
# have the newer `additional_data` column. Keep migration metadata in
|
|
# the same JSON document so the import remains idempotent.
|
|
data.update(additional)
|
|
lines.append(
|
|
"INSERT INTO `feedbacks` (`created_at`,`updated_at`,`data`,`channel_id`) "
|
|
f"SELECT {sql_string(row.get('created_at'))},{sql_string(row.get('updated_at') or row.get('created_at'))},"
|
|
f"{sql_json(data)},{channel_id} FROM DUAL "
|
|
"WHERE NOT EXISTS (SELECT 1 FROM `feedbacks` WHERE `channel_id` = "
|
|
f"{channel_id} AND JSON_UNQUOTE(JSON_EXTRACT(`data`,'$.migration_key')) = {sql_string(additional['migration_key'])});"
|
|
)
|
|
lines.extend(["COMMIT;", ""])
|
|
output.write_text("\n".join(lines), encoding="utf-8")
|
|
|
|
|
|
def write_support_secret_sql(posts: list[dict[str, Any]], output: Path) -> None:
|
|
lines = [
|
|
"-- Generated for baron_support.support_tickets.",
|
|
"-- Applies EGBIM qa_posts.is_secret OR qa_posts.is_private after ticket mappings exist.",
|
|
"SET NAMES utf8mb4;",
|
|
"USE `baron_support`;",
|
|
"START TRANSACTION;",
|
|
]
|
|
for row in posts:
|
|
is_secret = bool(int(row.get("is_secret") or 0))
|
|
is_private = bool(int(row.get("is_private") or 0))
|
|
if not (is_secret or is_private):
|
|
continue
|
|
source_id = str(row["post_id"])
|
|
lines.append(
|
|
"UPDATE support_tickets st JOIN migration_mappings mm ON mm.ticket_id=st.id "
|
|
"SET st.is_secret=1 WHERE mm.source_system='EGBIM_QA' "
|
|
"AND mm.source_entity_type='POST' "
|
|
f"AND mm.source_entity_id={sql_string(source_id)};"
|
|
)
|
|
lines.extend(["COMMIT;", ""])
|
|
output.write_text("\n".join(lines), encoding="utf-8")
|
|
|
|
|
|
|
|
def _support_category(value: Any) -> str:
|
|
category = str(value or "general").strip().lower()
|
|
return {
|
|
"error": "ERROR_QNA",
|
|
"improvement": "IMPROVEMENT_QNA",
|
|
"general": "GENERAL_QNA",
|
|
"notice": "GENERAL_QNA",
|
|
}.get(category, "GENERAL_QNA")
|
|
|
|
|
|
def _support_status(value: Any) -> str:
|
|
status = str(value or "").strip().lower().replace(" ", "_")
|
|
return {
|
|
"new": "RECEIVED",
|
|
"received": "RECEIVED",
|
|
"pending": "PENDING_APPROVAL",
|
|
"pending_approval": "PENDING_APPROVAL",
|
|
"approved": "APPROVED",
|
|
"rejected": "REJECTED",
|
|
"review": "IN_REVIEW",
|
|
"in_review": "IN_REVIEW",
|
|
"detailed_review": "DETAILED_REVIEW",
|
|
"in_progress": "IN_PROGRESS",
|
|
"processing": "IN_PROGRESS",
|
|
"resolved": "RESOLVED",
|
|
"complete": "RESOLVED",
|
|
"completed": "RESOLVED",
|
|
"closed": "CLOSED",
|
|
}.get(status, "RECEIVED")
|
|
|
|
|
|
ADMIN_CANDIDATE_EMAILS = {
|
|
"cjy627@hanmaceng.co.kr",
|
|
"b23072@hanmaceng.co.kr",
|
|
"kjy0426@hanmaceng.co.kr",
|
|
"b21367@hanmaceng.co.kr",
|
|
"rmsgud1202@hanmaceng.co.kr",
|
|
}
|
|
|
|
|
|
def write_support_ticket_comment_sql(
|
|
posts: list[dict[str, Any]],
|
|
comments: list[dict[str, Any]],
|
|
output: Path,
|
|
workspace_id: int,
|
|
) -> None:
|
|
"""Write idempotent support ticket/mapping/comment migration SQL."""
|
|
|
|
lines = [
|
|
"-- Generated from EGBIM qa_posts and qa_comments.",
|
|
"-- Run against baron_support after the support schema migration.",
|
|
"SET NAMES utf8mb4;",
|
|
"USE `baron_support`;",
|
|
f"SET @EGBIM_WORKSPACE_ID = {workspace_id};",
|
|
"INSERT INTO support_category_codes (code,name,sort_order) VALUES "
|
|
"('ERROR_QNA','오류 문의',51),('IMPROVEMENT_QNA','개선 문의',52) "
|
|
"ON DUPLICATE KEY UPDATE name=VALUES(name),sort_order=VALUES(sort_order);",
|
|
"START TRANSACTION;",
|
|
"INSERT INTO migration_batches (batch_name,source_system,status,executed_by,notes) "
|
|
"SELECT 'EGBIM_QA_POSTS_COMMENTS_V1','EGBIM_QA','RUNNING','migration:EGBIM_QA',"
|
|
"'Migrated qa_posts and qa_comments' FROM DUAL WHERE NOT EXISTS ("
|
|
"SELECT 1 FROM migration_batches WHERE batch_name='EGBIM_QA_POSTS_COMMENTS_V1' "
|
|
"AND source_system='EGBIM_QA' AND status='COMPLETED');",
|
|
"SET @EGBIM_BATCH_ID = (SELECT id FROM migration_batches "
|
|
"WHERE batch_name='EGBIM_QA_POSTS_COMMENTS_V1' AND source_system='EGBIM_QA' "
|
|
"ORDER BY id DESC LIMIT 1);",
|
|
]
|
|
|
|
for row in posts:
|
|
source_id = str(row["post_id"])
|
|
title = str(row.get("title") or f"EGBIM 문의 #{source_id}")
|
|
description = str(row.get("content") or "")
|
|
requester_id = str(row.get("login_id") or row.get("user_id") or f"legacy:{source_id}")
|
|
requester_email = str(row.get("login_id") or "") or None
|
|
source_is_secret = bool(int(row.get("is_secret") or 0))
|
|
source_is_private = bool(int(row.get("is_private") or 0))
|
|
migration_key = f"EGBIM_QA:POST:{source_id}"
|
|
extra_fields = {
|
|
"migration_key": migration_key,
|
|
"source_system": "EGBIM_QA",
|
|
"source_post_id": source_id,
|
|
"source_status": row.get("status"),
|
|
}
|
|
lines.append(
|
|
"INSERT INTO support_tickets (workspace_id,requester_id,requester_tenant_id,"
|
|
"requester_contact,requester_email,requester_name,requester_department,"
|
|
"requester_phone_number,ticket_type,source_system,title,description,category_code,"
|
|
"status_code,approval_status,sync_status,issue_link_status,is_secret,"
|
|
"requires_approval,priority,extra_fields,created_at,updated_at) "
|
|
f"SELECT @EGBIM_WORKSPACE_ID,{sql_string(requester_id)},'EGBIM_QA',"
|
|
f"{sql_string(row.get('phone'))},{sql_string(requester_email)},"
|
|
f"{sql_string(row.get('user_name'))},{sql_string(row.get('department'))},"
|
|
f"{sql_string(row.get('phone'))},'GENERAL','EGBIM_QA',{sql_string(title)},"
|
|
f"{sql_string(description)},{sql_string(_support_category(row.get('category')))},"
|
|
f"{sql_string(_support_status(row.get('status')))},'NOT_REQUIRED','SYNCED',"
|
|
f"'NOT_LINKED',{1 if (source_is_secret or source_is_private) else 0},0,'NORMAL',"
|
|
f"{sql_json(extra_fields)},{sql_string(row.get('created_at'))},"
|
|
f"{sql_string(row.get('updated_at') or row.get('created_at'))} FROM DUAL WHERE NOT EXISTS ("
|
|
"SELECT 1 FROM support_tickets WHERE workspace_id=@EGBIM_WORKSPACE_ID "
|
|
f"AND JSON_UNQUOTE(JSON_EXTRACT(extra_fields,'$.migration_key'))={sql_string(migration_key)});"
|
|
)
|
|
lines.append(
|
|
"INSERT INTO migration_mappings (batch_id,source_system,source_entity_type,"
|
|
"source_entity_id,source_parent_id,workspace_id,ticket_id,abc_feedback_id,"
|
|
"migration_status,error_message) "
|
|
"SELECT @EGBIM_BATCH_ID,'EGBIM_QA','POST',"
|
|
f"{sql_string(source_id)},NULL,@EGBIM_WORKSPACE_ID,st.id,NULL,'MIGRATED',NULL "
|
|
"FROM support_tickets st WHERE st.workspace_id=@EGBIM_WORKSPACE_ID "
|
|
f"AND JSON_UNQUOTE(JSON_EXTRACT(st.extra_fields,'$.migration_key'))={sql_string(migration_key)} "
|
|
"AND NOT EXISTS (SELECT 1 FROM migration_mappings mm WHERE mm.source_system='EGBIM_QA' "
|
|
"AND mm.source_entity_type='POST' "
|
|
f"AND mm.source_entity_id={sql_string(source_id)});"
|
|
)
|
|
|
|
lines.append(
|
|
"UPDATE migration_mappings mm JOIN support_tickets st ON st.workspace_id=mm.workspace_id "
|
|
"AND JSON_UNQUOTE(JSON_EXTRACT(st.extra_fields,'$.migration_key'))="
|
|
"CONCAT('EGBIM_QA:POST:',mm.source_entity_id) SET mm.ticket_id=st.id,"
|
|
"mm.migration_status='MIGRATED' WHERE mm.source_system='EGBIM_QA' "
|
|
"AND mm.source_entity_type='POST' AND mm.workspace_id=@EGBIM_WORKSPACE_ID "
|
|
"AND mm.ticket_id IS NULL;"
|
|
)
|
|
|
|
for row in comments:
|
|
source_id = str(row["comment_id"])
|
|
post_id = str(row["post_id"])
|
|
commenter = str(row.get("commenter") or f"legacy-commenter:{source_id}")
|
|
is_internal = commenter.strip().lower() in ADMIN_CANDIDATE_EMAILS
|
|
comment_key = f"EGBIM_QA:{source_id}"
|
|
lines.append(
|
|
"INSERT INTO ticket_comments (ticket_id,parent_comment_id,author_id,"
|
|
"author_tenant_id,author_name,comment_type,content,is_internal,abc_comment_id,"
|
|
"sync_status,edited_at,created_at,updated_at) SELECT mm.ticket_id,NULL,"
|
|
f"{sql_string(commenter)},'EGBIM_QA',{sql_string(row.get('user_name') or commenter)},"
|
|
f"{sql_string('ADMIN' if is_internal else 'COMMENT')},{sql_string(row.get('content') or '')},"
|
|
f"{1 if is_internal else 0},{sql_string(comment_key)},'SYNCED',"
|
|
f"{sql_string(row.get('updated_at'))},{sql_string(row.get('created_at'))},"
|
|
f"{sql_string(row.get('updated_at') or row.get('created_at'))} FROM migration_mappings mm "
|
|
"WHERE mm.source_system='EGBIM_QA' AND mm.source_entity_type='POST' "
|
|
f"AND mm.source_entity_id={sql_string(post_id)} AND mm.ticket_id IS NOT NULL "
|
|
"AND NOT EXISTS (SELECT 1 FROM ticket_comments WHERE abc_comment_id="
|
|
f"{sql_string(comment_key)});"
|
|
)
|
|
|
|
lines.extend([
|
|
"UPDATE migration_batches SET status='COMPLETED',completed_at=CURRENT_TIMESTAMP "
|
|
"WHERE id=@EGBIM_BATCH_ID;",
|
|
"COMMIT;",
|
|
"",
|
|
])
|
|
output.write_text("\n".join(lines), encoding="utf-8")
|
|
|
|
def write_attachment_sql(records: list[FileRecord], output: Path) -> None:
|
|
lines = [
|
|
"-- Generated for baron_support.attachments.",
|
|
"-- Run after EGBIM_QA POST/COMMENT migration mappings exist.",
|
|
"-- Comment migration must set ticket_comments.abc_comment_id = EGBIM_QA:<source comment id>.",
|
|
"SET NAMES utf8mb4;",
|
|
"USE `baron_support`;",
|
|
"SET @EGBIM_WORKSPACE_ID = (SELECT id FROM workspaces WHERE workspace_code='EGBIM');",
|
|
"SET @EGBIM_UPLOADER_ID = 'migration:EGBIM_QA';",
|
|
"SET @EGBIM_UPLOADER_TENANT_ID = 'migration:EGBIM_QA';",
|
|
"START TRANSACTION;",
|
|
]
|
|
for record in records:
|
|
if record.status != "READY" or record.entity_type == "COMMENT_THUMBNAIL":
|
|
continue
|
|
mime_type = mimetypes.guess_type(record.original_name)[0] or "application/octet-stream"
|
|
extension = safe_extension(record.original_name).lstrip(".") or None
|
|
attachment_key = f"EGBIM_QA:{record.entity_type}:{record.source_id}"
|
|
if record.entity_type == "POST_ATTACHMENT":
|
|
resolver = (
|
|
"SELECT mm.ticket_id AS ticket_id, NULL AS comment_id, @EGBIM_WORKSPACE_ID AS workspace_id FROM migration_mappings mm "
|
|
"WHERE mm.source_system='EGBIM_QA' AND mm.source_entity_type='POST' "
|
|
f"AND mm.source_entity_id={sql_string(record.parent_id)} AND mm.ticket_id IS NOT NULL LIMIT 1"
|
|
)
|
|
else:
|
|
resolver = (
|
|
"SELECT tc.ticket_id AS ticket_id, tc.id AS comment_id, @EGBIM_WORKSPACE_ID AS workspace_id FROM ticket_comments tc "
|
|
f"WHERE tc.abc_comment_id={sql_string('EGBIM_QA:' + record.parent_id)} LIMIT 1"
|
|
)
|
|
lines.append(
|
|
"INSERT INTO attachments (ticket_id,comment_id,workspace_id,uploader_id,uploader_tenant_id,"
|
|
"original_file_name,stored_file_name,storage_provider,storage_bucket,storage_key,mime_type,"
|
|
"file_extension,file_size,checksum_sha256,abc_attachment_id,attachment_status,created_at,updated_at) "
|
|
"SELECT resolved.ticket_id,resolved.comment_id,resolved.workspace_id,@EGBIM_UPLOADER_ID,"
|
|
f"@EGBIM_UPLOADER_TENANT_ID,{sql_string(record.original_name)},"
|
|
f"{sql_string(Path(record.storage_key).name)},'LOCAL','/app/uploads',{sql_string(record.storage_key)},"
|
|
f"{sql_string(mime_type)},{sql_string(extension)},{record.file_size or 0},{sql_string(record.checksum)},"
|
|
f"{sql_string(attachment_key)},'ACTIVE',{sql_string(record.uploaded_at)},{sql_string(record.uploaded_at)} "
|
|
f"FROM ({resolver}) resolved WHERE NOT EXISTS (SELECT 1 FROM attachments WHERE abc_attachment_id={sql_string(attachment_key)});"
|
|
)
|
|
lines.extend(["COMMIT;", ""])
|
|
output.write_text("\n".join(lines), encoding="utf-8")
|
|
|
|
|
|
def main() -> None:
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument("--source-sql", type=Path, default=Path("scripts/egbim_qa.sql"))
|
|
parser.add_argument("--source-uploads", type=Path, required=True)
|
|
parser.add_argument("--target-uploads", type=Path, default=Path("uploads"))
|
|
parser.add_argument("--output-dir", type=Path, default=Path("scripts/generated/egbim_migration"))
|
|
# EGBIM workspace is mapped to ABC userfeedback channel_id=2.
|
|
parser.add_argument("--channel-id", type=int, default=2)
|
|
parser.add_argument("--workspace-id", type=int, default=6)
|
|
parser.add_argument("--copy-files", action="store_true")
|
|
args = parser.parse_args()
|
|
|
|
text = args.source_sql.read_text(encoding="utf-8")
|
|
posts = load_table(text, "qa_posts")
|
|
attachments = load_table(text, "qa_attachments")
|
|
comment_images = load_table(text, "qa_comment_images")
|
|
comments = load_table(text, "qa_comments")
|
|
args.output_dir.mkdir(parents=True, exist_ok=True)
|
|
records = build_file_records(attachments, comment_images, args.source_uploads, args.target_uploads, args.copy_files)
|
|
write_manifest(records, args.output_dir / "uploads_manifest.csv")
|
|
write_feedback_sql(posts, args.output_dir / "userfeedback_feedbacks.sql", args.channel_id)
|
|
write_support_ticket_comment_sql(
|
|
posts, comments, args.output_dir / "support_tickets_comments.sql", args.workspace_id
|
|
)
|
|
write_support_secret_sql(posts, args.output_dir / "support_ticket_secrets.sql")
|
|
write_attachment_sql(records, args.output_dir / "support_attachments.sql")
|
|
ready = sum(record.status == "READY" for record in records)
|
|
missing = sum(record.status == "MISSING" for record in records)
|
|
print(f"posts={len(posts)} comments={len(comments)} post_attachments={len(attachments)} comment_images={len(comment_images)}")
|
|
print(f"file_records={len(records)} ready={ready} missing={missing}")
|
|
print(f"output={args.output_dir}")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|