# -*- coding: utf-8 -*- """验证各模块 KpiMaster + LayoutItem + kpi_value_l*_day 数据""" 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, ) MODULES = ["S1", "S2", "S3", "S4", "S5", "S6", "S7", "S9"] TENANTS = [1300000000001, 1300000000888] FACTORY_ID = 1 def main() -> None: conn = pymysql.connect(**CONN) with conn.cursor(pymysql.cursors.DictCursor) as cur: for tid in TENANTS: print(f"\n=== 租户 {tid} ===") print(f"{'模块':<6}{'KpiMaster':<12}{'LayoutItem(L1/L2+)':<22}{'L1日值':<10}{'L2日值':<10}{'L3日值':<10}") print("-" * 72) for m in MODULES: cur.execute( "SELECT COUNT(*) AS c FROM ado_smart_ops_kpi_master " "WHERE TenantId=%s AND ModuleCode=%s AND IsEnabled=1", (tid, m), ) km = cur.fetchone()['c'] cur.execute( "SELECT SUM(CASE WHEN MetricLevel=1 THEN 1 ELSE 0 END) AS l1, " "SUM(CASE WHEN MetricLevel>=2 THEN 1 ELSE 0 END) AS l2p " "FROM ado_smart_ops_layout_item " "WHERE TenantId=%s AND FactoryId=%s AND ModuleCode=%s AND IsEnabled=1", (tid, FACTORY_ID, m), ) lr = cur.fetchone() layout = f"{lr['l1'] or 0}/{lr['l2p'] or 0}" vals = {} for lvl in ('l1', 'l2', 'l3'): cur.execute( f"SELECT COUNT(*) AS c FROM ado_s9_kpi_value_{lvl}_day " "WHERE tenant_id=%s AND factory_id=%s AND module_code=%s AND is_deleted=0", (tid, FACTORY_ID, m), ) vals[lvl] = cur.fetchone()['c'] print(f"{m:<6}{km:<12}{layout:<22}{vals['l1']:<10}{vals['l2']:<10}{vals['l3']:<10}") conn.close() if __name__ == "__main__": main()