#!/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()