K2CrO4/NeoBot
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: 添加直接发送视频/图片功能并优化临时目录处理

Browse Source 
refactor(WS): 使用TYPE_CHECKING优化导入并延迟导入Bot类
refactor(image_manager): 使用系统临时目录替代自定义临时目录
feat(bili/douyin): 添加直接发送视频/图片功能
chore: 删除forward_test插件并添加furry插件
refactor(main): 移除JIT检查代码并优化插件重载逻辑
This commit is contained in:
K2cr2O1
2026-01-23 09:32:16 +08:00
parent 36a44f6e96
commit cd7ed30672
8 changed files with 89 additions and 119 deletions
Download Patch File Download Diff File Expand all files Collapse all files

9
core/WS.py
View File

@@ -13,15 +13,17 @@ WebSocket 连接。它是整个机器人框架的底层通信基础。
"""
import asyncio
import orjson
from typing import Any, Dict, Optional, cast
from typing import TYPE_CHECKING, Any, Dict, Optional, cast
import uuid
if TYPE_CHECKING:
from .bot import Bot
import websockets
from websockets.legacy.client import WebSocketClientProtocol
from models.events.factory import EventFactory
from .bot import Bot
from .config_loader import global_config
from .managers.command_manager import matcher
from .utils.executor import CodeExecutor
@@ -56,7 +58,7 @@ class WS:
# 初始化状态
self.ws: Optional[WebSocketClientProtocol] = None
self._pending_requests: Dict[str, asyncio.Future] = {} # echo: future
self.bot: Bot | None = None
self.bot: 'Bot' | None = None
self.self_id: int | None = None
self.code_executor = code_executor
self.use_pool = use_pool
@@ -249,6 +251,7 @@ class WS:
# 尝试初始化 Bot 实例 (如果尚未初始化且事件包含 self_id)
# 只要事件中包含 self_id，我们就可以初始化 Bot，不必非要等待 meta_event
if self.bot is None and hasattr(event, 'self_id'):
from .bot import Bot
self.self_id = event.self_id
self.bot = Bot(self)
self.logger.success(f"Bot 实例初始化完成: self_id={self.self_id}")

6
core/managers/image_manager.py
View File

@@ -5,6 +5,7 @@
"""
import os
import base64
import tempfile
from typing import Dict, Any, Optional
from jinja2 import Template
@@ -27,9 +28,8 @@ class ImageManager(Singleton):
# 模板目录
self.template_dir = os.path.join(os.path.dirname(os.path.dirname(os.path.dirname(__file__))), "templates")
# 临时文件目录
# core/managers/image_manager.py -> core/managers -> core -> core/data/temp
self.temp_dir = os.path.join(os.path.dirname(os.path.dirname(__file__)), "data", "temp")
# 临时文件目录 - 使用系统临时目录
self.temp_dir = os.path.join(tempfile.gettempdir(), "neobot_images")
os.makedirs(self.temp_dir, exist_ok=True)
# 模板缓存
self._template_cache: Dict[str, Template] = {}

41
main.py
View File

@@ -21,46 +21,7 @@ from core.managers.browser_manager import browser_manager
from core.utils.executor import run_in_thread_pool, initialize_executor
from core.config_loader import global_config as config
# 检查 JIT 编译状态
def check_jit_status():
"""
检查 Python JIT 编译状态
该函数用于检测当前 Python 解释器是否启用了 JIT 编译功能，
并打印相关信息，帮助用户了解运行环境的性能优化状态。
"""
print("\n=== Python JIT 编译状态检查 ===")
# 检查解释器信息
print(f"Python 版本: {sys.version}")
print(f"解释器路径: {sys.executable}")
# 检查优化级别
print(f"优化级别 (-O): {sys.flags.optimize}")
# 检查 JIT 相关模块和功能
if sys.version_info >= (3, 10):
try:
# 对于 CPython 3.10+，检查是否启用了 JIT
import _opcode
if hasattr(_opcode, 'jit'):
print("JIT 状态: 已启用 (_opcode.jit)")
else:
print("JIT 状态: 未启用 (_opcode.jit 不可用)")
except ImportError:
print("JIT 状态: 未启用 (_opcode 模块不可用)")
else:
print("JIT 状态: 不可用 (需要 Python 3.10+)")
# 检查是否使用了 PyPy
if hasattr(sys, 'pypy_version_info'):
print(f"PyPy 版本: {sys.pypy_version_info}")
print("JIT 状态: 已启用 (PyPy 内置 JIT)")
print("==============================\n")
# 执行 JIT 状态检查
check_jit_status()
# 尝试使用高性能事件循环
try:
@@ -148,7 +109,7 @@ class PluginReloadHandler(FileSystemEventHandler):
try:
# 使用线程安全的方式在主事件循环中运行异步的插件重载函数
asyncio.run_coroutine_threadsafe(run_in_thread_pool(plugin_manager.reload_plugin, module_name), self.loop)
asyncio.run_coroutine_threadsafe(reload_plugin_and_sync_help(module_name), self.loop)
logger.success(f"插件 {module_name} 重载任务已提交")
except Exception as e:
logger.exception(f"重载失败: {e}")

43
plugins/forward_test.py
View File

@@ -1,43 +0,0 @@
"""
合并转发消息测试插件
"""
from core.managers.command_manager import matcher
from core.bot import Bot
from models.events.message import MessageEvent
from models.message import MessageSegment
__plugin_meta__ = {
"name": "furry",
"description": "处理 /furry 指令，发送furry图片，同时也是bot.build_forward_node演示",
"usage": "/furry - 发送一条furry图",
}
@matcher.command("furry")
async def handle_forward_test(bot: Bot, event: MessageEvent, args: list[str]):
"""
处理 /furry 指令，发送furry图片，同时也是bot.build_forward_node实例
:param bot: Bot 实例
:param event: 消息事件对象
:param args: 指令参数
"""
# 1. 构建消息节点列表
nickname = event.sender.nickname if event.sender else "未知用户"
nodes = [
bot.build_forward_node(user_id=event.self_id, nickname="机器人", message="你要的furry来了"),
bot.build_forward_node(user_id=event.user_id, nickname=nickname, message="让我看看"),
bot.build_forward_node(
user_id=event.self_id,
nickname="机器人",
message=[
MessageSegment.from_text("你要的福瑞图"),
MessageSegment.image("https://api.furry.ist/furry-img/")
]
)
]
try:
# 2. 发送合并转发消息
await bot.send_forwarded_messages(event, nodes)
except Exception as e:
await event.reply(f"发送失败: {e}")

61
plugins/furry.py Normal file
View File

@@ -0,0 +1,61 @@
"""
thpic 插件
提供 /furry 指令，用于随机返回一个东方Project的图片。
"""
from core.managers.command_manager import matcher
from core.bot import Bot
from models.events.message import MessageEvent
from models.message import MessageSegment
__plugin_meta__ = {
"name": "furry",
"description": "处理 /furry 指令，发送furry出毛图片",
"usage": "/furry - 发送一条furry图",
}
@matcher.command("furry")
async def handle_echo(bot: Bot, event: MessageEvent, args: list[str]):
"""
处理 furry 指令，发送一张随机的东方furry图片。
:param bot: Bot 实例（未使用）。
:param event: 消息事件对象。
:param args: 指令参数列表（未使用）。
"""
parts = args
print(parts)
if not parts:
try:
await event.reply(
str(MessageSegment.image("https://api.furry.ist/furry-img/"))
)
except Exception as e:
await event.reply(f"报错了。。。{e}")
else:
if parts[0].isdigit():
nums = int(parts[0])
if nums <= 0:
await event.reply("请输入一个大于0的整数。")
return
elif nums > 10:
await event.reply("请输入一个不大于10的整数。")
return
try:
nodes = []
for _ in range(nums):
nodes.append(
bot.build_forward_node(
user_id=event.self_id,
nickname="机器人",
message=MessageSegment.image(
"https://api.furry.ist/furry-img/"
),
)
)
await bot.send_forwarded_messages(event, nodes)
except Exception as e:
await event.reply(f"报错了。。。{e}")
else:
await event.reply(f"用法不正确。\n\n{__plugin_meta__['usage']}")

30
plugins/github_parser.py
View File

@@ -196,33 +196,3 @@ async def handle_github_command(bot, event: MessageEvent):
else:
await event.reply("参数格式错误，请输入：/查仓库 作者/仓库名")
# 注册消息处理器
@matcher.on_message()
async def handle_github_link(event: MessageEvent):
"""
处理消息，检测GitHub仓库链接并自动解析
Args:
event (MessageEvent): 消息事件对象
"""
# 忽略机器人自己发送的消息，防止无限循环
if hasattr(event, "user_id") and hasattr(event, "self_id") and event.user_id == event.self_id:
return
# 提取消息文本
message_text = ""
for segment in event.message:
if segment.type == "text":
message_text += segment.data.get("text", "")
# 查找GitHub仓库链接
match = GITHUB_URL_PATTERN.search(message_text)
if match:
owner = match.group(1)
repo = match.group(2)
# 移除可能的.git后缀
repo = repo.replace(".git", "")
logger.info(f"[github_parser] 检测到GitHub仓库链接: {owner}/{repo}")
await process_github_repo(event, owner, repo)

8
plugins/web_parser/parsers/bili.py
View File

@@ -198,6 +198,7 @@ class BiliParser(BaseParser):
"""
# 检查视频时长
video_message: Union[str, MessageSegment]
direct_url = None
if data['duration'] > 1200: # 20分钟 = 1200秒
video_message = "视频时长超过20分钟，不进行解析。"
else:
@@ -244,6 +245,13 @@ class BiliParser(BaseParser):
event.bot.build_forward_node(user_id=event.self_id, nickname=self.nickname, message=video_message)
]
# 同时直接发送视频（如果获取到直链）
if direct_url:
try:
await event.reply(MessageSegment.video(direct_url))
except Exception as e:
logger.error(f"[{self.name}] 直接发送视频失败: {e}")
return nodes
def should_handle_url(self, url: str) -> bool:

10
plugins/web_parser/parsers/douyin.py
View File

@@ -210,15 +210,18 @@ class DouyinParser(BaseParser):
# 尝试添加视频直链（单独节点）
video_success = False
direct_message = None
try:
if data.get('video_url'):
video_url = data.get('video_url', '')
# 检查视频类型
if data.get('type') == 'video':
video_message = MessageSegment.video(video_url)
direct_message = video_message
video_type_text = "视频直链："
else: # image类型
video_message = MessageSegment.image(video_url) # 单个图片
direct_message = video_message
video_type_text = "图集首图："
# 构建视频/图片节点
@@ -244,6 +247,13 @@ class DouyinParser(BaseParser):
)
nodes.append(no_video_node)
# 同时直接发送视频/图片（如果获取到直链）
if direct_message:
try:
await event.reply(direct_message)
except Exception as e:
logger.error(f"[{self.name}] 直接发送视频/图片失败: {e}")
return nodes
def should_handle_url(self, url: str) -> bool: