mirror of
https://github.com/jxxghp/MoviePilot.git
synced 2026-05-08 22:33:16 +08:00
96 lines
3.8 KiB
Python
96 lines
3.8 KiB
Python
"""更新自定义识别词工具"""
|
|
|
|
import json
|
|
from typing import List, Optional, Type
|
|
|
|
from pydantic import BaseModel, Field
|
|
|
|
from app.agent.tools.base import MoviePilotTool
|
|
from app.db.systemconfig_oper import SystemConfigOper
|
|
from app.log import logger
|
|
from app.schemas.types import SystemConfigKey
|
|
|
|
|
|
class UpdateCustomIdentifiersInput(BaseModel):
|
|
"""更新自定义识别词工具的输入参数模型"""
|
|
|
|
explanation: str = Field(
|
|
...,
|
|
description="Clear explanation of why this tool is being used in the current context",
|
|
)
|
|
identifiers: List[str] = Field(
|
|
...,
|
|
description=(
|
|
"The complete list of custom identifier rules to save. "
|
|
"This REPLACES the entire existing list. "
|
|
"Always query existing identifiers first, merge new rules, then pass the full list."
|
|
),
|
|
)
|
|
|
|
|
|
class UpdateCustomIdentifiersTool(MoviePilotTool):
|
|
name: str = "update_custom_identifiers"
|
|
description: str = (
|
|
"Update the full list of custom identifiers (自定义识别词) used for preprocessing torrent/file names. "
|
|
"This tool REPLACES all existing identifier rules with the provided list. "
|
|
"IMPORTANT: Always use 'query_custom_identifiers' first to get existing rules, "
|
|
"then merge new rules into the list before calling this tool to avoid accidentally deleting existing rules. "
|
|
"Supported rule formats (spaces around operators are required): "
|
|
"1) Block word: just the word/regex to remove; "
|
|
"2) Replacement: '被替换词 => 替换词'; "
|
|
"3) Episode offset: '前定位词 <> 后定位词 >> EP±N'; "
|
|
"4) Combined: '被替换词 => 替换词 && 前定位词 <> 后定位词 >> EP±N'; "
|
|
"Lines starting with '#' are comments. "
|
|
"The replacement target supports: {[tmdbid=xxx;type=movie/tv;s=xxx;e=xxx]} for direct TMDB ID matching."
|
|
)
|
|
args_schema: Type[BaseModel] = UpdateCustomIdentifiersInput
|
|
|
|
def get_tool_message(self, **kwargs) -> Optional[str]:
|
|
"""生成友好的提示消息"""
|
|
identifiers = kwargs.get("identifiers", [])
|
|
return f"正在更新自定义识别词（共 {len(identifiers)} 条规则）"
|
|
|
|
async def run(self, identifiers: List[str] = None, **kwargs) -> str:
|
|
logger.info(
|
|
f"执行工具: {self.name}, 规则数量: {len(identifiers) if identifiers else 0}"
|
|
)
|
|
try:
|
|
if identifiers is None:
|
|
return json.dumps(
|
|
{"success": False, "message": "必须提供 identifiers 参数"},
|
|
ensure_ascii=False,
|
|
)
|
|
|
|
# 过滤空字符串
|
|
identifiers = [i for i in identifiers if i is not None]
|
|
|
|
system_config_oper = SystemConfigOper()
|
|
|
|
# 保存
|
|
value = identifiers if identifiers else None
|
|
success = await system_config_oper.async_set(
|
|
SystemConfigKey.CustomIdentifiers, value
|
|
)
|
|
if success:
|
|
return json.dumps(
|
|
{
|
|
"success": True,
|
|
"message": f"自定义识别词已更新，共 {len(identifiers)} 条规则",
|
|
"count": len(identifiers),
|
|
"identifiers": identifiers,
|
|
},
|
|
ensure_ascii=False,
|
|
indent=2,
|
|
)
|
|
else:
|
|
return json.dumps(
|
|
{"success": False, "message": "保存自定义识别词失败"},
|
|
ensure_ascii=False,
|
|
)
|
|
except Exception as e:
|
|
logger.error(f"更新自定义识别词失败: {e}")
|
|
return json.dumps(
|
|
{"success": False, "message": f"更新自定义识别词时发生错误: {str(e)}"},
|
|
ensure_ascii=False,
|
|
)
|