ai-rasa-manager/db/models.py

import logging
import json
from datetime import datetime
from sqlalchemy import Column, Integer, String, Text, DateTime, ForeignKey, Boolean, UniqueConstraint, Index
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import relationship
from sqlalchemy.exc import SQLAlchemyError

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

Base = declarative_base()

class Intent(Base):
    """意图表模型"""
    __tablename__ = 'intent'
    
    id = Column(Integer, primary_key=True, autoincrement=True, comment='意图ID')
    name = Column(String(100), nullable=False, unique=True, comment='意图名称')
    description = Column(String(500), comment='意图描述')
    created_time = Column(DateTime, default=datetime.now, nullable=False, comment='创建时间')
    updated_time = Column(DateTime, default=datetime.now, onupdate=datetime.now, nullable=False, comment='更新时间')
    
    examples = relationship('IntentExample', backref='intent', cascade='all, delete-orphan')
    
    __table_args__ = {'comment': '存储Rasa意图定义'}

class IntentExample(Base):
    """意图样本表模型"""
    __tablename__ = 'intent_example'
    
    id = Column(Integer, primary_key=True, autoincrement=True, comment='样本ID')
    intent_id = Column(Integer, ForeignKey('intent.id'), nullable=False, comment='意图ID')
    text = Column(String(1000), nullable=False, comment='样本文本')
    created_time = Column(DateTime, default=datetime.now, nullable=False, comment='创建时间')
    updated_time = Column(DateTime, default=datetime.now, onupdate=datetime.now, nullable=False, comment='更新时间')

    
    __table_args__ = {'comment': '存储意图对应的用户输入示例'}

class Story(Base):
    """故事表模型"""
    __tablename__ = 'story'
    
    id = Column(Integer, primary_key=True, autoincrement=True, comment='故事ID')
    name = Column(String(200), nullable=False, unique=True, comment='故事名称')
    description = Column(String(500), comment='故事描述')
    created_time = Column(DateTime, default=datetime.now, nullable=False, comment='创建时间')
    updated_time = Column(DateTime, default=datetime.now, onupdate=datetime.now, nullable=False, comment='更新时间')
    
    # 关联关系
    steps = relationship('StoryStep', backref='story', cascade='all, delete-orphan')
    
    __table_args__ = {'comment': '存储Rasa对话流程定义'}

class StoryStep(Base):
    """故事步骤表模型"""
    __tablename__ = 'story_step'
    
    id = Column(Integer, primary_key=True, autoincrement=True, comment='步骤ID')
    story_id = Column(Integer, ForeignKey('story.id'), nullable=False, comment='故事ID')
    step_order = Column(Integer, nullable=False, comment='步骤顺序')
    step_type = Column(String(20), nullable=False, comment='步骤类型')
    content = Column(Text, nullable=False, comment='步骤内容(JSON)')
    created_time = Column(DateTime, default=datetime.now, nullable=False, comment='创建时间')
    updated_time = Column(DateTime, default=datetime.now, onupdate=datetime.now, nullable=False, comment='更新时间')

    
    @property
    def content_dict(self):
        """将JSON字符串转换为字典"""
        try:
            if not self.content:
                return {}
            return json.loads(self.content)
        except json.JSONDecodeError as e:
            logger.error(f"解析StoryStep内容失败: {str(e)}, 内容: {self.content}")
            return {}
    
    __table_args__ = {'comment': '存储故事的具体步骤'}

class CustomAction(Base):
    """自定义动作表模型"""
    __tablename__ = 'custom_action'
    
    id = Column(Integer, primary_key=True, autoincrement=True, comment='动作ID')
    name = Column(String(100), nullable=False, unique=True, comment='动作名称')
    description = Column(String(500), comment='动作描述')
    http_method = Column(String(10), nullable=False, comment='HTTP方法')
    api_url = Column(String(500), nullable=False, comment='API地址')
    token = Column(String(500), comment='鉴权Token')
    headers = Column(Text, comment='请求头(JSON)')
    request_body = Column(Text, comment='请求体(JSON)')
    response_mapping = Column(Text, comment='响应映射规则(JSON)')
    created_time = Column(DateTime, default=datetime.now, nullable=False, comment='创建时间')
    updated_time = Column(DateTime, default=datetime.now, onupdate=datetime.now, nullable=False, comment='更新时间')
    
    @property
    def headers_dict(self):
        """将请求头JSON转换为字典"""
        try:
            if not self.headers:
                return {}
            return json.loads(self.headers)
        except json.JSONDecodeError as e:
            logger.error(f"解析CustomAction请求头失败: {str(e)}, 内容: {self.headers}")
            return {}
    
    @property
    def request_body_dict(self):
        """将请求体JSON转换为字典"""
        try:
            if not self.request_body:
                return {}
            return json.loads(self.request_body)
        except json.JSONDecodeError as e:
            logger.error(f"解析CustomAction请求体失败: {str(e)}, 内容: {self.request_body}")
            return {}
    
    @property
    def response_mapping_dict(self):
        """将响应映射JSON转换为字典"""
        try:
            if not self.response_mapping:
                return {}
            return json.loads(self.response_mapping)
        except json.JSONDecodeError as e:
            logger.error(f"解析CustomAction响应映射失败: {str(e)}, 内容: {self.response_mapping}")
            return {}
    
    __table_args__ = {'comment': '存储对接第三方API的自定义动作配置'}

class SystemConfig(Base):
    """系统配置表模型"""
    __tablename__ = 'system_config'
    
    id = Column(Integer, primary_key=True, autoincrement=True, comment='配置ID')
    config_key = Column(String(100), nullable=False, unique=True, comment='配置键')
    config_value = Column(String(500), nullable=False, comment='配置值')
    description = Column(String(500), comment='配置描述')
    updated_by = Column(String(100), comment='更新人')
    updated_time = Column(DateTime, default=datetime.now, onupdate=datetime.now, nullable=False, comment='更新时间')
    
    __table_args__ = {'comment': '存储系统配置信息'}
    
    @classmethod
    def get_value(cls, session, key, default=None):
        """
        获取配置值
        
        Args:
            session: 数据库会话
            key: 配置键
            default: 默认值
            
        Returns:
            配置值或默认值
        """
        try:
            config = session.query(cls).filter_by(config_key=key).first()
            return config.config_value if config else default
        except SQLAlchemyError as e:
            logger.error(f"获取系统配置 {key} 失败: {str(e)}")
            return default
    
    @classmethod
    def set_value(cls, session, key, value, user=None):
        """
        设置配置值
        
        Args:
            session: 数据库会话
            key: 配置键
            value: 配置值
            user: 操作人
            
        Returns:
            配置对象
        """
        try:
            config = session.query(cls).filter_by(config_key=key).first()
            if config:
                config.config_value = value
                config.updated_by = user
            else:
                config = cls(
                    config_key=key,
                    config_value=value,
                    updated_by=user
                )
                session.add(config)
            session.commit()
            return config
        except SQLAlchemyError as e:
            session.rollback()
            logger.error(f"设置系统配置 {key} 失败: {str(e)}")
            raise Exception(f"设置系统配置失败: {str(e)}")