From f7cd6eac50ad605ff4523cac20e311d341819703 Mon Sep 17 00:00:00 2001 From: jxxghp Date: Sun, 24 Aug 2025 19:17:41 +0800 Subject: [PATCH] =?UTF-8?q?feat=EF=BC=9A=E6=95=B4=E7=90=86=E6=89=8B?= =?UTF-8?q?=E5=8A=A8=E4=B8=AD=E6=AD=A2=E5=8A=9F=E8=83=BD?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- app/api/endpoints/transfer.py | 4 +++- app/core/config.py | 22 +++++++++++++++++++++- app/modules/filemanager/storages/alipan.py | 15 +++++++++------ app/modules/filemanager/storages/alist.py | 10 ++++++++-- app/modules/filemanager/storages/local.py | 22 +++++++++++++--------- app/modules/filemanager/storages/smb.py | 10 +++++++--- app/modules/filemanager/storages/u115.py | 9 +++++++-- 7 files changed, 68 insertions(+), 24 deletions(-) diff --git a/app/api/endpoints/transfer.py b/app/api/endpoints/transfer.py index ea3b10f9..e8a80040 100644 --- a/app/api/endpoints/transfer.py +++ b/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) diff --git a/app/core/config.py b/app/core/config.py index 58948a4f..c9d79022 100644 --- a/app/core/config.py +++ b/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() diff --git a/app/modules/filemanager/storages/alipan.py b/app/modules/filemanager/storages/alipan.py index 58936df0..10a6af49 100644 --- a/app/modules/filemanager/storages/alipan.py +++ b/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) diff --git a/app/modules/filemanager/storages/alist.py b/app/modules/filemanager/storages/alist.py index 95b903ee..3920b53d 100644 --- a/app/modules/filemanager/storages/alist.py +++ b/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,11 +604,14 @@ 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) if self.callback: - percent = (self.uploaded_size* 100) / self.file_size + percent = (self.uploaded_size * 100) / self.file_size self.callback(percent) return chunk diff --git a/app/modules/filemanager/storages/local.py b/app/modules/filemanager/storages/local.py index 725f742d..376d344f 100644 --- a/app/modules/filemanager/storages/local.py +++ b/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 diff --git a/app/modules/filemanager/storages/smb.py b/app/modules/filemanager/storages/smb.py index 063b2129..056b42bc 100644 --- a/app/modules/filemanager/storages/smb.py +++ b/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 diff --git a/app/modules/filemanager/storages/u115.py b/app/modules/filemanager/storages/u115.py index a149fafb..037deea1 100644 --- a/app/modules/filemanager/storages/u115.py +++ b/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