nonebot
Plugin: nonebot-plugin-rice-team
PyPI 项目名
nonebot-plugin-rice-team
插件模块名
nonebot_plugin_rice_team
标签
[{"label":"wow","color":"#ea5252"},{"label":"自用","color":"#ea5252"}]
插件配置项
#!/usr/bin/env python3
-- coding: utf-8 --
""" NoneBot2 大米组队插件 自动组队功能：当群里发送"有无米"时记录用户，凑满5人时艾特成员通知"组队成功 开始大米吧" 支持队列管理、超时重置和配置化 """ import asyncio from typing import Dict, List, Optional, Any, Set from datetime import datetime, timedelta from nonebot import get_driver, on_message, on_command from nonebot.adapters import Event from nonebot.adapters.onebot.v11 import ( GroupMessageEvent, Message, MessageSegment, Bot, GroupDecreaseNoticeEvent ) from nonebot.log import logger from nonebot.plugin import PluginMetadata from nonebot.params import CommandArg from nonebot.rule import to_me from nonebot.permission import SUPERUSER
导入插件配置
from .config import Config, plugin_config
尝试导入定时任务插件（可选依赖）
try: from nonebot_plugin_apscheduler import scheduler HAS_SCHEDULER = True logger.debug("定时任务插件可用") except ImportError: HAS_SCHEDULER = False logger.debug("未安装 nonebot-plugin-apscheduler，超时重置功能将不可用")
--- 插件元数据（商店审核必需）---
plugin_meta = PluginMetadata( name="大米组队", description="自动组队插件，凑满人数后通知成员开始游戏", usage=( "【基本命令】\n" "1. 发送关键词加入队列（默认：'有无米'）\n" "2. @机器人 '查看队列' - 查看当前队列状态\n" "3. @机器人 '退出队列' - 从队列中退出\n" "4. @机器人 '清空队列' - 清空当前队列（需要管理员权限）\n\n" "【配置命令】（需要管理员权限）\n" "1. @机器人 '设置人数 5' - 设置组队所需人数(2-20)\n" "2. @机器人 '设置超时 30' - 设置超时时间(5-240分钟)\n" "3. @机器人 '添加关键词 开车' - 添加触发关键词\n" "4. @机器人 '删除关键词 开车' - 删除触发关键词\n" "5. @机器人 '查看配置' - 查看当前配置\n\n" "【超级用户命令】\n" "@机器人 '队列统计' - 查看所有群的队列统计" ), type="application", homepage="https://github.com/你的用户名/nonebot-plugin-rice-team", config=Config, supported_adapters={"~onebot.v11"}, extra={ "author": "你的名字", "version": "1.0.0", "priority": 10, } )
--- 数据结构定义 ---
class TeamQueue: """组队队列管理类""" def init(self, group_id: int, team_size: Optional[int] = None, timeout_minutes: Optional[int] = None): self.group_id = group_id self.members: List[int] = [] self.member_names: Dict[int, str] = {} self.created_time = datetime.now() self.last_activity = datetime.now() self.team_size = team_size or plugin_config.team_size self.timeout_minutes = timeout_minutes or plugin_config.timeout_minutes self.reminders_sent: Set[int] = set() self.notification_sent = False
def add_member(self, user_id: int, user_name: Optional[str] = None) -> bool:
"""添加成员到队列"""
if user_id not in self.members and len(self.members) < self.team_size:
self.members.append(user_id)
if user_name:
self.member_names[user_id] = user_name
self.last_activity = datetime.now()
self.notification_sent = False
return True
return False
def remove_member(self, user_id: int) -> bool:
"""从队列移除成员"""
if user_id in self.members:
self.members.remove(user_id)
self.member_names.pop(user_id, None)
self.last_activity = datetime.now()
return True
return False
def clear(self) -> List[int]:
"""清空队列"""
removed = self.members.copy()
self.members.clear()
self.member_names.clear()
self.reminders_sent.clear()
return removed
def is_full(self) -> bool:
"""检查队列是否已满"""
return len(self.members) >= self.team_size
def is_empty(self) -> bool:
"""检查队列是否为空"""
return len(self.members) == 0
def get_progress(self) -> str:
"""获取队列进度"""
return f"{len(self.members)}/{self.team_size}"
def is_timed_out(self) -> bool:
"""检查是否超时"""
if self.is_empty():
return False
timeout_delta = timedelta(minutes=self.timeout_minutes)
return datetime.now() - self.last_activity > timeout_delta
def get_time_remaining(self) -> int:
"""获取剩余时间（分钟）"""
elapsed = datetime.now() - self.last_activity
remaining = timedelta(minutes=self.timeout_minutes) - elapsed
return max(0, int(remaining.total_seconds() / 60))
def get_elapsed_time(self) -> int:
"""获取已等待时间（分钟）"""
elapsed = datetime.now() - self.created_time
return int(elapsed.total_seconds() / 60)
def should_remind(self) -> Optional[int]:
"""检查是否需要发送提醒"""
if not plugin_config.enable_reminder or not plugin_config.reminder_intervals:
return None
elapsed_minutes = self.get_elapsed_time()
for interval in sorted(plugin_config.reminder_intervals, reverse=True):
if (interval <= self.timeout_minutes and
elapsed_minutes >= interval and
interval not in self.reminders_sent):
self.reminders_sent.add(interval)
return interval
return None
--- 全局存储 ---
team_queues: Dict[int, TeamQueue] = {} group_configs: Dict[int, Dict[str, Any]] = {}
--- 响应器定义 ---
keywords_matcher = on_message(priority=10, block=False) view_matcher = on_command("查看队列", aliases={"状态", "queue", "status"}, rule=to_me(), priority=5) quit_matcher = on_command("退出队列", aliases={"退出", "quit", "leave"}, rule=to_me(), priority=5) clear_matcher = on_command("清空队列", aliases={"清除队列", "clear"}, rule=to_me(), priority=5) config_matcher = on_command("设置人数", aliases={"人数设置"}, rule=to_me(), priority=5) timeout_matcher = on_command("设置超时", aliases={"超时设置"}, rule=to_me(), priority=5) add_keyword_matcher = on_command("添加关键词", aliases={"添加关键字"}, rule=to_me(), priority=5) remove_keyword_matcher = on_command("删除关键词", aliases={"删除关键字"}, rule=to_me(), priority=5) view_config_matcher = on_command("查看配置", aliases={"配置", "config"}, rule=to_me(), priority=5) reload_config_matcher = on_command("重载配置", aliases={"刷新配置"}, rule=to_me(), priority=5) stats_matcher = on_command("队列统计", rule=to_me(), permission=SUPERUSER, priority=5) help_matcher = on_command("大米帮助", rule=to_me(), priority=5)
--- 工具函数 ---
async def get_user_name(bot: Bot, user_id: int, group_id: Optional[int] = None) -> str: """获取用户显示名称""" try: if group_id: member_info = await bot.get_group_member_info( group_id=group_id, user_id=user_id ) return member_info.get("card", member_info.get("nickname", f"用户{user_id}")) except Exception: pass return f"用户{user_id}"
async def is_group_admin(bot: Bot, event: GroupMessageEvent) -> bool: """检查用户是否为群管理员""" try: member_info = await bot.get_group_member_info( group_id=event.group_id, user_id=event.user_id ) role = member_info.get("role", "member") return role in ["owner", "admin"] except Exception: return False
def get_group_config(group_id: int, key: str, default: Any = None) -> Any: """获取群组特定配置""" if group_id in group_configs and key in group_configs[group_id]: return group_configs[group_id][key] return default
def set_group_config(group_id: int, key: str, value: Any): """设置群组特定配置""" if group_id not in group_configs: group_configs[group_id] = {} group_configs[group_id][key] = value
def get_keywords_for_group(group_id: int) -> List[str]: """获取群组可用的关键词列表""" custom_keywords = get_group_config(group_id, "keywords") if custom_keywords: return custom_keywords return plugin_config.keywords
def format_time(minutes: int) -> str: """格式化时间显示""" if minutes >= 60: hours = minutes // 60 mins = minutes % 60 return f"{hours}小时{mins}分钟" if mins > 0 else f"{hours}小时" return f"{minutes}分钟"
--- 超时检查任务（可选功能）---
if HAS_SCHEDULER: @scheduler.scheduled_job("interval", seconds=60, id="check_team_timeout") async def check_timeout(): """定时检查超时队列""" try: bots = list(get_driver().bots.values()) if not bots: return bot = bots[0]
queues_to_remove = []
for group_id, queue in list(team_queues.items()):
# 检查超时
if queue.is_timed_out() and not queue.is_empty():
try:
at_members = Message()
for uid in queue.members:
at_members += MessageSegment.at(uid)
timeout_msg = (
at_members +
f"\n⏰ 组队已超时（{format_time(queue.timeout_minutes)}未满人），队列已清空。"
)
await bot.send_group_msg(group_id=group_id, message=timeout_msg)
logger.info(f"群 {group_id} 队列超时，已清空")
except Exception as e:
logger.error(f"发送超时通知失败: {e}")
finally:
queues_to_remove.append(group_id)
# 发送提醒
elif not queue.is_empty() and not queue.is_timed_out():
reminder_interval = queue.should_remind()
if reminder_interval is not None:
try:
at_members = Message()
for uid in queue.members:
at_members += MessageSegment.at(uid)
remaining = queue.get_time_remaining()
reminder_msg = (
at_members +
f"\n⏰ 组队提醒：已等待{format_time(reminder_interval)}，"
f"当前进度{queue.get_progress()}，"
f"剩余时间约{format_time(remaining)}。"
)
await bot.send_group_msg(group_id=group_id, message=reminder_msg)
except Exception as e:
logger.error(f"发送提醒失败: {e}")
# 移除超时队列
for group_id in queues_to_remove:
team_queues.pop(group_id, None)
except Exception as e:
logger.error(f"定时任务执行失败: {e}")
--- 核心响应器处理函数 ---
@keywords_matcher.handle() async def handle_keywords(event: GroupMessageEvent, bot: Bot): """处理组队关键词""" if not isinstance(event, GroupMessageEvent): return
msg = event.get_plaintext().strip()
group_id = event.group_id
user_id = event.user_id
# 检查是否为关键词
keywords = get_keywords_for_group(group_id)
if msg not in keywords:
return
# 获取用户名称
user_name = await get_user_name(bot, user_id, group_id)
# 获取或创建队列
if group_id not in team_queues:
team_size = get_group_config(group_id, "team_size", plugin_config.team_size)
timeout = get_group_config(group_id, "timeout_minutes", plugin_config.timeout_minutes)
team_queues[group_id] = TeamQueue(group_id, team_size, timeout)
queue = team_queues[group_id]
# 检查用户是否已在队列中
if user_id in queue.members:
await keywords_matcher.finish(
MessageSegment.at(user_id) +
f" 你已经在队列中啦！当前进度: {queue.get_progress()}"
)
return
# 添加用户到队列
if queue.add_member(user_id, user_name):
logger.info(f"群 {group_id} 用户 {user_name}({user_id}) 加入队列")
# 检查是否满员
if queue.is_full():
# 组队成功
at_members = Message()
member_names = []
for uid in queue.members:
at_members += MessageSegment.at(uid)
member_names.append(queue.member_names.get(uid, f"用户{uid}"))
success_msg = (
at_members +
f"\n🎉 组队成功！({queue.get_progress()})\n"
f"🚀 开始大米吧！"
)
await keywords_matcher.send(success_msg)
# 清空队列
team_queues.pop(group_id, None)
else:
# 发送加入成功提示
remaining = queue.get_time_remaining()
reply_msg = (
MessageSegment.at(user_id) +
f" 已加入组队队列！\n"
f"当前进度: {queue.get_progress()}，"
f"还需要{queue.team_size - len(queue.members)}人。\n"
f"超时时间: {format_time(queue.timeout_minutes)}"
)
await keywords_matcher.send(reply_msg)
@view_matcher.handle() async def handle_view_queue(event: GroupMessageEvent, bot: Bot): """处理查看队列命令""" group_id = event.group_id
if group_id not in team_queues or team_queues[group_id].is_empty():
await view_matcher.finish("当前没有进行中的组队队列。")
return
queue = team_queues[group_id]
# 构建成员列表
members_text = "\n".join([f" {i+1}. {name}" for i, name in enumerate(queue.get_member_names())])
time_elapsed = queue.get_elapsed_time()
reply_msg = (
f"【组队队列状态】\n"
f"进度: {queue.get_progress()}\n"
f"已等待: {format_time(time_elapsed)}\n"
f"成员列表:\n{members_text}"
)
await view_matcher.finish(reply_msg)
@quit_matcher.handle() async def handle_quit_queue(event: GroupMessageEvent): """处理退出队列命令""" group_id = event.group_id user_id = event.user_id
if group_id not in team_queues:
await quit_matcher.finish("你不在任何组队队列中。")
return
queue = team_queues[group_id]
if queue.remove_member(user_id):
# 检查队列是否为空
if queue.is_empty():
team_queues.pop(group_id, None)
await quit_matcher.finish(
MessageSegment.at(user_id) + " 已退出队列。"
)
else:
await quit_matcher.finish(
MessageSegment.at(user_id) + " 你不在当前组队队列中。"
)
@clear_matcher.handle() async def handle_clear_queue(event: GroupMessageEvent, bot: Bot): """处理清空队列命令""" if plugin_config.admin_only_clear: if not await is_group_admin(bot, event): await clear_matcher.finish("只有群管理员可以清空队列。") return
group_id = event.group_id
if group_id not in team_queues:
await clear_matcher.finish("当前没有可清空的队列。")
return
# 清空队列
team_queues.pop(group_id, None)
await clear_matcher.finish("✅ 队列已清空。")
@help_matcher.handle() async def handle_help(): """显示帮助信息""" help_text = """🍚 大米组队插件帮助 🍚
【基础功能】
- 发送'有无米'加入组队队列
- 满5人自动通知组队成功
【常用命令】 • @机器人 查看队列 - 查看当前队列状态 • @机器人 退出队列 - 退出当前队列 • @机器人 清空队列 - 清空队列（管理员）
【配置命令】（管理员） • @机器人 设置人数 [数字] - 设置组队人数 • @机器人 设置超时 [数字] - 设置超时时间(分钟) • @机器人 添加关键词 [词] - 添加触发关键词 • @机器人 查看配置 - 查看当前配置
默认组队人数: 5人 默认超时时间: 30分钟"""
await help_matcher.finish(help_text)
--- 配置管理命令 ---
@config_matcher.handle() async def handle_set_team_size(event: GroupMessageEvent, bot: Bot, args: Message = CommandArg()): """设置组队人数""" if plugin_config.admin_only_config and not await is_group_admin(bot, event): await config_matcher.finish("只有群管理员可以设置组队人数。") return
size_str = args.extract_plain_text().strip()
if not size_str:
await config_matcher.finish("请输入要设置的人数，例如: 设置人数 5")
return
try:
size = int(size_str)
if size < 2 or size > 20:
await config_matcher.finish("组队人数必须在2-20之间。")
return
set_group_config(event.group_id, "team_size", size)
# 更新现有队列
if event.group_id in team_queues:
team_queues[event.group_id].team_size = size
await config_matcher.finish(f"✅ 已设置组队人数为: {size}人")
except ValueError:
await config_matcher.finish("请输入有效的数字。")
@view_config_matcher.handle() async def handle_view_config(event: GroupMessageEvent): """查看配置""" group_id = event.group_id
# 获取配置
team_size = get_group_config(group_id, "team_size", plugin_config.team_size)
timeout = get_group_config(group_id, "timeout_minutes", plugin_config.timeout_minutes)
keywords = get_keywords_for_group(group_id)
reply_msg = (
f"【当前配置】\n"
f"组队人数: {team_size}人\n"
f"超时时间: {format_time(timeout)}\n"
f"触发关键词: {', '.join(keywords)}"
)
await view_config_matcher.finish(reply_msg)
--- 启动清理 ---
@get_driver().on_startup async def cleanup_on_start(): """启动时清理所有队列""" if plugin_config.enable_auto_cleanup: team_queues.clear() logger.info("✅ 启动时已清理所有组队队列")
--- 群成员离开处理 ---
@get_driver().on_notice async def handle_group_member_change(event: Event): """处理群成员离开，自动从队列中移除""" try: if isinstance(event, GroupDecreaseNoticeEvent): group_id = event.group_id user_id = event.user_id
if group_id in team_queues and user_id in team_queues[group_id].members:
team_queues[group_id].remove_member(user_id)
logger.info(f"群 {group_id} 用户 {user_id} 离开，已从队列中移除")
except Exception as e:
logger.error(f"处理群成员变动失败: {e}")
插件测试
- [ ] 如需重新运行插件测试，请勾选左侧勾选框
📃 商店发布检查结果
Plugin: nonebot-plugin-rice-team
⚠️ 在发布检查过程中，我们发现以下问题：
⚠️ 项目 nonebot-plugin-rice-team 未发布至 PyPI。请将你的项目发布至 PyPI。 ⚠️ 发布时间: 值不是合法的字符串 ⚠️ 版本号: 值不是合法的字符串 ⚠️ 插件加载测试未通过。测试输出
项目 nonebot-plugin-rice-team 创建失败：
Virtualenv
Python: 3.12.11
Implementation: CPython
Path: /app/plugin_test/.venv
Executable: /app/plugin_test/.venv/bin/python
Valid: True
Base
Platform: linux
OS: posix
Python: 3.12.11
Path: /root/.local/share/uv/python/cpython-3.12.11-linux-x86_64-gnu
Executable: /root/.local/share/uv/python/cpython-3.12.11-linux-x86_64-gnu/bin/python3.12
warning: The `tool.uv.dev-dependencies` field (used in `/app/pyproject.toml`) is deprecated and will be removed in a future release; use `dependency-groups.dev` instead
Downloading cpython-3.12.11-linux-x86_64-gnu (download) (31.1MiB)
Downloading cpython-3.12.11-linux-x86_64-gnu (download)
Using CPython 3.12.11
warning: The requested interpreter resolved to Python 3.12.11, which is incompatible with the project's Python requirement: `>=3.13.5` (from `project.requires-python`)
Creating virtual environment at: .venv
Could not find a matching version of package nonebot-plugin-rice-team ⚠️ 无法获取到插件元数据。请确保插件正常加载。 详情
✅ 标签: wow-#ea5252, 自用-#ea5252。 历史测试
⚠️ 2025-12-11 12:44:30 CST ⚠️ 2025-12-11 12:41:15 CST ⚠️ 2025-12-07 22:49:40 CST ⚠️ 2025-12-07 22:48:18 CST ⚠️ 2025-12-07 22:44:24 CST 💡 如需修改信息，请直接修改 issue，机器人会自动更新检查结果。 💡 当插件加载测试失败时，请发布新版本后勾选插件测试勾选框重新运行插件测试。
♻️ 评论已更新至最新检查结果
💪 Powered by NoneFlow
![noneflow[bot] avatar](https://avatars.githubusercontent.com/in/313205?v=4 "Published by a pub.dev verified publisher"){kind=small}
