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

#!/usr/bin/env python3
"""Idempotent migration of action_items_json into ado_smart_ops_improvement_action."""

from __future__ import annotations

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

import pymysql
from pymysql.constants import CLIENT


ROOT = Path(__file__).resolve().parents[4]
CONFIG = ROOT / "server" / "Admin.NET.Application" / "Configuration" / "Database.json"
EVIDENCE = ROOT / "doc" / "plan" / "UAT留证" / "2026-08-17-智慧诊断正式落地"
INVALID_TENANTS = {0, 1, 1300000000001}


def connect() -> pymysql.Connection:
    raw = CONFIG.read_text(encoding="utf-8-sig")
    value = next(
        item
        for item in re.findall(r'(?m)^\s*"ConnectionString"\s*:\s*"([^"]+)"', raw)
        if "Database=aidopdev" in item
    )
    parts = {
        item.split("=", 1)[0].strip().lower(): item.split("=", 1)[1].strip()
        for item in value.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=True,
        client_flag=CLIENT.MULTI_STATEMENTS,
        cursorclass=pymysql.cursors.DictCursor,
    )


def normalize_status(raw: str | None) -> str:
    value = (raw or "").strip().lower()
    if value in {"completed", "done"}:
        return "completed"
    if value in {"doing", "in_progress"}:
        return "in_progress"
    if value in {"cancelled", "canceled"}:
        return "cancelled"
    return "pending"


def parse_owner_user_id(owner: str | None) -> int | None:
    if owner is None:
        return None
    text = str(owner).strip()
    return int(text) if text.isdigit() else None


def parse_date(raw: str | None):
    if not raw:
        return None
    text = str(raw).strip()
    for fmt in ("%Y-%m-%d", "%Y/%m/%d", "%Y-%m-%dT%H:%M:%S"):
        try:
            return datetime.strptime(text[:19] if "T" in text else text[:10], fmt if "T" not in text else "%Y-%m-%dT%H:%M:%S")
        except ValueError:
            continue
    return None


def parse_drafts(raw: str | None) -> list[dict]:
    if not raw or raw.strip() in {"", "[]", "null"}:
        return []
    try:
        data = json.loads(raw)
    except json.JSONDecodeError:
        return []
    if not isinstance(data, list):
        return []
    items = []
    for item in data:
        if not isinstance(item, dict):
            continue
        content = (item.get("content") or item.get("Content") or "").strip()
        if not content:
            continue
        items.append(
            {
                "content": content,
                "owner": item.get("owner") or item.get("Owner"),
                "dueDate": item.get("dueDate") or item.get("DueDate"),
                "status": item.get("status") or item.get("Status"),
                "completedAt": item.get("completedAt") or item.get("CompletedAt"),
            }
        )
    return items


def main() -> None:
    conn = connect()
    report = {"migratedPlans": [], "skipped": [], "failed": []}
    try:
        with conn.cursor() as cur:
            cur.execute(
                """
                SELECT Id, TenantId, FactoryId, PlanNo, OwnerUserId, ActionItemsJson
                FROM ado_smart_ops_improvement_plan
                WHERE ActionItemsJson IS NOT NULL
                  AND ActionItemsJson NOT IN ('', '[]', 'null')
                """
            )
            plans = cur.fetchall()
            for plan in plans:
                tenant_id = int(plan["TenantId"] or 0)
                plan_id = int(plan["Id"])
                key = {"planId": str(plan_id), "tenantId": str(tenant_id), "planNo": plan["PlanNo"]}
                if tenant_id in INVALID_TENANTS:
                    report["skipped"].append({**key, "reason": "invalid_tenant"})
                    continue
                factory_id = int(plan["FactoryId"] or 0)
                if factory_id <= 0 or factory_id == tenant_id:
                    report["skipped"].append({**key, "reason": "invalid_factory"})
                    continue
                cur.execute(
                    "SELECT COUNT(*) AS n FROM ado_smart_ops_improvement_action WHERE tenant_id=%s AND plan_id=%s",
                    (tenant_id, plan_id),
                )
                if cur.fetchone()["n"] > 0:
                    report["skipped"].append({**key, "reason": "already_migrated"})
                    continue
                drafts = parse_drafts(plan["ActionItemsJson"])
                if not drafts:
                    report["skipped"].append({**key, "reason": "empty_or_unparseable"})
                    continue
                now = datetime.now()
                inserted = 0
                for idx, draft in enumerate(drafts, start=1):
                    cur.execute(
                        """
                        INSERT INTO ado_smart_ops_improvement_action
                          (tenant_id, factory_id, plan_id, sequence_no, content, owner_user_id, owner_name,
                           due_date, status, completed_at, version, create_user_id, update_user_id, create_time, update_time)
                        VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,1,%s,%s,%s,%s)
                        """,
                        (
                            tenant_id,
                            factory_id,
                            plan_id,
                            idx,
                            draft["content"][:500],
                            parse_owner_user_id(draft["owner"]),
                            (str(draft["owner"])[:100] if draft["owner"] else None),
                            parse_date(draft["dueDate"]),
                            normalize_status(draft["status"]),
                            parse_date(draft["completedAt"]),
                            plan["OwnerUserId"],
                            plan["OwnerUserId"],
                            now,
                            now,
                        ),
                    )
                    inserted += 1
                report["migratedPlans"].append({**key, "actions": inserted})
    finally:
        conn.close()

    EVIDENCE.mkdir(parents=True, exist_ok=True)
    out = EVIDENCE / "03-action-json-migration.json"
    out.write_text(json.dumps(report, ensure_ascii=False, indent=2), encoding="utf-8")
    print(json.dumps(report, ensure_ascii=False, indent=2))


if __name__ == "__main__":
    main()