82 lines
2.7 KiB
Python
82 lines
2.7 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
查询微信反馈数据表结构
|
|
"""
|
|
import os
|
|
import re
|
|
import pymysql
|
|
|
|
# 从secrets.md读取数据库配置
|
|
SECRETS_PATH = "/root/.openclaw/workspace-xiaokui/secrets.md"
|
|
|
|
def load_mysql_config():
|
|
if not os.path.exists(SECRETS_PATH):
|
|
raise RuntimeError(f"密钥文件不存在: {SECRETS_PATH}")
|
|
|
|
with open(SECRETS_PATH, 'r', encoding='utf-8') as f:
|
|
content = f.read()
|
|
|
|
# 匹配微信反馈数据行
|
|
pattern = r'\| 微信反馈数据.*?\| `([^`]+)` \| `([^`]+)` \| `([^`]+)` \| `([^`]+)` \| `([^`]+)` \|'
|
|
match = re.search(pattern, content, re.DOTALL)
|
|
if not match:
|
|
raise RuntimeError("secrets.md中未找到微信反馈数据库配置")
|
|
|
|
host = match.group(1)
|
|
port = int(match.group(2))
|
|
db = match.group(3)
|
|
user = match.group(4)
|
|
password = match.group(5)
|
|
|
|
return {
|
|
'host': host,
|
|
'port': port,
|
|
'user': user,
|
|
'password': password,
|
|
'database': db,
|
|
'charset': 'utf8mb4'
|
|
}
|
|
|
|
def main():
|
|
config = load_mysql_config()
|
|
print(f"连接数据库: {config['host']}:{config['port']}/{config['database']} (用户: {config['user']})")
|
|
|
|
conn = None
|
|
try:
|
|
conn = pymysql.connect(**config)
|
|
with conn.cursor() as cursor:
|
|
# 查看表结构
|
|
cursor.execute("DESCRIBE wechat_group_message")
|
|
columns = cursor.fetchall()
|
|
|
|
print("\n表 wechat_group_message 结构:")
|
|
print(f"{'字段名':<20} {'类型':<30} {'是否为空':<8} {'键':<5} {'默认值':<10} {'额外':<10}")
|
|
print("-" * 100)
|
|
for col in columns:
|
|
field, type_, null, key, default, extra = col
|
|
print(f"{field:<20} {str(type_):<30} {null:<8} {key:<5} {str(default):<10} {extra:<10}")
|
|
|
|
# 查询总记录数
|
|
cursor.execute("SELECT COUNT(*) FROM wechat_group_message")
|
|
count = cursor.fetchone()[0]
|
|
print(f"\n总记录数: {count}")
|
|
|
|
# 查询最新10条记录预览
|
|
print("\n最新10条记录预览:")
|
|
cursor.execute("SELECT * FROM wechat_group_message ORDER BY id DESC LIMIT 10")
|
|
rows = cursor.fetchall()
|
|
fields = [col[0] for col in cursor.description]
|
|
|
|
for row in rows:
|
|
row_dict = dict(zip(fields, row))
|
|
print(f"ID: {row_dict.get('id')} | 时间: {row_dict.get('create_time', '')} | 发送人: {row_dict.get('sender', '')[:10]} | 内容: {str(row_dict.get('content', ''))[:50]}...")
|
|
|
|
except Exception as e:
|
|
print(f"错误: {e}")
|
|
finally:
|
|
if conn:
|
|
conn.close()
|
|
|
|
if __name__ == '__main__':
|
|
main()
|