Optimize storage backends and improve task cleanup performance

apps/base/tasks.py

@@ -19,37 +19,65 @@
 from core.logger import logger
 from core.settings import settings, data_root
 from core.storage import FileStorageInterface, StoredFile, storages
-from apps.base.services import stored_file_of
 from apps.base.local_share import is_local_ref
 from core.utils import get_now
 
+# 清理任务单批处理量：限制单次 ORM 实例数量与 text 大字段峰值。
+CLEANUP_BATCH_SIZE = 200
+
+
+def _prune_empty_share_dirs(root: str) -> None:
+    """删除 share/data 下的空目录（阻塞操作，调用方需放入线程池）。"""
+    for current, dirs, files in os.walk(os.path.join(str(root), "share", "data")):
+        if not dirs and not files:
+            os.rmdir(current)
+
+
 async def delete_expire_files():
     while True:
         try:
             await refresh_settings()
             file_storage: FileStorageInterface = storages[settings.file_storage]()
             # 遍历 share目录下的所有文件夹，删除空的文件夹，并判断父目录是否为空，如果为空也删除
+            # 目录遍历是阻塞的文件系统调用，下放线程池避免卡住事件循环
             if settings.file_storage == "local":
-                for root, dirs, files in os.walk(f"{data_root}/share/data"):
-                    if not dirs and not files:
-                        os.rmdir(root)
+                await asyncio.to_thread(_prune_empty_share_dirs, data_root)
             await ip_limit["error"].remove_expired_ip()
             await ip_limit["metadata"].remove_expired_ip()
             await ip_limit["upload"].remove_expired_ip()
-            await StorageReservation.filter(expires_at__lte=await get_now()).delete()
-            expire_data = await FileCodes.filter(
-                Q(expired_at__lt=await get_now()) | Q(expired_count=0)
-            ).all()
-            for exp in expire_data:
-                try:
-                    if not is_local_ref(exp):
-                        await file_storage.delete_file(stored_file_of(exp))
-                except Exception as e:
-                    logger.error(f"删除过期文件失败 code={exp.code}: {e}")
-                try:
-                    await exp.delete()
-                except Exception as e:
-                    logger.error(f"删除记录失败 code={exp.code}: {e}")
+            now = await get_now()
+            await StorageReservation.filter(expires_at__lte=now).delete()
+            # 游标分批：只取必要字段（不含 text），避免一次性载入全部过期记录；
+            # 用 id 游标而非纯 limit，单条删除失败时也不会在同一批里反复重试卡死。
+            last_id = 0
+            while True:
+                expire_data = await FileCodes.filter(
+                    (Q(expired_at__lt=now) | Q(expired_count=0)) & Q(id__gt=last_id)
+                ).only(
+                    "id", "code", "prefix", "suffix", "file_path", "uuid_file_name"
+                ).order_by("id").limit(CLEANUP_BATCH_SIZE)
+                if not expire_data:
+                    break
+                for exp in expire_data:
+                    last_id = exp.id
+                    try:
+                        if not is_local_ref(exp):
+                            # 删除只需要路径字段；不加载 text 大字段
+                            await file_storage.delete_file(
+                                StoredFile(
+                                    file_path=exp.file_path,
+                                    uuid_file_name=exp.uuid_file_name,
+                                    code=exp.code,
+                                    prefix=exp.prefix,
+                                    suffix=exp.suffix,
+                                )
+                            )
+                    except Exception as e:
+                        logger.error(f"删除过期文件失败 code={exp.code}: {e}")
+                    try:
+                        await exp.delete()
+                    except Exception as e:
+                        logger.error(f"删除记录失败 code={exp.code}: {e}")
         except Exception as e:
             logger.error(e)
         finally:
@@ -64,33 +92,41 @@ async def clean_incomplete_uploads():
             expire_hours = getattr(settings, "chunk_expire_hours", 24)
             now = await get_now()
             expire_time = now - datetime.timedelta(hours=expire_hours)
-            expired_sessions = await UploadChunk.filter(
-                chunk_index=-1, created_at__lt=expire_time
-            ).all()
-
-            for session in expired_sessions:
-                try:
-                    save_path = session.save_path
-                    if not save_path:
-                        _, _, _, _, save_path = await get_chunk_file_path_name(
-                            session.file_name, session.upload_id
+            last_id = 0
+            while True:
+                expired_sessions = await UploadChunk.filter(
+                    chunk_index=-1,
+                    created_at__lt=expire_time,
+                    id__gt=last_id,
+                ).only("id", "upload_id", "save_path", "file_name").order_by(
+                    "id"
+                ).limit(CLEANUP_BATCH_SIZE)
+                if not expired_sessions:
+                    break
+                for session in expired_sessions:
+                    last_id = session.id
+                    try:
+                        save_path = session.save_path
+                        if not save_path:
+                            _, _, _, _, save_path = await get_chunk_file_path_name(
+                                session.file_name, session.upload_id
+                            )
+                        await file_storage.clean_chunks(session.upload_id, save_path)
+                    except Exception as e:
+                        logger.error(
+                            f"清理分片文件失败 upload_id={session.upload_id}: {e}"
                         )
-                    await file_storage.clean_chunks(session.upload_id, save_path)
-                except Exception as e:
-                    logger.error(
-                        f"清理分片文件失败 upload_id={session.upload_id}: {e}"
-                    )
 
-                try:
-                    await UploadChunk.filter(upload_id=session.upload_id).delete()
-                    await StorageReservation.filter(
-                        token=f"chunk:{session.upload_id}"
-                    ).delete()
-                    logger.info(f"已清理过期上传会话 upload_id={session.upload_id}")
-                except Exception as e:
-                    logger.error(
-                        f"删除分片记录失败 upload_id={session.upload_id}: {e}"
-                    )
+                    try:
+                        await UploadChunk.filter(upload_id=session.upload_id).delete()
+                        await StorageReservation.filter(
+                            token=f"chunk:{session.upload_id}"
+                        ).delete()
+                        logger.info(f"已清理过期上传会话 upload_id={session.upload_id}")
+                    except Exception as e:
+                        logger.error(
+                            f"删除分片记录失败 upload_id={session.upload_id}: {e}"
+                        )
 
         except Exception as e:
             logger.error(f"清理未完成上传任务异常: {e}")
@@ -104,28 +140,41 @@ async def clean_expired_presign_sessions():
             await refresh_settings()
             storage: FileStorageInterface = storages[settings.file_storage]()
             now = await get_now()
-            expired_sessions = await PresignUploadSession.filter(
-                expires_at__lt=now
-            ).all()
-            for session in expired_sessions:
-                if session.mode == "direct":
-                    try:
-                        if await storage.file_exists(session.save_path):
-                            await storage.delete_file(
-                                StoredFile(
-                                    file_path=os.path.dirname(session.save_path),
-                                    uuid_file_name=os.path.basename(session.save_path),
+            last_id = 0
+            while True:
+                expired_sessions = await PresignUploadSession.filter(
+                    expires_at__lt=now, id__gt=last_id
+                ).only("id", "upload_id", "mode", "save_path").order_by(
+                    "id"
+                ).limit(CLEANUP_BATCH_SIZE)
+                if not expired_sessions:
+                    break
+                for session in expired_sessions:
+                    last_id = session.id
+                    if session.mode == "direct":
+                        try:
+                            if await storage.file_exists(session.save_path):
+                                await storage.delete_file(
+                                    StoredFile(
+                                        file_path=os.path.dirname(session.save_path),
+                                        uuid_file_name=os.path.basename(session.save_path),
+                                    )
                                 )
+                        except Exception as e:
+                            logger.error(
+                                "清理过期直传文件失败 "
+                                f"upload_id={session.upload_id}: {e}"
                             )
+                    try:
+                        await StorageReservation.filter(
+                            token=f"presign:{session.upload_id}"
+                        ).delete()
+                        await session.delete()
                     except Exception as e:
                         logger.error(
-                            "清理过期直传文件失败 "
+                            "删除过期预签名会话记录失败 "
                             f"upload_id={session.upload_id}: {e}"
                         )
-                await StorageReservation.filter(
-                    token=f"presign:{session.upload_id}"
-                ).delete()
-                await session.delete()
         except Exception as e:
             logger.error(f"清理过期预签名会话异常: {e}")
         finally:

core/storage/__init__.py

@@ -2,24 +2,93 @@
 
 Public surface is re-exported here — `from core.storage import X` keeps
 working for every historical name.
+
+Backends load lazily: importing this package only pulls in ``_base`` (the
+framework-free contracts). Selecting ``local`` therefore never imports
+aioboto3/botocore/aiohttp, which is the bulk of the resident memory on a
+small VPS. ``storages[name]()`` keeps its historical shape; backends that
+opt in via ``cache_signature()`` are also reused across requests while their
+configuration is unchanged (previously every request rebuilt e.g. an
+``aioboto3.Session``).
 """
+from __future__ import annotations
+
+from importlib import import_module
+from typing import Callable
+
 from core.storage._base import (  # noqa: F401
     FileStorageInterface,
     S3_MIN_MULTIPART_PART_SIZE,
     StoredDownload,
     StoredFile,
     build_attachment_headers,
 )
-from core.storage.local import SystemFileStorage  # noqa: F401
-from core.storage.s3 import S3FileStorage  # noqa: F401
-from core.storage.onedrive import OneDriveFileStorage  # noqa: F401
-from core.storage.opendal import OpenDALFileStorage  # noqa: F401
-from core.storage.webdav import WebDAVFileStorage  # noqa: F401
-
-storages = {
-    "local": SystemFileStorage,
-    "s3": S3FileStorage,
-    "onedrive": OneDriveFileStorage,
-    "opendal": OpenDALFileStorage,
-    "webdav": WebDAVFileStorage,
+
+# 历史导出名 -> (模块, 类名)。模块级 __getattr__ 与惰性工厂共用这张表。
+_LAZY_EXPORTS: dict[str, tuple[str, str]] = {
+    "SystemFileStorage": ("core.storage.local", "SystemFileStorage"),
+    "S3FileStorage": ("core.storage.s3", "S3FileStorage"),
+    "OneDriveFileStorage": ("core.storage.onedrive", "OneDriveFileStorage"),
+    "OpenDALFileStorage": ("core.storage.opendal", "OpenDALFileStorage"),
+    "WebDAVFileStorage": ("core.storage.webdav", "WebDAVFileStorage"),
+}
+
+
+def _load_class(module_name: str, attr: str):
+    return getattr(import_module(module_name), attr)
+
+
+class _StorageFactory:
+    """Lazily import a backend class and reuse its instance while config holds.
+
+    Call sites continue to use ``storages[name]()``. Backends exposing a
+    ``cache_signature()`` classmethod are constructed once per configuration
+    and reused; a changed signature rebuilds the instance. ``local`` and
+    ``onedrive`` deliberately omit it (see their module notes).
+    """
+
+    def __init__(self, module_name: str, attr: str):
+        self._module_name = module_name
+        self._attr = attr
+        self._cached: tuple[object, FileStorageInterface] | None = None
+
+    @property
+    def storage_class(self):
+        return _load_class(self._module_name, self._attr)
+
+    def __call__(self) -> FileStorageInterface:
+        cls = self.storage_class
+        signature_fn = getattr(cls, "cache_signature", None)
+        if signature_fn is None:
+            return cls()
+        signature = signature_fn()
+        if self._cached is not None and self._cached[0] == signature:
+            return self._cached[1]
+        instance = cls()
+        self._cached = (signature, instance)
+        return instance
+
+    def __repr__(self) -> str:
+        return f"<storage factory {self._attr}>"
+
+
+storages: dict[str, Callable[[], FileStorageInterface]] = {
+    "local": _StorageFactory("core.storage.local", "SystemFileStorage"),
+    "s3": _StorageFactory("core.storage.s3", "S3FileStorage"),
+    "onedrive": _StorageFactory("core.storage.onedrive", "OneDriveFileStorage"),
+    "opendal": _StorageFactory("core.storage.opendal", "OpenDALFileStorage"),
+    "webdav": _StorageFactory("core.storage.webdav", "WebDAVFileStorage"),
 }
+
+
+def __getattr__(name: str):
+    target = _LAZY_EXPORTS.get(name)
+    if target is None:
+        raise AttributeError(f"module {__name__!r} has no attribute {name!r}")
+    value = _load_class(*target)
+    globals()[name] = value  # 缓存解析结果，后续访问不再经过 __getattr__
+    return value
+
+
+def __dir__() -> list[str]:
+    return sorted(set(globals()) | set(_LAZY_EXPORTS))

core/storage/_base.py

@@ -86,6 +86,17 @@ def _get_chunk_record(chunk_records: dict, index: int):
             raise ValueError(f"分片{index}记录不存在")
         return chunk_record
 
+    def _verify_chunk_hash(self, index: int, chunk_record, current_hash: str) -> None:
+        """Compare a chunk digest with its record using the shared error wording.
+
+        Streaming backends hash incrementally and call this once per chunk, so
+        the message stays byte-identical to the in-memory path.
+        """
+        if current_hash != chunk_record.chunk_hash:
+            raise ValueError(
+                f"分片{index}哈希不匹配: 期望 {chunk_record.chunk_hash}, 实际 {current_hash}"
+            )
+
     def _verify_and_hash_chunk(
         self,
         index: int,
@@ -95,11 +106,9 @@ def _verify_and_hash_chunk(
     ) -> None:
         """Verify a chunk against its recorded hash, then stripe it into the
         whole-file digest. Raises the shared ValueError wording on mismatch."""
-        current_hash = hashlib.sha256(chunk_data).hexdigest()
-        if current_hash != chunk_record.chunk_hash:
-            raise ValueError(
-                f"分片{index}哈希不匹配: 期望 {chunk_record.chunk_hash}, 实际 {current_hash}"
-            )
+        self._verify_chunk_hash(
+            index, chunk_record, hashlib.sha256(chunk_data).hexdigest()
+        )
         file_sha256.update(chunk_data)
 
     async def save_file(