from sqlalchemy import Column, String, Text, Float, DateTime, Integer, Boolean, func, ForeignKey, JSON from sqlalchemy.ext.declarative import declarative_base from sqlalchemy.schema import UniqueConstraint import datetime Base = declarative_base() class DeviceConfig(Base): __tablename__ = "device_configs" id = Column(Integer, primary_key=True, autoincrement=True) device_id = Column(String(64), unique=True, index=True, nullable=False) selected_role_key = Column(String(64), nullable=False) preferred_language = Column(String(10), nullable=True) volume = Column(Integer, nullable=True) last_update_time = Column(Float, nullable=False, default=lambda: datetime.datetime.now().timestamp()) created_at = Column(DateTime, nullable=False, server_default=func.now()) updated_at = Column(DateTime, nullable=False, server_default=func.now(), onupdate=func.now()) def to_dict(self): """将配置转换为可序列化的字典""" return { "selected_role_key": self.selected_role_key, "preferred_language": self.preferred_language, "volume": self.volume if hasattr(self, 'volume') else None, } class ConversationHistory(Base): __tablename__ = "conversation_histories" id = Column(Integer, primary_key=True, autoincrement=True) device_id = Column(String(64), nullable=False, index=True) role_key = Column(String(64), nullable=False, index=True) last_interaction_time = Column(Float, nullable=False, default=lambda: datetime.datetime.now().timestamp()) created_at = Column(DateTime, nullable=False, server_default=func.now()) updated_at = Column(DateTime, nullable=False, server_default=func.now(), onupdate=func.now()) __table_args__ = ( {'mysql_charset': 'utf8mb4', 'mysql_collate': 'utf8mb4_unicode_ci'} ) class ConversationMessage(Base): __tablename__ = "conversation_messages" id = Column(Integer, primary_key=True, autoincrement=True) conversation_id = Column(Integer, ForeignKey("conversation_histories.id", ondelete="CASCADE"), nullable=False, index=True) is_user = Column(Boolean, nullable=False, default=False) # True表示用户消息,False表示助手消息 content = Column(Text, nullable=False) timestamp = Column(Float, nullable=False, default=lambda: datetime.datetime.now().timestamp()) created_at = Column(DateTime, nullable=False, server_default=func.now()) updated_at = Column(DateTime, nullable=False, server_default=func.now(), onupdate=func.now()) __table_args__ = ( {'mysql_charset': 'utf8mb4', 'mysql_collate': 'utf8mb4_unicode_ci'} ) class Role(Base): __tablename__ = "roles" id = Column(Integer, primary_key=True, autoincrement=True) role_key = Column(String(64), unique=True, nullable=False, index=True) name = Column(String(128), nullable=False) description = Column(Text, nullable=True) content = Column(Text, nullable=False) default_language = Column(String(10), nullable=True) volcano_model_id = Column(String(64), nullable=True) minimax_voice_id = Column(String(64), nullable=True) url = Column(String(255), nullable=True) homophones = Column(JSON, nullable=True) enabled = Column(Boolean, default=True, nullable=False) created_at = Column(DateTime, nullable=False, server_default=func.now()) updated_at = Column(DateTime, nullable=False, server_default=func.now(), onupdate=func.now()) __table_args__ = ( {'mysql_charset': 'utf8mb4', 'mysql_collate': 'utf8mb4_unicode_ci'} ) class RoleLanguage(Base): __tablename__ = "role_languages" id = Column(Integer, primary_key=True, autoincrement=True) role_id = Column(Integer, ForeignKey("roles.id", ondelete="CASCADE"), nullable=False) language_code = Column(String(10), nullable=False) name = Column(String(128), nullable=True) content = Column(Text, nullable=True) minimax_voice_id = Column(String(64), nullable=True) url = Column(String(255), nullable=True) created_at = Column(DateTime, nullable=False, server_default=func.now()) updated_at = Column(DateTime, nullable=False, server_default=func.now(), onupdate=func.now()) __table_args__ = ( UniqueConstraint('role_id', 'language_code', name='uix_role_language'), {'mysql_charset': 'utf8mb4', 'mysql_collate': 'utf8mb4_unicode_ci'} ) class DeviceAuth(Base): __tablename__ = "device_auth" id = Column(Integer, primary_key=True, autoincrement=True) device_id = Column(String(64), nullable=False, unique=True, index=True) serial_number = Column(String(64), nullable=False) batch_id = Column(String(20), nullable=True, index=True) is_active = Column(Boolean, default=True, nullable=False) created_at = Column(DateTime, nullable=False, server_default=func.now()) updated_at = Column(DateTime, nullable=False, server_default=func.now(), onupdate=func.now()) __table_args__ = ( {'mysql_charset': 'utf8mb4', 'mysql_collate': 'utf8mb4_unicode_ci'} ) class DeviceFirmwareUpdate(Base): __tablename__ = "device_firmware_update" id = Column(Integer, primary_key=True, autoincrement=True) device_id = Column(String(64), nullable=False, unique=True, index=True) firmware_version = Column(String(64), nullable=False) update_status = Column(String(32), nullable=False, default="success") # 更新状态,如 updating/success/failed created_at = Column(DateTime, nullable=False, server_default=func.now()) updated_at = Column(DateTime, nullable=False, server_default=func.now(), onupdate=func.now()) progress = Column(Float, nullable=True, default=0.0) # 更新进度 __table_args__ = ( {'mysql_charset': 'utf8mb4', 'mysql_collate': 'utf8mb4_unicode_ci'} ) class SystemConfig(Base): __tablename__ = "system_config" id = Column(Integer, primary_key=True, autoincrement=True) config_key = Column(String(128), unique=True, nullable=False, index=True) config_value = Column(Text, nullable=True) created_at = Column(DateTime, nullable=False, server_default=func.now()) updated_at = Column(DateTime, nullable=False, server_default=func.now(), onupdate=func.now()) __table_args__ = ( {'mysql_charset': 'utf8mb4', 'mysql_collate': 'utf8mb4_unicode_ci'} )