iTi-Flask/iti/applications/models/sys/sys_file.py

from iti.applications.extensions import db
from iti.applications.common.crud import BaseModelMixin
from iti.applications.common.enums import StatusEnum, StorageTypeEnum
from iti.applications.common.utils import BaseSchema
from apiflask.fields import String, DateTime, Enum, Integer, Dict as DictField, Nested
from marshmallow import post_dump


class SysFile(BaseModelMixin):
    """文件记录表"""

    __tablename__ = "sys_file"

    filename = db.Column(db.String(255), nullable=False, comment="原始文件名")
    file_key = db.Column(
        db.String(512), nullable=False, unique=True, index=True, comment="存储路径"
    )
    file_hash = db.Column(db.String(64), nullable=True, index=True, comment="文件哈希")

    mime_type = db.Column(db.String(128), nullable=True, comment="MIME类型")
    file_size = db.Column(db.BigInteger, nullable=False, comment="文件大小(字节)")
    extension = db.Column(db.String(32), nullable=True, comment="文件扩展名")

    storage_type = db.Column(
        db.Enum(StorageTypeEnum, values_callable=lambda x: [e.value for e in x]),
        nullable=False,
        default=StorageTypeEnum.LOCAL.value,
        comment="存储类型",
    )
    storage_info = db.Column(db.JSON, nullable=True, comment="存储信息(bucket/region/endpoint/meta等)")

    directory_id = db.Column(db.String(36), nullable=True, index=True, comment="所属目录ID")

    metadata_ = db.Column("metadata", db.JSON, nullable=True, comment="扩展元数据")

    # 回收站相关
    is_deleted = db.Column(db.Boolean, default=False, nullable=False, comment="是否已删除（回收站）")
    deleted_at = db.Column(db.DateTime, nullable=True, comment="删除时间")
    deleted_by = db.Column(db.String(36), nullable=True, comment="删除人ID")

    # 分享相关
    share_code = db.Column(db.String(64), nullable=True, unique=True, index=True, comment="分享码")
    share_password = db.Column(db.String(64), nullable=True, comment="分享密码")
    share_expire_at = db.Column(db.DateTime, nullable=True, comment="分享过期时间")
    share_count = db.Column(db.Integer, default=0, nullable=False, comment="分享访问次数")

    status = db.Column(
        db.Enum(StatusEnum, values_callable=lambda x: [e.value for e in x]),
        nullable=False,
        default=StatusEnum.ENABLED.value,
        comment="状态",
    )

    # 关系
    directory = db.relationship(
        "SysFileDirectory",
        primaryjoin="SysFile.directory_id == SysFileDirectory.id",
        foreign_keys="SysFile.directory_id",
        uselist=False,
        lazy="noload",
    )


class SysFileDirectory(BaseModelMixin):
    """文件目录"""

    __tablename__ = "sys_file_directory"
    __table_args__ = (
        db.Index('ix_sys_file_directory_path', 'path', mysql_length=255),
    )

    name = db.Column(db.String(255), nullable=False, comment="目录名称")
    path = db.Column(db.String(1024), nullable=False, comment="完整路径")
    parent_id = db.Column(db.String(36), nullable=True, comment="父目录ID")
    level = db.Column(db.Integer, default=0, comment="层级")
    sort = db.Column(db.Integer, default=0, comment="排序")
    icon = db.Column(db.String(128), nullable=True, comment="目录图标")
    color = db.Column(db.String(32), nullable=True, comment="颜色标记")
    description = db.Column(db.Text, nullable=True, comment="目录描述")
    status = db.Column(
        db.Enum(StatusEnum, values_callable=lambda x: [e.value for e in x]),
        nullable=False,
        default=StatusEnum.ENABLED.value,
        comment="状态",
    )

    # 关系
    parent = db.relationship(
        "SysFileDirectory",
        primaryjoin="SysFileDirectory.parent_id == SysFileDirectory.id",
        foreign_keys="SysFileDirectory.parent_id",
        uselist=False,
        remote_side="SysFileDirectory.id",
        lazy="noload",
    )
    children = db.relationship(
        "SysFileDirectory",
        primaryjoin="SysFileDirectory.id == SysFileDirectory.parent_id",
        foreign_keys="SysFileDirectory.parent_id",
        uselist=True,
        lazy="noload",
        viewonly=True,
    )
    files = db.relationship(
        "SysFile",
        primaryjoin="SysFileDirectory.id == SysFile.directory_id",
        foreign_keys="SysFile.directory_id",
        uselist=True,
        lazy="noload",
        viewonly=True,
    )


class SysFileSchema(BaseSchema):
    id = String()
    filename = String()
    file_key = String()
    file_hash = String()
    mime_type = String()
    file_size = Integer()
    extension = String()
    storage_type = Enum(StorageTypeEnum, by_value=True)
    storage_info = DictField()
    directory_id = String()
    metadata_ = DictField(data_key="metadata")
    
    # 回收站相关
    is_deleted = String()
    deleted_at = DateTime(format="%Y-%m-%d %H:%M:%S")
    deleted_by = String()
    
    # 分享相关
    share_code = String()
    share_password = String()
    share_expire_at = DateTime(format="%Y-%m-%d %H:%M:%S")
    share_count = Integer()
    
    status = Enum(StatusEnum, by_value=True)
    created_at = DateTime(format="%Y-%m-%d %H:%M:%S")
    updated_at = DateTime(format="%Y-%m-%d %H:%M:%S")

    url = String(dump_only=True)
    preview_url = String(dump_only=True)
    thumbnail_url = String(dump_only=True)
    file_category = String(dump_only=True)
    file_size_text = String(dump_only=True)

    @post_dump
    def flatten_metadata(self, data, **kwargs):
        metadata = data.get("metadata") or {}
        for key, value in metadata.items():
            data.setdefault(key, value)

        from iti.applications.service.sys.sys_file import SysFileService

        file_id = data.get("id")
        if file_id:
            try:
                data["url"] = SysFileService.get_file_url(file_id)
                data["previewUrl"] = SysFileService.get_preview_url(file_id)
                # thumbnailUrl 不附带参数，由前端自行添加
                data["thumbnailUrl"] = SysFileService.get_thumbnail_url(file_id, include_params=False)
            except Exception:
                data["url"] = None
                data["previewUrl"] = None
                data["thumbnailUrl"] = None

        # 文件分类
        mime_type = data.get("mimeType", "") or data.get("mime_type", "") or ""
        if mime_type.startswith("image/"):
            data["fileCategory"] = "image"
        elif mime_type.startswith("video/"):
            data["fileCategory"] = "video"
        elif mime_type.startswith("audio/"):
            data["fileCategory"] = "audio"
        elif "pdf" in mime_type or "word" in mime_type or "document" in mime_type:
            data["fileCategory"] = "document"
        elif "zip" in mime_type or "rar" in mime_type or "7z" in mime_type:
            data["fileCategory"] = "archive"
        else:
            data["fileCategory"] = "other"

        # 文件大小文本
        file_size = data.get("fileSize") or data.get("file_size") or 0
        try:
            file_size = int(file_size)
        except (TypeError, ValueError):
            file_size = 0
            
        if file_size < 1024:
            data["fileSizeText"] = f"{file_size} B"
        elif file_size < 1024 * 1024:
            data["fileSizeText"] = f"{file_size / 1024:.2f} KB"
        elif file_size < 1024 * 1024 * 1024:
            data["fileSizeText"] = f"{file_size / (1024 * 1024):.2f} MB"
        else:
            data["fileSizeText"] = f"{file_size / (1024 * 1024 * 1024):.2f} GB"

        return data


class SysFileDirectorySchema(BaseSchema):
    id = String()
    name = String()
    path = String()
    parent_id = String()
    level = Integer()
    sort = Integer()
    icon = String()
    color = String()
    description = String()
    status = Enum(StatusEnum, by_value=True)
    created_at = DateTime(format="%Y-%m-%d %H:%M:%S")
    updated_at = DateTime(format="%Y-%m-%d %H:%M:%S")
    children = Nested(
        "SysFileDirectorySchema", many=True, dump_only=True, exclude=["children"]
    )