feat：整理手动中止功能

app/api/endpoints/transfer.py

@@ -8,7 +8,7 @@ from app import schemas
 from app.chain.media import MediaChain
 from app.chain.storage import StorageChain
 from app.chain.transfer import TransferChain
-from app.core.config import settings
+from app.core.config import settings, global_vars
 from app.core.metainfo import MetaInfoPath
 from app.core.security import verify_token, verify_apitoken
 from app.db import get_db
@@ -75,6 +75,8 @@ async def remove_queue(fileitem: schemas.FileItem, _: schemas.TokenPayload = Dep
     :param _: Token校验
     """
     TransferChain().remove_from_queue(fileitem)
+    # 取消整理
+    global_vars.stop_transfer(fileitem.path)
     return schemas.Response(success=True)
 
 

app/core/config.py

@@ -798,6 +798,8 @@ class GlobalVar(object):
     SUBSCRIPTIONS: List[dict] = []
     # 需应急停止的工作流
     EMERGENCY_STOP_WORKFLOWS: List[int] = []
+    # 需应急停止文件整理
+    EMERGENCY_STOP_TRANSFER: List[str] = []
 
     def stop_system(self):
         """
@@ -838,12 +840,30 @@ class GlobalVar(object):
         if workflow_id in self.EMERGENCY_STOP_WORKFLOWS:
             self.EMERGENCY_STOP_WORKFLOWS.remove(workflow_id)
 
-    def is_workflow_stopped(self, workflow_id: int):
+    def is_workflow_stopped(self, workflow_id: int) -> bool:
         """
         是否停止工作流
         """
         return self.is_system_stopped or workflow_id in self.EMERGENCY_STOP_WORKFLOWS
 
+    def stop_transfer(self, path: str):
+        """
+        停止文件整理
+        """
+        if path not in self.EMERGENCY_STOP_TRANSFER:
+            self.EMERGENCY_STOP_TRANSFER.append(path)
+
+    def is_transfer_stopped(self, path: str) -> bool:
+        """
+        是否停止文件整理
+        """
+        if self.is_system_stopped:
+            return True
+        if path in self.EMERGENCY_STOP_TRANSFER:
+            self.EMERGENCY_STOP_TRANSFER.remove(path)
+            return True
+        return False
+
 
 # 全局标识
 global_vars = GlobalVar()

app/modules/filemanager/storages/alipan.py

@@ -9,7 +9,7 @@ from typing import List, Optional, Tuple, Union
 import requests
 
 from app import schemas
-from app.core.config import settings
+from app.core.config import settings, global_vars
 from app.log import logger
 from app.modules.filemanager import StorageBase
 from app.modules.filemanager.storages import transfer_process
@@ -628,9 +628,12 @@ class AliPan(StorageBase, metaclass=WeakSingleton):
         uploaded_size = 0
         with open(local_path, 'rb') as f:
             for part_info in part_info_list:
-                part_num = part_info['part_number']
+                if global_vars.is_transfer_stopped(local_path.as_posix()):
+                    logger.info(f"【阿里云盘】{target_name} 上传已取消！")
+                    return None
 
                 # 计算分片参数
+                part_num = part_info['part_number']
                 start = (part_num - 1) * chunk_size
                 end = min(start + chunk_size, file_size)
                 current_chunk_size = end - start
@@ -656,7 +659,6 @@ class AliPan(StorageBase, metaclass=WeakSingleton):
                             upload_url = new_urls[0]['upload_url']
                         else:
                             upload_url = part_info['upload_url']
-
                         # 执行上传
                         logger.info(
                             f"【阿里云盘】开始 第{attempt + 1}次 上传 {target_name} 分片 {part_num} ...")
@@ -725,14 +727,15 @@ class AliPan(StorageBase, metaclass=WeakSingleton):
             with requests.get(download_url, stream=True) as r:
                 r.raise_for_status()
                 downloaded_size = 0
-
                 with open(local_path, "wb") as f:
                     for chunk in r.iter_content(chunk_size=self.chunk_size):
+                        if global_vars.is_transfer_stopped(fileitem.path):
+                            logger.info(f"【阿里云盘】{fileitem.path} 下载已取消！")
+                            return None
                         if chunk:
                             f.write(chunk)
-                            downloaded_size += len(chunk)
-
                             # 更新进度
+                            downloaded_size += len(chunk)
                             if file_size:
                                 progress = (downloaded_size * 100) / file_size
                                 progress_callback(progress)

app/modules/filemanager/storages/alist.py

@@ -7,7 +7,7 @@ import requests
 
 from app import schemas
 from app.core.cache import cached
-from app.core.config import settings
+from app.core.config import settings, global_vars
 from app.log import logger
 from app.modules.filemanager.storages import StorageBase, transfer_process
 from app.schemas.types import StorageSchema
@@ -561,6 +561,9 @@ class Alist(StorageBase, metaclass=WeakSingleton):
             r.raise_for_status()
             with open(local_path, "wb") as f:
                 for chunk in r.iter_content(chunk_size=8192):
+                    if global_vars.is_transfer_stopped(fileitem.path):
+                        logger.info(f"【OpenList】{fileitem.path} 下载已取消！")
+                        return None
                     f.write(chunk)
 
         if local_path.exists():
@@ -601,6 +604,9 @@ class Alist(StorageBase, metaclass=WeakSingleton):
                     self.file_size = file_path.stat().st_size
 
                 def read(self, size=-1):
+                    if global_vars.is_transfer_stopped(path.as_posix()):
+                        logger.info(f"【OpenList】{path} 上传已取消！")
+                        return None
                     chunk = self.file.read(size)
                     if chunk:
                         self.uploaded_size += len(chunk)

app/modules/filemanager/storages/local.py

@@ -3,6 +3,7 @@ from pathlib import Path
 from typing import Optional, List
 
 from app import schemas
+from app.core.config import global_vars
 from app.helper.directory import DirectoryHelper
 from app.log import logger
 from app.modules.filemanager.storages import StorageBase, transfer_process
@@ -98,7 +99,7 @@ class LocalStorage(StorageBase):
         # 遍历目录
         path_obj = Path(path)
         if not path_obj.exists():
-            logger.warn(f"【local】目录不存在：{path}")
+            logger.warn(f"【本地】目录不存在：{path}")
             return []
 
         # 如果是文件
@@ -170,7 +171,7 @@ class LocalStorage(StorageBase):
             else:
                 shutil.rmtree(path_obj, ignore_errors=True)
         except Exception as e:
-            logger.error(f"【local】删除文件失败：{e}")
+            logger.error(f"【本地】删除文件失败：{e}")
             return False
         return True
 
@@ -184,7 +185,7 @@ class LocalStorage(StorageBase):
         try:
             path_obj.rename(path_obj.parent / name)
         except Exception as e:
-            logger.error(f"【local】重命名文件失败：{e}")
+            logger.error(f"【本地】重命名文件失败：{e}")
             return False
         return True
 
@@ -204,6 +205,9 @@ class LocalStorage(StorageBase):
         try:
             with open(src, "rb") as fsrc, open(dest, "wb") as fdst:
                 while True:
+                    if global_vars.is_transfer_stopped(src.as_posix()):
+                        logger.info(f"【本地】{src} 复制已取消！")
+                        return False
                     buf = fsrc.read(self.chunk_size)
                     if not buf:
                         break
@@ -217,7 +221,7 @@ class LocalStorage(StorageBase):
             shutil.copystat(src, dest)
             return True
         except Exception as e:
-            logger.error(f"【local】复制文件 {src} 失败：{e}")
+            logger.error(f"【本地】复制文件 {src} 失败：{e}")
             return False
         finally:
             progress_callback(100)
@@ -239,7 +243,7 @@ class LocalStorage(StorageBase):
                 path.unlink()
                 return self.get_item(target_path)
         except Exception as err:
-            logger.error(f"【local】移动文件失败：{err}")
+            logger.error(f"【本地】移动文件失败：{err}")
         return None
 
     def copy(
@@ -257,7 +261,7 @@ class LocalStorage(StorageBase):
             if self._copy_with_progress(src, dest):
                 return True
         except Exception as err:
-            logger.error(f"【local】复制文件失败：{err}")
+            logger.error(f"【本地】复制文件失败：{err}")
         return False
 
     def move(
@@ -277,7 +281,7 @@ class LocalStorage(StorageBase):
                 src.unlink()
                 return True
         except Exception as err:
-            logger.error(f"【local】移动文件失败：{err}")
+            logger.error(f"【本地】移动文件失败：{err}")
         return False
 
     def link(self, fileitem: schemas.FileItem, target_file: Path) -> bool:
@@ -287,7 +291,7 @@ class LocalStorage(StorageBase):
         file_path = Path(fileitem.path)
         code, message = SystemUtils.link(file_path, target_file)
         if code != 0:
-            logger.error(f"【local】硬链接文件失败：{message}")
+            logger.error(f"【本地】硬链接文件失败：{message}")
             return False
         return True
 
@@ -298,7 +302,7 @@ class LocalStorage(StorageBase):
         file_path = Path(fileitem.path)
         code, message = SystemUtils.softlink(file_path, target_file)
         if code != 0:
-            logger.error(f"【local】软链接文件失败：{message}")
+            logger.error(f"【本地】软链接文件失败：{message}")
             return False
         return True
 

app/modules/filemanager/storages/smb.py

@@ -8,7 +8,7 @@ from smbclient import ClientConfig, register_session, reset_connection_cache
 from smbprotocol.exceptions import SMBException, SMBResponseException, SMBAuthenticationError
 
 from app import schemas
-from app.core.config import settings
+from app.core.config import settings, global_vars
 from app.log import logger
 from app.modules.filemanager import StorageBase
 from app.modules.filemanager.storages import transfer_process
@@ -438,12 +438,14 @@ class SMB(StorageBase, metaclass=WeakSingleton):
                 with open(local_path, "wb") as dst_file:
                     downloaded_size = 0
                     while True:
+                        if global_vars.is_transfer_stopped(fileitem.path):
+                            logger.info(f"【SMB】{fileitem.path} 下载已取消！")
+                            return None
                         chunk = src_file.read(self.chunk_size)
                         if not chunk:
                             break
                         dst_file.write(chunk)
                         downloaded_size += len(chunk)
-
                         # 更新进度
                         if file_size:
                             progress = (downloaded_size * 100) / file_size
@@ -485,12 +487,14 @@ class SMB(StorageBase, metaclass=WeakSingleton):
                 with smbclient.open_file(smb_path, mode="wb") as dst_file:
                     uploaded_size = 0
                     while True:
+                        if global_vars.is_transfer_stopped(path.as_posix()):
+                            logger.info(f"【SMB】{path} 上传已取消！")
+                            return None
                         chunk = src_file.read(self.chunk_size)
                         if not chunk:
                             break
                         dst_file.write(chunk)
                         uploaded_size += len(chunk)
-
                         # 更新进度
                         if file_size:
                             progress = (uploaded_size * 100) / file_size

app/modules/filemanager/storages/u115.py

@@ -12,7 +12,7 @@ from oss2 import SizedFileAdapter, determine_part_size
 from oss2.models import PartInfo
 
 from app import schemas
-from app.core.config import settings
+from app.core.config import settings, global_vars
 from app.log import logger
 from app.modules.filemanager import StorageBase
 from app.modules.filemanager.storages import transfer_process
@@ -532,6 +532,9 @@ class U115Pan(StorageBase, metaclass=WeakSingleton):
             part_number = 1
             offset = 0
             while offset < file_size:
+                if global_vars.is_transfer_stopped(local_path.as_posix()):
+                    logger.info(f"【115】{local_path} 上传已取消！")
+                    return None
                 num_to_upload = min(part_size, file_size - offset)
                 # 调用SizedFileAdapter(fileobj, size)方法会生成一个新的文件对象，重新计算起始追加位置。
                 logger.info(f"【115】开始上传 {target_name} 分片 {part_number}: {offset} -> {offset + num_to_upload}")
@@ -614,10 +617,12 @@ class U115Pan(StorageBase, metaclass=WeakSingleton):
 
                 with open(local_path, "wb") as f:
                     for chunk in r.iter_content(chunk_size=self.chunk_size):
+                        if global_vars.is_transfer_stopped(fileitem.path):
+                            logger.info(f"【115】{fileitem.path} 下载已取消！")
+                            return None
                         if chunk:
                             f.write(chunk)
                             downloaded_size += len(chunk)
-
                             # 更新进度
                             if file_size:
                                 progress = (downloaded_size * 100) / file_size