更新 version.py

Co-authored-by: jxxghp <51039935+jxxghp@users.noreply.github.com>

app/agent/__init__.py

@@ -1,12 +1,17 @@
 import asyncio
-from typing import Dict, List, Any
+from typing import Dict, List, Any, Union
+import json
+import tiktoken
 
-from langchain.agents import AgentExecutor, create_openai_tools_agent
+from langchain.agents import AgentExecutor
 from langchain.prompts import ChatPromptTemplate, MessagesPlaceholder
 from langchain_community.callbacks import get_openai_callback
 from langchain_core.chat_history import InMemoryChatMessageHistory
-from langchain_core.messages import HumanMessage, AIMessage, ToolCall, ToolMessage, SystemMessage
+from langchain_core.messages import HumanMessage, AIMessage, ToolCall, ToolMessage, SystemMessage, trim_messages
+from langchain_core.runnables import RunnablePassthrough, RunnableLambda
 from langchain_core.runnables.history import RunnableWithMessageHistory
+from langchain.agents.format_scratchpad.openai_tools import format_to_openai_tool_messages
+from langchain.agents.output_parsers.openai_tools import OpenAIToolsAgentOutputParser
 
 from app.agent.callback import StreamingCallbackHandler
 from app.agent.memory import conversation_manager
@@ -120,6 +125,7 @@ class MoviePilotAgent:
                     ))
                 elif msg.get("role") == "system":
                     chat_history.add_message(SystemMessage(content=msg.get("content", "")))
+        
         return chat_history
 
     @staticmethod
@@ -140,15 +146,140 @@ class MoviePilotAgent:
             logger.error(f"初始化提示词失败: {e}")
             raise e
 
+    @staticmethod
+    def _token_counter(messages: List[Union[HumanMessage, AIMessage, ToolMessage, SystemMessage]]) -> int:
+        """
+        通用的Token计数器
+        """
+        try:
+            # 尝试从模型获取编码集，如果失败则回退到 cl100k_base (大多数现代模型使用的编码)
+            try:
+                encoding = tiktoken.encoding_for_model(settings.LLM_MODEL)
+            except KeyError:
+                encoding = tiktoken.get_encoding("cl100k_base")
+
+            num_tokens = 0
+            for message in messages:
+                # 基础开销 (每个消息大约 3 个 token)
+                num_tokens += 3
+                
+                # 1. 处理文本内容 (content)
+                if isinstance(message.content, str):
+                    num_tokens += len(encoding.encode(message.content))
+                elif isinstance(message.content, list):
+                    for part in message.content:
+                        if isinstance(part, dict) and part.get("type") == "text":
+                            num_tokens += len(encoding.encode(part.get("text", "")))
+
+                # 2. 处理工具调用 (仅 AIMessage 包含 tool_calls)
+                if getattr(message, "tool_calls", None):
+                    for tool_call in message.tool_calls:
+                        # 函数名
+                        num_tokens += len(encoding.encode(tool_call.get("name", "")))
+                        # 参数 (转为 JSON 估算)
+                        args_str = json.dumps(tool_call.get("args", {}), ensure_ascii=False)
+                        num_tokens += len(encoding.encode(args_str))
+                        # 额外的结构开销 (ID 等)
+                        num_tokens += 3
+
+                # 3. 处理角色权重
+                num_tokens += 1
+
+            # 加上回复的起始 Token (大约 3 个 token)
+            num_tokens += 3
+            return num_tokens
+        except Exception as e:
+            logger.error(f"Token计数失败: {e}")
+            # 发生错误时返回一个保守的估算值
+            return len(str(messages)) // 4
+
     def _create_agent_executor(self) -> RunnableWithMessageHistory:
         """
         创建Agent执行器
         """
         try:
-            agent = create_openai_tools_agent(
-                llm=self.llm,
-                tools=self.tools,
-                prompt=self.prompt
+            # 消息裁剪器，防止上下文超出限制
+            base_trimmer = trim_messages(
+                max_tokens=settings.LLM_MAX_CONTEXT_TOKENS * 1000 * 0.8,
+                strategy="last",
+                token_counter=self._token_counter,
+                include_system=True,
+                allow_partial=False,
+                start_on="human",
+            )
+            
+            # 包装trimmer，在裁剪后验证工具调用的完整性
+            def validated_trimmer(messages):
+                # 如果输入是 PromptValue，转换为消息列表
+                if hasattr(messages, "to_messages"):
+                    messages = messages.to_messages()
+                trimmed = base_trimmer.invoke(messages)
+
+                # 二次校验：确保不出现 broken tool chains
+                # 1. AIMessage with tool_calls 必须紧跟着对应的 ToolMessage
+                # 2. ToolMessage 必须有对应的 AIMessage 前置
+                safe_messages = []
+                i = 0
+                while i < len(trimmed):
+                    msg = trimmed[i]
+
+                    if isinstance(msg, AIMessage) and getattr(msg, "tool_calls", None):
+                        # 检查工具调用序列是否完整
+                        tool_calls = msg.tool_calls
+                        is_valid_sequence = True
+                        tool_results = []
+                        
+                        # 向后查找对应的 ToolMessage
+                        temp_i = i + 1
+                        for tool_call in tool_calls:
+                            if temp_i >= len(trimmed):
+                                is_valid_sequence = False
+                                break
+                            
+                            next_msg = trimmed[temp_i]
+                            if isinstance(next_msg, ToolMessage) and next_msg.tool_call_id == tool_call.get("id"):
+                                tool_results.append(next_msg)
+                                temp_i += 1
+                            else:
+                                is_valid_sequence = False
+                                break
+                        
+                        if is_valid_sequence:
+                            # 序列完整，保留消息
+                            safe_messages.append(msg)
+                            safe_messages.extend(tool_results)
+                            i = temp_i  # 跳过已处理的工具结果
+                        else:
+                            # 序列不完整，丢弃该 AIMessage（后续的孤立 ToolMessage 会在下一次循环被当做 orphaned 处理掉）
+                            logger.warning(f"移除无效的工具调用链: {len(tool_calls)} calls, incomplete results")
+                            i += 1
+                        continue
+
+                    if isinstance(msg, ToolMessage):
+                        # 如果在这里遇到 ToolMessage，说明它没有被上面的逻辑消费，则是孤立的（或者顺序错乱）
+                        logger.warning("移除孤立的 ToolMessage")
+                        i += 1
+                        continue
+
+                    # 其他类型的消息直接保留
+                    safe_messages.append(msg)
+                    i += 1
+
+                if len(safe_messages) < len(messages):
+                    logger.info(f"LangChain消息上下文已裁剪: {len(messages)} -> {len(safe_messages)}")
+                return safe_messages
+            
+            # 创建Agent执行链
+            agent = (
+                RunnablePassthrough.assign(
+                    agent_scratchpad=lambda x: format_to_openai_tool_messages(
+                        x["intermediate_steps"]
+                    )
+                )
+                | self.prompt
+                | RunnableLambda(validated_trimmer)
+                | self.llm.bind_tools(self.tools)
+                | OpenAIToolsAgentOutputParser()
             )
             executor = AgentExecutor(
                 agent=agent,
@@ -169,11 +300,81 @@ class MoviePilotAgent:
             logger.error(f"创建Agent执行器失败: {e}")
             raise e
 
+    async def _summarize_history(self):
+        """
+        总结提炼之前的对话和工具执行情况，并把会话总结变成新的系统提示词取代之前的对话
+        """
+        try:
+            # 获取当前历史记录
+            chat_history = self.get_session_history(self.session_id)
+            messages = chat_history.messages
+            if not messages:
+                return
+
+            logger.info(f"会话 {self.session_id} 历史消息长度已超过 90%，开始总结并重置上下文...")
+
+            # 将消息转换为摘要所需的文本格式
+            history_text = ""
+            for msg in messages:
+                if isinstance(msg, HumanMessage):
+                    history_text += f"用户: {msg.content}\n"
+                elif isinstance(msg, AIMessage):
+                    history_text += f"智能体: {msg.content}\n"
+                    if getattr(msg, "tool_calls", None):
+                        for tool_call in msg.tool_calls:
+                            history_text += f"智能体调用工具: {tool_call.get('name')}，参数: {tool_call.get('args')}\n"
+                elif isinstance(msg, ToolMessage):
+                    history_text += f"工具响应: {msg.content}\n"
+                elif isinstance(msg, SystemMessage):
+                    history_text += f"系统: {msg.content}\n"
+
+            # 摘要提示词
+            summary_prompt = (
+                "Please provide a comprehensive and highly informational summary of the preceding conversation and tool executions. "
+                "Your goal is to condense the history while retaining all critical details for future reference. "
+                "Ensure you include:\n"
+                "1. User's core intents, specific requests, and any mentioned preferences.\n"
+                "2. Names of movies, TV shows, or other key entities discussed.\n"
+                "3. A concise log of tool calls made and their specific results/outcomes.\n"
+                "4. The current status of any tasks and any pending actions.\n"
+                "5. Any important context that would be necessary for the agent to continue the conversation seamlessly.\n"
+                "The summary should be dense with information and serve as the primary context for the next stage of the interaction."
+            )
+
+            # 调用 LLM 进行总结 (非流式)
+            summary_llm = LLMHelper.get_llm(streaming=False)
+            response = await summary_llm.ainvoke([
+                SystemMessage(content=summary_prompt),
+                HumanMessage(content=f"Here is the conversation history to summarize:\n{history_text}")
+            ])
+            summary_content = str(response.content)
+
+            if not summary_content:
+                logger.warning("总结生成失败，跳过重置逻辑。")
+                return
+
+            # 清空原有的会话记录并插入新的系统总结
+            await conversation_manager.clear_memory(self.session_id, self.user_id)
+            await conversation_manager.add_conversation(
+                session_id=self.session_id,
+                user_id=self.user_id,
+                role="system",
+                content=f"<history_summary>\n{summary_content}\n</history_summary>"
+            )
+            logger.info(f"会话 {self.session_id} 历史摘要替换完成。")
+        except Exception as e:
+            logger.error(f"执行会话总结出错: {str(e)}")
+
     async def process_message(self, message: str) -> str:
         """
         处理用户消息
         """
         try:
+            # 检查上下文长度是否超过 90%
+            history = self.get_session_history(self.session_id)
+            if self._token_counter(history.messages) > settings.LLM_MAX_CONTEXT_TOKENS * 1000 * 0.9:
+                await self._summarize_history()
+
             # 添加用户消息到记忆
             await conversation_manager.add_conversation(
                 self.session_id,
@@ -190,7 +391,8 @@ class MoviePilotAgent:
 
             # 执行Agent
             logger.info(f"Agent执行推理: session_id={self.session_id}, input={message}")
-            await self._execute_agent(input_context)
+
+            result = await self._execute_agent(input_context)
 
             # 获取Agent回复
             agent_message = await self.callback_handler.get_message()
@@ -208,7 +410,7 @@ class MoviePilotAgent:
                     content=agent_message
                 )
             else:
-                agent_message = "很抱歉，智能体出错了，未能生成回复内容。"
+                agent_message = result.get("output") or "很抱歉，智能体出错了，未能生成回复内容。"
                 await self.send_agent_message(agent_message)
 
             return agent_message
@@ -250,7 +452,7 @@ class MoviePilotAgent:
         except Exception as e:
             logger.error(f"Agent执行失败: {e}")
             return {
-                "output": f"执行过程中发生错误: {str(e)}",
+                "output": str(e),
                 "intermediate_steps": [],
                 "token_usage": {}
             }

app/agent/memory/__init__.py

@@ -232,7 +232,7 @@ class ConversationMemoryManager:
             return []
 
         # 获取所有消息
-        return memory.messages
+        return memory.messages[:-1]
 
     async def get_recent_messages(
             self,

app/agent/tools/base.py

@@ -1,4 +1,5 @@
 import json
+import uuid
 from abc import ABCMeta, abstractmethod
 from typing import Any, Optional
 
@@ -42,6 +43,9 @@ class MoviePilotTool(BaseTool, metaclass=ABCMeta):
         # 获取工具调用前的agent消息
         agent_message = await self._callback_handler.get_message()
 
+        # 生成唯一的工具调用ID
+        call_id = f"call_{str(uuid.uuid4())[:16]}"
+
         # 记忆工具调用
         await conversation_manager.add_conversation(
             session_id=self._session_id,
@@ -49,8 +53,8 @@ class MoviePilotTool(BaseTool, metaclass=ABCMeta):
             role="tool_call",
             content=agent_message,
             metadata={
-                "call_id": self.__class__.__name__,
-                "tool_name": self.__class__.__name__,
+                "call_id": call_id,
+                "tool_name": self.name,
                 "parameters": kwargs
             }
         )
@@ -61,22 +65,30 @@ class MoviePilotTool(BaseTool, metaclass=ABCMeta):
             explanation = kwargs.get("explanation")
             if explanation:
                 tool_message = explanation
-        
+
         # 合并agent消息和工具执行消息，一起发送
         messages = []
         if agent_message:
             messages.append(agent_message)
         if tool_message:
             messages.append(f"⚙️ => {tool_message}")
-        
+
         # 发送合并后的消息
         if messages:
             merged_message = "\n\n".join(messages)
             await self.send_tool_message(merged_message, title="MoviePilot助手")
 
         logger.debug(f'Executing tool {self.name} with args: {kwargs}')
-        result = await self.run(**kwargs)
-        logger.debug(f'Tool {self.name} executed with result: {result}')
+
+        # 执行工具，捕获异常确保结果总是被存储到记忆中
+        try:
+            result = await self.run(**kwargs)
+            logger.debug(f'Tool {self.name} executed with result: {result}')
+        except Exception as e:
+            # 记录异常详情
+            error_message = f"工具执行异常 ({type(e).__name__}): {str(e)}"
+            logger.error(f'Tool {self.name} execution failed: {e}', exc_info=True)
+            result = error_message
 
         # 记忆工具调用结果
         if isinstance(result, str):
@@ -85,13 +97,15 @@ class MoviePilotTool(BaseTool, metaclass=ABCMeta):
             formated_result = str(result)
         else:
             formated_result = json.dumps(result, ensure_ascii=False, indent=2)
+
         await conversation_manager.add_conversation(
             session_id=self._session_id,
             user_id=self._user_id,
             role="tool_result",
             content=formated_result,
             metadata={
-                "call_id": self.__class__.__name__
+                "call_id": call_id,
+                "tool_name": self.name,
             }
         )
 

app/agent/tools/factory.py

@@ -39,6 +39,7 @@ from app.agent.tools.impl.query_directory_settings import QueryDirectorySettings
 from app.agent.tools.impl.list_directory import ListDirectoryTool
 from app.agent.tools.impl.query_transfer_history import QueryTransferHistoryTool
 from app.agent.tools.impl.transfer_file import TransferFileTool
+from app.agent.tools.impl.execute_command import ExecuteCommandTool
 from app.core.plugin import PluginManager
 from app.log import logger
 from .base import MoviePilotTool
@@ -96,7 +97,8 @@ class MoviePilotToolFactory:
             QuerySchedulersTool,
             RunSchedulerTool,
             QueryWorkflowsTool,
-            RunWorkflowTool
+            RunWorkflowTool,
+            ExecuteCommandTool
         ]
         # 创建内置工具
         for ToolClass in tool_definitions:

app/agent/tools/impl/add_subscribe.py

@@ -108,6 +108,9 @@ class AddSubscribeTool(MoviePilotTool):
                 **subscribe_kwargs
             )
             if sid:
+                if message and "已存在" in message:
+                    return f"订阅已存在：{title} ({year})。如需修改参数请先删除旧订阅。"
+
                 result_msg = f"成功添加订阅：{title} ({year})"
                 if subscribe_kwargs:
                     params = []

app/agent/tools/impl/execute_command.py

@@ -0,0 +1,81 @@
+"""执行Shell命令工具"""
+
+import asyncio
+from typing import Optional, Type
+
+from pydantic import BaseModel, Field
+
+from app.agent.tools.base import MoviePilotTool
+from app.log import logger
+
+
+class ExecuteCommandInput(BaseModel):
+    """执行Shell命令工具的输入参数模型"""
+    explanation: str = Field(..., description="Clear explanation of why this command is being executed")
+    command: str = Field(..., description="The shell command to execute")
+    timeout: Optional[int] = Field(60, description="Max execution time in seconds (default: 60)")
+
+
+class ExecuteCommandTool(MoviePilotTool):
+    name: str = "execute_command"
+    description: str = "Safely execute shell commands on the server. Useful for system maintenance, checking status, or running custom scripts. Includes timeout and output limits."
+    args_schema: Type[BaseModel] = ExecuteCommandInput
+
+    def get_tool_message(self, **kwargs) -> Optional[str]:
+        """根据命令生成友好的提示消息"""
+        command = kwargs.get("command", "")
+        return f"正在执行系统命令: {command}"
+
+    async def run(self, command: str, timeout: Optional[int] = 60, **kwargs) -> str:
+        logger.info(f"执行工具: {self.name}, 参数: command={command}, timeout={timeout}")
+
+        # 简单安全过滤
+        forbidden_keywords = ["rm -rf /", ":(){ :|:& };:", "dd if=/dev/zero", "mkfs", "reboot", "shutdown"]
+        for keyword in forbidden_keywords:
+            if keyword in command:
+                return f"错误：命令包含禁止使用的关键字 '{keyword}'"
+
+        try:
+            # 执行命令
+            process = await asyncio.create_subprocess_shell(
+                command,
+                stdout=asyncio.subprocess.PIPE,
+                stderr=asyncio.subprocess.PIPE
+            )
+
+            try:
+                # 等待完成，带超时
+                stdout, stderr = await asyncio.wait_for(process.communicate(), timeout=timeout)
+                
+                # 处理输出
+                stdout_str = stdout.decode('utf-8', errors='replace').strip()
+                stderr_str = stderr.decode('utf-8', errors='replace').strip()
+                exit_code = process.returncode
+
+                result = f"命令执行完成 (退出码: {exit_code})"
+                if stdout_str:
+                    result += f"\n\n标准输出:\n{stdout_str}"
+                if stderr_str:
+                    result += f"\n\n错误输出:\n{stderr_str}"
+                
+                # 如果没有输出
+                if not stdout_str and not stderr_str:
+                    result += "\n\n(无输出内容)"
+                
+                # 限制输出长度，防止上下文过长
+                if len(result) > 3000:
+                    result = result[:3000] + "\n\n...(输出内容过长，已截断)"
+                
+                return result
+
+            except asyncio.TimeoutError:
+                # 超时处理
+                try:
+                    process.kill()
+                except ProcessLookupError:
+                    pass
+                return f"命令执行超时 (限制: {timeout}秒)"
+
+        except Exception as e:
+            logger.error(f"执行命令失败: {e}", exc_info=True)
+            return f"执行命令时发生错误: {str(e)}"

app/agent/tools/impl/query_library_exists.py

@@ -8,6 +8,7 @@ from pydantic import BaseModel, Field
 from app.agent.tools.base import MoviePilotTool
 from app.chain.mediaserver import MediaServerChain
 from app.core.context import MediaInfo
+from app.core.meta import MetaBase
 from app.log import logger
 from app.schemas.types import MediaType
 
@@ -51,47 +52,88 @@ class QueryLibraryExistsTool(MoviePilotTool):
         try:
             if not title:
                 return "请提供媒体标题进行查询"
-            
-            # 创建 MediaInfo 对象
-            mediainfo = MediaInfo()
-            mediainfo.title = title
-            mediainfo.year = year
-            
-            # 转换媒体类型
-            if media_type == "电影":
-                mediainfo.type = MediaType.MOVIE
-            elif media_type == "电视剧":
-                mediainfo.type = MediaType.TV
-            # media_type == "all" 时不设置类型，让媒体服务器自动判断
-            
-            # 调用媒体服务器接口实时查询
+
             media_chain = MediaServerChain()
+
+            # 1. 识别媒体信息（获取 TMDB ID 和各季的总集数等元数据）
+            meta = MetaBase(title=title)
+            if year:
+                meta.year = str(year)
+            if media_type == "电影":
+                meta.type = MediaType.MOVIE
+            elif media_type == "电视剧":
+                meta.type = MediaType.TV
+
+            # 使用识别方法补充信息
+            recognize_info = media_chain.recognize_media(meta=meta)
+            if recognize_info:
+                mediainfo = recognize_info
+            else:
+                # 识别失败，创建基本信息的 MediaInfo
+                mediainfo = MediaInfo()
+                mediainfo.title = title
+                mediainfo.year = year
+                if media_type == "电影":
+                    mediainfo.type = MediaType.MOVIE
+                elif media_type == "电视剧":
+                    mediainfo.type = MediaType.TV
+
+            # 2. 调用媒体服务器接口实时查询存在信息
             existsinfo = media_chain.media_exists(mediainfo=mediainfo)
-            
+
             if not existsinfo:
                 return "媒体库中未找到相关媒体"
-            
-            # 如果找到了，获取详细信息
+
+            # 3. 如果找到了，获取详细信息并组装结果
             result_items = []
             if existsinfo.itemid and existsinfo.server:
                 iteminfo = media_chain.iteminfo(server=existsinfo.server, item_id=existsinfo.itemid)
                 if iteminfo:
                     # 使用 model_dump() 转换为字典格式
                     item_dict = iteminfo.model_dump(exclude_none=True)
+
+                    # 对于电视剧，补充已存在的季集详情及进度统计
+                    if existsinfo.type == MediaType.TV:
+                        # 注入已存在集信息 (Dict[int, list])
+                        item_dict["seasoninfo"] = existsinfo.seasons
+
+                        # 统计库中已存在的季集总数
+                        if existsinfo.seasons:
+                            item_dict["existing_episodes_count"] = sum(len(e) for e in existsinfo.seasons.values())
+                            item_dict["seasons_existing_count"] = {str(s): len(e) for s, e in existsinfo.seasons.items()}
+
+                            # 如果识别到了元数据，补充总计对比和进度概览
+                            if mediainfo.seasons:
+                                item_dict["seasons_total_count"] = {str(s): len(e) for s, e in mediainfo.seasons.items()}
+                                # 进度概览，例如 "Season 1": "3/12"
+                                item_dict["seasons_progress"] = {
+                                    f"第{s}季": f"{len(existsinfo.seasons.get(s, []))}/{len(mediainfo.seasons.get(s, []))} 集"
+                                    for s in mediainfo.seasons.keys() if (s in existsinfo.seasons or s > 0)
+                                }
+
                     result_items.append(item_dict)
-            
+
             if result_items:
                 return json.dumps(result_items, ensure_ascii=False)
-            
-            # 如果找到了但没有详细信息，返回基本信息
+
+            # 如果找到了但没有获取到 iteminfo，返回基本信息
             result_dict = {
+                "title": mediainfo.title,
+                "year": mediainfo.year,
                 "type": existsinfo.type.value if existsinfo.type else None,
                 "server": existsinfo.server,
                 "server_type": existsinfo.server_type,
                 "itemid": existsinfo.itemid,
                 "seasons": existsinfo.seasons if existsinfo.seasons else {}
             }
+            if existsinfo.type == MediaType.TV and existsinfo.seasons:
+                result_dict["existing_episodes_count"] = sum(len(e) for e in existsinfo.seasons.values())
+                result_dict["seasons_existing_count"] = {str(s): len(e) for s, e in existsinfo.seasons.items()}
+                if mediainfo.seasons:
+                    result_dict["seasons_total_count"] = {str(s): len(e) for s, e in mediainfo.seasons.items()}
+
             return json.dumps([result_dict], ensure_ascii=False)
         except Exception as e:
             logger.error(f"查询媒体库失败: {e}", exc_info=True)
             return f"查询媒体库时发生错误: {str(e)}"
+

app/agent/tools/impl/query_subscribes.py

@@ -14,7 +14,7 @@ class QuerySubscribesInput(BaseModel):
     """查询订阅工具的输入参数模型"""
     explanation: str = Field(..., description="Clear explanation of why this tool is being used in the current context")
     status: Optional[str] = Field("all",
-                                  description="Filter subscriptions by status: 'R' for enabled subscriptions, 'P' for disabled ones, 'all' for all subscriptions")
+                                  description="Filter subscriptions by status: 'R' for enabled subscriptions, 'S' for paused ones, 'all' for all subscriptions")
     media_type: Optional[str] = Field("all",
                                       description="Filter by media type: '电影' for films, '电视剧' for television series, 'all' for all types")
 
@@ -33,7 +33,7 @@ class QuerySubscribesTool(MoviePilotTool):
         
         # 根据状态过滤条件生成提示
         if status != "all":
-            status_map = {"R": "已启用", "P": "已禁用"}
+            status_map = {"R": "已启用", "S": "已暂停"}
             parts.append(f"状态: {status_map.get(status, status)}")
         
         # 根据媒体类型过滤条件生成提示

app/agent/tools/impl/search_web.py

@@ -1,22 +1,26 @@
-"""搜索网络内容工具"""
-
+import asyncio
 import json
 import re
-from typing import Optional, Type
+from typing import Optional, Type, List, Dict
 
+import httpx
+from ddgs import DDGS
 from pydantic import BaseModel, Field
 
 from app.agent.tools.base import MoviePilotTool
 from app.core.config import settings
 from app.log import logger
-from app.utils.http import AsyncRequestUtils
+
+# 搜索超时时间（秒）
+SEARCH_TIMEOUT = 20
 
 
 class SearchWebInput(BaseModel):
     """搜索网络内容工具的输入参数模型"""
     explanation: str = Field(..., description="Clear explanation of why this tool is being used in the current context")
     query: str = Field(..., description="The search query string to search for on the web")
-    max_results: Optional[int] = Field(5, description="Maximum number of search results to return (default: 5, max: 10)")
+    max_results: Optional[int] = Field(5,
+                                       description="Maximum number of search results to return (default: 5, max: 10)")
 
 
 class SearchWebTool(MoviePilotTool):
@@ -33,151 +37,137 @@ class SearchWebTool(MoviePilotTool):
     async def run(self, query: str, max_results: Optional[int] = 5, **kwargs) -> str:
         """
         执行网络搜索
-        
-        Args:
-            query: 搜索查询字符串
-            max_results: 最大返回结果数（默认5，最大10）
-            
-        Returns:
-            格式化的搜索结果JSON字符串
         """
         logger.info(f"执行工具: {self.name}, 参数: query={query}, max_results={max_results}")
 
         try:
             # 限制最大结果数
             max_results = min(max(1, max_results or 5), 10)
-            
-            # 使用DuckDuckGo API进行搜索
-            search_results = await self._search_duckduckgo_api(query, max_results)
-            
-            if not search_results:
+            results = []
+
+            # 1. 优先使用 Tavily (如果配置了 API Key)
+            if settings.TAVILY_API_KEY:
+                logger.info("使用 Tavily 进行搜索...")
+                results = await self._search_tavily(query, max_results)
+
+            # 2. 如果没有结果或未配置 Tavily，使用 DuckDuckGo
+            if not results:
+                logger.info("使用 DuckDuckGo 进行搜索...")
+                results = await self._search_duckduckgo(query, max_results)
+
+            if not results:
                 return f"未找到与 '{query}' 相关的搜索结果"
-            
-            # 裁剪结果以避免占用过多上下文
-            formatted_results = self._format_and_truncate_results(search_results, max_results)
-            
-            result_json = json.dumps(formatted_results, ensure_ascii=False, indent=2)
-            return result_json
-            
+
+            # 格式化并裁剪结果
+            formatted_results = self._format_and_truncate_results(results, max_results)
+            return json.dumps(formatted_results, ensure_ascii=False, indent=2)
+
         except Exception as e:
             error_message = f"搜索网络内容失败: {str(e)}"
             logger.error(f"搜索网络内容失败: {e}", exc_info=True)
             return error_message
 
     @staticmethod
-    async def _search_duckduckgo_api(query: str, max_results: int) -> list:
-        """
-        使用DuckDuckGo API进行搜索
-        
-        Args:
-            query: 搜索查询
-            max_results: 最大结果数
-            
-        Returns:
-            搜索结果列表
-        """
+    async def _search_tavily(query: str, max_results: int) -> List[Dict]:
+        """使用 Tavily API 进行搜索"""
         try:
-            # DuckDuckGo Instant Answer API
-            api_url = "https://api.duckduckgo.com/"
-            params = {
-                "q": query,
-                "format": "json",
-                "no_html": "1",
-                "skip_disambig": "1"
-            }
-            
-            # 使用代理（如果配置了）
-            http_utils = AsyncRequestUtils(
-                proxies=settings.PROXY,
-                timeout=10
-            )
-            
-            data = await http_utils.get_json(api_url, params=params)
-            
-            results = []
-            
-            if data:
-                # 处理AbstractText（摘要）
-                if data.get("AbstractText"):
+            async with httpx.AsyncClient(timeout=SEARCH_TIMEOUT) as client:
+                response = await client.post(
+                    "https://api.tavily.com/search",
+                    json={
+                        "api_key": settings.TAVILY_API_KEY,
+                        "query": query,
+                        "search_depth": "basic",
+                        "max_results": max_results,
+                        "include_answer": False,
+                        "include_images": False,
+                        "include_raw_content": False,
+                    }
+                )
+                response.raise_for_status()
+                data = response.json()
+
+                results = []
+                for result in data.get("results", []):
                     results.append({
-                        "title": data.get("Heading", query),
-                        "snippet": data.get("AbstractText", ""),
-                        "url": data.get("AbstractURL", ""),
-                        "source": "DuckDuckGo Abstract"
+                        'title': result.get('title', ''),
+                        'snippet': result.get('content', ''),
+                        'url': result.get('url', ''),
+                        'source': 'Tavily'
                     })
-                
-                # 处理RelatedTopics（相关主题）
-                related_topics = data.get("RelatedTopics", [])
-                for topic in related_topics[:max_results - len(results)]:
-                    if isinstance(topic, dict):
-                        text = topic.get("Text", "")
-                        first_url = topic.get("FirstURL", "")
-                        if text and first_url:
-                            # 提取标题（通常在" - "之前）
-                            title = text.split(" - ")[0] if " - " in text else text[:100]
-                            snippet = text
-                            
-                            results.append({
-                                "title": title.strip(),
-                                "snippet": snippet,
-                                "url": first_url,
-                                "source": "DuckDuckGo Related"
-                            })
-                
-                # 处理Results（搜索结果）
-                api_results = data.get("Results", [])
-                for result in api_results[:max_results - len(results)]:
-                    if isinstance(result, dict):
-                        title = result.get("Text", "")
-                        url = result.get("FirstURL", "")
-                        if title and url:
-                            results.append({
-                                "title": title,
-                                "snippet": result.get("Text", ""),
-                                "url": url,
-                                "source": "DuckDuckGo Results"
-                            })
-            
-            return results[:max_results]
-            
+                return results
         except Exception as e:
-            logger.warning(f"DuckDuckGo API搜索失败: {e}")
+            logger.warning(f"Tavily 搜索失败: {e}")
             return []
 
     @staticmethod
-    def _format_and_truncate_results(results: list, max_results: int) -> dict:
-        """
-        格式化并裁剪搜索结果以避免占用过多上下文
-        
-        Args:
-            results: 原始搜索结果列表
-            max_results: 最大结果数
-            
-        Returns:
-            格式化后的结果字典
-        """
+    def _get_proxy_url(proxy_setting) -> Optional[str]:
+        """从代理设置中提取代理URL"""
+        if not proxy_setting:
+            return None
+        if isinstance(proxy_setting, dict):
+            return proxy_setting.get('http') or proxy_setting.get('https')
+        return proxy_setting
+
+    async def _search_duckduckgo(self, query: str, max_results: int) -> List[Dict]:
+        """使用 duckduckgo-search (DDGS) 进行搜索"""
+        try:
+            def sync_search():
+                results = []
+                ddgs_kwargs = {
+                    'timeout': SEARCH_TIMEOUT
+                }
+                proxy_url = self._get_proxy_url(settings.PROXY)
+                if proxy_url:
+                    ddgs_kwargs['proxy'] = proxy_url
+
+                try:
+                    with DDGS(**ddgs_kwargs) as ddgs:
+                        ddgs_gen = ddgs.text(
+                            query,
+                            max_results=max_results
+                        )
+                        if ddgs_gen:
+                            for result in ddgs_gen:
+                                results.append({
+                                    'title': result.get('title', ''),
+                                    'snippet': result.get('body', ''),
+                                    'url': result.get('href', ''),
+                                    'source': 'DuckDuckGo'
+                                })
+                except Exception as err:
+                    logger.warning(f"DuckDuckGo search process failed: {err}")
+                return results
+
+            loop = asyncio.get_running_loop()
+            return await loop.run_in_executor(None, sync_search)
+
+        except Exception as e:
+            logger.warning(f"DuckDuckGo 搜索失败: {e}")
+            return []
+
+    @staticmethod
+    def _format_and_truncate_results(results: List[Dict], max_results: int) -> Dict:
+        """格式化并裁剪搜索结果"""
         formatted = {
             "total_results": len(results),
             "results": []
         }
-        
-        # 限制结果数量
-        limited_results = results[:max_results]
-        
-        for idx, result in enumerate(limited_results, 1):
-            title = result.get("title", "")[:200]  # 限制标题长度
+
+        for idx, result in enumerate(results[:max_results], 1):
+            title = result.get("title", "")[:200]
             snippet = result.get("snippet", "")
             url = result.get("url", "")
             source = result.get("source", "Unknown")
-            
-            # 裁剪摘要，避免过长
-            max_snippet_length = 300  # 每个摘要最多300字符
+
+            # 裁剪摘要
+            max_snippet_length = 500  # 增加到500字符，提供更多上下文
             if len(snippet) > max_snippet_length:
                 snippet = snippet[:max_snippet_length] + "..."
-            
-            # 清理文本，移除多余的空白字符
+
+            # 清理文本
             snippet = re.sub(r'\s+', ' ', snippet).strip()
-            
+
             formatted["results"].append({
                 "rank": idx,
                 "title": title,
@@ -185,9 +175,8 @@ class SearchWebTool(MoviePilotTool):
                 "url": url,
                 "source": source
             })
-        
-        # 添加提示信息
+
         if len(results) > max_results:
-            formatted["note"] = f"注意：共找到 {len(results)} 条结果，为节省上下文空间，仅显示前 {max_results} 条结果。"
-        
+            formatted["note"] = f"仅显示前 {max_results} 条结果。"
+
         return formatted

app/agent/tools/impl/update_subscribe.py

@@ -29,7 +29,7 @@ class UpdateSubscribeInput(BaseModel):
     include: Optional[str] = Field(None, description="Include filter as regular expression (optional)")
     exclude: Optional[str] = Field(None, description="Exclude filter as regular expression (optional)")
     filter: Optional[str] = Field(None, description="Filter rule as regular expression (optional)")
-    state: Optional[str] = Field(None, description="Subscription state: 'R' for enabled, 'P' for pending, 'S' for stoped (optional)")
+    state: Optional[str] = Field(None, description="Subscription state: 'R' for enabled, 'P' for pending, 'S' for paused (optional)")
     sites: Optional[List[int]] = Field(None, description="List of site IDs to search from (optional)")
     downloader: Optional[str] = Field(None, description="Downloader name (optional)")
     save_path: Optional[str] = Field(None, description="Save path for downloaded files (optional)")

app/api/endpoints/history.py

@@ -4,6 +4,7 @@ import jieba
 from fastapi import APIRouter, Depends
 from sqlalchemy.ext.asyncio import AsyncSession
 from sqlalchemy.orm import Session
+from pathlib import Path
 
 from app import schemas
 from app.chain.storage import StorageChain
@@ -11,7 +12,7 @@ from app.core.event import eventmanager
 from app.core.security import verify_token
 from app.db import get_async_db, get_db
 from app.db.models import User
-from app.db.models.downloadhistory import DownloadHistory
+from app.db.models.downloadhistory import DownloadHistory, DownloadFiles
 from app.db.models.transferhistory import TransferHistory
 from app.db.user_oper import get_current_active_superuser_async, get_current_active_superuser
 from app.schemas.types import EventType
@@ -98,6 +99,8 @@ def delete_transfer_history(history_in: schemas.TransferHistory,
         state = StorageChain().delete_media_file(src_fileitem)
         if not state:
             return schemas.Response(success=False, message=f"{src_fileitem.path} 删除失败")
+        # 删除下载记录中关联的文件
+        DownloadFiles.delete_by_fullpath(db, Path(src_fileitem.path).as_posix())
         # 发送事件
         eventmanager.send_event(
             EventType.DownloadFileDeleted,

app/api/endpoints/media.py

@@ -11,7 +11,10 @@ from app.core.context import Context
 from app.core.event import eventmanager
 from app.core.metainfo import MetaInfo, MetaInfoPath
 from app.core.security import verify_token, verify_apitoken
+from app.db.models import User
+from app.db.user_oper import get_current_active_user, get_current_active_superuser
 from app.schemas import MediaType, MediaRecognizeConvertEventData
+from app.schemas.category import CategoryConfig
 from app.schemas.types import ChainEventType
 
 router = APIRouter()
@@ -131,6 +134,26 @@ def scrape(fileitem: schemas.FileItem,
     return schemas.Response(success=True, message=f"{fileitem.path} 刮削完成")
 
 
+@router.get("/category/config", summary="获取分类策略配置", response_model=schemas.Response)
+def get_category_config(_: User = Depends(get_current_active_user)):
+    """
+    获取分类策略配置
+    """
+    config = MediaChain().category_config()
+    return schemas.Response(success=True, data=config.model_dump())
+
+
+@router.post("/category/config", summary="保存分类策略配置", response_model=schemas.Response)
+def save_category_config(config: CategoryConfig, _: User = Depends(get_current_active_superuser)):
+    """
+    保存分类策略配置
+    """
+    if MediaChain().save_category_config(config):
+        return schemas.Response(success=True, message="保存成功")
+    else:
+        return schemas.Response(success=False, message="保存失败")
+
+
 @router.get("/category", summary="查询自动分类配置", response_model=dict)
 async def category(_: schemas.TokenPayload = Depends(verify_token)) -> Any:
     """

app/api/endpoints/storage.py

@@ -1,4 +1,4 @@
-from datetime import datetime
+import math
 from pathlib import Path
 from typing import Any, List, Optional
 
@@ -83,7 +83,7 @@ def list_files(fileitem: schemas.FileItem,
         if sort == "name":
             file_list.sort(key=lambda x: StringUtils.natural_sort_key(x.name or ""))
         else:
-            file_list.sort(key=lambda x: x.modify_time or datetime.min, reverse=True)
+            file_list.sort(key=lambda x: x.modify_time or -math.inf, reverse=True)
     return file_list
 
 

app/chain/__init__.py

@@ -26,6 +26,7 @@ from app.helper.service import ServiceConfigHelper
 from app.log import logger
 from app.schemas import TransferInfo, TransferTorrent, ExistMediaInfo, DownloadingTorrent, CommingMessage, Notification, \
     WebhookEventInfo, TmdbEpisode, MediaPerson, FileItem, TransferDirectoryConf
+from app.schemas.category import CategoryConfig
 from app.schemas.types import TorrentStatus, MediaType, MediaImageType, EventType, MessageChannel
 from app.utils.object import ObjectUtils
 
@@ -251,6 +252,7 @@ class ChainBase(metaclass=ABCMeta):
                     # 中止继续执行
                     break
             except Exception as err:
+                logger.error(traceback.format_exc())
                 self.__handle_system_error(err, module_id, module_name, method, **kwargs)
         return result
 
@@ -292,6 +294,7 @@ class ChainBase(metaclass=ABCMeta):
                     # 中止继续执行
                     break
             except Exception as err:
+                logger.error(traceback.format_exc())
                 self.__handle_system_error(err, module_id, module_name, method, **kwargs)
         return result
 
@@ -1060,6 +1063,18 @@ class ChainBase(metaclass=ABCMeta):
         """
         return self.run_module("media_category")
 
+    def category_config(self) -> CategoryConfig:
+        """
+        获取分类策略配置
+        """
+        return self.run_module("load_category_config")
+
+    def save_category_config(self, config: CategoryConfig) -> bool:
+        """
+        保存分类策略配置
+        """
+        return self.run_module("save_category_config", config=config)
+
     def register_commands(self, commands: Dict[str, dict]) -> None:
         """
         注册菜单命令

app/chain/download.py

@@ -292,10 +292,6 @@ class DownloadChain(ChainBase):
 
             # 登记下载记录
             downloadhis = DownloadHistoryOper()
-            # 获取应用的识别词（如果有）
-            custom_words_str = None
-            if hasattr(_meta, 'apply_words') and _meta.apply_words:
-                custom_words_str = '\n'.join(_meta.apply_words)
             downloadhis.add(
                 path=download_path.as_posix(),
                 type=_media.type.value,
@@ -319,7 +315,6 @@ class DownloadChain(ChainBase):
                 date=time.strftime("%Y-%m-%d %H:%M:%S", time.localtime()),
                 media_category=_media.category,
                 episode_group=_media.episode_group,
-                custom_words=custom_words_str,
                 note={"source": source}
             )
 

app/chain/message.py

@@ -40,7 +40,7 @@ class MessageChain(ChainBase):
     # 用户会话信息 {userid: (session_id, last_time)}
     _user_sessions: Dict[Union[str, int], tuple] = {}
     # 会话超时时间（分钟）
-    _session_timeout_minutes: int = 15
+    _session_timeout_minutes: int = 30
 
     @staticmethod
     def __get_noexits_info(
@@ -842,8 +842,7 @@ class MessageChain(ChainBase):
 
         return buttons
 
-    @staticmethod
-    def _get_or_create_session_id(userid: Union[str, int]) -> str:
+    def _get_or_create_session_id(self, userid: Union[str, int]) -> str:
         """
         获取或创建会话ID
         如果用户上次会话在15分钟内，则复用相同的会话ID；否则创建新的会话ID
@@ -851,34 +850,33 @@ class MessageChain(ChainBase):
         current_time = datetime.now()
 
         # 检查用户是否有已存在的会话
-        if userid in MessageChain._user_sessions:
-            session_id, last_time = MessageChain._user_sessions[userid]
+        if userid in self._user_sessions:
+            session_id, last_time = self._user_sessions[userid]
 
             # 计算时间差
             time_diff = current_time - last_time
 
-            # 如果时间差小于等于15分钟，复用会话ID
-            if time_diff <= timedelta(minutes=MessageChain._session_timeout_minutes):
+            # 如果时间差小于等于xx分钟，复用会话ID
+            if time_diff <= timedelta(minutes=self._session_timeout_minutes):
                 # 更新最后使用时间
-                MessageChain._user_sessions[userid] = (session_id, current_time)
+                self._user_sessions[userid] = (session_id, current_time)
                 logger.info(
                     f"复用会话ID: {session_id}, 用户: {userid}, 距离上次会话: {time_diff.total_seconds() / 60:.1f}分钟")
                 return session_id
 
         # 创建新的会话ID
         new_session_id = f"user_{userid}_{int(time.time())}"
-        MessageChain._user_sessions[userid] = (new_session_id, current_time)
+        self._user_sessions[userid] = (new_session_id, current_time)
         logger.info(f"创建新会话ID: {new_session_id}, 用户: {userid}")
         return new_session_id
 
-    @staticmethod
-    def clear_user_session(userid: Union[str, int]) -> bool:
+    def clear_user_session(self, userid: Union[str, int]) -> bool:
         """
         清除指定用户的会话信息
         返回是否成功清除
         """
-        if userid in MessageChain._user_sessions:
-            session_id, _ = MessageChain._user_sessions.pop(userid)
+        if userid in self._user_sessions:
+            session_id, _ = self._user_sessions.pop(userid)
             logger.info(f"已清除用户 {userid} 的会话: {session_id}")
             return True
         return False
@@ -889,8 +887,8 @@ class MessageChain(ChainBase):
         """
         # 获取并清除会话信息
         session_id = None
-        if userid in MessageChain._user_sessions:
-            session_id, _ = MessageChain._user_sessions.pop(userid)
+        if userid in self._user_sessions:
+            session_id, _ = self._user_sessions.pop(userid)
             logger.info(f"已清除用户 {userid} 的会话: {session_id}")
 
         # 如果有会话ID，同时清除智能体的会话记忆

app/chain/subscribe.py

@@ -292,7 +292,7 @@ class SubscribeChain(ChainBase):
             "description": mediainfo.overview
         })
         # 返回结果
-        return sid, ""
+        return sid, err_msg
 
     async def async_add(self, title: str, year: str,
                         mtype: MediaType = None,
@@ -469,7 +469,7 @@ class SubscribeChain(ChainBase):
             "description": mediainfo.overview
         })
         # 返回结果
-        return sid, ""
+        return sid, err_msg
 
     @staticmethod
     def exists(mediainfo: MediaInfo, meta: MetaBase = None):
@@ -1119,6 +1119,19 @@ class SubscribeChain(ChainBase):
             })
             logger.info(f'{subscribe.name} 订阅元数据更新完成')
 
+    def get_subscribe_by_source(self, source: str) -> Optional[Subscribe]:
+        """
+        从来源获取订阅
+        """
+        source_keyword = self.parse_subscribe_source_keyword(source)
+        if not source_keyword:
+            return None
+        # 只保留需要的字段动态获取订阅
+        valid_fields = {k: v for k, v in source_keyword.items()
+                        if k in ["type", "season", "tmdbid", "doubanid", "bangumiid"]}
+        # 暂时不考虑订阅历史, 若有必要再添加
+        return SubscribeOper().get_by(**valid_fields)
+
     @staticmethod
     def follow():
         """
@@ -1655,7 +1668,7 @@ class SubscribeChain(ChainBase):
         if download_his:
             for his in download_his:
                 # 查询下载文件
-                files = downloadhis.get_files_by_hash(his.download_hash)
+                files = downloadhis.get_files_by_hash(his.download_hash, state=1)
                 if files:
                     for file in files:
                         # 识别文件名
@@ -1828,8 +1841,9 @@ class SubscribeChain(ChainBase):
     def get_subscribe_source_keyword(subscribe: Subscribe) -> str:
         """
         构造用于订阅来源的关键字字符串
+
         :param subscribe: Subscribe 对象
-        :return: 格式化的订阅来源关键字字符串，格式为 "Subscribe|{...}"
+        :return str: 格式化的订阅来源关键字字符串，格式为 "Subscribe|{...}"
         """
         source_keyword = {
             'id': subscribe.id,
@@ -1844,3 +1858,24 @@ class SubscribeChain(ChainBase):
             'bangumiid': subscribe.bangumiid
         }
         return f"Subscribe|{json.dumps(source_keyword, ensure_ascii=False)}"
+
+    @staticmethod
+    def parse_subscribe_source_keyword(source_keyword_str: str) -> Optional[dict]:
+        """
+        解析订阅来源关键字字符串
+
+        :param source_keyword_str: 订阅来源关键字字符串，格式为 "Subscribe|{...}"
+        :return Dict: 如果解析失败则返回None
+        """
+        if not source_keyword_str or not source_keyword_str.startswith("Subscribe|"):
+            return None
+
+        try:
+            # 分割字符串获取JSON部分
+            json_part = source_keyword_str.split("|", 1)[1]
+            # 解析JSON字符串
+            source_keyword = json.loads(json_part)
+            return source_keyword
+        except (IndexError, json.JSONDecodeError, TypeError) as e:
+            logger.error(f"解析订阅来源关键字失败: {e}")
+            return None

app/chain/transfer.py

@@ -10,6 +10,7 @@ from app import schemas
 from app.chain import ChainBase
 from app.chain.media import MediaChain
 from app.chain.storage import StorageChain
+from app.chain.subscribe import SubscribeChain
 from app.chain.tmdb import TmdbChain
 from app.core.config import settings, global_vars
 from app.core.context import MediaInfo
@@ -26,7 +27,7 @@ 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, \
+from app.schemas import TransferInfo, Notification, EpisodeFormat, FileItem, TransferDirectoryConf, \
     TransferTask, TransferQueue, TransferJob, TransferJobTask
 from app.schemas.types import TorrentStatus, EventType, MediaType, ProgressKey, NotificationType, MessageChannel, \
     SystemConfigKey, ChainEventType, ContentType
@@ -46,6 +47,7 @@ task_lock = threading.Lock()
 class JobManager:
     """
     作业管理器
+    task任务负责一个文件的整理，job作业负责一个媒体的整理
     """
 
     # 整理中的作业
@@ -111,7 +113,7 @@ class JobManager:
 
     def add_task(self, task: TransferTask, state: Optional[str] = "waiting"):
         """
-        添加整理任务
+        添加整理任务，自动分组到对应的作业中
         """
         if not any([task, task.meta, task.fileitem]):
             return
@@ -165,7 +167,7 @@ class JobManager:
 
     def finish_task(self, task: TransferTask):
         """
-        设置任务为完成
+        设置任务为完成/成功
         """
         with job_lock:
             __mediaid__ = self.__get_id(task)
@@ -198,7 +200,7 @@ class JobManager:
 
     def remove_task(self, fileitem: FileItem) -> Optional[TransferJobTask]:
         """
-        移除任务
+        根据文件项移除任务
         """
         with job_lock:
             for mediaid in list(self._job_view):
@@ -219,10 +221,10 @@ class JobManager:
 
     def remove_job(self, task: TransferTask) -> Optional[TransferJob]:
         """
-        移除作业
+        移除任务对应的作业
         """
         with job_lock:
-            __mediaid__ = self.__get_media_id(media=task.mediainfo, season=task.meta.begin_season)
+            __mediaid__ = self.__get_id(task)
             if __mediaid__ in self._job_view:
                 # 移除季集信息
                 if __mediaid__ in self._season_episodes:
@@ -242,13 +244,10 @@ class JobManager:
             )
         else:
             meta_done = True
-        if __mediaid__ != __metaid__:
-            if __mediaid__ in self._job_view:
-                media_done = all(
-                    task.state in ["completed", "failed"] for task in self._job_view[__mediaid__].tasks
-                )
-            else:
-                media_done = False
+        if __mediaid__ in self._job_view:
+            media_done = all(
+                task.state in ["completed", "failed"] for task in self._job_view[__mediaid__].tasks
+            )
         else:
             media_done = True
         return meta_done and media_done
@@ -265,16 +264,13 @@ class JobManager:
             )
         else:
             meta_finished = True
-        if __mediaid__ != __metaid__:
-            if __mediaid__ in self._job_view:
-                tasks = self._job_view[__mediaid__].tasks
-                media_finished = all(
-                    task.state in ["completed", "failed"] for task in tasks
-                ) and any(
-                    task.state == "completed" for task in tasks
-                )
-            else:
-                media_finished = False
+        if __mediaid__ in self._job_view:
+            tasks = self._job_view[__mediaid__].tasks
+            media_finished = all(
+                task.state in ["completed", "failed"] for task in tasks
+            ) and any(
+                task.state == "completed" for task in tasks
+            )
         else:
             media_finished = True
         return meta_finished and media_finished
@@ -291,32 +287,64 @@ class JobManager:
             )
         else:
             meta_success = True
-        if __mediaid__ != __metaid__:
-            if __mediaid__ in self._job_view:
-                media_success = all(
-                    task.state in ["completed"] for task in self._job_view[__mediaid__].tasks
-                )
-            else:
-                media_success = False
+        if __mediaid__ in self._job_view:
+            media_success = all(
+                task.state in ["completed"] for task in self._job_view[__mediaid__].tasks
+            )
         else:
             media_success = True
         return meta_success and media_success
 
+    def get_all_torrent_hashes(self) -> set[str]:
+        """
+        获取所有种子的哈希值集合
+        """
+        with job_lock:
+            return {
+                task.download_hash
+                for job in self._job_view.values()
+                for task in job.tasks
+            }
+
+    def is_torrent_done(self, download_hash: str) -> bool:
+        """
+        检查指定种子的所有任务是否都已完成
+        """
+        with job_lock:
+            for job in self._job_view.values():
+                for task in job.tasks:
+                    if task.download_hash == download_hash:
+                        if task.state not in ["completed", "failed"]:
+                            return False
+            return True
+
+    def is_torrent_success(self, download_hash: str) -> bool:
+        """
+        检查指定种子的所有任务是否都已成功
+        """
+        with job_lock:
+            for job in self._job_view.values():
+                for task in job.tasks:
+                    if task.download_hash == download_hash:
+                        if task.state not in ["completed"]:
+                            return False
+            return True
+
     def has_tasks(self, meta: MetaBase, mediainfo: Optional[MediaInfo] = None, season: Optional[int] = None) -> bool:
         """
-        判断是否有任务正在处理
+        判断作业是否还有任务正在处理
         """
         if mediainfo:
-            __mediaid__ = self.__get_media_id(media=meta, season=season)
+            __mediaid__ = self.__get_media_id(media=mediainfo, season=season)
             if __mediaid__ in self._job_view:
                 return True
 
         __metaid__ = self.__get_meta_id(meta=meta, season=season)
-        return __metaid__ in self._job_view
+        return __metaid__ in self._job_view and len(self._job_view[__metaid__].tasks) > 0
 
     def success_tasks(self, media: MediaInfo, season: Optional[int] = None) -> List[TransferJobTask]:
         """
-        获取某项任务成功的任务
+        获取作业中所有成功的任务
         """
         __mediaid__ = self.__get_media_id(media=media, season=season)
         if __mediaid__ not in self._job_view:
@@ -325,7 +353,7 @@ class JobManager:
 
     def all_tasks(self, media: MediaInfo, season: Optional[int] = None) -> List[TransferJobTask]:
         """
-        获取全部任务
+        获取作业中全部任务
         """
         __mediaid__ = self.__get_media_id(media=media, season=season)
         if __mediaid__ not in self._job_view:
@@ -334,7 +362,7 @@ class JobManager:
 
     def count(self, media: MediaInfo, season: Optional[int] = None) -> int:
         """
-        获取某项任务成功总数
+        获取作业中成功总数
         """
         __mediaid__ = self.__get_media_id(media=media, season=season)
         if __mediaid__ not in self._job_view:
@@ -343,7 +371,7 @@ class JobManager:
 
     def size(self, media: MediaInfo, season: Optional[int] = None) -> int:
         """
-        获取某项任务成功文件总大小
+        获取作业中所有成功文件总大小
         """
         __mediaid__ = self.__get_media_id(media=media, season=season)
         if __mediaid__ not in self._job_view:
@@ -358,19 +386,19 @@ class JobManager:
 
     def total(self) -> int:
         """
-        获取所有task任务总数
+        获取所有任务总数
         """
         return sum([len(job.tasks) for job in self._job_view.values()])
 
     def list_jobs(self) -> List[TransferJob]:
         """
-        获取任务列表
+        获取所有作业的任务列表
         """
         return list(self._job_view.values())
 
     def season_episodes(self, media: MediaInfo, season: Optional[int] = None) -> List[int]:
         """
-        获取季集清单
+        获取作业的季集清单
         """
         __mediaid__ = self.__get_media_id(media=media, season=season)
         return self._season_episodes.get(__mediaid__) or []
@@ -389,10 +417,12 @@ class TransferChain(ChainBase, ConfigReloadMixin, metaclass=Singleton):
         super().__init__()
         # 主要媒体文件后缀
         self._media_exts = settings.RMT_MEDIAEXT
-        # 附加文件后缀
-        self._extra_exts = settings.RMT_SUBEXT + settings.RMT_AUDIOEXT
+        # 字幕文件后缀
+        self._subtitle_exts = settings.RMT_SUBEXT
+        # 音频文件后缀
+        self._audio_exts = settings.RMT_AUDIOEXT
         # 可处理的文件后缀（视频文件、字幕、音频文件）
-        self._allowed_exts = self._media_exts + self._extra_exts
+        self._allowed_exts = self._media_exts + self._audio_exts + self._subtitle_exts
         # 待整理任务队列
         self._queue = queue.Queue()
         # 文件整理线程
@@ -442,6 +472,33 @@ class TransferChain(ChainBase, ConfigReloadMixin, metaclass=Singleton):
         self.__stop()
         self.__init()
 
+    def __is_subtitle_file(self, fileitem: FileItem) -> bool:
+        """
+        判断是否为字幕文件
+        """
+        if not fileitem.extension:
+            return False
+        return True if f".{fileitem.extension.lower()}" in self._subtitle_exts else False
+
+    def __is_audio_file(self, fileitem: FileItem) -> bool:
+        """
+        判断是否为音频文件
+        """
+        if not fileitem.extension:
+            return False
+        return True if f".{fileitem.extension.lower()}" in self._audio_exts else False
+
+    def __is_media_file(self, fileitem: FileItem) -> bool:
+        """
+        判断是否为主要媒体文件
+        """
+        if fileitem.type == "dir":
+            # 蓝光原盘判断
+            return StorageChain().is_bluray_folder(fileitem)
+        if not fileitem.extension:
+            return False
+        return True if f".{fileitem.extension.lower()}" in self._media_exts else False
+
     def __is_allowed_file(self, fileitem: FileItem) -> bool:
         """
         判断是否允许的扩展名
@@ -450,22 +507,6 @@ class TransferChain(ChainBase, ConfigReloadMixin, metaclass=Singleton):
             return False
         return True if f".{fileitem.extension.lower()}" in self._allowed_exts else False
 
-    def __is_extra_file(self, fileitem: FileItem) -> bool:
-        """
-        判断是否额外的扩展名
-        """
-        if not fileitem.extension:
-            return False
-        return True if f".{fileitem.extension.lower()}" in self._extra_exts else False
-
-    def __is_media_file(self, fileitem: FileItem) -> bool:
-        """
-        判断是否为主要媒体文件
-        """
-        if not fileitem.extension:
-            return False
-        return True if f".{fileitem.extension.lower()}" in self._media_exts else False
-
     @staticmethod
     def __is_allow_filesize(fileitem: FileItem, min_filesize: int) -> bool:
         """
@@ -479,7 +520,12 @@ class TransferChain(ChainBase, ConfigReloadMixin, metaclass=Singleton):
         整理完成后处理
         """
 
-        def __all_finished():
+        # 状态
+        ret_status = True
+        # 错误信息
+        ret_message = ""
+
+        def __notify():
             """
             完成时发送消息、刮削事件、移除任务等
             """
@@ -501,6 +547,7 @@ class TransferChain(ChainBase, ConfigReloadMixin, metaclass=Singleton):
                         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,
@@ -517,16 +564,14 @@ class TransferChain(ChainBase, ConfigReloadMixin, metaclass=Singleton):
                     'overwrite': False
                 })
 
-            # 移除已完成的任务
-            self.jobview.remove_job(task)
-
         transferhis = TransferHistoryOper()
+
+        # 转移失败
         if not transferinfo.success:
-            # 转移失败
             logger.warn(f"{task.fileitem.name} 入库失败：{transferinfo.message}")
 
             # 新增转移失败历史记录
-            transferhis.add_fail(
+            history = transferhis.add_fail(
                 fileitem=task.fileitem,
                 mode=transferinfo.transfer_type if transferinfo else '',
                 downloader=task.downloader,
@@ -538,6 +583,7 @@ class TransferChain(ChainBase, ConfigReloadMixin, metaclass=Singleton):
 
             # 整理失败事件
             if self.__is_media_file(task.fileitem):
+                # 主要媒体文件整理失败事件
                 self.eventmanager.send_event(EventType.TransferFailed, {
                     'fileitem': task.fileitem,
                     'meta': task.meta,
@@ -545,6 +591,29 @@ class TransferChain(ChainBase, ConfigReloadMixin, metaclass=Singleton):
                     'transferinfo': transferinfo,
                     'downloader': task.downloader,
                     'download_hash': task.download_hash,
+                    'transfer_history_id': history.id if history else None,
+                })
+            elif self.__is_subtitle_file(task.fileitem):
+                # 字幕整理失败事件
+                self.eventmanager.send_event(EventType.SubtitleTransferFailed, {
+                    'fileitem': task.fileitem,
+                    'meta': task.meta,
+                    'mediainfo': task.mediainfo,
+                    'transferinfo': transferinfo,
+                    'downloader': task.downloader,
+                    'download_hash': task.download_hash,
+                    'transfer_history_id': history.id if history else None,
+                })
+            elif self.__is_audio_file(task.fileitem):
+                # 音频文件整理失败事件
+                self.eventmanager.send_event(EventType.AudioTransferFailed, {
+                    'fileitem': task.fileitem,
+                    'meta': task.meta,
+                    'mediainfo': task.mediainfo,
+                    'transferinfo': transferinfo,
+                    'downloader': task.downloader,
+                    'download_hash': task.download_hash,
+                    'transfer_history_id': history.id if history else None,
                 })
 
             # 发送失败消息
@@ -557,83 +626,103 @@ class TransferChain(ChainBase, ConfigReloadMixin, metaclass=Singleton):
                 link=settings.MP_DOMAIN('#/history')
             ))
 
-            # 整理失败
+            # 设置任务失败
             self.jobview.fail_task(task)
 
-            # 全部整理完成且有成功的任务时
-            if self.jobview.is_finished(task):
-                # 发送消息、刮削事件、移除任务
-                __all_finished()
+            # 返回失败
+            ret_status = False
+            ret_message = transferinfo.message
 
-            return False, transferinfo.message
+        else:
+            # 转移成功
+            logger.info(f"{task.fileitem.name} 入库成功：{transferinfo.target_diritem.path}")
 
-        # task转移成功
-        self.jobview.finish_task(task)
+            # 新增task转移成功历史记录
+            history = transferhis.add_success(
+                fileitem=task.fileitem,
+                mode=transferinfo.transfer_type if transferinfo else '',
+                downloader=task.downloader,
+                download_hash=task.download_hash,
+                meta=task.meta,
+                mediainfo=task.mediainfo,
+                transferinfo=transferinfo
+            )
 
-        logger.info(f"{task.fileitem.name} 入库成功：{transferinfo.target_diritem.path}")
+            # task整理完成事件
+            if self.__is_media_file(task.fileitem):
+                # 主要媒体文件整理完成事件
+                self.eventmanager.send_event(EventType.TransferComplete, {
+                    'fileitem': task.fileitem,
+                    'meta': task.meta,
+                    'mediainfo': task.mediainfo,
+                    'transferinfo': transferinfo,
+                    'downloader': task.downloader,
+                    'download_hash': task.download_hash,
+                    'transfer_history_id': history.id if history else None,
+                })
+            elif self.__is_subtitle_file(task.fileitem):
+                # 字幕整理完成事件
+                self.eventmanager.send_event(EventType.SubtitleTransferComplete, {
+                    'fileitem': task.fileitem,
+                    'meta': task.meta,
+                    'mediainfo': task.mediainfo,
+                    'transferinfo': transferinfo,
+                    'downloader': task.downloader,
+                    'download_hash': task.download_hash,
+                    'transfer_history_id': history.id if history else None,
+                })
+            elif self.__is_audio_file(task.fileitem):
+                # 音频文件整理完成事件
+                self.eventmanager.send_event(EventType.AudioTransferComplete, {
+                    'fileitem': task.fileitem,
+                    'meta': task.meta,
+                    'mediainfo': task.mediainfo,
+                    'transferinfo': transferinfo,
+                    'downloader': task.downloader,
+                    'download_hash': task.download_hash,
+                    'transfer_history_id': history.id if history else None,
+                })
 
-        # 新增task转移成功历史记录
-        transferhis.add_success(
-            fileitem=task.fileitem,
-            mode=transferinfo.transfer_type if transferinfo else '',
-            downloader=task.downloader,
-            download_hash=task.download_hash,
-            meta=task.meta,
-            mediainfo=task.mediainfo,
-            transferinfo=transferinfo
-        )
+            # task登记转移成功文件清单
+            target_dir_path = transferinfo.target_diritem.path
+            target_files = transferinfo.file_list_new
+            with job_lock:
+                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
 
-        # task整理完成事件
-        if self.__is_media_file(task.fileitem):
-            self.eventmanager.send_event(EventType.TransferComplete, {
-                'fileitem': task.fileitem,
-                'meta': task.meta,
-                'mediainfo': task.mediainfo,
-                'transferinfo': transferinfo,
-                'downloader': task.downloader,
-                'download_hash': task.download_hash,
-            })
+            # 设置任务成功
+            self.jobview.finish_task(task)
 
-        # task登记转移成功文件清单
-        target_dir_path = transferinfo.target_diritem.path
-        target_files = transferinfo.file_list_new
-        with job_lock:
-            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_finished(task):
+            __notify()
 
-        # 全部整理成功时
-        if self.jobview.is_success(task):
-            # 移动模式删除空目录
-            if transferinfo.transfer_type in ["move"]:
+        # 只要该种子的所有任务都已整理完成，则设置种子状态为已整理
+        if task.download_hash and self.jobview.is_torrent_done(task.download_hash):
+            self.transfer_completed(hashs=task.download_hash, downloader=task.downloader)
+
+        # 移动模式，全部成功时删除空目录和种子文件
+        if transferinfo.transfer_type in ["move"]:
+            # 全部整理成功时
+            if self.jobview.is_success(task):
                 # 所有成功的业务
                 tasks = self.jobview.success_tasks(task.mediainfo, task.meta.begin_season)
-                # 获取整理屏蔽词
-                transfer_exclude_words = SystemConfigOper().get(SystemConfigKey.TransferExcludeWords)
+                processed_hashes = set()
                 for t in tasks:
-                    if t.download_hash and self._can_delete_torrent(t.download_hash, t.downloader,
-                                                                    transfer_exclude_words):
-                        if self.remove_torrents(t.download_hash, downloader=t.downloader):
-                            logger.info(f"移动模式删除种子成功：{t.download_hash}")
-                    if t.fileitem:
+                    if t.download_hash and t.download_hash not in processed_hashes:
+                        # 检查该种子的所有任务（跨作业）是否都已成功
+                        if self.jobview.is_torrent_success(t.download_hash):
+                            processed_hashes.add(t.download_hash)
+                            # 移除种子及文件
+                            if self.remove_torrents(t.download_hash, downloader=t.downloader):
+                                logger.info(f"移动模式删除种子成功：{t.download_hash}")
+                    if not t.download_hash and t.fileitem:
+                        # 删除剩余空目录
                         StorageChain().delete_media_file(t.fileitem, delete_self=False)
 
-        # 全部整理完成且有成功的任务时
-        if self.jobview.is_finished(task):
-            # 发送消息、刮削事件、移除任务
-            __all_finished()
-
-        # 全部整理完成不管成功还是失败
-        if self.jobview.is_done(task):
-            # 所有任务
-            tasks = self.jobview.all_tasks()
-            for t in tasks:
-                if t.download_hash:
-                    # 设置种子状态为已整理
-                    self.transfer_completed(hashs=t.download_hash, downloader=t.downloader)
-
-        return True, ""
+        return ret_status, ret_message
 
     def put_to_queue(self, task: TransferTask):
         """
@@ -940,7 +1029,19 @@ class TransferChain(ChainBase, ConfigReloadMixin, metaclass=Singleton):
             logger.info("开始整理下载器中已经完成下载的文件 ...")
 
             # 从下载器获取种子列表
-            torrents: Optional[List[TransferTorrent]] = self.list_torrents(status=TorrentStatus.TRANSFER)
+            if torrents_list := self.list_torrents(status=TorrentStatus.TRANSFER):
+                seen = set()
+                existing_hashes = self.jobview.get_all_torrent_hashes()
+                torrents = [
+                    torrent
+                    for torrent in torrents_list
+                    if (h := torrent.hash) not in existing_hashes
+                       # 排除多下载器返回的重复种子
+                       and (h not in seen and (seen.add(h) or True))
+                ]
+            else:
+                torrents = []
+
             if not torrents:
                 logger.info("没有已完成下载但未整理的任务")
                 return False
@@ -975,20 +1076,11 @@ class TransferChain(ChainBase, ConfigReloadMixin, metaclass=Singleton):
                     # 查询下载记录识别情况
                     downloadhis: DownloadHistory = DownloadHistoryOper().get_by_hash(torrent.hash)
                     if downloadhis:
-                        # 获取自定义识别词
-                        custom_words_list = None
-                        if downloadhis.custom_words:
-                            custom_words_list = downloadhis.custom_words.split('\n')
-
                         # 类型
                         try:
                             mtype = MediaType(downloadhis.type)
                         except ValueError:
                             mtype = MediaType.TV
-
-                        # 识别元数据
-                        metainfo = MetaInfoPath(file_path, custom_words=custom_words_list)
-
                         # 识别媒体信息
                         mediainfo = self.recognize_media(mtype=mtype,
                                                          tmdbid=downloadhis.tmdbid,
@@ -1003,14 +1095,8 @@ class TransferChain(ChainBase, ConfigReloadMixin, metaclass=Singleton):
 
                     else:
                         # 非MoviePilot下载的任务，按文件识别
-                        metainfo = MetaInfoPath(file_path)
                         mediainfo = None
 
-                    # 检查是否已经有任务处理中，如有则跳过本次整理
-                    if self.jobview.has_tasks(meta=metainfo, mediainfo=mediainfo):
-                        logger.info(f"有任务正在整理中，跳过本次整理 ...")
-                        return False
-
                     # 执行异步整理，匹配源目录
                     self.do_transfer(
                         fileitem=FileItem(
@@ -1152,8 +1238,9 @@ class TransferChain(ChainBase, ConfigReloadMixin, metaclass=Singleton):
 
         # 过滤后缀和大小（蓝光目录、附加文件不过滤大小）
         file_items = [f for f in file_items if f[1] or
-                      self.__is_extra_file(f[0]) or
-                      (self.__is_allowed_file(f[0]) and self.__is_allow_filesize(f[0], min_filesize))]
+                      self.__is_subtitle_file(f[0]) or
+                      self.__is_audio_file(f[0]) or
+                      (self.__is_media_file(f[0]) and self.__is_allow_filesize(f[0], min_filesize))]
 
         if not file_items:
             logger.warn(f"{fileitem.path} 没有找到可整理的媒体文件")
@@ -1195,7 +1282,10 @@ class TransferChain(ChainBase, ConfigReloadMixin, metaclass=Singleton):
                 # 提前获取下载历史，以便获取自定义识别词
                 download_history = None
                 downloadhis = DownloadHistoryOper()
-                if bluray_dir:
+                if download_hash:
+                    # 先按hash查询
+                    download_history = downloadhis.get_by_hash(download_hash)
+                elif bluray_dir:
                     # 蓝光原盘，按目录名查询
                     download_history = downloadhis.get_by_path(file_path.as_posix())
                 else:
@@ -1204,14 +1294,15 @@ class TransferChain(ChainBase, ConfigReloadMixin, metaclass=Singleton):
                     if download_file:
                         download_history = downloadhis.get_by_hash(download_file.download_hash)
 
-                # 获取自定义识别词
-                custom_words_list = None
-                if download_history and download_history.custom_words:
-                    custom_words_list = download_history.custom_words.split('\n')
-
                 if not meta:
-                    # 文件元数据（传入自定义识别词）
-                    file_meta = MetaInfoPath(file_path, custom_words=custom_words_list)
+                    subscribe_custom_words = None
+                    if download_history and isinstance(download_history.note, dict):
+                        # 使用source动态获取订阅
+                        subscribe = SubscribeChain().get_subscribe_by_source(download_history.note.get("source"))
+                        subscribe_custom_words = subscribe.custom_words.split(
+                            "\n") if subscribe and subscribe.custom_words else None
+                    # 文件元数据(优先使用订阅识别词)
+                    file_meta = MetaInfoPath(file_path, custom_words=subscribe_custom_words)
                 else:
                     file_meta = meta
 
@@ -1549,46 +1640,3 @@ class TransferChain(ChainBase, ConfigReloadMixin, metaclass=Singleton):
                 logger.warn(f"{file_path} 命中屏蔽词 {keyword}")
                 return True
         return False
-
-    def _can_delete_torrent(self, download_hash: str, downloader: str, transfer_exclude_words) -> bool:
-        """
-        检查是否可以删除种子文件
-        :param download_hash: 种子Hash
-        :param downloader: 下载器名称
-        :param transfer_exclude_words: 整理屏蔽词
-        :return: 如果可以删除返回True，否则返回False
-        """
-        try:
-            # 获取种子信息
-            torrents = self.list_torrents(hashs=download_hash, downloader=downloader)
-            if not torrents:
-                return False
-
-            # 未下载完成
-            if torrents[0].progress < 100:
-                return False
-
-            # 获取种子文件列表
-            torrent_files = self.torrent_files(download_hash, downloader)
-            if not torrent_files:
-                return False
-
-            if not isinstance(torrent_files, list):
-                torrent_files = torrent_files.data
-
-            # 检查是否有媒体文件未被屏蔽且存在
-            save_path = torrents[0].path.parent
-            for file in torrent_files:
-                file_path = save_path / file.name
-                # 如果存在未被屏蔽的媒体文件，则不删除种子
-                if (file_path.suffix in self._allowed_exts
-                        and not self._is_blocked_by_exclude_words(file_path.as_posix(), transfer_exclude_words)
-                        and file_path.exists()):
-                    return False
-
-            # 所有媒体文件都被屏蔽或不存在，可以删除种子
-            return True
-
-        except Exception as e:
-            logger.error(f"检查种子 {download_hash} 是否需要删除失败：{e}")
-            return False

app/core/config.py

@@ -337,7 +337,7 @@ class ConfigModel(BaseModel):
                           "https://github.com/thsrite/MoviePilot-Plugins,"
                           "https://github.com/honue/MoviePilot-Plugins,"
                           "https://github.com/InfinityPacer/MoviePilot-Plugins,"
-                          "https://github.com/DDS-Derek/MoviePilot-Plugins,"
+                          "https://github.com/DDSRem-Dev/MoviePilot-Plugins,"
                           "https://github.com/madrays/MoviePilot-Plugins,"
                           "https://github.com/justzerock/MoviePilot-Plugins,"
                           "https://github.com/KoWming/MoviePilot-Plugins,"
@@ -347,7 +347,12 @@ class ConfigModel(BaseModel):
                           "https://github.com/Aqr-K/MoviePilot-Plugins,"
                           "https://github.com/hotlcc/MoviePilot-Plugins-Third,"
                           "https://github.com/gxterry/MoviePilot-Plugins,"
-                          "https://github.com/DzAvril/MoviePilot-Plugins")
+                          "https://github.com/DzAvril/MoviePilot-Plugins,"
+                          "https://github.com/mrtian2016/MoviePilot-Plugins,"
+                          "https://github.com/Hqyel/MoviePilot-Plugins-Third,"
+                          "https://github.com/xijin285/MoviePilot-Plugins,"
+                          "https://github.com/Seed680/MoviePilot-Plugins,"
+                          "https://github.com/imaliang/MoviePilot-Plugins")
     # 插件安装数据共享
     PLUGIN_STATISTIC_SHARE: bool = True
     # 是否开启插件热加载
@@ -427,10 +432,12 @@ class ConfigModel(BaseModel):
     LLM_API_KEY: Optional[str] = None
     # LLM基础URL（用于自定义API端点）
     LLM_BASE_URL: Optional[str] = "https://api.deepseek.com"
+    # LLM最大上下文Token数量（K）
+    LLM_MAX_CONTEXT_TOKENS: int = 64
     # LLM温度参数
     LLM_TEMPERATURE: float = 0.1
     # LLM最大迭代次数
-    LLM_MAX_ITERATIONS: int = 15
+    LLM_MAX_ITERATIONS: int = 128
     # LLM工具调用超时时间（秒）
     LLM_TOOL_TIMEOUT: int = 300
     # 是否启用详细日志
@@ -445,10 +452,14 @@ class ConfigModel(BaseModel):
     AI_RECOMMEND_ENABLED: bool = False
     # AI推荐用户偏好
     AI_RECOMMEND_USER_PREFERENCE: str = ""
+    # Tavily API密钥（用于网络搜索）
+    TAVILY_API_KEY: str = "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"
+
     # AI推荐条目数量限制
     AI_RECOMMEND_MAX_ITEMS: int = 50
 
 
+
 class Settings(BaseSettings, ConfigModel, LogConfigModel):
     """
     系统配置类

app/core/meta/metabase.py

@@ -535,7 +535,7 @@ class MetaBase(object):
 
     def merge(self, meta: Self):
         """
-        全并Meta信息
+        合并Meta信息
         """
         # 类型
         if self.type == MediaType.UNKNOWN \

app/db/models/downloadhistory.py

@@ -209,7 +209,7 @@ class DownloadFiles(Base):
     @classmethod
     @db_query
     def get_by_hash(cls, db: Session, download_hash: str, state: Optional[int] = None):
-        if state:
+        if state is not None:
             return db.query(cls).filter(cls.download_hash == download_hash,
                                         cls.state == state).all()
         else:

app/db/models/subscribe.py

@@ -227,6 +227,66 @@ class Subscribe(Base):
         )
         return result.scalars().first()
 
+    @classmethod
+    @db_query
+    def get_by(cls, db: Session, type: str, season: Optional[str] = None,
+                tmdbid: Optional[int] = None, doubanid: Optional[str] = None, bangumiid: Optional[str] = None):
+        """
+        根据条件查询订阅
+        """
+        # TMDBID
+        if tmdbid:
+            if season is not None:
+                result = db.query(cls).filter(
+                    cls.tmdbid == tmdbid, cls.type == type, cls.season == season
+                )
+            else:
+                result = db.query(cls).filter(cls.tmdbid == tmdbid, cls.type == type)
+        # 豆瓣ID
+        elif doubanid:
+            result = db.query(cls).filter(cls.doubanid == doubanid, cls.type == type)
+        # BangumiID
+        elif bangumiid:
+            result = db.query(cls).filter(cls.bangumiid == bangumiid, cls.type == type)
+        else:
+            return None
+
+        return result.first()
+
+    @classmethod
+    @async_db_query
+    async def async_get_by(cls, db: AsyncSession, type: str, season: Optional[str] = None,
+                tmdbid: Optional[int] = None, doubanid: Optional[str] = None, bangumiid: Optional[str] = None):
+        """
+        根据条件查询订阅
+        """
+        # TMDBID
+        if tmdbid:
+            if season is not None:
+                result = await db.execute(
+                    select(cls).filter(
+                        cls.tmdbid == tmdbid, cls.type == type, cls.season == season
+                    )
+                )
+            else:
+                result = await db.execute(
+                    select(cls).filter(cls.tmdbid == tmdbid, cls.type == type)
+                )
+        # 豆瓣ID
+        elif doubanid:
+            result = await db.execute(
+                select(cls).filter(cls.doubanid == doubanid, cls.type == type)
+            )
+        # BangumiID
+        elif bangumiid:
+            result = await db.execute(
+                select(cls).filter(cls.bangumiid == bangumiid, cls.type == type)
+            )
+        else:
+            return None
+
+        return result.scalars().first()
+
     @db_update
     def delete_by_tmdbid(self, db: Session, tmdbid: int, season: int):
         subscrbies = self.get_by_tmdbid(db, tmdbid, season)

app/db/subscribe_oper.py

@@ -71,6 +71,7 @@ class SubscribeOper(DbOper):
             "backdrop": mediainfo.get_backdrop_image(),
             "vote": mediainfo.vote_average,
             "description": mediainfo.overview,
+            "search_imdbid": 1 if kwargs.get('search_imdbid') else 0,
             "date": time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())
         })
         if not subscribe:
@@ -111,6 +112,20 @@ class SubscribeOper(DbOper):
         """
         return await Subscribe.async_get(self._db, rid=sid)
 
+    def get_by(self, type: str, season: Optional[str] = None, tmdbid: Optional[int] = None,
+               doubanid: Optional[str] = None, bangumiid: Optional[str] = None) -> Optional[Subscribe]:
+        """
+        根据条件查询订阅
+        """
+        return Subscribe.get_by(self._db, type, season, tmdbid, doubanid, bangumiid)
+
+    async def async_get_by(self, type: str, season: Optional[str] = None, tmdbid: Optional[int] = None,
+                           doubanid: Optional[str] = None, bangumiid: Optional[str] = None) -> Optional[Subscribe]:
+        """
+        根据条件查询订阅
+        """
+        return await Subscribe.async_get_by(self._db, type, season, tmdbid, doubanid, bangumiid)
+
     def list(self, state: Optional[str] = None) -> List[Subscribe]:
         """
         获取订阅列表

app/helper/message.py

@@ -539,7 +539,7 @@ class MessageTemplateHelper:
         获取消息模板
         """
         template_dict: dict[str, str] = SystemConfigOper().get(SystemConfigKey.NotificationTemplates)
-        return template_dict.get(f"{message.ctype.value}")
+        return template_dict.get(message.ctype.value)
 
 
 class MessageQueueManager(metaclass=SingletonClass):

app/modules/filemanager/__init__.py

@@ -36,7 +36,7 @@ class FileManagerModule(_ModuleBase):
         self._storage_schemas = ModuleHelper.load('app.modules.filemanager.storages',
                                                   filter_func=lambda _, obj: hasattr(obj, 'schema') and obj.schema)
         # 获取存储类型
-        self._support_storages = [storage.schema.value for storage in self._storage_schemas]
+        self._support_storages = [storage.schema.value for storage in self._storage_schemas if storage.schema]
 
     @staticmethod
     def get_name() -> str:
@@ -464,7 +464,7 @@ class FileManagerModule(_ModuleBase):
         else:
             # 未找到有效的媒体库目录
             logger.error(
-                f"{mediainfo.type.value} {mediainfo.title_year} 未找到有效的媒体库目录，无法整理文件，源路径：{fileitem.path}")
+                f"{mediainfo.type.value if mediainfo.type else '未知类型'} {mediainfo.title_year} 未找到有效的媒体库目录，无法整理文件，源路径：{fileitem.path}")
             return TransferInfo(success=False,
                                 fileitem=fileitem,
                                 message="未找到有效的媒体库目录")

app/modules/filemanager/storages/u115.py

@@ -268,7 +268,7 @@ class U115Pan(StorageBase, metaclass=WeakSingleton):
 
         # 返回数据
         ret_data = resp.json()
-        if ret_data.get("code") != 0:
+        if ret_data.get("code") not in (0, 20004):
             error_msg = ret_data.get("message")
             if not no_error_log:
                 logger.warn(f"【115】{method} 请求 {endpoint} 出错：{error_msg}")
@@ -386,7 +386,10 @@ class U115Pan(StorageBase, metaclass=WeakSingleton):
         resp = self._request_api(
             "POST",
             "/open/folder/add",
-            data={"pid": int(parent_item.fileid or "0"), "file_name": name},
+            data={
+                "pid": 0 if parent_item.path == "/" else int(parent_item.fileid or 0),
+                "file_name": name,
+            },
         )
         if not resp:
             return None

app/modules/filemanager/transhandler.py

@@ -150,10 +150,9 @@ class TransHandler:
                 if stream_fileitem := source_oper.get_item(
                         Path(fileitem.path) / "BDMV" / "STREAM"
                 ):
-                    fileitem.size = 0
-                    files = source_oper.list(stream_fileitem) or []
-                    for file in files:
-                        fileitem.size += file.size
+                    fileitem.size = sum(
+                        file.size for file in source_oper.list(stream_fileitem) or []
+                    )
                 # 整理目录
                 new_diritem, errmsg = self.__transfer_dir(fileitem=fileitem,
                                                           mediainfo=mediainfo,
@@ -708,7 +707,7 @@ class TransHandler:
         """
         获取目标路径
         """
-        if need_type_folder:
+        if need_type_folder and mediainfo.type:
             target_path = target_path / mediainfo.type.value
         if need_category_folder and mediainfo.category:
             target_path = target_path / mediainfo.category
@@ -728,7 +727,7 @@ class TransHandler:
             need_type_folder = target_dir.library_type_folder
         if need_category_folder is None:
             need_category_folder = target_dir.library_category_folder
-        if not target_dir.media_type and need_type_folder:
+        if not target_dir.media_type and need_type_folder and mediainfo.type:
             # 一级自动分类
             library_dir = Path(target_dir.library_path) / mediainfo.type.value
         elif target_dir.media_type and need_type_folder:

app/modules/filter/__init__.py

@@ -7,11 +7,12 @@ from app.helper.rule import RuleHelper
 from app.log import logger
 from app.modules import _ModuleBase
 from app.modules.filter.RuleParser import RuleParser
-from app.schemas.types import ModuleType, OtherModulesType
+from app.schemas.types import ModuleType, OtherModulesType, SystemConfigKey
 from app.utils.string import StringUtils
 
 
 class FilterModule(_ModuleBase):
+    CONFIG_WATCH = {SystemConfigKey.CustomFilterRules.value}
     # 规则解析器
     parser: RuleParser = None
     # 媒体信息
@@ -44,7 +45,8 @@ class FilterModule(_ModuleBase):
             "include": [
                 r'[中国國繁简](/|\s|\\|\|)?[繁简英粤]|[英简繁](/|\s|\\|\|)?[中繁简]'
                 r'|繁體|简体|[中国國][字配]|国语|國語|中文|中字|简日|繁日|简繁|繁体'
-                r'|([\s,.-\[])(CHT|CHS|cht|chs)(|[\s,.-\]])'],
+                r'|([\s,.-\[])(chs|cht)(|[\s,.-\]])'
+                r'|(?<![a-z0-9])(gb|big5)(?![a-z0-9])'],
             "exclude": [],
             "tmdb": {
                 "original_language": "zh,cn"
@@ -203,8 +205,6 @@ class FilterModule(_ModuleBase):
         if not rule_groups:
             return torrent_list
         self.media = mediainfo
-        # 重新加载自定义规则
-        self.__init_custom_rules()
         # 查询规则表详情
         groups = self.rulehelper.get_rule_group_by_media(media=mediainfo, group_names=rule_groups)
         if groups:
@@ -227,7 +227,7 @@ class FilterModule(_ModuleBase):
         for torrent in torrent_list:
             # 能命中优先级的才返回
             if not self.__get_order(torrent, rule_string):
-                logger.debug(f"种子 {torrent.site_name} - {torrent.title} {torrent.description} "
+                logger.debug(f"种子 {torrent.site_name} - {torrent.title} {torrent.description or ''} "
                              f"不匹配 {rule_name} 过滤规则")
                 continue
             ret_torrents.append(torrent)

app/modules/indexer/__init__.py

@@ -434,7 +434,7 @@ class IndexerModule(_ModuleBase):
             获取站点解析器
             """
             for site_schema in self._site_schemas:
-                if site_schema.schema.value == site.get("schema"):
+                if site_schema.schema and site_schema.schema.value == site.get("schema"):
                     return site_schema(
                         site_name=site.get("name"),
                         url=site.get("url"),

app/modules/indexer/parser/rousi.py

@@ -197,13 +197,13 @@ class RousiSiteUserInfo(SiteParserBase):
                 url=urljoin(self._base_url, "api/messages"),
                 params=params
             )
-            if not res or res.status_code != 200 or not res.text:
+            if not res or res.status_code != 200 or res.json().get("code", -1) != 0:
                 logger.warn(f"{self._site_name} 站点解析消息失败，状态码: {res.status_code if res else '无响应'}")
                 return {
                     "messages": [],
                     "total_pages": 0
                 }
-            return res.json()
+            return res.json().get("data")
         
         # 分页获取所有未读消息
         page = 0

app/modules/themoviedb/__init__.py

@@ -14,10 +14,12 @@ from app.modules.themoviedb.category import CategoryHelper
 from app.modules.themoviedb.scraper import TmdbScraper
 from app.modules.themoviedb.tmdb_cache import TmdbCache
 from app.modules.themoviedb.tmdbapi import TmdbApi
+from app.schemas.category import CategoryConfig
 from app.schemas.types import MediaType, MediaImageType, ModuleType, MediaRecognizeType
 from app.utils.http import RequestUtils
 
 
+
 class TheMovieDbModule(_ModuleBase):
     """
     TMDB媒体信息匹配
@@ -1290,3 +1292,15 @@ class TheMovieDbModule(_ModuleBase):
         self.tmdb.clear_cache()
         self.cache.clear()
         logger.info("TMDB缓存清除完成")
+
+    def load_category_config(self) -> CategoryConfig:
+        """
+        加载分类配置
+        """
+        return self.category.load()
+
+    def save_category_config(self, config: CategoryConfig) -> bool:
+        """
+        保存分类配置
+        """
+        return self.category.save(config)

app/modules/themoviedb/category.py

@@ -7,8 +7,23 @@ from ruamel.yaml import CommentedMap
 
 from app.core.config import settings
 from app.log import logger
+from app.schemas.category import CategoryConfig
 from app.utils.singleton import WeakSingleton
 
+HEADER_COMMENTS = """####### 配置说明 #######
+# 1. 该配置文件用于配置电影和电视剧的分类策略，配置后程序会按照配置的分类策略名称进行分类，配置文件采用yaml格式，需要严格符合语法规则
+# 2. 配置文件中的一级分类名称：`movie`、`tv` 为固定名称不可修改，二级名称同时也是目录名称，会按先后顺序匹配，匹配后程序会按这个名称建立二级目录
+# 3. 支持的分类条件：
+#   `original_language` 语种，具体含义参考下方字典
+#   `production_countries` 国家或地区（电影）、`origin_country` 国家或地区（电视剧），具体含义参考下方字典
+#   `genre_ids` 内容类型，具体含义参考下方字典
+#   `release_year` 发行年份，格式：YYYY，电影实际对应`release_date`字段，电视剧实际对应`first_air_date`字段，支持范围设定，如：`YYYY-YYYY`
+#   themoviedb 详情API返回的其它一级字段
+# 4. 配置多项条件时需要同时满足，一个条件需要匹配多个值是使用`,`分隔
+# 5. !条件值表示排除该值
+
+"""
+
 
 class CategoryHelper(metaclass=WeakSingleton):
     """
@@ -31,8 +46,8 @@ class CategoryHelper(metaclass=WeakSingleton):
                 shutil.copy(settings.INNER_CONFIG_PATH / "category.yaml", self._category_path)
             with open(self._category_path, mode='r', encoding='utf-8') as f:
                 try:
-                    yaml = ruamel.yaml.YAML()
-                    self._categorys = yaml.load(f)
+                    yaml_loader = ruamel.yaml.YAML()
+                    self._categorys = yaml_loader.load(f)
                 except Exception as e:
                     logger.warn(f"二级分类策略配置文件格式出现严重错误！请检查：{str(e)}")
                     self._categorys = {}
@@ -44,6 +59,40 @@ class CategoryHelper(metaclass=WeakSingleton):
             self._tv_categorys = self._categorys.get('tv')
         logger.info(f"已加载二级分类策略 category.yaml")
 
+    def load(self) -> CategoryConfig:
+        """
+        加载配置
+        """
+        config = CategoryConfig()
+        if not self._category_path.exists():
+            return config
+        try:
+            with open(self._category_path, 'r', encoding='utf-8') as f:
+                yaml_loader = ruamel.yaml.YAML()
+                data = yaml_loader.load(f)
+                if data:
+                    config = CategoryConfig(**data)
+        except Exception as e:
+            logger.error(f"Load category config failed: {e}")
+        return config
+
+    def save(self, config: CategoryConfig) -> bool:
+        """
+        保存配置
+        """
+        data = config.model_dump(exclude_none=True)
+        try:
+            with open(self._category_path, 'w', encoding='utf-8') as f:
+                f.write(HEADER_COMMENTS)
+                yaml_dumper = ruamel.yaml.YAML()
+                yaml_dumper.dump(data, f)
+            # 保存后重新加载配置
+            self.init()
+            return True
+        except Exception as e:
+            logger.error(f"Save category config failed: {e}")
+            return False
+
     @property
     def is_movie_category(self) -> bool:
         """

app/schemas/category.py

@@ -0,0 +1,31 @@
+from typing import Dict, Optional
+
+from pydantic import BaseModel, ConfigDict
+
+
+class CategoryRule(BaseModel):
+    """
+    分类规则详情
+    """
+    # 内容类型
+    genre_ids: Optional[str] = None
+    # 语种
+    original_language: Optional[str] = None
+    # 国家或地区（电视剧）
+    origin_country: Optional[str] = None
+    # 国家或地区（电影）
+    production_countries: Optional[str] = None
+    # 发行年份
+    release_year: Optional[str] = None
+    # 允许接收其他动态字段
+    model_config = ConfigDict(extra='allow')
+
+
+class CategoryConfig(BaseModel):
+    """
+    分类策略配置
+    """
+    # 电影分类策略
+    movie: Optional[Dict[str, Optional[CategoryRule]]] = {}
+    # 电视剧分类策略
+    tv: Optional[Dict[str, Optional[CategoryRule]]] = {}

app/schemas/transfer.py

@@ -3,11 +3,11 @@ from typing import Optional, List, Any, Callable
 
 from pydantic import BaseModel, Field
 
-from app.schemas.tmdb import TmdbEpisode
-from app.schemas.history import DownloadHistory
 from app.schemas.context import MetaInfo, MediaInfo
 from app.schemas.file import FileItem
+from app.schemas.history import DownloadHistory
 from app.schemas.system import TransferDirectoryConf
+from app.schemas.tmdb import TmdbEpisode
 
 
 class TransferTorrent(BaseModel):
@@ -124,14 +124,6 @@ class TransferInfo(BaseModel):
     total_size: Optional[int] = Field(default=0)
     # 失败清单
     fail_list: Optional[list] = Field(default_factory=list)
-    # 处理字幕文件清单
-    subtitle_list: Optional[list] = Field(default_factory=list)
-    # 目标字幕文件清单
-    subtitle_list_new: Optional[list] = Field(default_factory=list)
-    # 处理音频文件清单
-    audio_list: Optional[list] = Field(default_factory=list)
-    # 目标音频文件清单
-    audio_list_new: Optional[list] = Field(default_factory=list)
     # 错误信息
     message: Optional[str] = None
     # 是否需要刮削

app/schemas/types.py

@@ -38,10 +38,18 @@ class EventType(Enum):
     SiteUpdated = "site.updated"
     # 站点已刷新
     SiteRefreshed = "site.refreshed"
-    # 整理完成
+    # 媒体文件整理完成
     TransferComplete = "transfer.complete"
-    # 整理失败
+    # 媒体文件整理失败
     TransferFailed = "transfer.failed"
+    # 字幕整理完成
+    SubtitleTransferComplete = "transfer.subtitle.complete"
+    # 字幕整理失败
+    SubtitleTransferFailed = "transfer.subtitle.failed"
+    # 音频文件整理完成
+    AudioTransferComplete = "transfer.audio.complete"
+    # 音频文件整理失败
+    AudioTransferFailed = "transfer.audio.failed"
     # 下载已添加
     DownloadAdded = "download.added"
     # 删除历史记录
@@ -88,6 +96,11 @@ EVENT_TYPE_NAMES = {
     EventType.SiteUpdated: "站点已更新",
     EventType.SiteRefreshed: "站点已刷新",
     EventType.TransferComplete: "整理完成",
+    EventType.TransferFailed: "整理失败",
+    EventType.SubtitleTransferComplete: "字幕整理完成",
+    EventType.SubtitleTransferFailed: "字幕整理失败",
+    EventType.AudioTransferComplete: "音频整理完成",
+    EventType.AudioTransferFailed: "音频整理失败",
     EventType.DownloadAdded: "添加下载",
     EventType.HistoryDeleted: "删除历史记录",
     EventType.DownloadFileDeleted: "删除下载源文件",

database/versions/41ef1dd7467c_2_2_2.py

@@ -6,8 +6,9 @@ Create Date: 2026-01-13 13:02:41.614029
 
 """
 
-from app.db import ScopedSession
-from app.db.models.systemconfig import SystemConfig
+from alembic import op
+from sqlalchemy import text
+
 from app.log import logger
 
 # revision identifiers, used by Alembic.
@@ -19,22 +20,28 @@ depends_on = None
 
 def upgrade() -> None:
     # systemconfig表 去重
-    with ScopedSession() as db:
-        try:
-            seen_keys = set()
-            # 按ID降序查询，以便保留最新的配置
-            for item in db.query(SystemConfig).order_by(SystemConfig.id.desc()).all():
-                if item.key in seen_keys:
-                    logger.warn(
-                        f"已删除重复的SystemConfig项：{item.key} 值:{item.value}"
-                    )
-                    db.delete(item)
-                else:
-                    seen_keys.add(item.key)
-            db.commit()
-        except Exception as e:
-            logger.error(e)
-            db.rollback()
+    connection = op.get_bind()
+
+    select_stmt = text(
+        """
+        SELECT id, key, value
+        FROM SystemConfig
+        WHERE id NOT IN (
+            SELECT MAX(id)
+            FROM SystemConfig
+            GROUP BY key
+        )
+    """
+    )
+    to_delete = connection.execute(select_stmt).fetchall()
+    for row in to_delete:
+        logger.warn(
+            f"已删除重复的 SystemConfig 项：key={row.key}, value={row.value}, id={row.id}"
+        )
+        delete_stmt = text("DELETE FROM SystemConfig WHERE id = :id")
+        connection.execute(delete_stmt, {"id": row.id})
+
+    logger.info("SystemConfig 表去重操作已完成。")
 
 
 def downgrade() -> None:

requirements.in

@@ -91,3 +91,4 @@ langchain-deepseek~=0.1.4
 langchain-experimental~=0.3.4
 openai~=1.108.2
 google-generativeai~=0.8.5
+ddgs~=9.10.0

version.py

@@ -1,2 +1,2 @@
-APP_VERSION = 'v2.9.5'
-FRONTEND_VERSION = 'v2.9.5'
+APP_VERSION = 'v2.9.7'
+FRONTEND_VERSION = 'v2.9.7'