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

feat: 整合开发历史

Browse Source
This commit is contained in:
K2cr2O1
2026-01-04 19:38:47 +08:00
parent 0965123c1d
commit bbdeecb89b
25 changed files with 2199 additions and 506 deletions
Download Patch File Download Diff File Expand all files Collapse all files

147
plugins/admin.py
View File

@@ -1,115 +1,74 @@
from core import PluginDataManager
"""
管理员管理插件
提供通过聊天指令动态添加或移除机器人管理员的功能。
"""
from core.bot import Bot
from core.command_manager import matcher
from models import GroupMessageEvent
from core.admin_manager import admin_manager
from models.events.message import MessageEvent
__plugin_meta__ = {
"name": "admin",
"description": "机器人权限管理插件",
"usage": "/admin",
"name": "管理员管理",
"description": "管理机器人的全局管理员",
"usage": (
"/admin list - 列出所有管理员\n"
"/admin add <QQ号> - 添加管理员\n"
"/admin remove <QQ号> - 移除管理员"
),
}
data = PluginDataManager("admin")
@matcher.command("admin", permission=MessageEvent.ADMIN)
async def handle_admin_command(bot: Bot, event: MessageEvent, args: list[str]):
"""
处理 /admin 指令
@matcher.command("admin")
async def handle_permission(bot: Bot, event: GroupMessageEvent, args: list[str]):
:param bot: Bot 实例
:param event: 消息事件实例
:param args: 指令参数列表
"""
if not args:
await event.reply(
"机器人权限管理插件指令:\n/admin list 列出所有权限\n/admin add member <QQ号> 添加群成员权限\n/admin remove member <QQ号> 删除群成员权限\n/admin add group <群号> 添加群权限\n/admin remove group <群号> 删除群权限\n/admin clear member 清空群成员权限\n/admin clear group 清空群权限\n/admin clear all 清空所有权限"
)
return
if str(event.user_id) not in data.get("members", []):
await event.reply("你没有权限使用此命令。")
return
if str(event.group_id) not in data.get("groups", []):
await event.reply("群聊不在权限中")
await event.reply(__plugin_meta__["usage"])
return
action = args[0].lower()
# ensure storage keys exist
members = data.get("members", []) or []
groups = data.get("groups", []) or []
if action == "list":
msg_lines = ["当前权限列表:"]
msg_lines.append(
f"群成员权限 ({len(members)}): {', '.join(members) if members else '无'}"
)
msg_lines.append(
f"群权限 ({len(groups)}): {', '.join(groups) if groups else '无'}"
)
await event.reply("\n".join(msg_lines))
admins = await admin_manager.get_all_admins()
if not admins:
await event.reply("当前没有设置任何管理员。")
return
admin_list_str = "\n".join(str(admin_id) for admin_id in admins)
await event.reply(f"当前管理员列表 ({len(admins)}):\n{admin_list_str}")
return
if action in ("add", "remove"):
if len(args) < 3:
await event.reply("参数错误，示例：/admin add member 123456")
if len(args) < 2 or not args[1].isdigit():
await event.reply("参数错误，请提供一个有效的 QQ 号。\n示例: /admin add 123456")
return
target = args[1].lower()
value = args[2]
if target == "member":
# operate on members list
if action == "add":
if str(value) in members:
await event.reply(f"成员 {value} 已存在，无需重复添加。")
return
members.append(str(value))
data.set("members", members)
await event.reply(f"已添加群成员权限：{value}")
return
else: # remove
if str(value) not in members:
await event.reply(f"成员 {value} 不在权限列表中。")
return
members = [m for m in members if m != str(value)]
data.set("members", members)
await event.reply(f"已移除群成员权限：{value}")
return
if target == "group":
if action == "add":
if str(value) in groups:
await event.reply(f"群 {value} 已存在，无需重复添加。")
return
groups.append(str(value))
data.set("groups", groups)
await event.reply(f"已添加群权限：{value}")
return
else: # remove
if str(value) not in groups:
await event.reply(f"群 {value} 不在权限列表中。")
return
groups = [g for g in groups if g != str(value)]
data.set("groups", groups)
await event.reply(f"已移除群权限：{value}")
return
await event.reply("未知目标类型，请使用 member 或 group")
return
if action == "clear":
if len(args) < 2:
await event.reply("参数错误，示例：/admin clear member")
try:
user_id = int(args[1])
except ValueError:
await event.reply("无效的 QQ 号，请输入纯数字。")
return
target = args[1].lower()
if target == "member":
data.set("members", [])
await event.reply("已清空群成员权限。")
return
if target == "group":
data.set("groups", [])
await event.reply("已清空群权限。")
return
if target == "all":
data.clear()
await event.reply("已清空所有权限。")
return
await event.reply("未知清空目标，请使用 member/group/all")
return
await event.reply("未知指令，使用 /admin 查看帮助")
if action == "add":
success = await admin_manager.add_admin(user_id)
if success:
await event.reply(f"成功添加管理员: {user_id}")
else:
await event.reply(f"管理员 {user_id} 已存在，无需重复添加。")
return
elif action == "remove":
success = await admin_manager.remove_admin(user_id)
if success:
await event.reply(f"成功移除管理员: {user_id}")
else:
await event.reply(f"管理员 {user_id} 不存在。")
return
await event.reply(f"未知的指令: {action}\n\n{__plugin_meta__['usage']}")

13
plugins/code_py.py
View File

@@ -13,6 +13,7 @@ from typing import Tuple, Set
from core.bot import Bot
from core.command_manager import matcher
from core.executor import run_in_thread_pool
from models import MessageEvent
__plugin_meta__ = {
@@ -38,9 +39,11 @@ def is_code_safe(code: str) -> Tuple[bool, str]:
statements = STATEMENT_SPLIT_PATTERN.split(code)
for statement in statements:
statement = statement.strip()
if not statement: continue
if not statement:
continue
parts = statement.split()
if not parts: continue
if not parts:
continue
if parts[0] == 'from' and len(parts) > 1:
module_name = parts[1].strip()
if module_name in DANGEROUS_MODULES:
@@ -83,7 +86,7 @@ async def process_and_reply(bot: Bot, event: MessageEvent, code: str):
"""
核心处理逻辑：安全检查、执行代码并回复结果。
"""
safe, message = is_code_safe(code)
safe, message = await run_in_thread_pool(is_code_safe, code)
if not safe:
await event.reply(f"代码安全检查未通过：\n{message}")
return
@@ -150,11 +153,11 @@ async def handle_code_input(bot: Bot, event: MessageEvent):
# 处理取消操作
if event.raw_message.strip() == "取消":
await event.reply("已取消输入。")
return True # 消费事件
return True # 消费事件
# 执行代码
await process_and_reply(bot, event, event.raw_message)
return True # 消费事件，防止被其他指令匹配
return True # 消费事件，防止被其他指令匹配
# 如果用户不在等待状态，则不处理
return False

1
plugins/data/admin.json
View File

@@ -1 +0,0 @@
{}

18
plugins/echo.py
View File

@@ -29,18 +29,26 @@ async def handle_echo(bot: Bot, event: MessageEvent, args: list[str]):
await event.reply(reply_msg)
@matcher.command("赞我")
async def handle_poke(bot: Bot, event: MessageEvent, args: list[str]):
@matcher.command(
"赞我",
permission=MessageEvent.ADMIN,
override_permission_check=True
)
async def handle_poke(bot: Bot, event: MessageEvent, permission_granted: bool):
"""
处理 赞我 指令，发送点赞
:param bot: Bot 实例
:param event: 消息事件对象
:param args: 指令参数列表（本指令不使用参数）
:param permission_granted: 权限检查结果
"""
if not permission_granted:
await event.reply("只有我的操作员才能让我点赞哦！(｡•ˇ‸ˇ•｡)")
return
try:
# 尝试发送赞
await bot.send_like(event.user_id, times=10)
await event.reply("戳一戳发送成功！")
await event.reply("好感度+10！(〃'▽'〃)")
except Exception as e:
await event.reply(f"戳一戳发送失败：{str(e)}")
await event.reply(f"点赞失败了 >_<: {str(e)}")

7
plugins/jrcd.py
View File

@@ -9,6 +9,7 @@ from datetime import datetime
from core.bot import Bot
from core.command_manager import matcher
from core.executor import run_in_thread_pool
from models import MessageEvent, MessageSegment
__plugin_meta__ = {
@@ -77,7 +78,7 @@ async def handle_jrcd(bot: Bot, event: MessageEvent, args: list[str]):
:param args: 指令参数列表（未使用）。
"""
user_id = event.user_id
jrcd = get_jrcd(user_id)
jrcd = await run_in_thread_pool(get_jrcd, user_id)
msg = [MessageSegment.at(user_id)]
if jrcd <= 9:
msg.append(MessageSegment.text(random.choice(JRCDMSG_1) % jrcd))
@@ -112,8 +113,8 @@ async def handle_bbcd(bot: Bot, event: MessageEvent, args: list[str]):
await event.reply("不能和自己比！")
return
jrcd1 = get_jrcd(user_id1)
jrcd2 = get_jrcd(user_id2)
jrcd1 = await run_in_thread_pool(get_jrcd, user_id1)
jrcd2 = await run_in_thread_pool(get_jrcd, user_id2)
jrcz = jrcd1 - jrcd2

88
plugins/sync_async_test_plugin.py Normal file
View File

@@ -0,0 +1,88 @@
"""
同步/异步函数测试插件
用于演示 SyncHandlerError 异常以及如何将同步函数放入线程池执行。
"""
import time
from typing import Any
from core.command_manager import matcher
from core.executor import run_in_thread_pool
from core.bot import Bot
from core.logger import logger
# 插件元数据
__plugin_meta__ = {
"name": "SyncAsyncTestPlugin",
"description": "用于测试同步/异步函数处理的插件。",
"usage": (
"/test_sync_error - 尝试注册一个同步函数作为异步处理器，会触发错误。\n"
"/test_blocking_task <duration> - 演示将同步阻塞任务放入线程池执行。"
),
}
# --- 示例 1: 触发 SyncHandlerError (此函数不会被成功注册) ---
# 这是一个同步函数，如果直接用 @matcher.message_handler 装饰，
# 并且 event_handler 检查到它是同步的，就会抛出 SyncHandlerError。
# 注意：为了演示错误，我们不会真正注册它，因为注册会失败。
def _sync_function_that_should_fail(bot: Bot, event: Any):
"""
一个同步函数，如果直接作为异步事件处理器注册，会触发 SyncHandlerError。
"""
logger.info("这个同步函数不应该被直接调用。")
return "这是一个同步函数的结果。"
# --- 示例 2: 将同步阻塞任务放入线程池运行 ---
def _blocking_task(duration: int) -> str:
"""
一个模拟耗时操作的同步函数。
Args:
duration (int): 模拟阻塞的秒数。
Returns:
str: 任务完成消息。
"""
logger.info(f"同步阻塞任务开始，持续 {duration} 秒...")
time.sleep(duration)
logger.info("同步阻塞任务结束。")
return f"阻塞任务完成，耗时 {duration} 秒。"
@matcher.message_handler.command("test_blocking_task")
async def test_blocking_task_handler(bot: Bot, event: Any, args: list):
"""
处理 /test_blocking_task 命令，将同步阻塞任务放入线程池执行。
Args:
bot (Bot): 机器人实例。
event (Any): 接收到的事件对象。
args (list): 命令参数列表。
"""
if not args:
await bot.send(event, "请提供阻塞时长，例如：/test_blocking_task 5")
return
try:
duration = int(args[0])
if duration <= 0:
raise ValueError("时长必须是正整数。")
except ValueError:
await bot.send(event, "无效的时长，请提供一个正整数。")
return
await bot.send(event, f"开始执行同步阻塞任务，预计耗时 {duration} 秒...")
# 将同步函数放入线程池执行
result = await run_in_thread_pool(_blocking_task, duration)
await bot.send(event, f"同步阻塞任务已完成：{result}")
# --- 示例 3: 尝试注册一个同步函数作为异步处理器 (会失败) ---
# 这个函数不会被成功注册，因为 event_handler 会检测到它是同步的并抛出 SyncHandlerError。
# 插件管理器会捕获这个错误并跳过加载此插件。
# 为了演示，我们故意尝试注册它。
# @matcher.message_handler.command("test_sync_error")
# def test_sync_error_handler(bot: Bot, event: Any):
# """
# 这个同步函数尝试作为异步处理器注册，会触发 SyncHandlerError。
# """
# logger.error("这个同步函数不应该被直接注册为异步处理器。")
# return "这个消息不应该被看到。"