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"] )