""" 数据模型定义模块 定义AI换脸换装系统的所有数据表结构 用户表：存储用户信息（用户ID，用户名, 密码哈希，是否激活，是否管理员，创建时间，更新时间，最后登录时间） 图片记录表：存储图片信息（图片ID，用户ID，图片类型，原始文件名，存储路径，文件大小，图片哈希，是否已删除，创建时间，更新时间） 处理记录表：存储处理记录信息（处理记录ID，用户ID，人脸图片ID，服装图片ID，生成结果图片ID, AI生成的文案内容，完成时间） 系统配置表：存储系统配置信息（配置ID，配置键，配置值，配置类型，配置描述，是否公开配置，创建时间，更新时间） 数据库版本表：存储数据库版本信息（版本ID，版本号，版本描述，应用时间，迁移文件名） """ from datetime import datetime from typing import Optional, List from sqlalchemy import ( Column, Integer, String, Text, DateTime, Boolean, Float, ForeignKey, Index, UniqueConstraint, CheckConstraint ) from sqlalchemy.ext.declarative import declarative_base from sqlalchemy.orm import relationship, Mapped from sqlalchemy.sql import func # 定义数据库模型 Base = declarative_base() class User(Base): """用户表""" __tablename__ = "users" id = Column(Integer, primary_key=True, autoincrement=True, comment="用户ID") username = Column(String(50), unique=True, nullable=False, comment="用户名") password_hash = Column(String(255), nullable=False, comment="密码哈希") is_active = Column(Boolean, default=True, comment="是否激活") is_admin = Column(Boolean, default=False, comment="是否管理员") created_at = Column(DateTime, default=func.now(), comment="创建时间") updated_at = Column(DateTime, default=func.now(), onupdate=func.now(), comment="更新时间") last_login = Column(DateTime, nullable=True, comment="最后登录时间") # 关联关系: 用户与图片记录、处理记录一对多关系（一个用户可以有多张图片和多个处理记录） image_records: Mapped[List["ImageRecord"]] = relationship("ImageRecord", back_populates="user") process_records: Mapped[List["ProcessRecord"]] = relationship("ProcessRecord", back_populates="user") text_records: Mapped[List["TextRecord"]] = relationship("TextRecord", back_populates="user") __table_args__ = ( Index('idx_username', 'username'), ) class ImageRecord(Base): """图片记录表""" __tablename__ = "image_records" id = Column(Integer, primary_key=True, autoincrement=True, comment="图片记录ID") user_id = Column(Integer, ForeignKey("users.id"), nullable=False, comment="用户ID") image_type = Column(String(20), nullable=False, comment="图片类型(face/cloth/result/original)") original_filename = Column(String(255), nullable=False, comment="原始文件名") stored_path = Column(String(500), nullable=False, comment="存储路径") file_size = Column(Integer, nullable=True, comment="文件大小(字节)") image_hash = Column(String(64), nullable=True, comment="图片哈希值") is_deleted = Column(Boolean, default=False, comment="是否已删除") created_at = Column(DateTime, default=func.now(), comment="创建时间") updated_at = Column(DateTime, default=func.now(), onupdate=func.now(), comment="更新时间") # 关联关系：图片记录与用户、处理记录多对一关系；一个人脸图片可以有多个处理记录 user: Mapped["User"] = relationship("User", back_populates="image_records") face_processes: Mapped[List["ProcessRecord"]] = relationship( "ProcessRecord", foreign_keys="ProcessRecord.face_image_id", back_populates="face_image" ) cloth_processes: Mapped[List["ProcessRecord"]] = relationship( "ProcessRecord", foreign_keys="ProcessRecord.cloth_image_id", back_populates="cloth_image" ) result_processes: Mapped[List["ProcessRecord"]] = relationship( "ProcessRecord", foreign_keys="ProcessRecord.result_image_id", back_populates="result_image" ) __table_args__ = ( Index('idx_image_user_id', 'user_id'), Index('idx_image_type', 'image_type'), Index('idx_image_created_at', 'created_at'), CheckConstraint("image_type IN ('face', 'cloth', 'result', 'original')", name="ck_image_type"), ) class TextRecord(Base): """文本模板记录表""" __tablename__ = "text_records" id = Column(Integer, primary_key=True, autoincrement=True, comment="文本模板记录ID") user_id = Column(Integer, ForeignKey("users.id"), nullable=False, comment="用户ID") text_type = Column(String(20), nullable=False, comment="文本模型类型(prompt/copywrite)") text_name = Column(String(20), nullable=False, comment="文本名称") text_label = Column(String(20), nullable=False, comment="文本模板标签") text_content = Column(Text, nullable=False, comment="文本模板内容") created_at = Column(DateTime, default=func.now(), comment="创建时间") # 关联关系：文本模型与用户多对一关系 user: Mapped["User"] = relationship("User", back_populates="text_records") __table_args__ = ( Index('idx_text_user_id', 'user_id'), Index('idx_text_type', 'text_type'), Index('idx_text_name', 'text_name'), Index('idx_text_label', 'text_label'), CheckConstraint("text_type IN ('prompt', 'copywrite')", name="ck_text_type"), ) class ProcessRecord(Base): """AI换脸换装处理记录表""" __tablename__ = "process_records" id = Column(Integer, primary_key=True, autoincrement=True, comment="处理记录ID") user_id = Column(Integer, ForeignKey("users.id"), nullable=False, comment="用户ID") # 输入图片信息 face_image_id = Column(Integer, ForeignKey("image_records.id"), nullable=False, comment="人脸图片ID") cloth_image_id = Column(Integer, ForeignKey("image_records.id"), nullable=False, comment="服装图片ID") # 输出图片信息 result_image_id = Column(Integer, ForeignKey("image_records.id"), nullable=True, comment="生成结果图片ID") # 文案信息 generated_text = Column(Text, nullable=True, comment="AI生成的文案内容") # 状态信息 status = Column(String(20), nullable=False, default="待审核", comment="处理记录状态") # 完成时间 completed_at = Column(DateTime, nullable=True, onupdate=func.now(), comment="完成时间") # 任务类型 task_type = Column(String(40), nullable=False, comment="任务类型") # 提示词 prompt = Column(Text, nullable=False, comment="提示词") # 关联关系：处理记录与用户、图片记录多对一关系 user: Mapped["User"] = relationship("User", back_populates="process_records") face_image: Mapped["ImageRecord"] = relationship( "ImageRecord", foreign_keys=[face_image_id], back_populates="face_processes" ) cloth_image: Mapped["ImageRecord"] = relationship( "ImageRecord", foreign_keys=[cloth_image_id], back_populates="cloth_processes" ) result_image: Mapped[Optional["ImageRecord"]] = relationship( "ImageRecord", foreign_keys=[result_image_id], back_populates="result_processes" ) __table_args__ = ( Index('idx_process_user_id', 'user_id'), Index('idx_face_image_id', 'face_image_id'), Index('idx_cloth_image_id', 'cloth_image_id'), Index('idx_result_image_id', 'result_image_id'), Index('idx_completed_at', 'completed_at'), Index('idx_status', 'status'), CheckConstraint( "status IN ('已审核', '已发布', '待审核', '待发布', '已拒绝')", name="ck_process_status" ), CheckConstraint( "task_type IN ('swap_face', 'swap_cloth', 'swap_bg', 'swap_all')", name="ck_process_task_type" ) ) class SystemConfig(Base): """系统配置表""" __tablename__ = "system_configs" id = Column(Integer, primary_key=True, autoincrement=True, comment="配置ID") config_key = Column(String(100), nullable=False, comment="配置键") config_value = Column(Text, nullable=False, comment="配置值") config_type = Column(String(20), nullable=False, comment="配置类型(string/int/float/bool/json)") config_description = Column(Text, nullable=True, comment="配置描述") is_public = Column(Boolean, default=False, comment="是否公开配置") created_at = Column(DateTime, default=func.now(), comment="创建时间") updated_at = Column(DateTime, default=func.now(), onupdate=func.now(), comment="更新时间") __table_args__ = ( Index('idx_config_key', 'config_key'), Index('idx_is_public', 'is_public'), CheckConstraint("config_type IN ('string', 'int', 'float', 'bool', 'json')", name="ck_config_type"), ) class DatabaseVersion(Base): """数据库版本表""" __tablename__ = "database_versions" id = Column(Integer, primary_key=True, autoincrement=True, comment="版本ID") version = Column(String(20), unique=True, nullable=False, comment="版本号") description = Column(Text, nullable=True, comment="版本描述") applied_at = Column(DateTime, default=func.now(), comment="应用时间") migration_file = Column(String(255), nullable=True, comment="迁移文件名") __table_args__ = ( Index('idx_version', 'version'), Index('idx_applied_at', 'applied_at'), )