ai-swap/backend/modules/database/models.py

"""
数据模型定义模块
定义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'),
    )