nas/MoviePilot
1
0
Fork 0
mirror of https://github.com/jxxghp/MoviePilot.git synced 2026-03-19 19:46:55 +08:00
Code Issues Packages Projects Releases Wiki Activity

fix agent

Browse Source
This commit is contained in:
jxxghp
2025-11-01 11:41:22 +08:00
parent 182c46037b
commit 247208b8a9
10 changed files with 114 additions and 121 deletions
Download Patch File Download Diff File Expand all files Collapse all files

43
app/agent/tools/impl/search_media.py
View File

@@ -3,24 +3,24 @@
import json
from typing import Optional
from app.agent.tools.base import MoviePilotTool
from app.chain.media import MediaChain
from app.core.metainfo import MetaInfo
from app.log import logger
from app.schemas.types import MediaType
from app.agent.tools.base import MoviePilotTool
class SearchMediaTool(MoviePilotTool):
name: str = "search_media"
description: str = "搜索媒体资源，包括电影、电视剧、动漫等。可以根据标题、年份、类型等条件进行搜索。"
async def _arun(self, title: str, explanation: str, year: Optional[str] = None,
async def _arun(self, title: str, explanation: str, year: Optional[str] = None,
media_type: Optional[str] = None, season: Optional[int] = None, **kwargs) -> str:
logger.info(f"执行工具: {self.name}, 参数: title={title}, year={year}, media_type={media_type}, season={season}")
logger.info(
f"执行工具: {self.name}, 参数: title={title}, year={year}, media_type={media_type}, season={season}")
# 发送工具执行说明
self._send_tool_message(f"正在搜索媒体资源: {title}" + (f" ({year})" if year else ""), title="搜索中")
self.send_tool_message(f"正在搜索媒体资源: {title}" + (f" ({year})" if year else ""), title="搜索中")
try:
media_chain = MediaChain()
# 构建搜索标题
@@ -31,10 +31,10 @@ class SearchMediaTool(MoviePilotTool):
search_title = f"{search_title} {media_type}"
if season:
search_title = f"{search_title} S{season:02d}"
# 使用 MediaChain.search 方法
meta, results = media_chain.search(title=search_title)
# 过滤结果
if results:
filtered_results = []
@@ -42,35 +42,32 @@ class SearchMediaTool(MoviePilotTool):
if year and result.year != year:
continue
if media_type:
try:
if result.type != MediaType(media_type):
continue
except:
pass
if result.type != MediaType(media_type):
continue
if season and result.season != season:
continue
filtered_results.append(result)
if filtered_results:
result_message = f"找到 {len(filtered_results)} 个相关媒体资源"
self._send_tool_message(result_message, title="搜索成功")
self.send_tool_message(result_message, title="搜索成功")
# 发送详细结果
for i, result in enumerate(filtered_results[:5]): # 只显示前5个结果
media_info = f"{i+1}. {result.title} ({result.year}) - {result.type.value if result.type else '未知'}"
self._send_tool_message(media_info, title="搜索结果")
media_info = f"{i + 1}. {result.title} ({result.year}) - {result.type.value if result.type else '未知'}"
self.send_tool_message(media_info, title="搜索结果")
return json.dumps([r.to_dict() for r in filtered_results], ensure_ascii=False, indent=2)
else:
error_message = f"未找到符合条件的媒体资源: {title}"
self._send_tool_message(error_message, title="搜索完成")
self.send_tool_message(error_message, title="搜索完成")
return error_message
else:
error_message = f"未找到相关媒体资源: {title}"
self._send_tool_message(error_message, title="搜索完成")
self.send_tool_message(error_message, title="搜索完成")
return error_message
except Exception as e:
error_message = f"搜索媒体失败: {str(e)}"
logger.error(f"搜索媒体失败: {e}", exc_info=True)
self._send_tool_message(error_message, title="搜索失败")
self.send_tool_message(error_message, title="搜索失败")
return error_message