iTi-Flask/iti/applications/models/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="扩展元数据")

    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"

    name = db.Column(db.String(255), nullable=False, comment="目录名称")
    path = db.Column(db.String(1024), nullable=False, index=True, 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="目录描述")
    default_storage_type = db.Column(
        db.String(32), 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")
    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 ..service.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)
                data["thumbnailUrl"] = SysFileService.get_thumbnail_url(file_id)
            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()
    default_storage_type = 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"]
    )