ai-rasa-manager/scripts/init_db.py

import logging
import os
import sys
from sqlalchemy import create_engine
from sqlalchemy.exc import SQLAlchemyError

# 添加项目根目录到Python路径
sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))

from config import config
from db import Base

# 初始化日志
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

def init_database():
    """初始化数据库：创建表结构并执行初始SQL脚本"""
    try:
        # 获取数据库连接字符串
        connection_string = config.get("database.connection_string")
        if not connection_string:
            raise ValueError("数据库连接字符串未配置")
        
        # 创建数据库引擎
        engine = create_engine(connection_string)
        
        # 创建所有表
        logger.info("开始创建数据库表结构...")
        Base.metadata.create_all(engine)
        logger.info("数据库表结构创建成功")
        
        # 执行SQL初始化脚本
        sql_script_path = os.path.join(os.path.dirname(os.path.dirname(__file__)), "sql", "schema.sql")
        if os.path.exists(sql_script_path):
            logger.info(f"开始执行初始化SQL脚本: {sql_script_path}")
            
            with engine.connect() as connection:
                with open(sql_script_path, 'r', encoding='utf-8') as f:
                    # 读取并执行SQL脚本
                    sql = f.read()
                    # 分割SQL语句（简单处理，适用于大多数情况）
                    statements = sql.split(';')
                    for stmt in statements:
                        stmt = stmt.strip()
                        if stmt and not stmt.startswith('--'):
                            connection.execute(stmt)
                connection.commit()
            logger.info("SQL初始化脚本执行成功")
        else:
            logger.warning(f"未找到SQL初始化脚本: {sql_script_path}")
        
        logger.info("数据库初始化完成")
        
    except SQLAlchemyError as e:
        logger.error(f"数据库操作错误: {str(e)}")
        sys.exit(1)
    except Exception as e:
        logger.error(f"初始化数据库失败: {str(e)}")
        sys.exit(1)

if __name__ == "__main__":
    init_database()