AiDOPWarehouse/ai-dop-platform/tools/_probe_tenant_isolation.py

# -*- coding: utf-8 -*-
"""对比两个租户在 AiDOP 核心表中的数据差异，验证租户隔离是否落地。"""
from __future__ import annotations
import sys
import pymysql

sys.stdout.reconfigure(encoding='utf-8')

CONN = dict(
    host='123.60.180.165', port=3306,
    user='aidopremote', password='1234567890aiDOP#',
    database='aidopdev', charset='utf8mb4', autocommit=True,
)

# Admin.NET 框架默认租户 ID（SqlSugarConst.DefaultTenantId）
DEFAULT_TENANT = 1300000000001
DEMO_TENANT = 1300000000888

TENANTS = [
    ("superAdmin 默认租户", DEFAULT_TENANT),
    ("DemoAdmin Demo 租户", DEMO_TENANT),
]


def table_exists(cur, table: str) -> bool:
    cur.execute(
        "SELECT COUNT(*) AS c FROM information_schema.tables "
        "WHERE table_schema=DATABASE() AND table_name=%s",
        (table,),
    )
    row = cur.fetchone()
    return bool(row and row.get('c', 0) > 0)


def count_by_tenant(cur, table: str, tenant_col: str, tenant_id: int, extra_where: str = "") -> int:
    sql = f"SELECT COUNT(*) AS c FROM `{table}` WHERE `{tenant_col}` = %s"
    if extra_where:
        sql += f" AND {extra_where}"
    cur.execute(sql, (tenant_id,))
    row = cur.fetchone()
    return int(row['c']) if row else 0


def sample_s4_l1(cur, tenant_id: int, limit: int = 3):
    # 最新 biz_date 的 S4 L1 值
    cur.execute(
        """
        SELECT biz_date, metric_code, metric_value, target_value, status_color
        FROM ado_s9_kpi_value_l1_day
        WHERE tenant_id = %s AND module_code = 'S4' AND is_deleted = 0
          AND biz_date = (
              SELECT MAX(biz_date) FROM ado_s9_kpi_value_l1_day
              WHERE tenant_id = %s AND module_code = 'S4' AND is_deleted = 0
          )
        ORDER BY metric_code
        LIMIT %s
        """,
        (tenant_id, tenant_id, limit),
    )
    return cur.fetchall()


def list_kpi_master_s4(cur, tenant_id: int, limit: int = 3):
    cur.execute(
        """
        SELECT MetricCode, MetricName, MetricLevel, YellowThreshold, RedThreshold
        FROM ado_smart_ops_kpi_master
        WHERE TenantId = %s AND ModuleCode = 'S4' AND MetricLevel = 1
        ORDER BY SortNo
        LIMIT %s
        """,
        (tenant_id, limit),
    )
    return cur.fetchall()


def main() -> None:
    conn = pymysql.connect(**CONN)
    with conn.cursor(pymysql.cursors.DictCursor) as cur:
        tables = [
            ("ado_smart_ops_kpi_master",    "TenantId",  ""),
            ("ado_smart_ops_layout_item",   "TenantId",  ""),
            ("ado_smart_ops_home_module",   "TenantId",  ""),
            ("ado_s9_kpi_value_l1_day",     "tenant_id", "is_deleted=0"),
            ("ado_s9_kpi_value_l2_day",     "tenant_id", "is_deleted=0"),
            ("ado_s9_kpi_value_l3_day",     "tenant_id", "is_deleted=0"),
        ]

        print("=" * 88)
        print(f"{'表':<32} {'过滤':<18} " + "  ".join(f"{n:<24}" for n, _ in TENANTS))
        print("-" * 88)
        for t, tcol, extra in tables:
            if not table_exists(cur, t):
                print(f"{t:<32} [表不存在]")
                continue
            counts = [count_by_tenant(cur, t, tcol, tid, extra) for _, tid in TENANTS]
            extra_lbl = extra or "-"
            print(f"{t:<32} {extra_lbl:<18} " + "  ".join(f"{c:<24}" for c in counts))

        # S4 KpiMaster 指标名称对比（第一个 L1 指标）
        print("\n" + "=" * 88)
        print("S4 L1 KpiMaster（前 3 条）:")
        for name, tid in TENANTS:
            print(f"\n  [{name}] TenantId={tid}")
            rows = list_kpi_master_s4(cur, tid, 3)
            if not rows:
                print("    (无数据)")
            for r in rows:
                print(f"    {r['MetricCode']:<20} | {r['MetricName']:<24} | 黄={r['YellowThreshold']} 红={r['RedThreshold']}")

        # S4 L1 最新日值对比
        print("\n" + "=" * 88)
        print("S4 L1 日表最新 biz_date 的前 3 条（目的：两个租户数值应不相同）:")
        for name, tid in TENANTS:
            print(f"\n  [{name}] TenantId={tid}")
            rows = sample_s4_l1(cur, tid, 3)
            if not rows:
                print("    (无数据)")
            for r in rows:
                print(f"    {str(r['biz_date']):<12} {r['metric_code']:<20} 实际={r['metric_value']} 目标={r['target_value']} 颜色={r['status_color']}")

    conn.close()


if __name__ == "__main__":
    main()