47 lines
1.4 KiB
Python
47 lines
1.4 KiB
Python
"""
|
|
检查数据库表结构脚本
|
|
查看各表的实际字段结构
|
|
"""
|
|
|
|
import sys
|
|
import os
|
|
sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
|
|
|
from src.storage.database import db_manager
|
|
from sqlalchemy import text
|
|
import logging
|
|
|
|
# 配置日志
|
|
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
|
|
logger = logging.getLogger(__name__)
|
|
|
|
def check_table_structure():
|
|
"""检查表结构"""
|
|
try:
|
|
# 获取数据库会话
|
|
session = db_manager.get_session()
|
|
|
|
logger.info("=== 检查表结构 ===")
|
|
|
|
# 检查各表结构
|
|
tables = ['stock_basic', 'daily_kline', 'financial_report', 'data_source', 'system_log']
|
|
|
|
for table in tables:
|
|
logger.info(f"=== 表 {table} 结构 ===")
|
|
result = session.execute(text(f"DESCRIBE {table}"))
|
|
columns = result.fetchall()
|
|
|
|
for column in columns:
|
|
logger.info(f"字段: {column[0]}, 类型: {column[1]}, 是否为空: {column[2]}, 键: {column[3]}")
|
|
|
|
logger.info("")
|
|
|
|
session.close()
|
|
logger.info("=== 表结构检查完成 ===")
|
|
|
|
except Exception as e:
|
|
logger.error(f"检查表结构失败: {str(e)}")
|
|
raise
|
|
|
|
if __name__ == "__main__":
|
|
check_table_structure() |