65 lines
2.5 KiB
Python
65 lines
2.5 KiB
Python
# 只读探测甄房知识图谱库（zhenos_kg）——不做任何写入
|
||
# 用法: python _design-test/kg-probe.py
|
||
import os, re, sys, pymysql
|
||
|
||
ZF = r'e:\KF_7\ZhenOS-ZF'
|
||
env = {}
|
||
with open(os.path.join(ZF, '.env.local'), encoding='utf-8') as f:
|
||
for ln in f:
|
||
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("'")
|
||
|
||
user = env.get('DB_USERNAME')
|
||
pwd = env.get('DB_PASSWORD')
|
||
if not user or not pwd:
|
||
sys.exit('未在 .env.local 取到 DB_USERNAME / DB_PASSWORD')
|
||
|
||
HOST = 'rm-0jlbgr2rv6dj3t6jngo.mysql.rds.aliyuncs.com'
|
||
PORT = 3306
|
||
|
||
try:
|
||
conn = pymysql.connect(host=HOST, port=PORT, user=user, password=pwd, charset='utf8mb4',
|
||
connect_timeout=12, read_timeout=30)
|
||
except Exception as e:
|
||
sys.exit('连接失败（可能是 RDS IP 白名单未放行本机）：%s' % e)
|
||
|
||
cur = conn.cursor()
|
||
cur.execute("SHOW DATABASES")
|
||
dbs = [r[0] for r in cur.fetchall()]
|
||
print('可见数据库:', ', '.join(dbs))
|
||
print('zhenos_kg 存在:', 'zhenos_kg' in dbs)
|
||
if 'zhenos_kg' not in dbs:
|
||
conn.close(); sys.exit(0)
|
||
|
||
cur.execute("USE zhenos_kg")
|
||
cur.execute("SHOW TABLES")
|
||
tabs = [r[0] for r in cur.fetchall()]
|
||
print('表:', ', '.join(tabs))
|
||
for t in ('kg_entities', 'kg_relations', 'dev_knowledge'):
|
||
if t in tabs:
|
||
cur.execute("SELECT COUNT(*) FROM `%s`" % t)
|
||
print(' %-16s %d 行' % (t, cur.fetchone()[0]))
|
||
|
||
if 'kg_entities' in tabs:
|
||
cur.execute("SELECT entity_type, COUNT(*) FROM kg_entities GROUP BY entity_type ORDER BY 2 DESC")
|
||
print('实体类型分布:', ', '.join('%s=%d' % r for r in cur.fetchall()))
|
||
cur.execute("SELECT COUNT(*) FROM kg_entities WHERE entity_name LIKE 'ZhenosSH:%%'")
|
||
print("前缀 ZhenosSH: 实体数:", cur.fetchone()[0])
|
||
cur.execute("SELECT COUNT(*) FROM kg_entities WHERE entity_name LIKE 'XiaoHu:%%'")
|
||
print("前缀 XiaoHu: 实体数（应为 0）:", cur.fetchone()[0])
|
||
cur.execute("SELECT entity_name FROM kg_entities WHERE entity_name LIKE '%%ui%%' OR entity_name LIKE '%%样式%%' OR entity_name LIKE '%%主题%%' LIMIT 15")
|
||
rows = cur.fetchall()
|
||
print('与 UI/样式/主题 相关的已有实体:')
|
||
for r in rows:
|
||
print(' ', r[0])
|
||
if not rows:
|
||
print(' （无）')
|
||
|
||
if 'dev_knowledge' in tabs:
|
||
cur.execute("SHOW COLUMNS FROM dev_knowledge")
|
||
print('dev_knowledge 列:', ', '.join(c[0] for c in cur.fetchall()))
|
||
|
||
conn.close()
|
||
print('\n只读探测完成，未做任何写入。')
|