AiDOPWarehouse/doc/plan/sql/uat-data/run_demo_s8_migration.py

#!/usr/bin/env python3
"""Curated, credential-free S8 migration into UATDEMO."""

from __future__ import annotations

import json
import re
from datetime import datetime
from pathlib import Path

import pymysql
from pymysql.cursors import DictCursor


ROOT = Path(__file__).resolve().parents[4]
DATABASE_JSON = ROOT / "server" / "Admin.NET.Application" / "Configuration" / "Database.json"
SOURCE_TENANT = 797403760988230
SOURCE_FACTORY = 797403760988231
TARGET_TENANT = 838257237606469
TARGET_FACTORY = 838257237676101


def connect() -> pymysql.Connection:
    raw = DATABASE_JSON.read_text(encoding="utf-8-sig")
    connection_string = next(
        value
        for value in re.findall(
            r'(?m)^\s*"ConnectionString"\s*:\s*"([^"]+)"', raw
        )
        if "Database=aidopdev" in value
    )
    parts = {
        item.split("=", 1)[0].strip().lower(): item.split("=", 1)[1].strip()
        for item in connection_string.split(";")
        if "=" in item
    }
    return pymysql.connect(
        host=parts["server"],
        port=int(parts["port"]),
        user=parts["uid"],
        password=parts["pwd"],
        database=parts["database"],
        charset="utf8mb4",
        autocommit=False,
        cursorclass=DictCursor,
    )


def columns(cur: DictCursor, table: str) -> list[str]:
    cur.execute(f"SHOW COLUMNS FROM `{table}`")
    return [
        row["Field"]
        for row in cur.fetchall()
        if "auto_increment" not in (row["Extra"] or "")
    ]


def insert(cur: DictCursor, table: str, row: dict[str, object]) -> int:
    writable = columns(cur, table)
    values = {key: row.get(key) for key in writable}
    cur.execute(
        f"INSERT INTO `{table}` ({','.join(f'`{key}`' for key in writable)}) "
        f"VALUES ({','.join(['%s'] * len(writable))})",
        tuple(values[key] for key in writable),
    )
    return int(cur.lastrowid)


def source_rows(cur: DictCursor, table: str) -> list[dict[str, object]]:
    cur.execute(
        f"""
        SELECT * FROM `{table}`
        WHERE tenant_id=%s AND factory_id=%s
        ORDER BY id
        """,
        (SOURCE_TENANT, SOURCE_FACTORY),
    )
    return list(cur.fetchall())


def existing_id(
    cur: DictCursor, table: str, key: str, value: object
) -> int | None:
    cur.execute(
        f"""
        SELECT id FROM `{table}`
        WHERE tenant_id=%s AND factory_id=%s AND `{key}`=%s
        ORDER BY id LIMIT 1
        """,
        (TARGET_TENANT, TARGET_FACTORY, value),
    )
    row = cur.fetchone()
    return int(row["id"]) if row else None


def target_scope(row: dict[str, object]) -> dict[str, object]:
    result = dict(row)
    result["tenant_id"] = TARGET_TENANT
    result["factory_id"] = TARGET_FACTORY
    return result


def ensure_unassigned_department(cur: DictCursor) -> int:
    cur.execute(
        """
        SELECT RecID FROM DepartmentMaster
        WHERE tenant_id=%s AND factory_ref_id=%s AND Department='未分配'
        ORDER BY RecID LIMIT 1
        """,
        (TARGET_TENANT, TARGET_FACTORY),
    )
    row = cur.fetchone()
    if row:
        return int(row["RecID"])
    cur.execute(
        """
        INSERT INTO DepartmentMaster
        (company_ref_id,factory_ref_id,tenant_id,Domain,Department,Descr,
         IsActive,CreateTime,CreateUser)
        VALUES(%s,%s,%s,'UATDEMO','未分配','S8 migration fallback',1,NOW(),
               'uat-s8-migration')
        """,
        (TARGET_FACTORY, TARGET_FACTORY, TARGET_TENANT),
    )
    return int(cur.lastrowid)


def copy_simple(
    cur: DictCursor,
    table: str,
    natural_key: str,
    transform,
) -> tuple[int, dict[int, int]]:
    inserted = 0
    id_map: dict[int, int] = {}
    for source in source_rows(cur, table):
        old_id = int(source["id"])
        existing = existing_id(cur, table, natural_key, source[natural_key])
        if existing:
            id_map[old_id] = existing
            continue
        row = transform(target_scope(source))
        row.pop("id", None)
        new_id = insert(cur, table, row)
        id_map[old_id] = new_id
        inserted += 1
    return inserted, id_map


def main() -> None:
    conn = connect()
    summary: dict[str, int] = {}
    try:
        with conn.cursor() as cur:
            department_id = ensure_unassigned_department(cur)

            summary["scenes_inserted"], _ = copy_simple(
                cur,
                "ado_s8_scene_config",
                "scene_code",
                lambda row: row,
            )

            def sanitize_source(row: dict[str, object]) -> dict[str, object]:
                row.update(
                    endpoint=None,
                    auth_type="NONE",
                    enabled=0,
                    last_check_at=None,
                    last_check_status=None,
                )
                return row

            summary["data_sources_inserted"], data_source_map = copy_simple(
                cur,
                "ado_s8_data_source",
                "data_source_code",
                sanitize_source,
            )

            def sanitize_rule(row: dict[str, object]) -> dict[str, object]:
                old_ds = row.get("data_source_id")
                row["data_source_id"] = (
                    data_source_map.get(int(old_ds)) if old_ds else None
                )
                row.update(
                    enabled=0,
                    next_run_at=None,
                    last_run_at=None,
                    last_status=None,
                    last_error=None,
                    last_duration_ms=None,
                    last_run_id=None,
                    lock_token=None,
                    locked_by=None,
                    lock_until=None,
                    running_started_at=None,
                    consecutive_failure_count=0,
                    paused_until=None,
                    pause_reason=None,
                )
                return row

            summary["rules_inserted"], rule_map = copy_simple(
                cur,
                "ado_s8_watch_rule",
                "rule_code",
                sanitize_rule,
            )

            cur.execute(
                """
                SELECT type_code FROM ado_s8_exception_type
                WHERE (tenant_id=0 OR tenant_id=%s)
                  AND (factory_id=0 OR factory_id=%s)
                """,
                (TARGET_TENANT, TARGET_FACTORY),
            )
            valid_types = {str(row["type_code"]) for row in cur.fetchall()}
            exception_map: dict[int, int] = {}
            exceptions_inserted = 0
            for source in source_rows(cur, "ado_s8_exception"):
                old_id = int(source["id"])
                type_code = str(source.get("exception_type_code") or "")
                if old_id in {329, 331, 332} or type_code not in valid_types:
                    continue
                existing = existing_id(
                    cur, "ado_s8_exception", "exception_code", source["exception_code"]
                )
                if existing:
                    exception_map[old_id] = existing
                    continue
                row = target_scope(source)
                row.pop("id", None)
                old_ds = row.get("source_data_source_id")
                old_rule = row.get("source_rule_id")
                row["source_data_source_id"] = (
                    data_source_map.get(int(old_ds)) if old_ds else None
                )
                row["source_rule_id"] = (
                    rule_map.get(int(old_rule)) if old_rule else None
                )
                row["occurrence_dept_id"] = department_id
                row["responsible_dept_id"] = department_id
                for field in (
                    "responsible_group_id",
                    "assignee_id",
                    "reporter_id",
                    "created_by",
                    "updated_by",
                    "verifier_id",
                    "active_flow_instance_id",
                    "active_flow_biz_type",
                ):
                    row[field] = None
                new_id = insert(cur, "ado_s8_exception", row)
                exception_map[old_id] = new_id
                exceptions_inserted += 1
            summary["exceptions_inserted"] = exceptions_inserted

            timeline_inserted = 0
            cur.execute(
                """
                SELECT t.* FROM ado_s8_exception_timeline t
                JOIN ado_s8_exception e ON e.id=t.exception_id
                WHERE e.tenant_id=%s AND e.factory_id=%s
                ORDER BY t.id
                """,
                (SOURCE_TENANT, SOURCE_FACTORY),
            )
            for source in cur.fetchall():
                old_exception = int(source["exception_id"])
                if old_exception not in exception_map:
                    continue
                target_exception = exception_map[old_exception]
                cur.execute(
                    """
                    SELECT 1 FROM ado_s8_exception_timeline
                    WHERE exception_id=%s AND action_code=%s AND created_at=%s
                    LIMIT 1
                    """,
                    (
                        target_exception,
                        source["action_code"],
                        source["created_at"],
                    ),
                )
                if cur.fetchone():
                    continue
                row = dict(source)
                row.pop("id", None)
                row["exception_id"] = target_exception
                row["operator_id"] = None
                row["operator_name"] = "Demo演示"
                insert(cur, "ado_s8_exception_timeline", row)
                timeline_inserted += 1
            summary["timelines_inserted"] = timeline_inserted

            detection_inserted = 0
            for source in source_rows(cur, "ado_s8_detection_log"):
                old_rule = source.get("rule_id")
                if not old_rule or int(old_rule) not in rule_map:
                    continue
                old_exception = source.get("exception_id")
                row = target_scope(source)
                row.pop("id", None)
                row["rule_id"] = rule_map[int(old_rule)]
                row["exception_id"] = (
                    exception_map.get(int(old_exception)) if old_exception else None
                )
                cur.execute(
                    """
                    SELECT 1 FROM ado_s8_detection_log
                    WHERE tenant_id=%s AND factory_id=%s
                      AND rule_id=%s AND detected_at=%s
                      AND COALESCE(run_id,'')=COALESCE(%s,'')
                      AND COALESCE(source_object_id,'')=COALESCE(%s,'')
                    LIMIT 1
                    """,
                    (
                        TARGET_TENANT,
                        TARGET_FACTORY,
                        row["rule_id"],
                        row["detected_at"],
                        row.get("run_id"),
                        row.get("source_object_id"),
                    ),
                )
                if cur.fetchone():
                    continue
                insert(cur, "ado_s8_detection_log", row)
                detection_inserted += 1
            summary["detections_inserted"] = detection_inserted

            states_inserted = 0
            for source in source_rows(cur, "ado_s8_rule_detection_state"):
                cur.execute(
                    """
                    SELECT id FROM ado_s8_rule_detection_state
                    WHERE tenant_id=%s AND factory_id=%s
                      AND rule_code=%s AND dedup_key=%s
                    LIMIT 1
                    """,
                    (
                        TARGET_TENANT,
                        TARGET_FACTORY,
                        source["rule_code"],
                        source["dedup_key"],
                    ),
                )
                if cur.fetchone():
                    continue
                row = target_scope(source)
                row.pop("id", None)
                active = row.get("active_exception_id")
                row["active_exception_id"] = (
                    exception_map.get(int(active)) if active else None
                )
                insert(cur, "ado_s8_rule_detection_state", row)
                states_inserted += 1
            summary["states_inserted"] = states_inserted

            conn.commit()

            checks = {
                "scenes": "ado_s8_scene_config",
                "data_sources": "ado_s8_data_source",
                "rules": "ado_s8_watch_rule",
                "exceptions": "ado_s8_exception",
                "detections": "ado_s8_detection_log",
                "states": "ado_s8_rule_detection_state",
                "notifications": "ado_s8_notification_log",
            }
            for label, table in checks.items():
                cur.execute(
                    f"SELECT COUNT(*) count FROM `{table}` "
                    "WHERE tenant_id=%s AND factory_id=%s",
                    (TARGET_TENANT, TARGET_FACTORY),
                )
                summary[f"{label}_total"] = int(cur.fetchone()["count"])
            cur.execute(
                """
                SELECT COUNT(*) count FROM ado_s8_exception_timeline t
                JOIN ado_s8_exception e ON e.id=t.exception_id
                WHERE e.tenant_id=%s AND e.factory_id=%s
                """,
                (TARGET_TENANT, TARGET_FACTORY),
            )
            summary["timelines_total"] = int(cur.fetchone()["count"])
    except Exception:
        conn.rollback()
        raise
    finally:
        conn.close()

    evidence = Path(__file__).with_name("WP4-DEMO-S8-execution-evidence.json")
    evidence.write_text(
        json.dumps(
            {
                "executed_at": datetime.now().isoformat(timespec="seconds"),
                "summary": summary,
            },
            ensure_ascii=False,
            indent=2,
        ),
        encoding="utf-8",
    )
    print(json.dumps(summary, ensure_ascii=False, indent=2))


if __name__ == "__main__":
    main()