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

#!/usr/bin/env python3
"""WP-SD8 audit reset for Smart Diagnosis improvement-loop test batches.

Default is dry-run. Only specified UAT tenants and RootCause batch markers
are touched. KPI / DWD business facts are never deleted.
"""

from __future__ import annotations

import argparse
import json
import re
from datetime import date, datetime
from decimal import Decimal
from pathlib import Path

import pymysql


ROOT = Path(__file__).resolve().parents[4]
CONFIG = ROOT / "server" / "Admin.NET.Application" / "Configuration" / "Database.json"
EVIDENCE = ROOT / "doc" / "plan" / "UAT留证" / "2026-08-17-智慧诊断正式落地"
DEFAULT_TENANTS = (838257186181189, 838257212780613, 838257237606469)
BATCH_PREFIX = "[WP-SD7/"


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",
        cursorclass=pymysql.cursors.DictCursor,
        autocommit=False,
    )


def json_safe(value: object) -> object:
    if isinstance(value, datetime):
        return value.isoformat(timespec="seconds")
    if isinstance(value, date):
        return value.isoformat()
    if isinstance(value, Decimal):
        return str(value)
    if isinstance(value, dict):
        return {str(key): json_safe(item) for key, item in value.items()}
    if isinstance(value, (list, tuple)):
        return [json_safe(item) for item in value]
    if isinstance(value, int) and abs(value) > 2**53:
        return str(value)
    return value


def fetch_all(cur: pymysql.cursors.DictCursor, sql: str, params: tuple = ()) -> list[dict]:
    cur.execute(sql, params)
    return [json_safe(row) for row in cur.fetchall()]


def table_exists(cur: pymysql.cursors.DictCursor, name: str) -> bool:
    cur.execute(
        """
        SELECT COUNT(*) c FROM information_schema.tables
        WHERE table_schema=DATABASE() AND table_name=%s
        """,
        (name,),
    )
    return int((cur.fetchone() or {}).get("c") or 0) > 0


def main() -> None:
    parser = argparse.ArgumentParser()
    parser.add_argument("--batch", default="WP-SD7", help="Batch marker inside RootCause")
    parser.add_argument("--apply", action="store_true", help="Actually delete; default is dry-run")
    parser.add_argument(
        "--tenants",
        default=",".join(str(x) for x in DEFAULT_TENANTS),
        help="Comma-separated tenant IDs",
    )
    args = parser.parse_args()
    tenant_ids = tuple(int(x.strip()) for x in args.tenants.split(",") if x.strip())
    marker = f"[{args.batch}/"
    placeholders = ",".join(["%s"] * len(tenant_ids))

    EVIDENCE.mkdir(parents=True, exist_ok=True)
    conn = connect()
    try:
        cur = conn.cursor()
        plans = fetch_all(
            cur,
            f"""
            SELECT * FROM ado_smart_ops_improvement_plan
            WHERE TenantId IN ({placeholders})
              AND RootCause LIKE %s
            """,
            (*tenant_ids, f"%{marker}%"),
        )
        plan_ids = [int(row["Id"]) for row in plans]
        backup = {
            "work_package": "WP-SD8",
            "dry_run": not args.apply,
            "batch": args.batch,
            "generated_at": datetime.now().isoformat(timespec="seconds"),
            "tenant_ids": [str(x) for x in tenant_ids],
            "plan_count": len(plans),
            "plans": plans,
            "actions": [],
            "action_logs": [],
            "action_notices": [],
            "verify_logs": [],
            "flow_instances": [],
        }
        if plan_ids:
            id_ph = ",".join(["%s"] * len(plan_ids))
            if table_exists(cur, "ado_smart_ops_improvement_action"):
                backup["actions"] = fetch_all(
                    cur,
                    f"SELECT * FROM ado_smart_ops_improvement_action WHERE tenant_id IN ({placeholders}) AND plan_id IN ({id_ph})",
                    (*tenant_ids, *plan_ids),
                )
                action_ids = [int(row["id"]) for row in backup["actions"]]
                if action_ids and table_exists(cur, "ado_smart_ops_improvement_action_log"):
                    a_ph = ",".join(["%s"] * len(action_ids))
                    backup["action_logs"] = fetch_all(
                        cur,
                        f"SELECT * FROM ado_smart_ops_improvement_action_log WHERE tenant_id IN ({placeholders}) AND action_id IN ({a_ph})",
                        (*tenant_ids, *action_ids),
                    )
                if action_ids and table_exists(cur, "ado_smart_ops_improvement_action_notice"):
                    a_ph = ",".join(["%s"] * len(action_ids))
                    backup["action_notices"] = fetch_all(
                        cur,
                        f"SELECT * FROM ado_smart_ops_improvement_action_notice WHERE tenant_id IN ({placeholders}) AND action_id IN ({a_ph})",
                        (*tenant_ids, *action_ids),
                    )
            if table_exists(cur, "ado_smart_ops_improvement_verify_log"):
                backup["verify_logs"] = fetch_all(
                    cur,
                    f"SELECT * FROM ado_smart_ops_improvement_verify_log WHERE tenant_id IN ({placeholders}) AND plan_id IN ({id_ph})",
                    (*tenant_ids, *plan_ids),
                )
            flow_ids = [int(row["FlowInstanceId"]) for row in plans if row.get("FlowInstanceId")]
            for table in ("af_flow_instance", "ado_flow_instance", "ApprovalFlowInstance"):
                if flow_ids and table_exists(cur, table):
                    f_ph = ",".join(["%s"] * len(flow_ids))
                    backup["flow_instances"] = fetch_all(
                        cur,
                        f"SELECT * FROM {table} WHERE Id IN ({f_ph})",
                        tuple(flow_ids),
                    )
                    backup["flow_table"] = table
                    break

        stamp = datetime.now().strftime("%Y%m%d-%H%M%S")
        backup_path = EVIDENCE / f"08-reset-backup-{args.batch}-{stamp}.json"
        backup_path.write_text(json.dumps(json_safe(backup), ensure_ascii=False, indent=2), encoding="utf-8")

        deleted = {
            "notices": 0,
            "action_logs": 0,
            "actions": 0,
            "verify_logs": 0,
            "plans": 0,
        }
        if args.apply and plan_ids:
            id_ph = ",".join(["%s"] * len(plan_ids))
            action_ids = [int(x["id"]) for x in backup["actions"]]
            if action_ids and table_exists(cur, "ado_smart_ops_improvement_action_notice"):
                a_ph = ",".join(["%s"] * len(action_ids))
                cur.execute(
                    f"DELETE FROM ado_smart_ops_improvement_action_notice WHERE tenant_id IN ({placeholders}) AND action_id IN ({a_ph})",
                    (*tenant_ids, *action_ids),
                )
                deleted["notices"] = cur.rowcount
            if action_ids and table_exists(cur, "ado_smart_ops_improvement_action_log"):
                a_ph = ",".join(["%s"] * len(action_ids))
                cur.execute(
                    f"DELETE FROM ado_smart_ops_improvement_action_log WHERE tenant_id IN ({placeholders}) AND action_id IN ({a_ph})",
                    (*tenant_ids, *action_ids),
                )
                deleted["action_logs"] = cur.rowcount
            if backup["actions"]:
                cur.execute(
                    f"DELETE FROM ado_smart_ops_improvement_action WHERE tenant_id IN ({placeholders}) AND plan_id IN ({id_ph})",
                    (*tenant_ids, *plan_ids),
                )
                deleted["actions"] = cur.rowcount
            if table_exists(cur, "ado_smart_ops_improvement_verify_log"):
                cur.execute(
                    f"DELETE FROM ado_smart_ops_improvement_verify_log WHERE tenant_id IN ({placeholders}) AND plan_id IN ({id_ph})",
                    (*tenant_ids, *plan_ids),
                )
                deleted["verify_logs"] = cur.rowcount
            cur.execute(
                f"DELETE FROM ado_smart_ops_improvement_plan WHERE TenantId IN ({placeholders}) AND Id IN ({id_ph})",
                (*tenant_ids, *plan_ids),
            )
            deleted["plans"] = cur.rowcount
            conn.commit()
        else:
            conn.rollback()

        summary = {
            "work_package": "WP-SD8",
            "dry_run": not args.apply,
            "batch": args.batch,
            "backup": str(backup_path.relative_to(ROOT)).replace("\\", "/"),
            "selected_plans": [str(x) for x in plan_ids],
            "deleted": deleted,
            "kpi_dwd_untouched": True,
        }
        summary_path = EVIDENCE / "08-reset-last.json"
        summary_path.write_text(json.dumps(summary, ensure_ascii=False, indent=2), encoding="utf-8")
        print(json.dumps(summary, ensure_ascii=False, indent=2))
    finally:
        conn.close()


if __name__ == "__main__":
    main()