50 lines
1.6 KiB
Python
50 lines
1.6 KiB
Python
"""验证预警规则系统"""
|
|
import sys, os, json
|
|
sys.path.insert(0, '/root/cma-management/backend')
|
|
os.chdir('/root/cma-management/backend')
|
|
from dotenv import load_dotenv
|
|
load_dotenv()
|
|
from app.database import get_session_local
|
|
from app.models import KPIDefinition
|
|
from sqlalchemy import text, func
|
|
|
|
db = get_session_local()()
|
|
|
|
# 检查 alert_rules 表
|
|
try:
|
|
total = db.execute(text("SELECT COUNT(*) FROM alert_rules")).scalar()
|
|
type_counts = db.execute(text("SELECT rule_type, COUNT(*) FROM alert_rules GROUP BY rule_type")).fetchall()
|
|
print(f"Total alert rules: {total}")
|
|
for t, c in type_counts:
|
|
print(f" {t}: {c}")
|
|
except Exception as e:
|
|
print(f"表不存在: {e}")
|
|
|
|
# 检查 dynamic_threshold_cache
|
|
try:
|
|
dc = db.execute(text("SELECT COUNT(*) FROM dynamic_threshold_cache")).scalar()
|
|
print(f"Dynamic threshold caches: {dc}")
|
|
except Exception as e:
|
|
print(f"dynamic_threshold_cache表不存在: {e}")
|
|
|
|
# 取样显示预警规则
|
|
try:
|
|
samples = db.execute(text("""
|
|
SELECT ar.id, ar.kpi_id, ar.rule_type, ar.enabled, ar.params, k.kpi_code, k.kpi_name
|
|
FROM alert_rules ar
|
|
LEFT JOIN kpi_definitions k ON k.id = ar.kpi_id
|
|
LIMIT 6
|
|
""")).fetchall()
|
|
print("\nSample rules:")
|
|
for s in samples:
|
|
print(f" [{s.id}] {s.kpi_code}({s.kpi_name}) type={s.rule_type} enabled={s.enabled}")
|
|
except Exception as e:
|
|
print(f"查询失败: {e}")
|
|
|
|
# 总KPI数验证
|
|
total_kpis = db.execute(text("SELECT COUNT(*) FROM kpi_definitions WHERE status='active'")).scalar()
|
|
print(f"\nTotal active KPIs: {total_kpis}")
|
|
|
|
db.close()
|
|
print("\n✅ 验证完成")
|