63 lines
3.0 KiB
Python
63 lines
3.0 KiB
Python
# 甄房知识图谱健康检查（只读）
|
|
# 检查项：孤立实体 / 悬空关系 / 前缀合规（不得出现 XiaoHu:）
|
|
# 用法: python _design-test/kg-health.py [关键词]
|
|
import os, re, sys, pymysql
|
|
|
|
ZF = r'e:\KF_7\ZhenOS-ZF'
|
|
env = {}
|
|
for ln in open(os.path.join(ZF, '.env.local'), encoding='utf-8'):
|
|
m = re.match(r'^\s*([A-Z_][A-Z0-9_]*)\s*=\s*(.*?)\s*$', ln)
|
|
if m:
|
|
env[m.group(1)] = m.group(2).strip().strip('"').strip("'")
|
|
|
|
conn = pymysql.connect(host='rm-0jlbgr2rv6dj3t6jngo.mysql.rds.aliyuncs.com', port=3306,
|
|
user=env['DB_USERNAME'], password=env['DB_PASSWORD'],
|
|
charset='utf8mb4', connect_timeout=12)
|
|
cur = conn.cursor()
|
|
cur.execute('USE zhenos_kg')
|
|
|
|
def one(sql):
|
|
cur.execute(sql); return cur.fetchone()[0]
|
|
|
|
print('=== 甄房知识图谱健康检查 ===')
|
|
print(' 实体总数 ', one('SELECT COUNT(*) FROM kg_entities'))
|
|
print(' 关系总数 ', one('SELECT COUNT(*) FROM kg_relations'))
|
|
print(' 孤立实体（无任何关系）', one(
|
|
'SELECT COUNT(*) FROM kg_entities e WHERE NOT EXISTS('
|
|
'SELECT 1 FROM kg_relations r WHERE r.from_entity=e.entity_name OR r.to_entity=e.entity_name)'))
|
|
print(' 悬空关系 from ', one(
|
|
'SELECT COUNT(*) FROM kg_relations WHERE from_entity NOT IN (SELECT entity_name FROM kg_entities)'))
|
|
print(' 悬空关系 to ', one(
|
|
'SELECT COUNT(*) FROM kg_relations WHERE to_entity NOT IN (SELECT entity_name FROM kg_entities)'))
|
|
print(' 非 ZhenosSH: 前缀实体', one(
|
|
"SELECT COUNT(*) FROM kg_entities WHERE entity_name NOT LIKE 'ZhenosSH:%'"))
|
|
print(" 旧前缀 XiaoHu: ", one("SELECT COUNT(*) FROM kg_entities WHERE entity_name LIKE 'XiaoHu:%'"))
|
|
print(' dev_knowledge 条数 ', one('SELECT COUNT(*) FROM dev_knowledge'))
|
|
print(' 其中 source=AI_SYNC ', one("SELECT COUNT(*) FROM dev_knowledge WHERE source='AI_SYNC'"))
|
|
print(' 实体中 AI_SYNC ', one("SELECT COUNT(*) FROM kg_entities WHERE source='AI_SYNC'"))
|
|
print(' 关系中 AI_SYNC 模块 ', one("SELECT COUNT(*) FROM kg_relations WHERE module IN ('ruoyi-ui','zhenfang-uni')"))
|
|
|
|
cur.execute("SELECT entity_type, entity_name FROM kg_entities WHERE source='AI_SYNC' ORDER BY entity_type, entity_name")
|
|
_rows = cur.fetchall()
|
|
if _rows:
|
|
print('\n source=AI_SYNC 的新增实体（%d 个）：' % len(_rows))
|
|
for t, n in _rows:
|
|
print(' %-12s %s' % (t, n))
|
|
|
|
if len(sys.argv) > 1:
|
|
kw = sys.argv[1]
|
|
cur.execute("SELECT entity_type, entity_name FROM kg_entities WHERE entity_name LIKE %s ORDER BY entity_type",
|
|
('%' + kw + '%',))
|
|
rows = cur.fetchall()
|
|
print('\n 含「%s」的实体 %d 个：' % (kw, len(rows)))
|
|
for t, n in rows:
|
|
print(' %-14s %s' % (t, n))
|
|
cur.execute("SELECT type, title FROM dev_knowledge WHERE title LIKE %s OR tags LIKE %s",
|
|
('%' + kw + '%', '%' + kw + '%'))
|
|
rows = cur.fetchall()
|
|
print(' 含「%s」的知识条目 %d 条：' % (kw, len(rows)))
|
|
for t, n in rows:
|
|
print(' %-12s %s' % (t, n))
|
|
|
|
conn.close()
|