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

#!/usr/bin/env python3
"""Remove reconstructible invalid-tenant DWD/KPI rows and add hard guards."""

from __future__ import annotations

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

import pymysql


ROOT = Path(__file__).resolve().parents[4]
CONFIG = ROOT / "server" / "Admin.NET.Application" / "Configuration" / "Database.json"
TABLES = (
    "dwd_material_readiness",
    "dwd_material_shortage",
    "dwd_order_schedule_trans",
    "dwd_supplier_delivery",
    "dwd_supplier_risk",
    "dwd_supply_demand",
    "ado_s9_kpi_value_l1_day",
    "ado_s9_kpi_value_l2_day",
    "ado_s9_kpi_value_l3_day",
    "ado_s9_kpi_value_l4_day",
)


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,
    )


def main() -> None:
    conn = connect()
    results: list[dict[str, object]] = []
    try:
        with conn.cursor() as cur:
            cur.execute(
                """
                CREATE TABLE IF NOT EXISTS uat_data_cleanup_audit (
                  id BIGINT NOT NULL AUTO_INCREMENT PRIMARY KEY,
                  batch_code VARCHAR(100) NOT NULL,
                  table_name VARCHAR(128) NOT NULL,
                  invalid_rows_before BIGINT NOT NULL,
                  deleted_rows BIGINT NOT NULL,
                  invalid_rows_after BIGINT NOT NULL,
                  executed_at DATETIME(3) NOT NULL
                ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4
                """
            )
            for index, table in enumerate(TABLES, 1):
                constraint = f"ck_wp5_valid_tenant_{index:02d}"
                invalid = "tenant_id IS NULL OR tenant_id IN (0,1,1300000000001)"
                cur.execute(f"SELECT COUNT(*) FROM `{table}` WHERE {invalid}")
                before = int(cur.fetchone()[0])
                cur.execute(f"DELETE FROM `{table}` WHERE {invalid}")
                deleted = int(cur.rowcount)
                cur.execute(f"SELECT COUNT(*) FROM `{table}` WHERE {invalid}")
                after = int(cur.fetchone()[0])
                cur.execute(
                    """
                    INSERT INTO uat_data_cleanup_audit
                    (batch_code,table_name,invalid_rows_before,deleted_rows,
                     invalid_rows_after,executed_at)
                    VALUES('WP5_INVALID_TENANT_20260817',%s,%s,%s,%s,NOW(3))
                    """,
                    (table, before, deleted, after),
                )
                cur.execute(
                    """
                    SELECT COUNT(*) FROM information_schema.TABLE_CONSTRAINTS
                    WHERE CONSTRAINT_SCHEMA=DATABASE()
                      AND TABLE_NAME=%s AND CONSTRAINT_NAME=%s
                    """,
                    (table, constraint),
                )
                if int(cur.fetchone()[0]) == 0:
                    cur.execute(
                        f"""
                        ALTER TABLE `{table}`
                        ADD CONSTRAINT `{constraint}`
                        CHECK (tenant_id IS NOT NULL
                               AND tenant_id NOT IN (0,1,1300000000001))
                        """
                    )
                results.append(
                    {
                        "table": table,
                        "before": before,
                        "deleted": deleted,
                        "after": after,
                        "constraint": constraint,
                    }
                )
    finally:
        conn.close()
    payload = {
        "executed_at": datetime.now().isoformat(timespec="seconds"),
        "results": results,
    }
    Path(__file__).with_name("WP5-pollution-cleanup-evidence.json").write_text(
        json.dumps(payload, ensure_ascii=False, indent=2), encoding="utf-8"
    )
    print(json.dumps(payload, ensure_ascii=False, indent=2))


if __name__ == "__main__":
    main()