From eeeb82ae9bcf4be43bf0deee41246cb6b3cd9270 Mon Sep 17 00:00:00 2001 From: NoahLan <6995syu@163.com> Date: Thu, 6 Nov 2025 02:02:10 +0800 Subject: [PATCH] =?UTF-8?q?fix:=20=E4=BF=AE=E5=A4=8D=E6=9C=AC=E5=9C=B0?= =?UTF-8?q?=E6=96=87=E4=BB=B6=E6=97=A0=E6=B3=95=E4=B8=8A=E4=BC=A0=E7=9A=84?= =?UTF-8?q?=E9=97=AE=E9=A2=98=EF=BC=9B=E4=BF=AE=E5=A4=8D=E7=A7=92=E4=BC=A0?= =?UTF-8?q?=E9=94=99=E8=AF=AF=E7=9A=84=E9=97=AE=E9=A2=98=E3=80=82?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- iti/applications/common/storage/local.py | 2 ++ iti/applications/service/sys_file.py | 43 +++++++++++++++--------- 2 files changed, 29 insertions(+), 16 deletions(-) diff --git a/iti/applications/common/storage/local.py b/iti/applications/common/storage/local.py index a9b94d9..0ed51b9 100644 --- a/iti/applications/common/storage/local.py +++ b/iti/applications/common/storage/local.py @@ -23,6 +23,8 @@ class LocalStorage(StorageInterface): def upload(self, file_stream: BinaryIO, key: str, mime_type: Optional[str] = None) -> Dict: """上传文件到本地""" + # 本地文件存储需要移除前缀 + key = key.lstrip(self.storage_type + ":") abs_path = self._abs_path(key) os.makedirs(os.path.dirname(abs_path), exist_ok=True) diff --git a/iti/applications/service/sys_file.py b/iti/applications/service/sys_file.py index 4844895..3e41836 100644 --- a/iti/applications/service/sys_file.py +++ b/iti/applications/service/sys_file.py @@ -7,7 +7,7 @@ from io import BytesIO from typing import Dict, Optional from flask import current_app, url_for -from sqlalchemy import select +from sqlalchemy import select, exists from iti.applications.common.enums import StatusEnum from iti.applications.common.exceptions.biz_exp import BizException @@ -40,9 +40,12 @@ class SysFileService: resolved_storage_type = cls._resolve_storage_type(storage_type, directory_id) existing = db.session.scalar( - select(SysFile) - .filter_by(file_hash=file_hash, status=StatusEnum.ENABLED.value) - .limit(1) + select( + exists(SysFile).where( + SysFile.file_hash == file_hash, + SysFile.status == StatusEnum.ENABLED.value, + ) + ) ) if existing and existing.storage_type == resolved_storage_type: # 秒传：更新已有记录 @@ -56,7 +59,9 @@ class SysFileService: ext = os.path.splitext(file.filename or "")[1] # 为支持多存储类型，前缀加上存储类型，使用冒号分隔，避免唯一索引冲突 file_key = f"{resolved_storage_type}:{datetime.now():%Y%m%d}/{file_hash}{ext}" - upload_result = storage.upload(BytesIO(file_bytes), file_key, getattr(file, "mimetype", None)) + upload_result = storage.upload( + BytesIO(file_bytes), file_key, getattr(file, "mimetype", None) + ) new_file = SysFile( filename=file.filename, @@ -175,7 +180,9 @@ class SysFileService: file_size=file_size, extension=os.path.splitext(upload_data.get("filename") or "")[1], storage_type=storage_type, - storage_info=upload_data.get("storage_info") if storage_type != "local" else None, + storage_info=upload_data.get("storage_info") + if storage_type != "local" + else None, directory_id=upload_data.get("directory_id"), metadata_=metadata if metadata else None, status=StatusEnum.ENABLED.value, @@ -250,7 +257,9 @@ class SysFileService: return storage.get_preview_url(file_obj.file_key, expires=3600) @classmethod - def get_thumbnail_url(cls, file_id: str, width: int = 200, height: int = 200, mode: str = "fit") -> Optional[str]: + def get_thumbnail_url( + cls, file_id: str, width: int = 200, height: int = 200, mode: str = "fit" + ) -> Optional[str]: """获取缩略图URL - local: 返回后端缩略图路由 @@ -278,31 +287,33 @@ class SysFileService: ) @classmethod - def get_thumbnail(cls, file_id: str, width: int = 200, height: int = 200, mode: str = "fit") -> BytesIO: + def get_thumbnail( + cls, file_id: str, width: int = 200, height: int = 200, mode: str = "fit" + ) -> BytesIO: """生成缩略图""" file_obj = cls.get_file_by_id(file_id) - + # 非图片类型不支持缩略图 if not file_obj.mime_type or not file_obj.mime_type.startswith("image/"): raise BizException("该文件类型不支持缩略图", code=400) - + # OSS 存储：使用 OSS 图片处理 if file_obj.storage_type != "local": # 对于 OSS，直接重定向到带图片处理参数的URL # TODO: 各OSS需要实现 get_thumbnail_url 方法 return cls.download_file(file_id) - + # 本地存储：使用 Pillow 生成缩略图 try: from PIL import Image from io import BytesIO as PILBytesIO except ImportError: raise BizException("需要安装 Pillow: pip install Pillow", code=500) - + # 下载原图 file_stream = cls.download_file(file_id) img = Image.open(file_stream) - + # 转换为 RGB（处理 PNG 透明通道等） if img.mode in ("RGBA", "LA", "P"): background = Image.new("RGB", img.size, (255, 255, 255)) @@ -312,7 +323,7 @@ class SysFileService: img = background elif img.mode != "RGB": img = img.convert("RGB") - + # 根据模式调整大小 if mode == "fill": # 填充模式：裁剪居中 @@ -330,12 +341,12 @@ class SysFileService: else: # fit (默认) # 适应模式：保持比例 img.thumbnail((width, height), Image.Resampling.LANCZOS) - + # 保存为 JPEG output = PILBytesIO() img.save(output, format="JPEG", quality=85, optimize=True) output.seek(0) - + return output @classmethod