54 lines
1.8 KiB
Python
54 lines
1.8 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 simple_check():
|
||
"""简单检查数据状态"""
|
||
try:
|
||
# 获取数据库会话
|
||
session = db_manager.get_session()
|
||
|
||
logger.info("=== 简单数据检查 ===")
|
||
|
||
# 检查各表是否存在数据
|
||
tables = ['stock_basic', 'daily_kline', 'financial_report', 'data_source', 'system_log']
|
||
|
||
for table in tables:
|
||
# 先检查表是否存在
|
||
result = session.execute(text(f"SHOW TABLES LIKE '{table}'"))
|
||
if result.fetchone():
|
||
# 检查数据量
|
||
result = session.execute(text(f"SELECT COUNT(*) FROM {table}"))
|
||
count = result.fetchone()[0]
|
||
logger.info(f"表 {table}: {count} 条记录")
|
||
|
||
# 如果是stock_basic表且有数据，显示前几条
|
||
if table == 'stock_basic' and count > 0:
|
||
result = session.execute(text(f"SELECT code, name FROM {table} LIMIT 5"))
|
||
stocks = result.fetchall()
|
||
for stock in stocks:
|
||
logger.info(f" 股票: {stock[0]}, 名称: {stock[1]}")
|
||
else:
|
||
logger.info(f"表 {table} 不存在")
|
||
|
||
session.close()
|
||
logger.info("=== 检查完成 ===")
|
||
|
||
except Exception as e:
|
||
logger.error(f"检查失败: {str(e)}")
|
||
raise
|
||
|
||
if __name__ == "__main__":
|
||
simple_check() |