feat(napcat_adapter): 增强视频处理配置,添加最大大小和超时设置,并更新消息处理器以支持新配置

This commit is contained in:
LuiKlee
2025-12-16 15:01:35 +08:00
parent dc57e7fcf9
commit a8903e73e1
4 changed files with 452 additions and 6 deletions

View File

@@ -414,7 +414,22 @@ class NapcatAdapterPlugin(BasePlugin):
"enable_emoji_like": ConfigField(type=bool, default=True, description="是否启用群聊表情回复处理"),
"enable_reply_at": ConfigField(type=bool, default=True, description="是否在回复时自动@原消息发送者"),
"reply_at_rate": ConfigField(type=float, default=0.5, description="回复时@的概率0.0-1.0"),
"enable_video_processing": ConfigField(type=bool, default=True, description="是否启用视频消息处理(下载和解析)"),
# ========== 视频消息处理配置 ==========
"enable_video_processing": ConfigField(
type=bool,
default=True,
description="是否启用视频消息处理(下载和解析)。关闭后视频消息将显示为 [视频消息] 占位符,不会进行下载"
),
"video_max_size_mb": ConfigField(
type=int,
default=100,
description="允许下载的视频文件最大大小MB超过此大小的视频将被跳过"
),
"video_download_timeout": ConfigField(
type=int,
default=60,
description="视频下载超时时间(秒),若超时将中止下载"
),
},
}

View File

@@ -37,10 +37,21 @@ class MessageHandler:
def __init__(self, adapter: "NapcatAdapter"):
self.adapter = adapter
self.plugin_config: dict[str, Any] | None = None
self._video_downloader = None
def set_plugin_config(self, config: dict[str, Any]) -> None:
"""设置插件配置"""
"""设置插件配置,并根据配置初始化视频下载器"""
self.plugin_config = config
# 如果启用了视频处理,根据配置初始化视频下载器
if config_api.get_plugin_config(config, "features.enable_video_processing", True):
from ..video_handler import VideoDownloader
max_size = config_api.get_plugin_config(config, "features.video_max_size_mb", 100)
timeout = config_api.get_plugin_config(config, "features.video_download_timeout", 60)
self._video_downloader = VideoDownloader(max_size_mb=max_size, download_timeout=timeout)
logger.debug(f"视频下载器已初始化: max_size={max_size}MB, timeout={timeout}s")
async def handle_raw_message(self, raw: dict[str, Any]):
"""
@@ -325,10 +336,13 @@ class MessageHandler:
},
}
elif video_url:
# URL下载处理
from ..video_handler import get_video_downloader
video_downloader = get_video_downloader()
download_result = await video_downloader.download_video(video_url)
# URL下载处理 - 使用配置中的下载器实例
downloader = self._video_downloader
if not downloader:
from ..video_handler import get_video_downloader
downloader = get_video_downloader()
download_result = await downloader.download_video(video_url)
if not download_result["success"]:
logger.warning(f"视频下载失败: {download_result.get('error', '未知错误')}")