"""
|
数据库初始化工具
|
"""
|
from sqlalchemy import text
|
from loguru import logger
|
from app.models.database import engine, SessionLocal
|
from app.models.contact import Contact
|
from app.models.conversation import Conversation
|
|
|
def init_database():
|
"""初始化数据库"""
|
try:
|
# 测试数据库连接
|
with engine.connect() as conn:
|
result = conn.execute(text("SELECT 1"))
|
logger.info("数据库连接测试成功")
|
|
# 创建所有表
|
Contact.metadata.create_all(bind=engine)
|
Conversation.metadata.create_all(bind=engine)
|
|
logger.info("数据库表创建成功")
|
return True
|
|
except Exception as e:
|
logger.error(f"数据库初始化失败: {str(e)}")
|
return False
|
|
|
def check_database_health():
|
"""检查数据库健康状态"""
|
try:
|
db = SessionLocal()
|
|
# 测试查询
|
contact_count = db.query(Contact).count()
|
conversation_count = db.query(Conversation).count()
|
|
db.close()
|
|
logger.info(f"数据库健康检查通过: contacts={contact_count}, conversations={conversation_count}")
|
return True
|
|
except Exception as e:
|
logger.error(f"数据库健康检查失败: {str(e)}")
|
return False
|
|
|
if __name__ == "__main__":
|
# 直接运行此文件可以初始化数据库
|
init_database()
|
