rollback size

Co-authored-by: jxxghp <jxxghp@163.com>

app/actions/note.py

@@ -0,0 +1,30 @@
+from app.actions import BaseAction
+from app.schemas import ActionContext
+
+
+class NoteAction(BaseAction):
+    """
+    备注
+    """
+
+    @classmethod
+    @property
+    def name(cls) -> str: # noqa
+        return "备注"
+
+    @classmethod
+    @property
+    def description(cls) -> str: # noqa
+        return "给工作流添加备注"
+
+    @classmethod
+    @property
+    def data(cls) -> dict: # noqa
+        return {}
+
+    @property
+    def success(self) -> bool:
+        return True
+
+    def execute(self, workflow_id: int, params: dict, context: ActionContext) -> ActionContext:
+        return context

app/api/endpoints/system.py

@@ -11,6 +11,7 @@ from typing import Optional, Union, Annotated
 import aiofiles
 import pillow_avif  # noqa 用于自动注册AVIF支持
 from PIL import Image
+from app.helper.sites import SitesHelper
 from fastapi import APIRouter, Body, Depends, HTTPException, Header, Request, Response
 from fastapi.responses import StreamingResponse
 
@@ -18,10 +19,10 @@ from app import schemas
 from app.chain.search import SearchChain
 from app.chain.system import SystemChain
 from app.core.config import global_vars, settings
+from app.core.event import eventmanager
 from app.core.metainfo import MetaInfo
 from app.core.module import ModuleManager
 from app.core.security import verify_apitoken, verify_resource_token, verify_token
-from app.core.event import eventmanager
 from app.db.models import User
 from app.db.systemconfig_oper import SystemConfigOper
 from app.db.user_oper import get_current_active_superuser
@@ -29,7 +30,6 @@ from app.helper.mediaserver import MediaServerHelper
 from app.helper.message import MessageHelper
 from app.helper.progress import ProgressHelper
 from app.helper.rule import RuleHelper
-from app.helper.sites import SitesHelper
 from app.helper.subscribe import SubscribeHelper
 from app.helper.system import SystemHelper
 from app.log import logger
@@ -187,9 +187,11 @@ def get_global_setting(token: str):
                  "COOKIECLOUD_KEY", "COOKIECLOUD_PASSWORD", "GITHUB_TOKEN", "REPO_GITHUB_TOKEN"}
     )
     # 追加用户唯一ID和订阅分享管理权限
+    share_admin = SubscribeHelper().is_admin_user()
     info.update({
         "USER_UNIQUE_ID": SubscribeHelper().get_user_uuid(),
-        "SUBSCRIBE_SHARE_MANAGE": SubscribeHelper().is_admin_user(),
+        "SUBSCRIBE_SHARE_MANAGE": share_admin,
+        "WORKFLOW_SHARE_MANAGE": share_admin
     })
     return schemas.Response(success=True,
                             data=info)
@@ -290,9 +292,9 @@ def get_setting(key: str,
 
 @router.post("/setting/{key}", summary="更新系统设置", response_model=schemas.Response)
 def set_setting(
-    key: str,
-    value: Annotated[Union[list, dict, bool, int, str] | None, Body()] = None,
-    _: User = Depends(get_current_active_superuser),
+        key: str,
+        value: Annotated[Union[list, dict, bool, int, str] | None, Body()] = None,
+        _: User = Depends(get_current_active_superuser),
 ):
     """
     更新系统设置（仅管理员）
@@ -452,10 +454,10 @@ def ruletest(title: str,
 
 @router.get("/nettest", summary="测试网络连通性")
 def nettest(
-    url: str,
-    proxy: bool,
-    include: Optional[str] = None,
-    _: schemas.TokenPayload = Depends(verify_token),
+        url: str,
+        proxy: bool,
+        include: Optional[str] = None,
+        _: schemas.TokenPayload = Depends(verify_token),
 ):
     """
     测试网络连通性

app/api/endpoints/transfer.py

@@ -8,11 +8,13 @@ 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.metainfo import MetaInfoPath
 from app.core.security import verify_token, verify_apitoken
 from app.db import get_db
 from app.db.models.transferhistory import TransferHistory
 from app.db.user_oper import get_current_active_superuser
+from app.helper.directory import DirectoryHelper
 from app.schemas import MediaType, FileItem, ManualTransferItem
 
 router = APIRouter()
@@ -35,11 +37,23 @@ def query_name(path: str, filetype: str,
     if not new_path:
         return schemas.Response(success=False, message="未识别到新名称")
     if filetype == "dir":
-        parents = Path(new_path).parents
-        if len(parents) > 2:
-            new_name = parents[1].name
+        media_path = DirectoryHelper.get_media_root_path(
+            rename_format=(
+                settings.TV_RENAME_FORMAT
+                if mediainfo.type == MediaType.TV
+                else settings.MOVIE_RENAME_FORMAT
+            ),
+            rename_path=Path(new_path),
+        )
+        if media_path:
+            new_name = media_path.name
         else:
-            new_name = parents[0].name
+            # fallback
+            parents = Path(new_path).parents
+            if len(parents) > 2:
+                new_name = parents[1].name
+            else:
+                new_name = parents[0].name
     else:
         new_name = Path(new_path).name
     return schemas.Response(success=True, data={

app/api/endpoints/workflow.py

@@ -1,3 +1,4 @@
+import json
 from datetime import datetime
 from typing import List, Any, Optional
 
@@ -5,14 +6,15 @@ from fastapi import APIRouter, Depends
 from sqlalchemy.orm import Session
 
 from app import schemas
+from app.chain.workflow import WorkflowChain
 from app.core.config import global_vars
 from app.core.plugin import PluginManager
 from app.core.workflow import WorkFlowManager
 from app.db import get_db
-from app.db.models.workflow import Workflow
+from app.db.models import Workflow
 from app.db.systemconfig_oper import SystemConfigOper
 from app.db.user_oper import get_current_active_user
-from app.chain.workflow import WorkflowChain
+from app.helper.workflow import WorkflowHelper
 from app.scheduler import Scheduler
 
 router = APIRouter()
@@ -24,7 +26,8 @@ def list_workflows(db: Session = Depends(get_db),
     """
     获取工作流列表
     """
-    return Workflow.list(db)
+    from app.db.workflow_oper import WorkflowOper
+    return WorkflowOper(db).list()
 
 
 @router.post("/", summary="创建工作流", response_model=schemas.Response)
@@ -34,13 +37,15 @@ def create_workflow(workflow: schemas.Workflow,
     """
     创建工作流
     """
-    if Workflow.get_by_name(db, workflow.name):
+    from app.db.workflow_oper import WorkflowOper
+    if workflow.name and WorkflowOper(db).get_by_name(workflow.name):
         return schemas.Response(success=False, message="已存在相同名称的工作流")
     if not workflow.add_time:
         workflow.add_time = datetime.strftime(datetime.now(), "%Y-%m-%d %H:%M:%S")
     if not workflow.state:
         workflow.state = "P"
-    Workflow(**workflow.dict()).create(db)
+    from app.db.models.workflow import Workflow as WorkflowModel
+    WorkflowModel(**workflow.dict()).create(db)
     return schemas.Response(success=True, message="创建工作流成功")
 
 
@@ -60,47 +65,97 @@ def list_actions(_: schemas.TokenPayload = Depends(get_current_active_user)) ->
     return WorkFlowManager().list_actions()
 
 
-@router.get("/{workflow_id}", summary="工作流详情", response_model=schemas.Workflow)
-def get_workflow(workflow_id: int,
-                 db: Session = Depends(get_db),
-                 _: schemas.TokenPayload = Depends(get_current_active_user)) -> Any:
+@router.post("/share", summary="分享工作流", response_model=schemas.Response)
+def workflow_share(
+        workflow: schemas.WorkflowShare,
+        _: schemas.TokenPayload = Depends(get_current_active_user)) -> Any:
     """
-    获取工作流详情
+    分享工作流
     """
-    return Workflow.get(db, workflow_id)
+    if not workflow.id or not workflow.share_title or not workflow.share_user:
+        return schemas.Response(success=False, message="请填写工作流ID、分享标题和分享人")
+
+    state, errmsg = WorkflowHelper().workflow_share(workflow_id=workflow.id,
+                                                    share_title=workflow.share_title or "",
+                                                    share_comment=workflow.share_comment or "",
+                                                    share_user=workflow.share_user or "")
+    return schemas.Response(success=state, message=errmsg)
 
 
-@router.put("/{workflow_id}", summary="更新工作流", response_model=schemas.Response)
-def update_workflow(workflow: schemas.Workflow,
-                    db: Session = Depends(get_db),
-                    _: schemas.TokenPayload = Depends(get_current_active_user)) -> Any:
+@router.delete("/share/{share_id}", summary="删除分享", response_model=schemas.Response)
+def workflow_share_delete(
+        share_id: int,
+        _: schemas.TokenPayload = Depends(get_current_active_user)) -> Any:
     """
-    更新工作流
+    删除分享
     """
-    wf = Workflow.get(db, workflow.id)
-    if not wf:
-        return schemas.Response(success=False, message="工作流不存在")
-    wf.update(db, workflow.dict())
-    return schemas.Response(success=True, message="更新成功")
+    state, errmsg = WorkflowHelper().share_delete(share_id=share_id)
+    return schemas.Response(success=state, message=errmsg)
 
 
-@router.delete("/{workflow_id}", summary="删除工作流", response_model=schemas.Response)
-def delete_workflow(workflow_id: int,
-                    db: Session = Depends(get_db),
-                    _: schemas.TokenPayload = Depends(get_current_active_user)) -> Any:
+@router.post("/fork", summary="复用工作流", response_model=schemas.Response)
+def workflow_fork(
+        workflow: schemas.WorkflowShare,
+        db: Session = Depends(get_db),
+        _: schemas.User = Depends(get_current_active_user)) -> Any:
     """
-    删除工作流
+    复用工作流
     """
-    workflow = Workflow.get(db, workflow_id)
-    if not workflow:
-        return schemas.Response(success=False, message="工作流不存在")
-    # 删除定时任务
-    Scheduler().remove_workflow_job(workflow)
-    # 删除工作流
-    Workflow.delete(db, workflow_id)
-    # 删除缓存
-    SystemConfigOper().delete(f"WorkflowCache-{workflow_id}")
-    return schemas.Response(success=True, message="删除成功")
+    if not workflow.name:
+        return schemas.Response(success=False, message="工作流名称不能为空")
+
+    # 解析JSON数据，添加错误处理
+    try:
+        actions = json.loads(workflow.actions or "[]")
+    except json.JSONDecodeError:
+        return schemas.Response(success=False, message="actions字段JSON格式错误")
+
+    try:
+        flows = json.loads(workflow.flows or "[]")
+    except json.JSONDecodeError:
+        return schemas.Response(success=False, message="flows字段JSON格式错误")
+
+    try:
+        context = json.loads(workflow.context or "{}")
+    except json.JSONDecodeError:
+        return schemas.Response(success=False, message="context字段JSON格式错误")
+
+    # 创建工作流
+    workflow_dict = {
+        "name": workflow.name,
+        "description": workflow.description,
+        "timer": workflow.timer,
+        "actions": actions,
+        "flows": flows,
+        "context": context,
+        "state": "P"  # 默认暂停状态
+    }
+
+    # 检查名称是否重复
+    if Workflow.get_by_name(db, workflow_dict["name"]):
+        return schemas.Response(success=False, message="已存在相同名称的工作流")
+
+    # 创建新工作流
+    workflow = Workflow(**workflow_dict)
+    workflow.create(db)
+
+    # 更新复用次数
+    if workflow.id:
+        WorkflowHelper().workflow_fork(share_id=workflow.id)
+
+    return schemas.Response(success=True, message="复用成功")
+
+
+@router.get("/shares", summary="查询分享的工作流", response_model=List[schemas.WorkflowShare])
+def workflow_shares(
+        name: Optional[str] = None,
+        page: Optional[int] = 1,
+        count: Optional[int] = 30,
+        _: schemas.TokenPayload = Depends(get_current_active_user)) -> Any:
+    """
+    查询分享的工作流
+    """
+    return WorkflowHelper().get_shares(name=name, page=page, count=count)
 
 
 @router.post("/{workflow_id}/run", summary="执行工作流", response_model=schemas.Response)
@@ -123,7 +178,8 @@ def start_workflow(workflow_id: int,
     """
     启用工作流
     """
-    workflow = Workflow.get(db, workflow_id)
+    from app.db.workflow_oper import WorkflowOper
+    workflow = WorkflowOper(db).get(workflow_id)
     if not workflow:
         return schemas.Response(success=False, message="工作流不存在")
     # 添加定时任务
@@ -140,7 +196,8 @@ def pause_workflow(workflow_id: int,
     """
     停用工作流
     """
-    workflow = Workflow.get(db, workflow_id)
+    from app.db.workflow_oper import WorkflowOper
+    workflow = WorkflowOper(db).get(workflow_id)
     if not workflow:
         return schemas.Response(success=False, message="工作流不存在")
     # 删除定时任务
@@ -159,7 +216,8 @@ def reset_workflow(workflow_id: int,
     """
     重置工作流
     """
-    workflow = Workflow.get(db, workflow_id)
+    from app.db.workflow_oper import WorkflowOper
+    workflow = WorkflowOper(db).get(workflow_id)
     if not workflow:
         return schemas.Response(success=False, message="工作流不存在")
     # 停止工作流
@@ -169,3 +227,52 @@ def reset_workflow(workflow_id: int,
     # 删除缓存
     SystemConfigOper().delete(f"WorkflowCache-{workflow_id}")
     return schemas.Response(success=True)
+
+
+@router.get("/{workflow_id}", summary="工作流详情", response_model=schemas.Workflow)
+def get_workflow(workflow_id: int,
+                 db: Session = Depends(get_db),
+                 _: schemas.TokenPayload = Depends(get_current_active_user)) -> Any:
+    """
+    获取工作流详情
+    """
+    from app.db.workflow_oper import WorkflowOper
+    return WorkflowOper(db).get(workflow_id)
+
+
+@router.put("/{workflow_id}", summary="更新工作流", response_model=schemas.Response)
+def update_workflow(workflow: schemas.Workflow,
+                    db: Session = Depends(get_db),
+                    _: schemas.TokenPayload = Depends(get_current_active_user)) -> Any:
+    """
+    更新工作流
+    """
+    from app.db.workflow_oper import WorkflowOper
+    if not workflow.id:
+        return schemas.Response(success=False, message="工作流ID不能为空")
+    wf = WorkflowOper(db).get(workflow.id)
+    if not wf:
+        return schemas.Response(success=False, message="工作流不存在")
+    wf.update(db, workflow.dict())
+    return schemas.Response(success=True, message="更新成功")
+
+
+@router.delete("/{workflow_id}", summary="删除工作流", response_model=schemas.Response)
+def delete_workflow(workflow_id: int,
+                    db: Session = Depends(get_db),
+                    _: schemas.TokenPayload = Depends(get_current_active_user)) -> Any:
+    """
+    删除工作流
+    """
+    from app.db.workflow_oper import WorkflowOper
+    workflow = WorkflowOper(db).get(workflow_id)
+    if not workflow:
+        return schemas.Response(success=False, message="工作流不存在")
+    # 删除定时任务
+    Scheduler().remove_workflow_job(workflow)
+    # 删除工作流
+    from app.db.models.workflow import Workflow as WorkflowModel
+    WorkflowModel.delete(db, workflow_id)
+    # 删除缓存
+    SystemConfigOper().delete(f"WorkflowCache-{workflow_id}")
+    return schemas.Response(success=True, message="删除成功")

app/chain/download.py

@@ -188,6 +188,9 @@ class DownloadChain(ChainBase):
                     f"Resource download canceled by event: {event_data.source},"
                     f"Reason: {event_data.reason}")
                 return None
+            # 如果事件修改了下载路径，使用新路径
+            if event_data.options and event_data.options.get("save_path"):
+                save_path = event_data.options.get("save_path")
 
         # 补充完整的media数据
         if not _media.genre_ids:

app/chain/media.py

@@ -367,12 +367,12 @@ class MediaChain(ChainBase):
                                      overwrite=overwrite)
             else:
                 # 检查目的目录下是否已经有nfo刮削文件
-                sub_files = storagechain.list_files(fileitem)
-                if any(f.name.endswith('.nfo') for f in sub_files):
+                has_nfo_file = storagechain.any_files(fileitem, extensions=['.nfo'])
+                if has_nfo_file and file_list:
                     logger.info(f"目录 {fileitem.path} 已有NFO文件，开始增量刮削...")
                     for file_path in file_list:
                         file_item = storagechain.get_file_item(storage=fileitem.storage,
-                                                              path=Path(file_path))
+                                                               path=Path(file_path))
                         if file_item:
                             # 对于电视剧文件，应该保存到与视频文件相同的目录
                             # 而不是电视剧根目录

app/chain/storage.py

@@ -180,13 +180,13 @@ class StorageChain(ChainBase):
             # 重命名格式
             rename_format = settings.TV_RENAME_FORMAT \
                 if mtype == MediaType.TV else settings.MOVIE_RENAME_FORMAT
-            # 计算重命名中的文件夹层数
-            rename_format_level = len(rename_format.split("/")) - 1
-            if rename_format_level < 1:
+            media_path = DirectoryHelper.get_media_root_path(
+                rename_format, rename_path=Path(fileitem.path)
+            )
+            if not media_path:
                 return True
             # 处理媒体文件根目录
-            dir_item = self.get_file_item(storage=fileitem.storage,
-                                          path=Path(fileitem.path).parents[rename_format_level - 1])
+            dir_item = self.get_file_item(storage=fileitem.storage, path=media_path)
         else:
             # 处理上级目录
             dir_item = self.get_parent_item(fileitem)

app/chain/transfer.py

@@ -5,7 +5,6 @@ import threading
 import traceback
 from copy import deepcopy
 from pathlib import Path
-from queue import Queue
 from time import sleep
 from typing import List, Optional, Tuple, Union, Dict, Callable
 
@@ -16,9 +15,9 @@ from app.chain.storage import StorageChain
 from app.chain.tmdb import TmdbChain
 from app.core.config import settings, global_vars
 from app.core.context import MediaInfo
+from app.core.event import eventmanager
 from app.core.meta import MetaBase
 from app.core.metainfo import MetaInfoPath
-from app.core.event import eventmanager
 from app.db.downloadhistory_oper import DownloadHistoryOper
 from app.db.models.downloadhistory import DownloadHistory
 from app.db.models.transferhistory import TransferHistory
@@ -28,11 +27,11 @@ from app.helper.directory import DirectoryHelper
 from app.helper.format import FormatParser
 from app.helper.progress import ProgressHelper
 from app.log import logger
+from app.schemas import StorageOperSelectionEventData
 from app.schemas import TransferInfo, TransferTorrent, Notification, EpisodeFormat, FileItem, TransferDirectoryConf, \
     TransferTask, TransferQueue, TransferJob, TransferJobTask
 from app.schemas.types import TorrentStatus, EventType, MediaType, ProgressKey, NotificationType, MessageChannel, \
     SystemConfigKey, ChainEventType, ContentType
-from app.schemas import StorageOperSelectionEventData
 from app.utils.singleton import Singleton
 from app.utils.string import StringUtils
 
@@ -213,6 +212,7 @@ class JobManager:
                                 set(self._season_episodes[mediaid]) - set(task.meta.episode_list)
                             )
                         return task
+            return None
 
     def remove_job(self, task: TransferTask) -> Optional[TransferJob]:
         """
@@ -226,6 +226,7 @@ class JobManager:
                 if __mediaid__ in self._season_episodes:
                     self._season_episodes.pop(__mediaid__)
                 return self._job_view.pop(__mediaid__)
+            return None
 
     def is_done(self, task: TransferTask) -> bool:
         """
@@ -311,7 +312,7 @@ class JobManager:
 
     def count(self, media: MediaInfo, season: Optional[int] = None) -> int:
         """
-        获取某项任务总数
+        获取某项任务成功总数
         """
         __mediaid__ = self.__get_media_id(media=media, season=season)
         with job_lock:
@@ -322,7 +323,7 @@ class JobManager:
 
     def size(self, media: MediaInfo, season: Optional[int] = None) -> int:
         """
-        获取某项任务总大小
+        获取某项任务成功文件总大小
         """
         __mediaid__ = self.__get_media_id(media=media, season=season)
         with job_lock:
@@ -359,22 +360,20 @@ class TransferChain(ChainBase, metaclass=Singleton):
     文件整理处理链
     """
 
-    # 可处理的文件后缀
-    all_exts = settings.RMT_MEDIAEXT
-
-    # 待整理任务队列
-    _queue = Queue()
-
-    # 文件整理线程
-    _transfer_thread = None
-
-    # 队列间隔时间（秒）
-    _transfer_interval = 15
-
     def __init__(self):
         super().__init__()
+        # 可处理的文件后缀
+        self.all_exts = settings.RMT_MEDIAEXT
+        # 待整理任务队列
+        self._queue = queue.Queue()
+        # 文件整理线程
+        self._transfer_thread = None
+        # 队列间隔时间（秒）
+        self._transfer_interval = 15
+        # 事件管理器
         self.jobview = JobManager()
-
+        # 车移成功的文件清单
+        self._success_target_files: Dict[str, List[str]] = {}
         # 启动整理任务
         self.__init()
 
@@ -391,6 +390,44 @@ class TransferChain(ChainBase, metaclass=Singleton):
         """
         整理完成后处理
         """
+
+        def __do_finished():
+            """
+            完成时发送消息、刮削事件、移除任务等
+            """
+            # 更新文件数量
+            transferinfo.file_count = self.jobview.count(task.mediainfo, task.meta.begin_season) or 1
+            # 更新文件大小
+            transferinfo.total_size = self.jobview.size(task.mediainfo,
+                                                        task.meta.begin_season) or task.fileitem.size
+            # 更新文件清单
+            transferinfo.file_list_new = self._success_target_files.pop(transferinfo.target_diritem.path, [])
+            # 发送通知，实时手动整理时不发
+            if transferinfo.need_notify and (task.background or not task.manual):
+                se_str = None
+                if task.mediainfo.type == MediaType.TV:
+                    season_episodes = self.jobview.season_episodes(task.mediainfo, task.meta.begin_season)
+                    if season_episodes:
+                        se_str = f"{task.meta.season} {StringUtils.format_ep(season_episodes)}"
+                    else:
+                        se_str = f"{task.meta.season}"
+                self.send_transfer_message(meta=task.meta,
+                                           mediainfo=task.mediainfo,
+                                           transferinfo=transferinfo,
+                                           season_episode=se_str,
+                                           username=task.username)
+            # 刮削事件
+            if transferinfo.need_scrape:
+                self.eventmanager.send_event(EventType.MetadataScrape, {
+                    'meta': task.meta,
+                    'mediainfo': task.mediainfo,
+                    'fileitem': transferinfo.target_diritem,
+                    'file_list': transferinfo.file_list_new,
+                    'overwrite': False
+                })
+            # 移除已完成的任务
+            self.jobview.remove_job(task)
+
         transferhis = TransferHistoryOper()
         if not transferinfo.success:
             # 转移失败
@@ -416,6 +453,10 @@ class TransferChain(ChainBase, metaclass=Singleton):
             ))
             # 整理失败
             self.jobview.fail_task(task)
+            with task_lock:
+                # 整理完成且有成功的任务时
+                if self.jobview.is_finished(task):
+                    __do_finished()
             return False, transferinfo.message
 
         # 转移成功
@@ -444,6 +485,13 @@ class TransferChain(ChainBase, metaclass=Singleton):
         })
 
         with task_lock:
+            # 登记转移成功文件清单
+            target_dir_path = transferinfo.target_diritem.path
+            target_files = transferinfo.file_list_new
+            if self._success_target_files.get(target_dir_path):
+                self._success_target_files[target_dir_path].extend(target_files)
+            else:
+                self._success_target_files[target_dir_path] = target_files
             # 全部整理成功时
             if self.jobview.is_success(task):
                 # 移动模式删除空目录
@@ -464,37 +512,7 @@ class TransferChain(ChainBase, metaclass=Singleton):
                             storagechain.delete_media_file(t.fileitem, delete_self=False)
             # 整理完成且有成功的任务时
             if self.jobview.is_finished(task):
-                # 发送通知，实时手动整理时不发
-                if transferinfo.need_notify and (task.background or not task.manual):
-                    se_str = None
-                    if task.mediainfo.type == MediaType.TV:
-                        season_episodes = self.jobview.season_episodes(task.mediainfo, task.meta.begin_season)
-                        if season_episodes:
-                            se_str = f"{task.meta.season} {StringUtils.format_ep(season_episodes)}"
-                        else:
-                            se_str = f"{task.meta.season}"
-                    # 更新文件数量
-                    transferinfo.file_count = self.jobview.count(task.mediainfo, task.meta.begin_season) or 1
-                    # 更新文件大小
-                    transferinfo.total_size = self.jobview.size(task.mediainfo,
-                                                                task.meta.begin_season) or task.fileitem.size
-                    self.send_transfer_message(meta=task.meta,
-                                               mediainfo=task.mediainfo,
-                                               transferinfo=transferinfo,
-                                               season_episode=se_str,
-                                               username=task.username)
-                # 刮削事件
-                if transferinfo.need_scrape:
-                    self.eventmanager.send_event(EventType.MetadataScrape, {
-                        'meta': task.meta,
-                        'mediainfo': task.mediainfo,
-                        'fileitem': transferinfo.target_diritem,
-                        'file_list': transferinfo.file_list_new,
-                        'overwrite': False
-                    })
-
-                # 移除已完成的任务
-                self.jobview.remove_job(task)
+                __do_finished()
 
         return True, ""
 
@@ -1101,7 +1119,8 @@ class TransferChain(ChainBase, metaclass=Singleton):
                 # 自定义识别
                 if formaterHandler:
                     # 开始集、结束集、PART
-                    begin_ep, end_ep, part = formaterHandler.split_episode(file_name=file_path.name, file_meta=file_meta)
+                    begin_ep, end_ep, part = formaterHandler.split_episode(file_name=file_path.name,
+                                                                           file_meta=file_meta)
                     if begin_ep is not None:
                         file_meta.begin_episode = begin_ep
                         file_meta.part = part

app/core/config.py

@@ -274,12 +274,6 @@ class ConfigModel(BaseModel):
     REPO_GITHUB_TOKEN: Optional[str] = None
     # 大内存模式
     BIG_MEMORY_MODE: bool = False
-    # 是否启用内存监控
-    MEMORY_ANALYSIS: bool = False
-    # 内存快照间隔（分钟）
-    MEMORY_SNAPSHOT_INTERVAL: int = 30
-    # 保留的内存快照文件数量
-    MEMORY_SNAPSHOT_KEEP_COUNT: int = 20
     # 全局图片缓存，将媒体图片缓存到本地
     GLOBAL_IMAGE_CACHE: bool = False
     # 是否启用编码探测的性能模式
@@ -311,6 +305,8 @@ class ConfigModel(BaseModel):
     DEFAULT_SUB: Optional[str] = "zh-cn"
     # Docker Client API地址
     DOCKER_CLIENT_API: Optional[str] = "tcp://127.0.0.1:38379"
+    # 工作流数据共享
+    WORKFLOW_STATISTIC_SHARE: bool = True
 
 
 class Settings(BaseSettings, ConfigModel, LogConfigModel):

app/core/event.py

@@ -69,9 +69,6 @@ class EventManager(metaclass=Singleton):
     EventManager 负责管理和调度广播事件和链式事件，包括订阅、发送和处理事件
     """
 
-    # 退出事件
-    __event = threading.Event()
-
     def __init__(self):
         self.__executor = ThreadHelper()  # 动态线程池，用于消费事件
         self.__consumer_threads = []  # 用于保存启动的事件消费者线程
@@ -81,6 +78,7 @@ class EventManager(metaclass=Singleton):
         self.__disabled_handlers = set()  # 禁用的事件处理器集合
         self.__disabled_classes = set()  # 禁用的事件处理器类集合
         self.__lock = threading.Lock()  # 线程锁
+        self.__event = threading.Event()  # 退出事件
 
     def start(self):
         """

app/core/meta/customization.py

@@ -9,8 +9,6 @@ class CustomizationMatcher(metaclass=Singleton):
     """
     识别自定义占位符
     """
-    customization = None
-    custom_separator = None
 
     def __init__(self):
         self.systemconfig = SystemConfigOper()

app/core/meta/releasegroup.py

@@ -9,7 +9,6 @@ class ReleaseGroupsMatcher(metaclass=Singleton):
     """
     识别制作组、字幕组
     """
-    __release_groups: str = None
     # 内置组
     RELEASE_GROUPS: dict = {
         "0ff": ['FF(?:(?:A|WE)B|CD|E(?:DU|B)|TV)'],

app/core/module.py

@@ -16,14 +16,14 @@ class ModuleManager(metaclass=Singleton):
     模块管理器
     """
 
-    # 模块列表
-    _modules: dict = {}
-    # 运行态模块列表
-    _running_modules: dict = {}
     # 子模块类型集合
     SubType = Union[DownloaderType, MediaServerType, MessageChannel, StorageSchema, OtherModulesType]
 
     def __init__(self):
+        # 模块列表
+        self._modules: dict = {}
+        # 运行态模块列表
+        self._running_modules: dict = {}
         self.load_modules()
 
     def load_modules(self):

app/core/plugin.py

@@ -10,6 +10,7 @@ from concurrent.futures import ThreadPoolExecutor, as_completed
 from pathlib import Path
 from typing import Any, Dict, List, Optional, Type, Union, Callable, Tuple
 
+from app.helper.sites import SitesHelper
 from fastapi import HTTPException
 from starlette import status
 from watchdog.events import FileSystemEventHandler
@@ -21,7 +22,6 @@ from app.core.event import eventmanager, Event
 from app.db.plugindata_oper import PluginDataOper
 from app.db.systemconfig_oper import SystemConfigOper
 from app.helper.plugin import PluginHelper
-from app.helper.sites import SitesHelper
 from app.log import logger
 from app.schemas.types import EventType, SystemConfigKey
 from app.utils.crypto import RSAUtils
@@ -88,16 +88,15 @@ class PluginManager(metaclass=Singleton):
     插件管理器
     """
 
-    # 插件列表
-    _plugins: dict = {}
-    # 运行态插件列表
-    _running_plugins: dict = {}
-    # 配置Key
-    _config_key: str = "plugin.%s"
-    # 监听器
-    _observer: Observer = None
-
     def __init__(self):
+        # 插件列表
+        self._plugins: dict = {}
+        # 运行态插件列表
+        self._running_plugins: dict = {}
+        # 配置Key
+        self._config_key: str = "plugin.%s"
+        # 监听器
+        self._observer: Observer = None
         # 开发者模式监测插件修改
         if settings.DEV or settings.PLUGIN_AUTO_RELOAD:
             self.__start_monitor()

app/core/workflow.py

@@ -13,10 +13,9 @@ class WorkFlowManager(metaclass=Singleton):
     工作流管理器
     """
 
-    # 所有动作定义
-    _actions: Dict[str, Any] = {}
-
     def __init__(self):
+        # 所有动作定义
+        self._actions: Dict[str, Any] = {}
         self.init()
 
     def init(self):

app/db/models/workflow.py

@@ -37,6 +37,11 @@ class Workflow(Base):
     # 最后执行时间
     last_time = Column(String)
 
+    @staticmethod
+    @db_query
+    def list(db):
+        return db.query(Workflow).all()
+
     @staticmethod
     @db_query
     def get_enabled_workflows(db):

app/db/workflow_oper.py

@@ -25,6 +25,12 @@ class WorkflowOper(DbOper):
         """
         return Workflow.get(self._db, wid)
 
+    def list(self) -> List[Workflow]:
+        """
+        获取所有工作流列表
+        """
+        return Workflow.list(self._db)
+
     def list_enabled(self) -> List[Workflow]:
         """
         获取启用的工作流列表

app/helper/directory.py

@@ -4,6 +4,7 @@ from typing import List, Optional
 from app import schemas
 from app.core.context import MediaInfo
 from app.db.systemconfig_oper import SystemConfigOper
+from app.log import logger
 from app.schemas.types import SystemConfigKey
 from app.utils.system import SystemUtils
 
@@ -109,3 +110,36 @@ class DirectoryHelper:
                         return matched_dir
             return matched_dirs[0]
         return None
+
+    @staticmethod
+    def get_media_root_path(rename_format: str, rename_path: Path) -> Optional[Path]:
+        """
+        获取重命名后的媒体文件根路径
+
+        :param rename_format: 重命名格式
+        :param rename_path: 重命名后的路径
+        :return: 媒体文件根路径
+        """
+        # 计算重命名中的文件夹层数
+        rename_list = rename_format.split("/")
+        rename_format_level = len(rename_list) - 1
+        if rename_format_level <= 0:
+            # 无效重命名格式
+            logger.error(f"重命名格式 {rename_format} 不正确")
+            return None
+        for level, name in enumerate(rename_list):
+            # 处理特例，有的人重命名的第一层是年份、分辨率
+            if "{{title}}" in name:
+                # 找出含标题的这一层作为媒体根目录
+                rename_format_level -= level
+                break
+        else:
+            # 假定第一层目录是媒体根目录
+            logger.warn(f"重命名格式 {rename_format} 缺少 {{{{title}}}}")
+        if rename_format_level > len(rename_path.parents):
+            # 通常因为路径以/结尾，被Path规范化删除了
+            logger.error(f"路径 {rename_path} 不匹配重命名格式 {rename_format}")
+            return None
+        # 媒体根路径
+        media_root = rename_path.parents[rename_format_level - 1]
+        return media_root

app/helper/display.py

@@ -8,7 +8,6 @@ import os
 
 
 class DisplayHelper(metaclass=Singleton):
-    _display: Display = None
 
     def __init__(self):
         if not SystemUtils.is_docker():

app/helper/doh.py

@@ -70,6 +70,9 @@ def enable_doh(enable: bool):
 
 
 class DohHelper(metaclass=Singleton):
+    """
+    DoH帮助类，用于处理DNS over HTTPS解析。
+    """
     def __init__(self):
         enable_doh(settings.DOH_ENABLE)
 

app/helper/memory.py

@@ -1,457 +0,0 @@
-import gc
-import sys
-import threading
-import time
-from datetime import datetime
-from typing import Optional
-
-import psutil
-from pympler import muppy, summary, asizeof
-
-from app.core.config import settings
-from app.core.event import eventmanager, Event
-from app.log import logger
-from app.schemas import ConfigChangeEventData
-from app.schemas.types import EventType
-from app.utils.singleton import Singleton
-
-
-class MemoryHelper(metaclass=Singleton):
-    """
-    内存管理工具类，用于监控和优化内存使用
-    """
-
-    def __init__(self):
-        # 检查间隔(秒) - 从配置获取，默认5分钟
-        self._check_interval = settings.MEMORY_SNAPSHOT_INTERVAL * 60
-        self._monitoring = False
-        self._monitor_thread: Optional[threading.Thread] = None
-        # 内存快照保存目录
-        self._memory_snapshot_dir = settings.LOG_PATH / "memory_snapshots"
-        # 保留的快照文件数量
-        self._keep_count = settings.MEMORY_SNAPSHOT_KEEP_COUNT
-
-    @eventmanager.register(EventType.ConfigChanged)
-    def handle_config_changed(self, event: Event):
-        """
-        处理配置变更事件，更新内存监控设置
-        :param event: 事件对象
-        """
-        if not event:
-            return
-        event_data: ConfigChangeEventData = event.event_data
-        if event_data.key not in ['MEMORY_ANALYSIS', 'MEMORY_SNAPSHOT_INTERVAL', 'MEMORY_SNAPSHOT_KEEP_COUNT']:
-            return
-
-        # 更新配置
-        if event_data.key == 'MEMORY_SNAPSHOT_INTERVAL':
-            self._check_interval = settings.MEMORY_SNAPSHOT_INTERVAL * 60
-        elif event_data.key == 'MEMORY_SNAPSHOT_KEEP_COUNT':
-            self._keep_count = settings.MEMORY_SNAPSHOT_KEEP_COUNT
-        self.stop_monitoring()
-        self.start_monitoring()
-
-    def start_monitoring(self):
-        """
-        开始内存监控
-        """
-        if not settings.MEMORY_ANALYSIS:
-            return
-        if self._monitoring:
-            return
-
-        # 创建内存快照目录
-        self._memory_snapshot_dir.mkdir(parents=True, exist_ok=True)
-
-        # 初始化内存分析器
-        self._monitoring = True
-        self._monitor_thread = threading.Thread(target=self._monitor_loop, daemon=True)
-        self._monitor_thread.start()
-        logger.info("内存监控已启动")
-
-    def stop_monitoring(self):
-        """
-        停止内存监控
-        """
-        self._monitoring = False
-        if self._monitor_thread:
-            self._monitor_thread.join(timeout=5)
-            logger.info("内存监控已停止")
-
-    def _monitor_loop(self):
-        """
-        内存监控循环
-        """
-        logger.info("内存监控循环开始")
-        while self._monitoring:
-            try:
-                # 生成内存快照
-                self._create_memory_snapshot()
-                time.sleep(self._check_interval)
-            except Exception as e:
-                logger.error(f"内存监控出错: {e}")
-                # 出错后等待1分钟再继续
-                time.sleep(60)
-        logger.info("内存监控循环结束")
-
-    def _create_memory_snapshot(self):
-        """
-        创建内存快照并保存到文件
-        """
-        try:
-            # 获取当前时间戳
-            timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
-            snapshot_file = self._memory_snapshot_dir / f"memory_snapshot_{timestamp}.txt"
-
-            # 获取系统内存使用情况
-            memory_usage = psutil.Process().memory_info().rss
-
-            logger.info(f"开始创建内存快照: {snapshot_file}")
-
-            # 第一步：写入基本信息和对象类型统计
-            self._write_basic_info(snapshot_file, memory_usage)
-
-            # 第二步：分析并写入类实例内存使用情况
-            self._append_class_analysis(snapshot_file)
-
-            # 第三步：分析并写入大内存变量详情
-            self._append_variable_analysis(snapshot_file)
-
-            logger.info(f"内存快照已保存: {snapshot_file}, 当前内存使用: {memory_usage / 1024 / 1024:.2f} MB")
-
-            # 清理过期的快照文件（保留最近30个）
-            self._cleanup_old_snapshots()
-
-        except Exception as e:
-            logger.error(f"创建内存快照失败: {e}")
-
-    @staticmethod
-    def _write_basic_info(snapshot_file, memory_usage):
-        """
-        写入基本信息和对象类型统计
-        """
-        # 获取当前进程的内存使用情况
-        all_objects = muppy.get_objects()
-        sum1 = summary.summarize(all_objects)
-
-        with open(snapshot_file, 'w', encoding='utf-8') as f:
-            f.write(f"内存快照时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}\n")
-            f.write(f"当前进程内存使用: {memory_usage / 1024 / 1024:.2f} MB\n")
-            f.write("=" * 80 + "\n")
-            f.write("对象类型统计:\n")
-            f.write("-" * 80 + "\n")
-
-            # 写入对象统计信息
-            for line in summary.format_(sum1):
-                f.write(line + "\n")
-
-            # 立即刷新到磁盘
-            f.flush()
-
-        logger.debug("基本信息已写入快照文件")
-
-    def _append_class_analysis(self, snapshot_file):
-        """
-        分析并追加类实例内存使用情况
-        """
-        with open(snapshot_file, 'a', encoding='utf-8') as f:
-            f.write("\n" + "=" * 80 + "\n")
-            f.write("类实例内存使用情况 (按内存大小排序):\n")
-            f.write("-" * 80 + "\n")
-            f.write("正在分析中...\n")
-            # 立即刷新，让用户知道这部分开始了
-            f.flush()
-
-        try:
-            logger.debug("开始分析类实例内存使用情况")
-            class_objects = self._get_class_memory_usage()
-
-            # 重新打开文件，移除"正在分析中..."并写入实际结果
-            with open(snapshot_file, 'r', encoding='utf-8') as f:
-                content = f.read()
-
-            # 替换"正在分析中..."
-            content = content.replace("正在分析中...\n", "")
-
-            with open(snapshot_file, 'w', encoding='utf-8') as f:
-                f.write(content)
-
-                if class_objects:
-                    # 只显示前100个类
-                    for i, class_info in enumerate(class_objects[:100], 1):
-                        f.write(f"{i:3d}. {class_info['name']:<50} "
-                                f"{class_info['size_mb']:>8.2f} MB ({class_info['count']} 个实例)\n")
-                else:
-                    f.write("未找到有效的类实例信息\n")
-
-                f.flush()
-
-        except Exception as e:
-            logger.error(f"获取类实例信息失败: {e}")
-
-            # 即使出错也要更新文件
-            with open(snapshot_file, 'r', encoding='utf-8') as f:
-                content = f.read()
-
-            content = content.replace("正在分析中...\n", f"获取类实例信息失败: {e}\n")
-
-            with open(snapshot_file, 'w', encoding='utf-8') as f:
-                f.write(content)
-                f.flush()
-
-        logger.debug("类实例分析已完成并写入")
-
-    def _append_variable_analysis(self, snapshot_file):
-        """
-        分析并追加大内存变量详情
-        """
-        with open(snapshot_file, 'a', encoding='utf-8') as f:
-            f.write("\n" + "=" * 80 + "\n")
-            f.write("大内存变量详情 (前100个):\n")
-            f.write("-" * 80 + "\n")
-            f.write("正在分析中...\n")
-            # 立即刷新，让用户知道这部分开始了
-            f.flush()
-
-        try:
-            logger.debug("开始分析大内存变量")
-            large_variables = self._get_large_variables(100)
-
-            # 重新打开文件，移除"正在分析中..."并写入实际结果
-            with open(snapshot_file, 'r', encoding='utf-8') as f:
-                content = f.read()
-
-            # 替换最后的"正在分析中..."
-            content = content.replace("正在分析中...\n", "")
-
-            with open(snapshot_file, 'w', encoding='utf-8') as f:
-                f.write(content)
-
-                if large_variables:
-                    for i, var_info in enumerate(large_variables, 1):
-                        f.write(
-                            f"{i:3d}. {var_info['name']:<30} {var_info['type']:<15} {var_info['size_mb']:>8.2f} MB\n")
-                else:
-                    f.write("未找到大内存变量\n")
-
-                f.flush()
-
-        except Exception as e:
-            logger.error(f"获取大内存变量信息失败: {e}")
-
-            # 即使出错也要更新文件
-            with open(snapshot_file, 'r', encoding='utf-8') as f:
-                content = f.read()
-
-            content = content.replace("正在分析中...\n", f"获取变量信息失败: {e}\n")
-
-            with open(snapshot_file, 'w', encoding='utf-8') as f:
-                f.write(content)
-                f.flush()
-
-        logger.debug("大内存变量分析已完成并写入")
-
-    def _cleanup_old_snapshots(self):
-        """
-        清理过期的内存快照文件，只保留最近的指定数量文件
-        """
-        try:
-            snapshot_files = list(self._memory_snapshot_dir.glob("memory_snapshot_*.txt"))
-            if len(snapshot_files) > self._keep_count:
-                # 按修改时间排序，删除最旧的文件
-                snapshot_files.sort(key=lambda x: x.stat().st_mtime)
-                for old_file in snapshot_files[:-self._keep_count]:
-                    old_file.unlink()
-                    logger.debug(f"已删除过期内存快照: {old_file}")
-        except Exception as e:
-            logger.error(f"清理过期快照失败: {e}")
-
-    @staticmethod
-    def _get_class_memory_usage():
-        """
-        获取所有类实例的内存使用情况，按内存大小排序
-        """
-        class_info = {}
-        processed_count = 0
-        error_count = 0
-
-        # 获取所有对象
-        all_objects = muppy.get_objects()
-        logger.debug(f"开始分析 {len(all_objects)} 个对象的类实例内存使用情况")
-
-        for obj in all_objects:
-            try:
-                # 跳过类对象本身，统计类的实例
-                if isinstance(obj, type):
-                    continue
-
-                # 获取对象的类名 - 这里可能会出错
-                obj_class = type(obj)
-
-                # 安全地获取类名
-                try:
-                    if hasattr(obj_class, '__module__') and hasattr(obj_class, '__name__'):
-                        class_name = f"{obj_class.__module__}.{obj_class.__name__}"
-                    else:
-                        class_name = str(obj_class)
-                except Exception as e:
-                    # 如果获取类名失败，使用简单的类型描述
-                    class_name = f"<unknown_class_{id(obj_class)}>"
-                    logger.debug(f"获取类名失败: {e}")
-
-                # 计算对象本身的内存使用（不包括引用对象，避免重复计算）
-                size_bytes = sys.getsizeof(obj)
-                if size_bytes < 100:  # 跳过太小的对象
-                    continue
-
-                size_mb = size_bytes / 1024 / 1024
-                processed_count += 1
-
-                if class_name in class_info:
-                    class_info[class_name]['size_mb'] += size_mb
-                    class_info[class_name]['count'] += 1
-                else:
-                    class_info[class_name] = {
-                        'name': class_name,
-                        'size_mb': size_mb,
-                        'count': 1
-                    }
-
-            except Exception as e:
-                # 捕获所有可能的异常，包括SQLAlchemy、ORM等框架的异常
-                error_count += 1
-                if error_count <= 5:  # 只记录前5个错误，避免日志过多
-                    logger.debug(f"分析对象时出错: {e}")
-                continue
-
-        logger.debug(f"类实例分析完成: 处理了 {processed_count} 个对象, 遇到 {error_count} 个错误")
-
-        # 按内存大小排序
-        sorted_classes = sorted(class_info.values(), key=lambda x: x['size_mb'], reverse=True)
-        return sorted_classes
-
-    def _get_large_variables(self, limit=100):
-        """
-        获取大内存变量信息，按内存大小排序
-        使用已计算对象集合避免重复计算
-        """
-        large_vars = []
-        processed_count = 0
-        calculated_objects = set()  # 避免重复计算
-
-        # 获取所有对象
-        all_objects = muppy.get_objects()
-        logger.debug(f"开始分析 {len(all_objects)} 个对象的内存使用情况")
-
-        for obj in all_objects:
-            # 跳过类对象
-            if isinstance(obj, type):
-                continue
-
-            # 跳过已经计算过的对象
-            obj_id = id(obj)
-            if obj_id in calculated_objects:
-                continue
-
-            try:
-                # 首先使用 sys.getsizeof 快速筛选
-                shallow_size = sys.getsizeof(obj)
-                if shallow_size < 1024:  # 只处理大于1KB的对象
-                    continue
-
-                # 对于较大的对象，使用 asizeof 进行深度计算
-                size_bytes = asizeof.asizeof(obj)
-
-                # 只处理大于10KB的对象，提高分析效率
-                if size_bytes < 10240:
-                    continue
-
-                size_mb = size_bytes / 1024 / 1024
-                processed_count += 1
-                calculated_objects.add(obj_id)
-
-                # 获取对象信息
-                var_info = self._get_variable_info(obj, size_mb)
-                if var_info:
-                    large_vars.append(var_info)
-
-                # 如果已经找到足够多的大对象，可以提前结束
-                if len(large_vars) >= limit * 2:  # 多收集一些，后面排序筛选
-                    break
-
-            except Exception as e:
-                # 更广泛的异常捕获
-                logger.debug(f"分析对象失败: {e}")
-                continue
-
-        logger.debug(f"处理了 {processed_count} 个大对象，找到 {len(large_vars)} 个有效变量")
-
-        # 按内存大小排序并返回前N个
-        large_vars.sort(key=lambda x: x['size_mb'], reverse=True)
-        return large_vars[:limit]
-
-    def _get_variable_info(self, obj, size_mb):
-        """
-        获取变量的描述信息
-        """
-        try:
-            obj_type = type(obj).__name__
-
-            # 尝试获取变量名
-            var_name = self._get_variable_name(obj)
-
-            # 生成描述性信息
-            if isinstance(obj, dict):
-                key_count = len(obj)
-                if key_count > 0:
-                    sample_keys = list(obj.keys())[:3]
-                    var_name += f" ({key_count}项, 键: {sample_keys})"
-            elif isinstance(obj, (list, tuple, set)):
-                var_name += f" ({len(obj)}个元素)"
-            elif isinstance(obj, str):
-                if len(obj) > 50:
-                    var_name += f" (长度: {len(obj)}, 内容: '{obj[:50]}...')"
-                else:
-                    var_name += f" ('{obj}')"
-            elif hasattr(obj, '__class__') and hasattr(obj.__class__, '__name__'):
-                if hasattr(obj, '__dict__'):
-                    attr_count = len(obj.__dict__)
-                    var_name += f" ({attr_count}个属性)"
-
-            return {
-                'name': var_name,
-                'type': obj_type,
-                'size_mb': size_mb
-            }
-
-        except Exception as e:
-            logger.debug(f"获取变量信息失败: {e}")
-            return None
-
-    @staticmethod
-    def _get_variable_name(obj):
-        """
-        尝试获取变量名
-        """
-        try:
-            # 尝试通过gc获取引用该对象的变量名
-            referrers = gc.get_referrers(obj)
-
-            for referrer in referrers:
-                if isinstance(referrer, dict):
-                    # 检查是否在某个模块的全局变量中
-                    for name, value in referrer.items():
-                        if value is obj and isinstance(name, str):
-                            return name
-                elif hasattr(referrer, '__dict__'):
-                    # 检查是否在某个实例的属性中
-                    for name, value in referrer.__dict__.items():
-                        if value is obj and isinstance(name, str):
-                            return f"{type(referrer).__name__}.{name}"
-
-            # 如果找不到变量名，返回对象类型和id
-            return f"{type(obj).__name__}_{id(obj)}"
-
-        except Exception as e:
-            logger.debug(f"获取变量名失败: {e}")
-            return f"{type(obj).__name__}_{id(obj)}"

app/helper/message.py

@@ -541,8 +541,6 @@ class MessageQueueManager(metaclass=SingletonClass):
     消息发送队列管理器
     """
 
-    schedule_periods: List[tuple[int, int, int, int]] = []
-
     def __init__(
             self,
             send_callback: Optional[Callable] = None,
@@ -554,6 +552,8 @@ class MessageQueueManager(metaclass=SingletonClass):
         :param send_callback: 实际发送消息的回调函数
         :param check_interval: 时间检查间隔（秒）
         """
+        self.schedule_periods: List[tuple[int, int, int, int]] = []
+
         self.init_config()
 
         self.queue: queue.Queue[Any] = queue.Queue()

app/helper/plugin.py

@@ -18,14 +18,14 @@ from app.db.systemconfig_oper import SystemConfigOper
 from app.log import logger
 from app.schemas.types import SystemConfigKey
 from app.utils.http import RequestUtils
-from app.utils.singleton import Singleton
+from app.utils.singleton import WeakSingleton
 from app.utils.system import SystemUtils
 from app.utils.url import UrlUtils
 
 PLUGIN_DIR = Path(settings.ROOT_PATH) / "app" / "plugins"
 
 
-class PluginHelper(metaclass=Singleton):
+class PluginHelper(metaclass=WeakSingleton):
     """
     插件市场管理，下载安装插件到本地
     """

app/helper/progress.py

@@ -1,12 +1,11 @@
 from enum import Enum
-from typing import Union, Dict, Optional
+from typing import Union, Optional
 
 from app.schemas.types import ProgressKey
-from app.utils.singleton import Singleton
+from app.utils.singleton import WeakSingleton
 
 
-class ProgressHelper(metaclass=Singleton):
-    _process_detail: Dict[str, dict] = {}
+class ProgressHelper(metaclass=WeakSingleton):
 
     def __init__(self):
         self._process_detail = {}

app/helper/subscribe.py

@@ -8,11 +8,11 @@ from app.db.systemconfig_oper import SystemConfigOper
 from app.log import logger
 from app.schemas.types import SystemConfigKey
 from app.utils.http import RequestUtils
-from app.utils.singleton import Singleton
+from app.utils.singleton import WeakSingleton
 from app.utils.system import SystemUtils
 
 
-class SubscribeHelper(metaclass=Singleton):
+class SubscribeHelper(metaclass=WeakSingleton):
     """
     订阅数据统计/订阅分享等
     """

app/helper/torrent.py

@@ -1,10 +1,9 @@
 import datetime
 import re
 from pathlib import Path
-from typing import Tuple, Optional, List, Union, Dict
+from typing import Tuple, Optional, List, Union, Dict, Any
 from urllib.parse import unquote
 
-from requests import Response
 from torrentool.api import Torrent
 
 from app.core.config import settings
@@ -16,17 +15,17 @@ from app.db.systemconfig_oper import SystemConfigOper
 from app.log import logger
 from app.schemas.types import MediaType, SystemConfigKey
 from app.utils.http import RequestUtils
-from app.utils.singleton import Singleton
+from app.utils.singleton import WeakSingleton
 from app.utils.string import StringUtils
 
 
-class TorrentHelper(metaclass=Singleton):
+class TorrentHelper(metaclass=WeakSingleton):
     """
     种子帮助类
     """
 
-    # 失败的种子：站点链接
-    _invalid_torrents = []
+    def __init__(self):
+        self._invalid_torrents = []
 
     def download_torrent(self, url: str,
                          cookie: Optional[str] = None,
@@ -170,7 +169,7 @@ class TorrentHelper(metaclass=Singleton):
             return "", []
 
     @staticmethod
-    def get_url_filename(req: Response, url: str) -> str:
+    def get_url_filename(req: Any, url: str) -> str:
         """
         从下载请求中获取种子文件名
         """

app/helper/workflow.py

@@ -0,0 +1,132 @@
+import json
+from typing import List, Tuple, Optional
+
+from app.core.cache import cached, cache_backend
+from app.core.config import settings
+from app.db.workflow_oper import WorkflowOper
+from app.log import logger
+from app.utils.http import RequestUtils
+from app.utils.singleton import WeakSingleton
+from app.utils.system import SystemUtils
+
+
+class WorkflowHelper(metaclass=WeakSingleton):
+    """
+    工作流分享等
+    """
+
+    _workflow_share = f"{settings.MP_SERVER_HOST}/workflow/share"
+
+    _workflow_shares = f"{settings.MP_SERVER_HOST}/workflow/shares"
+
+    _workflow_fork = f"{settings.MP_SERVER_HOST}/workflow/fork/%s"
+
+    _shares_cache_region = "workflow_share"
+
+    _share_user_id = None
+
+    def __init__(self):
+        self.get_user_uuid()
+
+    def workflow_share(self, workflow_id: int,
+                       share_title: str, share_comment: str, share_user: str) -> Tuple[bool, str]:
+        """
+        分享工作流
+        """
+        if not settings.WORKFLOW_STATISTIC_SHARE:  # 使用独立的工作流分享开关
+            return False, "当前没有开启工作流数据共享功能"
+        
+        # 获取工作流信息
+        workflow = WorkflowOper().get(workflow_id)
+        if not workflow:
+            return False, "工作流不存在"
+
+        if not workflow.actions or not workflow.flows:
+            return False, "请分享有动作和流程的工作流"
+
+        workflow_dict = workflow.to_dict()
+        workflow_dict.pop("id", None)
+        workflow_dict.pop("context", None)
+        workflow_dict['actions'] = json.dumps(workflow_dict['actions'] or [])
+        workflow_dict['flows'] = json.dumps(workflow_dict['flows'] or [])
+
+        # 发送分享请求
+        res = RequestUtils(proxies=settings.PROXY or {}, content_type="application/json",
+                           timeout=10).post(self._workflow_share,
+                                            json={
+                                                "share_title": share_title,
+                                                "share_comment": share_comment,
+                                                "share_user": share_user,
+                                                "share_uid": self._share_user_id,
+                                                **workflow_dict
+                                            })
+        if res is None:
+            return False, "连接MoviePilot服务器失败"
+        if res.ok:
+            # 清除 get_shares 的缓存，以便实时看到结果
+            cache_backend.clear(region=self._shares_cache_region)
+            return True, ""
+        else:
+            return False, res.json().get("message")
+
+    def share_delete(self, share_id: int) -> Tuple[bool, str]:
+        """
+        删除分享
+        """
+        if not settings.WORKFLOW_STATISTIC_SHARE:  # 使用独立的工作流分享开关
+            return False, "当前没有开启工作流数据共享功能"
+        
+        res = RequestUtils(proxies=settings.PROXY or {},
+                           timeout=5).delete_res(f"{self._workflow_share}/{share_id}",
+                                                 params={"share_uid": self._share_user_id})
+        if res is None:
+            return False, "连接MoviePilot服务器失败"
+        if res.ok:
+            # 清除 get_shares 的缓存，以便实时看到结果
+            cache_backend.clear(region=self._shares_cache_region)
+            return True, ""
+        else:
+            return False, res.json().get("message")
+
+    def workflow_fork(self, share_id: int) -> Tuple[bool, str]:
+        """
+        复用分享的工作流
+        """
+        if not settings.WORKFLOW_STATISTIC_SHARE:  # 使用独立的工作流分享开关
+            return False, "当前没有开启工作流数据共享功能"
+        
+        res = RequestUtils(proxies=settings.PROXY or {}, timeout=5, headers={
+            "Content-Type": "application/json"
+        }).get_res(self._workflow_fork % share_id)
+        if res is None:
+            return False, "连接MoviePilot服务器失败"
+        if res.ok:
+            return True, ""
+        else:
+            return False, res.json().get("message")
+
+    @cached(region=_shares_cache_region)
+    def get_shares(self, name: Optional[str] = None, page: Optional[int] = 1, count: Optional[int] = 30) -> List[dict]:
+        """
+        获取工作流分享数据
+        """
+        if not settings.WORKFLOW_STATISTIC_SHARE:  # 使用独立的工作流分享开关
+            return []
+        
+        res = RequestUtils(proxies=settings.PROXY or {}, timeout=15).get_res(self._workflow_shares, params={
+            "name": name,
+            "page": page,
+            "count": count
+        })
+        if res and res.status_code == 200:
+            return res.json()
+        return []
+
+    def get_user_uuid(self) -> str:
+        """
+        获取用户uuid
+        """
+        if not self._share_user_id:
+            self._share_user_id = SystemUtils.generate_user_unique_id()
+            logger.info(f"当前用户UUID: {self._share_user_id}")
+        return self._share_user_id or ""

app/modules/douban/apiv2.py

@@ -12,10 +12,10 @@ import requests
 from app.core.cache import cached
 from app.core.config import settings
 from app.utils.http import RequestUtils
-from app.utils.singleton import Singleton
+from app.utils.singleton import WeakSingleton
 
 
-class DoubanApi(metaclass=Singleton):
+class DoubanApi(metaclass=WeakSingleton):
     _urls = {
         # 搜索类
         # sort=U:近期热门 T:标记最多 S:评分最高 R:最新上映
@@ -151,7 +151,6 @@ class DoubanApi(metaclass=Singleton):
     _api_key2 = "0ab215a8b1977939201640fa14c66bab"
     _base_url = "https://frodo.douban.com/api/v2"
     _api_url = "https://api.douban.com/v2"
-    _session = None
 
     def __init__(self):
         self._session = requests.Session()

app/modules/douban/douban_cache.py

@@ -10,7 +10,7 @@ from app.core.config import settings
 from app.core.meta import MetaBase
 from app.core.metainfo import MetaInfo
 from app.log import logger
-from app.utils.singleton import Singleton
+from app.utils.singleton import WeakSingleton
 from app.schemas.types import MediaType
 
 lock = RLock()
@@ -19,7 +19,7 @@ CACHE_EXPIRE_TIMESTAMP_STR = "cache_expire_timestamp"
 EXPIRE_TIMESTAMP = settings.CONF.meta
 
 
-class DoubanCache(metaclass=Singleton):
+class DoubanCache(metaclass=WeakSingleton):
     """
     豆瓣缓存数据
     {
@@ -29,9 +29,6 @@ class DoubanCache(metaclass=Singleton):
         "type": MediaType
     }
     """
-    _meta_data: dict = {}
-    # 缓存文件路径
-    _meta_path: Path = None
     # TMDB缓存过期
     _tmdb_cache_expire: bool = True
 
@@ -233,3 +230,6 @@ class DoubanCache(metaclass=Singleton):
         if not cache_media_info:
             return
         self._meta_data[key]['title'] = cn_title
+
+    def __del__(self):
+        self.save()

app/modules/filemanager/__init__.py

@@ -1,6 +1,7 @@
 from pathlib import Path
 from typing import Optional, List, Tuple, Union, Dict, Callable
 
+from app.chain.tmdb import TmdbChain
 from app.core.config import settings
 from app.core.context import MediaInfo
 from app.core.meta import MetaBase
@@ -141,11 +142,28 @@ class FileManagerModule(_ModuleBase):
         # 重命名格式
         rename_format = settings.TV_RENAME_FORMAT \
             if mediainfo.type == MediaType.TV else settings.MOVIE_RENAME_FORMAT
+        # 获取集信息
+        episodes_info: Optional[List[TmdbEpisode]] = None
+        if mediainfo.type == MediaType.TV:
+            # 判断注意season为0的情况
+            season_num = mediainfo.season
+            if season_num is None and meta.season_seq:
+                if meta.season_seq.isdigit():
+                    season_num = int(meta.season_seq)
+            # 默认值1
+            if season_num is None:
+                season_num = 1
+            episodes_info = TmdbChain().tmdb_episodes(
+                tmdbid=mediainfo.tmdb_id,
+                season=season_num,
+                episode_group=mediainfo.episode_group,
+            )
         # 获取重命名后的名称
         path = handler.get_rename_path(
             template_string=rename_format,
             rename_dict=handler.get_naming_dict(meta=meta,
                                                 mediainfo=mediainfo,
+                                                episodes_info=episodes_info,
                                                 file_ext=Path(meta.title).suffix)
         )
         return str(path)
@@ -529,17 +547,13 @@ class FileManagerModule(_ModuleBase):
                 rename_dict=handler.get_naming_dict(meta=meta,
                                                     mediainfo=mediainfo)
             )
-            # 计算重命名中的文件夹层数
-            rename_list = rename_format.split("/")
-            rename_format_level = len(rename_list) - 1
-            for level, name in enumerate(rename_list):
-                # 处理特例，有的人重命名第一层是年份、分辨率
-                if "{{title}}" in name:
-                    # 找出含标题的这一层作为扫描路径
-                    rename_format_level -= level
-                    break
-            # 取相对路径的第1层目录
-            media_path = target_path.parents[rename_format_level - 1]
+            # 获取重命名后的媒体文件根路径
+            media_path = DirectoryHelper.get_media_root_path(
+                rename_format, rename_path=target_path
+            )
+            if not media_path:
+                # 忽略
+                continue
             if dir_path.is_relative_to(media_path):
                 # 兜底检查，避免不必要的扫盘
                 logger.warn(f"{media_path} 是媒体库目录 {dir_path} 的父目录，忽略获取媒体文件列表，请检查重命名格式！")

app/modules/filemanager/storages/alipan.py

@@ -15,7 +15,7 @@ from app.core.config import settings
 from app.log import logger
 from app.modules.filemanager import StorageBase
 from app.schemas.types import StorageSchema
-from app.utils.singleton import Singleton
+from app.utils.singleton import WeakSingleton
 from app.utils.string import StringUtils
 
 lock = threading.Lock()
@@ -29,7 +29,7 @@ class SessionInvalidException(Exception):
     pass
 
 
-class AliPan(StorageBase, metaclass=Singleton):
+class AliPan(StorageBase, metaclass=WeakSingleton):
     """
     阿里云盘相关操作
     """
@@ -43,17 +43,12 @@ class AliPan(StorageBase, metaclass=Singleton):
         "copy": "复制"
     }
 
-    # 验证参数
-    _auth_state = {}
-
-    # 上传进度值
-    _last_progress = 0
-
     # 基础url
     base_url = "https://openapi.alipan.com"
 
     def __init__(self):
         super().__init__()
+        self._auth_state = {}
         self.session = requests.Session()
         self._init_session()
 
@@ -244,6 +239,7 @@ class AliPan(StorageBase, metaclass=Singleton):
         conf = self.get_conf()
         conf.update(result)
         self.set_config(conf)
+        return None
 
     def _request_api(self, method: str, endpoint: str,
                      result_key: Optional[str] = None, **kwargs) -> Optional[Union[dict, list]]:
@@ -369,7 +365,7 @@ class AliPan(StorageBase, metaclass=Singleton):
                 break
             next_marker = resp.get("next_marker")
             for item in resp.get("items", []):
-                items.append(self.__get_fileitem(item, parent=fileitem.path))
+                items.append(self.__get_fileitem(item, parent=str(fileitem.path)))
             if len(resp.get("items")) < 100:
                 break
         return items

app/modules/filemanager/storages/alist.py

@@ -12,11 +12,11 @@ from app.log import logger
 from app.modules.filemanager.storages import StorageBase
 from app.schemas.types import StorageSchema
 from app.utils.http import RequestUtils
-from app.utils.singleton import Singleton
+from app.utils.singleton import WeakSingleton
 from app.utils.url import UrlUtils
 
 
-class Alist(StorageBase, metaclass=Singleton):
+class Alist(StorageBase, metaclass=WeakSingleton):
     """
     Alist相关操作
     api文档：https://oplist.org/zh/
@@ -38,7 +38,7 @@ class Alist(StorageBase, metaclass=Singleton):
         """
         初始化
         """
-        self.__generate_token.clear_cache()
+        self.__generate_token.clear_cache() # noqa
 
     @property
     def __get_base_url(self) -> str:

app/modules/filemanager/storages/smb.py

@@ -12,17 +12,19 @@ from app.core.config import settings
 from app.log import logger
 from app.modules.filemanager import StorageBase
 from app.schemas.types import StorageSchema
-from app.utils.singleton import Singleton
+from app.utils.singleton import WeakSingleton
 
 lock = threading.Lock()
 
 
 class SMBConnectionError(Exception):
-    """SMB 连接错误"""
+    """
+    SMB 连接错误
+    """
     pass
 
 
-class SMB(StorageBase, metaclass=Singleton):
+class SMB(StorageBase, metaclass=WeakSingleton):
     """
     SMB网络挂载存储相关操作 - 使用 smbclient 高级接口
     """

app/modules/filemanager/storages/u115.py

@@ -18,7 +18,7 @@ from app.core.config import settings
 from app.log import logger
 from app.modules.filemanager import StorageBase
 from app.schemas.types import StorageSchema
-from app.utils.singleton import Singleton
+from app.utils.singleton import WeakSingleton
 from app.utils.string import StringUtils
 
 lock = threading.Lock()
@@ -28,7 +28,7 @@ class NoCheckInException(Exception):
     pass
 
 
-class U115Pan(StorageBase, metaclass=Singleton):
+class U115Pan(StorageBase, metaclass=WeakSingleton):
     """
     115相关操作
     """
@@ -41,18 +41,12 @@ class U115Pan(StorageBase, metaclass=Singleton):
         "move": "移动",
         "copy": "复制"
     }
-
-    # 验证参数
-    _auth_state = {}
-
-    # 上传进度值
-    _last_progress = 0
-
     # 基础url
     base_url = "https://proapi.115.com"
 
     def __init__(self):
         super().__init__()
+        self._auth_state = {}
         self.session = requests.Session()
         self._init_session()
 
@@ -492,7 +486,8 @@ class U115Pan(StorageBase, metaclass=Singleton):
                         type="file" if info_resp["file_category"] == "1" else "dir",
                         name=info_resp["file_name"],
                         basename=Path(info_resp["file_name"]).stem,
-                        extension=Path(info_resp["file_name"]).suffix[1:] if info_resp["file_category"] == "1" else None,
+                        extension=Path(info_resp["file_name"]).suffix[1:] if info_resp[
+                                                                                 "file_category"] == "1" else None,
                         pickcode=info_resp["pick_code"],
                         size=StringUtils.num_filesize(info_resp['size']) if info_resp["file_category"] == "1" else None,
                         modify_time=info_resp["utime"]

app/modules/filemanager/transhandler.py

@@ -10,6 +10,7 @@ from app.core.context import MediaInfo
 from app.core.event import eventmanager
 from app.core.meta import MetaBase
 from app.core.metainfo import MetaInfoPath
+from app.helper.directory import DirectoryHelper
 from app.helper.message import TemplateHelper
 from app.log import logger
 from app.modules.filemanager.storages import StorageBase
@@ -26,11 +27,10 @@ class TransHandler:
     文件转移整理类
     """
 
-    result: Optional[TransferInfo] = None
     inner_lock: Lock = Lock()
 
     def __init__(self):
-        self.__reset_result()
+        self.result = None
 
     def __reset_result(self):
         """
@@ -103,185 +103,212 @@ class TransHandler:
         # 重置结果
         self.__reset_result()
 
-        # 重命名格式
-        rename_format = settings.TV_RENAME_FORMAT \
-            if mediainfo.type == MediaType.TV else settings.MOVIE_RENAME_FORMAT
+        try:
 
-        # 判断是否为文件夹
-        if fileitem.type == "dir":
-            # 整理整个目录，一般为蓝光原盘
-            if need_rename:
-                new_path = self.get_rename_path(
-                    path=target_path,
-                    template_string=rename_format,
-                    rename_dict=self.get_naming_dict(meta=in_meta,
-                                                     mediainfo=mediainfo)
-                ).parent
-            else:
-                new_path = target_path / fileitem.name
-            # 整理目录
-            new_diritem, errmsg = self.__transfer_dir(fileitem=fileitem,
-                                                      mediainfo=mediainfo,
-                                                      source_oper=source_oper,
-                                                      target_oper=target_oper,
-                                                      target_storage=target_storage,
-                                                      target_path=new_path,
-                                                      transfer_type=transfer_type)
-            if not new_diritem:
-                logger.error(f"文件夹 {fileitem.path} 整理失败：{errmsg}")
-                self.__set_result(success=False,
-                                  message=errmsg,
-                                  fileitem=fileitem,
-                                  transfer_type=transfer_type,
-                                  need_notify=need_notify)
-                return self.result
+            # 重命名格式
+            rename_format = settings.TV_RENAME_FORMAT \
+                if mediainfo.type == MediaType.TV else settings.MOVIE_RENAME_FORMAT
 
-            logger.info(f"文件夹 {fileitem.path} 整理成功")
-            # 计算目录下所有文件大小
-            total_size = sum(file.stat().st_size for file in Path(fileitem.path).rglob('*') if file.is_file())
-            # 返回整理后的路径
-            self.__set_result(success=True,
-                              fileitem=fileitem,
-                              target_item=new_diritem,
-                              target_diritem=new_diritem,
-                              total_size=total_size,
-                              need_scrape=need_scrape,
-                              need_notify=need_notify,
-                              transfer_type=transfer_type)
-            return self.result
-        else:
-            # 整理单个文件
-            if mediainfo.type == MediaType.TV:
-                # 电视剧
-                if in_meta.begin_episode is None:
-                    logger.warn(f"文件 {fileitem.path} 整理失败：未识别到文件集数")
+            # 判断是否为文件夹
+            if fileitem.type == "dir":
+                # 整理整个目录，一般为蓝光原盘
+                if need_rename:
+                    new_path = self.get_rename_path(
+                        path=target_path,
+                        template_string=rename_format,
+                        rename_dict=self.get_naming_dict(meta=in_meta,
+                                                         mediainfo=mediainfo)
+                    )
+                    new_path = DirectoryHelper.get_media_root_path(
+                        rename_format, rename_path=new_path
+                    )
+                    if not new_path:
+                        self.__set_result(
+                            success=False,
+                            message="重命名格式无效",
+                            fileitem=fileitem,
+                            transfer_type=transfer_type,
+                            need_notify=need_notify,
+                        )
+                        return self.result.copy()
+                else:
+                    new_path = target_path / fileitem.name
+                # 整理目录
+                new_diritem, errmsg = self.__transfer_dir(fileitem=fileitem,
+                                                          mediainfo=mediainfo,
+                                                          source_oper=source_oper,
+                                                          target_oper=target_oper,
+                                                          target_storage=target_storage,
+                                                          target_path=new_path,
+                                                          transfer_type=transfer_type)
+                if not new_diritem:
+                    logger.error(f"文件夹 {fileitem.path} 整理失败：{errmsg}")
                     self.__set_result(success=False,
-                                      message=f"未识别到文件集数",
+                                      message=errmsg,
+                                      fileitem=fileitem,
+                                      transfer_type=transfer_type,
+                                      need_notify=need_notify)
+                    return self.result.copy()
+
+                logger.info(f"文件夹 {fileitem.path} 整理成功")
+                # 计算目录下所有文件大小
+                total_size = sum(file.stat().st_size for file in Path(fileitem.path).rglob('*') if file.is_file())
+                # 返回整理后的路径
+                self.__set_result(success=True,
+                                  fileitem=fileitem,
+                                  target_item=new_diritem,
+                                  target_diritem=new_diritem,
+                                  total_size=total_size,
+                                  need_scrape=need_scrape,
+                                  need_notify=need_notify,
+                                  transfer_type=transfer_type)
+                return self.result.copy()
+            else:
+                # 整理单个文件
+                if mediainfo.type == MediaType.TV:
+                    # 电视剧
+                    if in_meta.begin_episode is None:
+                        logger.warn(f"文件 {fileitem.path} 整理失败：未识别到文件集数")
+                        self.__set_result(success=False,
+                                          message="未识别到文件集数",
+                                          fileitem=fileitem,
+                                          fail_list=[fileitem.path],
+                                          transfer_type=transfer_type,
+                                          need_notify=need_notify)
+                        return self.result.copy()
+
+                    # 文件结束季为空
+                    in_meta.end_season = None
+                    # 文件总季数为1
+                    if in_meta.total_season:
+                        in_meta.total_season = 1
+                    # 文件不可能超过2集
+                    if in_meta.total_episode > 2:
+                        in_meta.total_episode = 1
+                        in_meta.end_episode = None
+
+                # 目的文件名
+                if need_rename:
+                    new_file = self.get_rename_path(
+                        path=target_path,
+                        template_string=rename_format,
+                        rename_dict=self.get_naming_dict(
+                            meta=in_meta,
+                            mediainfo=mediainfo,
+                            episodes_info=episodes_info,
+                            file_ext=f".{fileitem.extension}"
+                        )
+                    )
+                    folder_path = DirectoryHelper.get_media_root_path(
+                        rename_format, rename_path=new_file
+                    )
+                    if not folder_path:
+                        self.__set_result(
+                            success=False,
+                            message="重命名格式无效",
+                            fileitem=fileitem,
+                            fail_list=[fileitem.path],
+                            transfer_type=transfer_type,
+                            need_notify=need_notify,
+                        )
+                        return self.result.copy()
+                else:
+                    new_file = target_path / fileitem.name
+                    folder_path = target_path
+
+                # 判断是否要覆盖
+                overflag = False
+                # 目标目录
+                target_diritem = target_oper.get_folder(folder_path)
+                if not target_diritem:
+                    logger.error(f"目标目录 {folder_path} 获取失败")
+                    self.__set_result(success=False,
+                                      message=f"目标目录 {folder_path} 获取失败",
                                       fileitem=fileitem,
                                       fail_list=[fileitem.path],
                                       transfer_type=transfer_type,
                                       need_notify=need_notify)
-                    return self.result
-
-                # 文件结束季为空
-                in_meta.end_season = None
-                # 文件总季数为1
-                if in_meta.total_season:
-                    in_meta.total_season = 1
-                # 文件不可能超过2集
-                if in_meta.total_episode > 2:
-                    in_meta.total_episode = 1
-                    in_meta.end_episode = None
-
-            # 目的文件名
-            if need_rename:
-                new_file = self.get_rename_path(
-                    path=target_path,
-                    template_string=rename_format,
-                    rename_dict=self.get_naming_dict(
-                        meta=in_meta,
-                        mediainfo=mediainfo,
-                        episodes_info=episodes_info,
-                        file_ext=f".{fileitem.extension}"
-                    )
-                )
-            else:
-                new_file = target_path / fileitem.name
-
-            # 判断是否要覆盖
-            overflag = False
-            # 计算重命名中的文件夹层级
-            rename_format_level = len(rename_format.split("/")) - 1
-            folder_path = new_file.parents[rename_format_level - 1]
-            # 目标目录
-            target_diritem = target_oper.get_folder(folder_path)
-            if not target_diritem:
-                logger.error(f"目标目录 {folder_path} 获取失败")
-                self.__set_result(success=False,
-                                  message=f"目标目录 {folder_path} 获取失败",
-                                  fileitem=fileitem,
-                                  fail_list=[fileitem.path],
-                                  transfer_type=transfer_type,
-                                  need_notify=need_notify)
-                return self.result
-            # 目标文件
-            target_item = target_oper.get_item(new_file)
-            if target_item:
-                # 目标文件已存在
-                target_file = new_file
-                if target_storage == "local" and new_file.is_symlink():
-                    target_file = new_file.readlink()
-                    if not target_file.exists():
-                        overflag = True
-                if not overflag:
+                    return self.result.copy()
+                # 目标文件
+                target_item = target_oper.get_item(new_file)
+                if target_item:
                     # 目标文件已存在
-                    logger.info(f"目的文件系统中已经存在同名文件 {target_file}，当前整理覆盖模式设置为 {overwrite_mode}")
-                    if overwrite_mode == 'always':
-                        # 总是覆盖同名文件
-                        overflag = True
-                    elif overwrite_mode == 'size':
-                        # 存在时大覆盖小
-                        if target_item.size < fileitem.size:
-                            logger.info(f"目标文件文件大小更小，将覆盖：{new_file}")
+                    target_file = new_file
+                    if target_storage == "local" and new_file.is_symlink():
+                        target_file = new_file.readlink()
+                        if not target_file.exists():
                             overflag = True
-                        else:
+                    if not overflag:
+                        # 目标文件已存在
+                        logger.info(f"目的文件系统中已经存在同名文件 {target_file}，当前整理覆盖模式设置为 {overwrite_mode}")
+                        if overwrite_mode == 'always':
+                            # 总是覆盖同名文件
+                            overflag = True
+                        elif overwrite_mode == 'size':
+                            # 存在时大覆盖小
+                            if target_item.size < fileitem.size:
+                                logger.info(f"目标文件文件大小更小，将覆盖：{new_file}")
+                                overflag = True
+                            else:
+                                self.__set_result(success=False,
+                                                  message=f"媒体库存在同名文件，且质量更好",
+                                                  fileitem=fileitem,
+                                                  target_item=target_item,
+                                                  target_diritem=target_diritem,
+                                                  fail_list=[fileitem.path],
+                                                  transfer_type=transfer_type,
+                                                  need_notify=need_notify)
+                                return self.result.copy()
+                        elif overwrite_mode == 'never':
+                            # 存在不覆盖
                             self.__set_result(success=False,
-                                              message=f"媒体库存在同名文件，且质量更好",
+                                              message=f"媒体库存在同名文件，当前覆盖模式为不覆盖",
                                               fileitem=fileitem,
                                               target_item=target_item,
                                               target_diritem=target_diritem,
                                               fail_list=[fileitem.path],
                                               transfer_type=transfer_type,
                                               need_notify=need_notify)
-                            return self.result
-                    elif overwrite_mode == 'never':
-                        # 存在不覆盖
-                        self.__set_result(success=False,
-                                          message=f"媒体库存在同名文件，当前覆盖模式为不覆盖",
-                                          fileitem=fileitem,
-                                          target_item=target_item,
-                                          target_diritem=target_diritem,
-                                          fail_list=[fileitem.path],
-                                          transfer_type=transfer_type,
-                                          need_notify=need_notify)
-                        return self.result
-                    elif overwrite_mode == 'latest':
-                        # 仅保留最新版本
-                        logger.info(f"当前整理覆盖模式设置为仅保留最新版本，将覆盖：{new_file}")
-                        overflag = True
-            else:
-                if overwrite_mode == 'latest':
-                    # 文件不存在，但仅保留最新版本
-                    logger.info(f"当前整理覆盖模式设置为 {overwrite_mode}，仅保留最新版本，正在删除已有版本文件 ...")
-                    self.__delete_version_files(target_oper, new_file)
-            # 整理文件
-            new_item, err_msg = self.__transfer_file(fileitem=fileitem,
-                                                     mediainfo=mediainfo,
-                                                     target_storage=target_storage,
-                                                     target_file=new_file,
-                                                     transfer_type=transfer_type,
-                                                     over_flag=overflag,
-                                                     source_oper=source_oper,
-                                                     target_oper=target_oper)
-            if not new_item:
-                logger.error(f"文件 {fileitem.path} 整理失败：{err_msg}")
-                self.__set_result(success=False,
-                                  message=err_msg,
+                            return self.result.copy()
+                        elif overwrite_mode == 'latest':
+                            # 仅保留最新版本
+                            logger.info(f"当前整理覆盖模式设置为仅保留最新版本，将覆盖：{new_file}")
+                            overflag = True
+                else:
+                    if overwrite_mode == 'latest':
+                        # 文件不存在，但仅保留最新版本
+                        logger.info(f"当前整理覆盖模式设置为 {overwrite_mode}，仅保留最新版本，正在删除已有版本文件 ...")
+                        self.__delete_version_files(target_oper, new_file)
+                # 整理文件
+                new_item, err_msg = self.__transfer_file(fileitem=fileitem,
+                                                         mediainfo=mediainfo,
+                                                         target_storage=target_storage,
+                                                         target_file=new_file,
+                                                         transfer_type=transfer_type,
+                                                         over_flag=overflag,
+                                                         source_oper=source_oper,
+                                                         target_oper=target_oper)
+                if not new_item:
+                    logger.error(f"文件 {fileitem.path} 整理失败：{err_msg}")
+                    self.__set_result(success=False,
+                                      message=err_msg,
+                                      fileitem=fileitem,
+                                      fail_list=[fileitem.path],
+                                      transfer_type=transfer_type,
+                                      need_notify=need_notify)
+                    return self.result.copy()
+
+                logger.info(f"文件 {fileitem.path} 整理成功")
+                self.__set_result(success=True,
                                   fileitem=fileitem,
-                                  fail_list=[fileitem.path],
+                                  target_item=new_item,
+                                  target_diritem=target_diritem,
+                                  need_scrape=need_scrape,
                                   transfer_type=transfer_type,
                                   need_notify=need_notify)
-                return self.result
-
-            logger.info(f"文件 {fileitem.path} 整理成功")
-            self.__set_result(success=True,
-                              fileitem=fileitem,
-                              target_item=new_item,
-                              target_diritem=target_diritem,
-                              need_scrape=need_scrape,
-                              transfer_type=transfer_type,
-                              need_notify=need_notify)
-            return self.result
+                return self.result.copy()
+        finally:
+            self.result = None
 
     @staticmethod
     def __transfer_command(fileitem: FileItem, target_storage: str,

app/modules/themoviedb/category.py

@@ -7,19 +7,19 @@ from ruamel.yaml import CommentedMap
 
 from app.core.config import settings
 from app.log import logger
-from app.utils.singleton import Singleton
+from app.utils.singleton import WeakSingleton
 
 
-class CategoryHelper(metaclass=Singleton):
+class CategoryHelper(metaclass=WeakSingleton):
     """
     二级分类
     """
-    _categorys = {}
-    _movie_categorys = {}
-    _tv_categorys = {}
 
     def __init__(self):
         self._category_path: Path = settings.CONFIG_PATH / "category.yaml"
+        self._categorys = {}
+        self._movie_categorys = {}
+        self._tv_categorys = {}
         self.init()
 
     def init(self):
@@ -69,7 +69,7 @@ class CategoryHelper(metaclass=Singleton):
         """
         if not self._movie_categorys:
             return []
-        return self._movie_categorys.keys()
+        return list(self._movie_categorys.keys())
 
     @property
     def tv_categorys(self) -> list:
@@ -78,7 +78,7 @@ class CategoryHelper(metaclass=Singleton):
         """
         if not self._tv_categorys:
             return []
-        return self._tv_categorys.keys()
+        return list(self._tv_categorys.keys())
 
     def get_movie_category(self, tmdb_info) -> str:
         """
@@ -127,7 +127,7 @@ class CategoryHelper(metaclass=Singleton):
                     continue
                 elif attr == "production_countries":
                     # 制片国家
-                    info_values = [str(val.get("iso_3166_1")).upper() for val in info_value]
+                    info_values = [str(val.get("iso_3166_1")).upper() for val in info_value] # type: ignore
                 else:
                     if isinstance(info_value, list):
                         info_values = [str(val).upper() for val in info_value]

app/modules/themoviedb/tmdb_cache.py

@@ -9,7 +9,7 @@ from typing import Optional
 from app.core.config import settings
 from app.core.meta import MetaBase
 from app.log import logger
-from app.utils.singleton import Singleton
+from app.utils.singleton import WeakSingleton
 from app.schemas.types import MediaType
 
 lock = RLock()
@@ -18,7 +18,7 @@ CACHE_EXPIRE_TIMESTAMP_STR = "cache_expire_timestamp"
 EXPIRE_TIMESTAMP = settings.CONF.meta
 
 
-class TmdbCache(metaclass=Singleton):
+class TmdbCache(metaclass=WeakSingleton):
     """
     TMDB缓存数据
     {
@@ -28,9 +28,6 @@ class TmdbCache(metaclass=Singleton):
         "type": MediaType
     }
     """
-    _meta_data: dict = {}
-    # 缓存文件路径
-    _meta_path: Path = None
     # TMDB缓存过期
     _tmdb_cache_expire: bool = True
 
@@ -218,3 +215,6 @@ class TmdbCache(metaclass=Singleton):
         if not cache_media_info:
             return
         self._meta_data[key]['title'] = cn_title
+
+    def __del__(self):
+        self.save()

app/modules/trimemedia/api.py

@@ -320,7 +320,7 @@ class Api:
         types=None,
         exclude_grouped_video=True,
         page=1,
-        page_size=22,
+        page_size=20,
         sort_by="create_time",
         sort="DESC",
     ) -> Optional[list[Item]]:

app/monitor.py

@@ -59,26 +59,23 @@ class Monitor(metaclass=Singleton):
     目录监控处理链，单例模式
     """
 
-    # 退出事件
-    _event = threading.Event()
 
-    # 监控服务
-    _observers = []
-
-    # 定时服务
-    _scheduler = None
-
-    # 存储快照缓存目录
-    _snapshot_cache_dir = None
-
-    # 存储过照间隔（分钟）
-    _snapshot_interval = 5
-
-    # TTL缓存，10秒钟有效
-    _cache = TTLCache(maxsize=1024, ttl=10)
 
     def __init__(self):
         super().__init__()
+        # 退出事件
+        self._event = threading.Event()
+        # 监控服务
+        self._observers = []
+        # 定时服务
+        self._scheduler = None
+        # 存储快照缓存目录
+        self._snapshot_cache_dir = None
+        # 存储过照间隔（分钟）
+        self._snapshot_interval = 5
+        # TTL缓存，10秒钟有效
+        self._cache = TTLCache(maxsize=1024, ttl=10)
+        # 监控的文件扩展名
         self.all_exts = settings.RMT_MEDIAEXT
         # 初始化快照缓存目录
         self._snapshot_cache_dir = settings.TEMP_PATH / "snapshots"

app/scheduler.py

@@ -40,20 +40,20 @@ class Scheduler(metaclass=Singleton):
     """
     定时任务管理
     """
-    # 定时服务
-    _scheduler = None
-    # 退出事件
-    _event = threading.Event()
-    # 锁
-    _lock = threading.RLock()
-    # 各服务的运行状态
-    _jobs = {}
-    # 用户认证失败次数
-    _auth_count = 0
-    # 用户认证失败消息发送
-    _auth_message = False
 
     def __init__(self):
+        # 定时服务
+        self._scheduler = None
+        # 退出事件
+        self._event = threading.Event()
+        # 锁
+        self._lock = threading.RLock()
+        # 各服务的运行状态
+        self._jobs = {}
+        # 用户认证失败次数
+        self._auth_count = 0
+        # 用户认证失败消息发送
+        self._auth_message = False
         self.init()
 
     @eventmanager.register(EventType.ConfigChanged)
@@ -443,7 +443,7 @@ class Scheduler(metaclass=Singleton):
             return
         with self._lock:
             job_id = f"workflow-{workflow.id}"
-            service = self._jobs.pop(job_id, None)
+            service = self._jobs.pop(job_id, {})
             if not service:
                 return
             try:

app/schemas/event.py

@@ -214,7 +214,7 @@ class ResourceDownloadEventData(ChainEventData):
     channel: Optional[MessageChannel] = Field(None, description="通知渠道")
     origin: Optional[str] = Field(None, description="来源")
     downloader: Optional[str] = Field(None, description="下载器")
-    options: Optional[dict] = Field(None, description="其他参数")
+    options: Optional[dict] = Field(default={}, description="其他参数")
 
     # 输出参数
     cancel: bool = Field(default=False, description="是否取消下载")

app/schemas/workflow.py

@@ -82,3 +82,25 @@ class ActionFlow(BaseModel):
     source: Optional[str] = Field(default=None, description="源动作")
     target: Optional[str] = Field(default=None, description="目标动作")
     animated: Optional[bool] = Field(default=True, description="是否动画流程")
+
+
+class WorkflowShare(BaseModel):
+    """
+    工作流分享信息
+    """
+    id: Optional[int] = Field(default=None, description="分享ID")
+    share_title: Optional[str] = Field(default=None, description="分享标题")
+    share_comment: Optional[str] = Field(default=None, description="分享说明")
+    share_user: Optional[str] = Field(default=None, description="分享人")
+    share_uid: Optional[str] = Field(default=None, description="分享人唯一ID")
+    name: Optional[str] = Field(default=None, description="工作流名称")
+    description: Optional[str] = Field(default=None, description="工作流描述")
+    timer: Optional[str] = Field(default=None, description="定时器")
+    actions: Optional[str] = Field(default=None, description="任务列表(JSON字符串)")
+    flows: Optional[str] = Field(default=None, description="任务流(JSON字符串)")
+    context: Optional[str] = Field(default=None, description="执行上下文(JSON字符串)")
+    date: Optional[str] = Field(default=None, description="分享时间")
+    count: Optional[int] = Field(default=0, description="复用人次")
+
+    class Config:
+        orm_mode = True

app/startup/lifecycle.py

@@ -5,7 +5,6 @@ from fastapi import FastAPI
 
 from app.chain.system import SystemChain
 from app.startup.command_initializer import init_command, stop_command, restart_command
-from app.startup.memory_initializer import init_memory_manager, stop_memory_manager
 from app.startup.modules_initializer import init_modules, stop_modules
 from app.startup.monitor_initializer import stop_monitor, init_monitor
 from app.startup.plugins_initializer import init_plugins, stop_plugins, sync_plugins
@@ -52,8 +51,6 @@ async def lifespan(app: FastAPI):
     init_command()
     # 初始化工作流
     init_workflow()
-    # 初始化内存管理
-    init_memory_manager()
     # 插件同步到本地
     sync_plugins_task = asyncio.create_task(init_extra())
     try:
@@ -71,8 +68,6 @@ async def lifespan(app: FastAPI):
             print(str(e))
         # 备份插件
         SystemChain().backup_plugins()
-        # 停止内存管理器
-        stop_memory_manager()
         # 停止工作流
         stop_workflow()
         # 停止命令

app/startup/memory_initializer.py

@@ -1,15 +0,0 @@
-from app.helper.memory import MemoryHelper
-
-
-def init_memory_manager():
-    """
-    初始化内存监控器
-    """
-    MemoryHelper().start_monitoring()
-
-
-def stop_memory_manager():
-    """
-    停止内存监控器
-    """
-    MemoryHelper().stop_monitoring()

app/utils/singleton.py

@@ -1,4 +1,6 @@
 import abc
+import threading
+import weakref
 
 
 class Singleton(abc.ABCMeta, type):
@@ -40,3 +42,17 @@ class AbstractSingletonClass(abc.ABC, metaclass=SingletonClass):
     抽像类单例模式（按类）
     """
     pass
+
+
+class WeakSingleton(abc.ABCMeta, type):
+    """
+    弱引用单例模式 - 当没有强引用时自动清理
+    """
+    _instances: weakref.WeakKeyDictionary = weakref.WeakKeyDictionary()
+    _lock = threading.RLock()
+
+    def __call__(cls, *args, **kwargs):
+        with cls._lock:
+            if cls not in cls._instances:
+                cls._instances[cls] = super().__call__(*args, **kwargs)
+            return cls._instances[cls]

version.py

@@ -1,2 +1,2 @@
-APP_VERSION = 'v2.6.3'
-FRONTEND_VERSION = 'v2.6.3'
+APP_VERSION = 'v2.6.4'
+FRONTEND_VERSION = 'v2.6.4'