Diuse/zhenxun_bot
1
0
Fork 0
mirror of https://github.com/zhenxun-org/zhenxun_bot.git synced 2026-10-04 03:10:01 +08:00
Code Issues Actions Packages Projects Releases Wiki Activity

✨ feat(core): 增强定时任务与群组标签管理，重构调度核心

Browse Source 
✨ 新功能

* **标签 (tags)**: 引入群组标签服务。
    * 支持静态标签和动态标签 (基于 Alconna 规则自动匹配群信息)。
    * 支持黑名单模式及 `@all` 特殊标签。
    * 提供 `tag_manage` 超级用户插件 (list, create, edit, delete 等)。
    * 群成员变动时自动失效动态标签缓存。
* **调度 (scheduler)**: 增强定时任务。
    * 重构 `ScheduledJob` 模型，支持 `TAG`, `ALL_GROUPS` 等多种目标类型。
    * 新增任务别名 (`name`)、创建者、权限、来源等字段。
    * 支持一次性任务 (`schedule_once`) 和 Alconna 命令行参数 (`--params-cli`)。
    * 新增执行选项 (`jitter`, `spread`) 和并发策略 (`ALLOW`, `SKIP`, `QUEUE`)。
    * 支持批量获取任务状态。

♻️ 重构优化

* **调度器核心**:
    * 拆分 `service.py` 为 `manager.py` (API) 和 `types.py` (模型)。
    * 合并 `adapter.py` / `job.py` 至 `engine.py` (统一调度引擎)。
    * 引入 `targeting.py` 模块管理任务目标解析。
* **调度器插件 (scheduler_admin)**:
    * 迁移命令参数校验逻辑至 `ArparmaBehavior`。
    * 引入 `dependencies.py` 和 `data_source.py` 解耦业务逻辑与依赖注入。
    * 适配新的任务目标类型展示。
This commit is contained in:
webjoin111
2025-10-20 14:58:01 +08:00
parent ae07e5f329
commit da4123363c
25 changed files with 3103 additions and 1252 deletions
Download Patch File Download Diff File Expand all files Collapse all files
zhenxun/builtin_plugins/admin/group_member_update/__init__.py
+6
View File
@@ -14,6 +14,7 @@ from nonebot_plugin_session import EventSession
from zhenxun.configs.config import BotConfig
from zhenxun.configs.utils import PluginExtraData
from zhenxun.services.log import logger
from zhenxun.services.tags import tag_manager
from zhenxun.utils.enum import PluginType
from zhenxun.utils.message import MessageUtils
from zhenxun.utils.platform import PlatformUtils
@@ -96,6 +97,7 @@ async def _update_all_groups_task(bot: Bot, session: EventSession):
)
return
await tag_manager._invalidate_cache()
summary_message = (
f"🤖 Bot {bot_id} 所有群组信息更新任务完成！\n"
f"总计群组: {total_count}\n"
@@ -120,6 +122,7 @@ async def _(bot: Bot, session: EventSession, arparma: Arparma):
logger.info("更新群组成员信息", arparma.header_result, session=session)
result = await MemberUpdateManage.update_group_member(bot, gid)
await MessageUtils.build_message(result).finish(reply_to=True)
await tag_manager._invalidate_cache()
await MessageUtils.build_message("群组id为空...").send()
@@ -133,6 +136,7 @@ async def _(bot: Bot, event: GroupIncreaseNoticeEvent):
session=event.user_id,
group_id=event.group_id,
)
await tag_manager._invalidate_cache()
@scheduler.scheduled_job(
@@ -160,3 +164,5 @@ async def _():
except Exception as e:
logger.error(f"Bot: {bot.self_id} 自动更新群组信息", e=e)
logger.debug(f"自动 Bot: {bot.self_id} 更新群组成员信息成功...")
await tag_manager._invalidate_cache()
zhenxun/builtin_plugins/scheduler_admin/__init__.py
+63 -32
View File
@@ -10,47 +10,54 @@ __all__ = ["commands", "handlers"]
__plugin_meta__ = PluginMetadata(
name="定时任务管理",
description="查看和管理由 SchedulerManager 控制的定时任务。",
usage="""
📋 定时任务管理 - 支持群聊和私聊操作
usage="""### 📋 定时任务管理
---
#### 🔍 **查看任务**
- **命令**: `定时任务 查看 [选项]` (别名: `ls`, `list`)
- **选项**:
- `--all`: 查看所有群组的任务 **(SUPERUSER)**。
- `-g <群号>`: 查看指定群组的任务 **(SUPERUSER)**。
- `-p <插件名>`: 按插件名筛选。
- `--page <页码>`: 指定页码。
- **说明**:
- 在群聊中不带选项使用，默认查看本群任务。
- 在私聊中必须使用 `-g <群号>` 或 `--all`。
🔍 查看任务:
定时任务 查看 [-all] [-g <群号>] [-p <插件>] [--page <页码>]
• 群聊中: 查看本群任务
• 私聊中: 必须使用 -g <群号> 或 -all 选项 (SUPERUSER)
#### 📊 **任务状态**
- **命令**: `定时任务 状态 <任务ID>` (别名: `status`, `info`, `任务状态`)
- **说明**: 查看单个任务的详细信息和状态。
📊 任务状态:
定时任务 状态 <任务ID> 或 任务状态 <任务ID>
• 查看单个任务的详细信息和状态
#### ⚙️ **任务管理 (SUPERUSER)**
- **设置**: `定时任务 设置 <插件>` (别名: `add`, `开启`)
- **选项**:
- `<时间选项>`: 详见下文。
- `-g <群号|all>`: 指定目标群组。
- `--kwargs "<参数>"`: 设置任务参数 (例: `"key=value"`)。
- **删除**: `定时任务 删除 <ID>` (别名: `del`, `rm`, `remove`, `关闭`, `取消`)
- **暂停**: `定时任务 暂停 <ID>` (别名: `pause`)
- **恢复**: `定时任务 恢复 <ID>` (别名: `resume`)
- **执行**: `定时任务 执行 <ID>` (别名: `trigger`, `run`)
- **更新**: `定时任务 更新 <ID>` (别名: `update`, `modify`, `修改`)
- **选项**:
- `<时间选项>`: 详见下文。
- `--kwargs "<参数>"`: 更新任务参数。
- **批量操作**: `删除/暂停/恢复` 命令支持通过 `-p <插件名>` 或 `--all`
(当前群) 进行批量操作。
⚙️ 任务管理 (SUPERUSER):
定时任务 设置 <插件> [时间选项] [-g <群号> | -g all] [--kwargs <参数>]
定时任务 删除 <任务ID> | -p <插件> [-g <群号>] | -all
定时任务 暂停 <任务ID> | -p <插件> [-g <群号>] | -all
定时任务 恢复 <任务ID> | -p <插件> [-g <群号>] | -all
定时任务 执行 <任务ID>
定时任务 更新 <任务ID> [时间选项] [--kwargs <参数>]
# [修改] 增加说明
• 说明: -p 选项可单独使用，用于操作指定插件的所有任务
#### 📝 **时间选项 (设置/更新时三选一)**
- `--cron "<分> <时> <日> <月> <周>"` (例: `--cron "0 8 * * *"`)
- `--interval <时间间隔>` (例: `--interval 30m`, `2h`, `10s`)
- `--date "<YYYY-MM-DD HH:MM:SS>"` (例: `--date "2024-01-01 08:00:00"`)
- `--daily "<HH:MM>"` (例: `--daily "08:30"`)
📝 时间选项 (三选一):
--cron "<分> <时> <日> <月> <周>" # 例: --cron "0 8 * * *"
--interval <时间间隔> # 例: --interval 30m, 2h, 10s
--date "<YYYY-MM-DD HH:MM:SS>" # 例: --date "2024-01-01 08:00:00"
--daily "<HH:MM>" # 例: --daily "08:30"
📚 其他功能:
定时任务 插件列表 # 查看所有可设置定时任务的插件 (SUPERUSER)
🏷️ 别名支持:
查看: ls, list | 设置: add, 开启 | 删除: del, rm, remove, 关闭, 取消
暂停: pause | 恢复: resume | 执行: trigger, run | 状态: status, info
更新: update, modify, 修改 | 插件列表: plugins
#### 📚 **其他功能**
- **命令**: `定时任务 插件列表` (别名: `plugins`)
- **说明**: 查看所有可设置定时任务的插件 **(SUPERUSER)**。
""".strip(),
extra=PluginExtraData(
author="HibiKier",
version="0.1.2",
plugin_type=PluginType.SUPERUSER,
is_show=False,
configs=[
RegisterConfig(
module="SchedulerManager",
@@ -80,6 +87,30 @@ __plugin_meta__ = PluginMetadata(
help="定时任务使用的时区，默认为 Asia/Shanghai",
type=str,
),
RegisterConfig(
module="SchedulerManager",
key="SCHEDULE_ADMIN_LEVEL",
value=5,
help="设置'定时任务'系列命令的基础使用权限等级",
default_value=5,
type=int,
),
RegisterConfig(
module="SchedulerManager",
key="DEFAULT_JITTER_SECONDS",
value=60,
help="为多目标定时任务（如 --all, -t）设置的默认触发抖动秒数，避免所有任务同时启动。", # noqa: E501
default_value=60,
type=int,
),
RegisterConfig(
module="SchedulerManager",
key="DEFAULT_SPREAD_SECONDS",
value=300,
help="为多目标定时任务设置的默认执行分散秒数，将任务执行分散在一个时间窗口内。",
default_value=300,
type=int,
),
],
).to_dict(),
)
zhenxun/builtin_plugins/scheduler_admin/commands.py
+158 -220
View File
@@ -1,33 +1,101 @@
import re
from nonebot.adapters import Event
from nonebot.adapters.onebot.v11 import Bot
from nonebot.params import Depends
from nonebot.permission import SUPERUSER
from arclet.alconna import ArparmaBehavior
from nonebot_plugin_alconna import (
Alconna,
AlconnaMatch,
Args,
Match,
Arparma,
Field,
MultiVar,
Option,
Query,
Subcommand,
on_alconna,
store_true,
)
from zhenxun.configs.config import Config
from zhenxun.services.scheduler import scheduler_manager
from zhenxun.services.scheduler.targeter import ScheduleTargeter
from zhenxun.utils.rules import admin_check
def create_time_options() -> list[Option]:
"""创建一组用于定义任务执行时间的通用选项"""
return [
Option("--cron", Args["cron_expr", str], help_text="设置 cron 表达式"),
Option("--interval", Args["interval_expr", str], help_text="设置时间间隔"),
Option("--date", Args["date_expr", str], help_text="设置特定执行日期"),
Option(
"--daily",
Args["daily_expr", str],
help_text="设置每天执行的时间 (如 08:20)",
),
]
def create_targeting_options() -> list[Option]:
"""创建一组用于定位定时任务的通用选项"""
return [
Option("-p", Args["plugin_name", str], help_text="按插件名筛选"),
Option("-u", Args["user_id", str], help_text="指定用户ID"),
Option(
"-g",
Args["group_ids", MultiVar(str)],
help_text="指定一个或多个群组ID (SUPERUSER)",
),
Option("-t", Args["tag_name", str], help_text="指定标签"),
Option("--all", action=store_true, help_text="对所有群生效"),
Option("--global", action=store_true, help_text="操作全局任务"),
Option("--bot", Args["bot_id", str], help_text="指定操作的Bot ID (SUPERUSER)"),
]
class SchedulerAdminBehavior(ArparmaBehavior):
"""对定时任务命令的参数进行复杂的复合验证。"""
def _validate_time_options(self, interface: Arparma, subcommand: str):
"""验证时间选项 (--cron, --interval, --date, --daily) 的互斥性。"""
time_options = ["cron", "interval", "date", "daily"]
provided_options = [
f"--{opt}" for opt in time_options if interface.query(f"{subcommand}.{opt}")
]
if len(provided_options) > 1:
interface.behave_fail(
f"时间选项 {', '.join(provided_options)} 不能同时使用，请只选择一个。"
)
def _validate_target_options(self, interface: Arparma, subcommand: str):
"""验证目标选项 (-u, -g, -t, --all, --global) 的互斥性。"""
target_flags = {
"-u": "u",
"-g": "g",
"-t": "t",
"--all": "all",
"--global": "global",
}
provided_flags = [
flag
for flag, name in target_flags.items()
if interface.query(f"{subcommand}.{name}")
]
if len(provided_flags) > 1:
interface.behave_fail(
f"目标选项 {', '.join(provided_flags)} 是互斥的，请只选择一个。"
)
def operate(self, interface: Arparma):
subcommand = next(iter(interface.subcommands.keys()), None)
if not subcommand:
return
if subcommand in {"设置", "更新"}:
self._validate_time_options(interface, subcommand)
if subcommand in {"查看", "设置", "删除", "暂停", "恢复"}:
self._validate_target_options(interface, subcommand)
schedule_cmd = on_alconna(
Alconna(
"定时任务",
Subcommand(
"查看",
Option("-g", Args["target_group_id", str]),
Option("-all", help_text="查看所有群聊 (SUPERUSER)"),
Option("-p", Args["plugin_name", str], help_text="按插件名筛选"),
*create_targeting_options(),
Option("--page", Args["page", int, 1], help_text="指定页码"),
alias=["ls", "list"],
help_text="查看定时任务",
@@ -35,17 +103,36 @@ schedule_cmd = on_alconna(
Subcommand(
"设置",
Args["plugin_name", str],
Option("--cron", Args["cron_expr", str], help_text="设置 cron 表达式"),
Option("--interval", Args["interval_expr", str], help_text="设置时间间隔"),
Option("--date", Args["date_expr", str], help_text="设置特定执行日期"),
*create_time_options(),
Option(
"--daily",
Args["daily_expr", str],
help_text="设置每天执行的时间 (如 08:20)",
"-g", Args["group_ids", MultiVar(str)], help_text="指定一个或多个群组ID"
),
Option("-g", Args["group_id", str], help_text="指定群组ID或'all'"),
Option("-all", help_text="对所有群生效 (等同于 -g all)"),
Option("-u", Args["user_id", str], help_text="指定用户ID"),
Option("-t", Args["tag_name", str], help_text="指定一个群组标签"),
Option("--all", action=store_true, help_text="对所有群生效"),
Option("--global", action=store_true, help_text="设置为全局任务"),
Option("--name", Args["job_name", str], help_text="为任务设置一个别名"),
Option("--kwargs", Args["kwargs_str", str], help_text="设置任务参数"),
Option(
"--params-cli",
Args["cli_string", str],
help_text="传递给插件任务的原始命令行参数字符串",
),
Option(
"--jitter",
Args["jitter_seconds", int],
help_text="设置触发时间抖动(秒)",
),
Option(
"--spread",
Args["spread_seconds", int],
help_text="设置多目标执行的分散延迟(秒)",
),
Option(
"--permission",
Args["perm_level", int],
help_text="设置任务的管理权限等级",
),
Option(
"--bot", Args["bot_id", str], help_text="指定操作的Bot ID (SUPERUSER)"
),
@@ -54,64 +141,75 @@ schedule_cmd = on_alconna(
),
Subcommand(
"删除",
Args["schedule_id?", int],
Option("-p", Args["plugin_name", str], help_text="指定插件名"),
Option("-g", Args["group_id", str], help_text="指定群组ID"),
Option("-all", help_text="对所有群生效"),
Option(
"--bot", Args["bot_id", str], help_text="指定操作的Bot ID (SUPERUSER)"
),
Args[
"schedule_id?",
int,
Field(unmatch_tips=lambda text: f"任务ID '{text}' 必须是数字！"),
],
*create_targeting_options(),
alias=["del", "rm", "remove", "关闭", "取消"],
help_text="删除一个或多个定时任务",
),
Subcommand(
"暂停",
Args["schedule_id?", int],
Option("-all", help_text="对当前群所有任务生效"),
Option("-p", Args["plugin_name", str], help_text="指定插件名"),
Option("-g", Args["group_id", str], help_text="指定群组ID (SUPERUSER)"),
Option(
"--bot", Args["bot_id", str], help_text="指定操作的Bot ID (SUPERUSER)"
),
Args[
"schedule_id?",
int,
Field(unmatch_tips=lambda text: f"任务ID '{text}' 必须是数字！"),
],
*create_targeting_options(),
alias=["pause"],
help_text="暂停一个或多个定时任务",
),
Subcommand(
"恢复",
Args["schedule_id?", int],
Option("-all", help_text="对当前群所有任务生效"),
Option("-p", Args["plugin_name", str], help_text="指定插件名"),
Option("-g", Args["group_id", str], help_text="指定群组ID (SUPERUSER)"),
Option(
"--bot", Args["bot_id", str], help_text="指定操作的Bot ID (SUPERUSER)"
),
Args[
"schedule_id?",
int,
Field(unmatch_tips=lambda text: f"任务ID '{text}' 必须是数字！"),
],
*create_targeting_options(),
alias=["resume"],
help_text="恢复一个或多个定时任务",
),
Subcommand(
"执行",
Args["schedule_id", int],
Args[
"schedule_id",
int,
Field(
missing_tips=lambda: "请提供要立即执行的任务ID！",
unmatch_tips=lambda text: f"任务ID '{text}' 必须是数字！",
),
],
alias=["trigger", "run"],
help_text="立即执行一次任务",
),
Subcommand(
"更新",
Args["schedule_id", int],
Option("--cron", Args["cron_expr", str], help_text="设置 cron 表达式"),
Option("--interval", Args["interval_expr", str], help_text="设置时间间隔"),
Option("--date", Args["date_expr", str], help_text="设置特定执行日期"),
Option(
"--daily",
Args["daily_expr", str],
help_text="更新每天执行的时间 (如 08:20)",
),
Args[
"schedule_id",
int,
Field(
missing_tips=lambda: "请提供要更新的任务ID！",
unmatch_tips=lambda text: f"任务ID '{text}' 必须是数字！",
),
],
*create_time_options(),
Option("--kwargs", Args["kwargs_str", str], help_text="更新参数"),
alias=["update", "modify", "修改"],
help_text="更新任务配置",
),
Subcommand(
"状态",
Args["schedule_id", int],
Args[
"schedule_id",
int,
Field(
missing_tips=lambda: "请提供要查看状态的任务ID！",
unmatch_tips=lambda text: f"任务ID '{text}' 必须是数字！",
),
],
alias=["status", "info"],
help_text="查看单个任务的详细状态",
),
@@ -120,179 +218,19 @@ schedule_cmd = on_alconna(
alias=["plugins"],
help_text="列出所有可用的插件",
),
behaviors=[SchedulerAdminBehavior()],
),
priority=5,
block=True,
rule=admin_check(1),
skip_for_unmatch=False,
aliases={"schedule", "cron", "job"},
rule=admin_check("SchedulerManager", "SCHEDULE_ADMIN_LEVEL"),
)
schedule_cmd.shortcut(
"任务状态",
command="定时任务",
arguments=["状态", "{%0}"],
prefix=True,
)
class ScheduleTarget:
pass
class TargetByID(ScheduleTarget):
def __init__(self, id: int):
self.id = id
class TargetByPlugin(ScheduleTarget):
def __init__(
self, plugin: str, group_id: str | None = None, all_groups: bool = False
):
self.plugin = plugin
self.group_id = group_id
self.all_groups = all_groups
class TargetAll(ScheduleTarget):
def __init__(self, for_group: str | None = None):
self.for_group = for_group
TargetScope = TargetByID | TargetByPlugin | TargetAll | None
def create_target_parser(subcommand_name: str):
async def dependency(
event: Event,
schedule_id: Match[int] = AlconnaMatch("schedule_id"),
plugin_name: Match[str] = AlconnaMatch("plugin_name"),
group_id: Match[str] = AlconnaMatch("group_id"),
all_enabled: Query[bool] = Query(f"{subcommand_name}.all"),
) -> TargetScope:
if schedule_id.available:
return TargetByID(schedule_id.result)
if plugin_name.available:
p_name = plugin_name.result
if all_enabled.available:
return TargetByPlugin(plugin=p_name, all_groups=True)
elif group_id.available:
gid = group_id.result
if gid.lower() == "all":
return TargetByPlugin(plugin=p_name, all_groups=True)
return TargetByPlugin(plugin=p_name, group_id=gid)
else:
current_group_id = getattr(event, "group_id", None)
return TargetByPlugin(
plugin=p_name,
group_id=str(current_group_id) if current_group_id else None,
)
if all_enabled.available:
current_group_id = getattr(event, "group_id", None)
if not current_group_id:
await schedule_cmd.finish(
"私聊中单独使用 -all 选项时，必须使用 -g <群号> 指定目标。"
)
return TargetAll(for_group=str(current_group_id))
return None
return dependency
def parse_interval(interval_str: str) -> dict:
match = re.match(r"(\d+)([smhd])", interval_str.lower())
if not match:
raise ValueError("时间间隔格式错误, 请使用如 '30m', '2h', '1d', '10s' 的格式。")
value, unit = int(match.group(1)), match.group(2)
if unit == "s":
return {"seconds": value}
if unit == "m":
return {"minutes": value}
if unit == "h":
return {"hours": value}
if unit == "d":
return {"days": value}
return {}
def parse_daily_time(time_str: str) -> dict:
if match := re.match(r"^(\d{1,2}):(\d{1,2})(?::(\d{1,2}))?$", time_str):
hour, minute, second = match.groups()
hour, minute = int(hour), int(minute)
if not (0 <= hour <= 23 and 0 <= minute <= 59):
raise ValueError("小时或分钟数值超出范围。")
cron_config = {
"minute": str(minute),
"hour": str(hour),
"day": "*",
"month": "*",
"day_of_week": "*",
"timezone": Config.get_config("SchedulerManager", "SCHEDULER_TIMEZONE"),
}
if second is not None:
if not (0 <= int(second) <= 59):
raise ValueError("秒数值超出范围。")
cron_config["second"] = str(second)
return cron_config
else:
raise ValueError("时间格式错误，请使用 'HH:MM' 或 'HH:MM:SS' 格式。")
async def GetBotId(bot: Bot, bot_id_match: Match[str] = AlconnaMatch("bot_id")) -> str:
if bot_id_match.available:
return bot_id_match.result
return bot.self_id
def GetTargeter(subcommand: str):
"""
依赖注入函数，用于解析命令参数并返回一个配置好的 ScheduleTargeter 实例。
"""
async def dependency(
event: Event,
bot: Bot,
schedule_id: Match[int] = AlconnaMatch("schedule_id"),
plugin_name: Match[str] = AlconnaMatch("plugin_name"),
group_id: Match[str] = AlconnaMatch("group_id"),
all_enabled: Query[bool] = Query(f"{subcommand}.all"),
bot_id_to_operate: str = Depends(GetBotId),
) -> ScheduleTargeter:
if schedule_id.available:
return scheduler_manager.target(id=schedule_id.result)
if plugin_name.available:
if all_enabled.available:
return scheduler_manager.target(plugin_name=plugin_name.result)
current_group_id = getattr(event, "group_id", None)
gid = group_id.result if group_id.available else current_group_id
return scheduler_manager.target(
plugin_name=plugin_name.result,
group_id=str(gid) if gid else None,
bot_id=bot_id_to_operate,
)
if all_enabled.available:
current_group_id = getattr(event, "group_id", None)
gid = group_id.result if group_id.available else current_group_id
is_su = await SUPERUSER(bot, event)
if not gid and not is_su:
await schedule_cmd.finish(
f"在私聊中对所有任务进行'{subcommand}'操作需要超级用户权限。"
)
if (gid and str(gid).lower() == "all") or (not gid and is_su):
return scheduler_manager.target()
return scheduler_manager.target(
group_id=str(gid) if gid else None, bot_id=bot_id_to_operate
)
await schedule_cmd.finish(
f"'{subcommand}'操作失败：请提供任务ID，"
f"或通过 -p <插件名> 或 -all 指定要操作的任务。"
)
return Depends(dependency)
zhenxun/builtin_plugins/scheduler_admin/data_source.py
+311
View File
@@ -0,0 +1,311 @@
from typing import Any
from pydantic import BaseModel, ValidationError
from zhenxun.models.level_user import LevelUser
from zhenxun.models.scheduled_job import ScheduledJob
from zhenxun.services import scheduler_manager
from zhenxun.services.log import logger
from zhenxun.services.scheduler.repository import ScheduleRepository
from zhenxun.utils.pydantic_compat import model_dump, model_validate
from . import presenters
class SchedulerAdminService:
"""封装定时任务管理的所有业务逻辑"""
async def get_schedules_view(
self,
user_id: str,
group_id: str | None,
is_superuser: bool,
filters: dict[str, Any],
page: int,
) -> bytes | str:
"""获取任务列表视图"""
page_size = 30
schedules, total_items = await scheduler_manager.get_schedules(
page=page, page_size=page_size, **filters
)
if not schedules:
return "没有找到任何相关的定时任务。"
permitted_schedules = schedules
skipped_count = 0
if not is_superuser:
permitted_schedules, skipped_count = await self._filter_schedules_for_user(
schedules, user_id, group_id
)
if not permitted_schedules:
return (
f"您没有权限查看任何匹配的任务。（因权限不足跳过 {skipped_count} 个）"
)
title = self._generate_view_title(filters)
return await presenters.format_schedule_list_as_image(
schedules=permitted_schedules,
title=title,
current_page=page,
total_items=total_items,
)
async def set_schedule(
self,
targets: list[str],
creator_permission_level: int,
plugin_name: str,
trigger_info: tuple[str, dict],
job_kwargs: dict,
permission: int,
bot_id: str,
job_name: str | None,
jitter: int | None,
spread: int | None,
created_by: str,
) -> str:
"""创建或更新一个定时任务"""
trigger_type, trigger_config = trigger_info
success_targets = []
failed_targets = []
permission_denied_targets = []
execution_options = {}
if jitter is not None:
execution_options["jitter"] = jitter
if spread is not None:
execution_options["spread"] = spread
for target_desc in targets:
target_type, target_id = self._resolve_target_descriptor(target_desc)
existing_schedule = await ScheduleRepository.filter(
plugin_name=plugin_name,
target_type=target_type,
target_identifier=target_id,
bot_id=bot_id,
).first()
if (
existing_schedule
and creator_permission_level < existing_schedule.required_permission
):
permission_denied_targets.append(
(
target_desc,
f"需要 {existing_schedule.required_permission} 级权限",
)
)
continue
if target_type in ["TAG", "ALL_GROUPS"]:
logger.debug(
f"检测到多目标任务 (类型: {target_type})，"
f"将所需权限强制提升至超级用户级别。"
)
permission = 9
try:
schedule = await scheduler_manager.add_schedule(
plugin_name=plugin_name,
target_type=target_type,
target_identifier=target_id,
trigger_type=trigger_type,
trigger_config=trigger_config,
job_kwargs=job_kwargs,
bot_id=bot_id,
required_permission=permission,
name=job_name,
created_by=created_by,
execution_options=execution_options if execution_options else None,
)
if schedule:
success_targets.append((target_desc, schedule.id))
else:
failed_targets.append((target_desc, "服务返回失败"))
except Exception as e:
failed_targets.append((target_desc, str(e)))
return self._format_set_result_message(
targets, success_targets, failed_targets, permission_denied_targets
)
async def perform_bulk_operation(
self,
operation_name: str,
user_id: str,
group_id: str | None,
is_superuser: bool,
targeter,
all_flag: bool,
global_flag: bool,
) -> str:
"""执行批量操作（删除、暂停、恢复）"""
if not is_superuser:
permission_denied = False
if all_flag or global_flag:
permission_denied = True
elif targeter._filters.get("target_type") in ["TAG", "ALL_GROUPS"]:
permission_denied = True
if permission_denied:
return "权限不足，只有超级用户才能对所有群组或通过标签进行批量操作。"
schedules_to_operate = await targeter._get_schedules()
if not schedules_to_operate:
return "没有找到符合条件的可操作任务。"
permitted_schedules, skipped_count = (
(schedules_to_operate, 0)
if is_superuser
else await self._filter_schedules_for_user(
schedules_to_operate, user_id, group_id
)
)
if not permitted_schedules:
return (
f"您没有权限{operation_name}任何匹配的任务。"
f"（因权限不足跳过 {skipped_count} 个）"
)
permitted_ids = [s.id for s in permitted_schedules]
final_targeter = scheduler_manager.target(id__in=permitted_ids)
operation_map = {
"删除": final_targeter.remove,
"暂停": final_targeter.pause,
"恢复": final_targeter.resume,
}
operation_func = operation_map.get(operation_name)
if not operation_func:
return f"未知的批量操作: {operation_name}"
count, _ = await operation_func()
msg = f"批量{operation_name}操作完成：\n - 成功: {count} 个"
if skipped_count > 0:
msg += f"\n - 因权限不足跳过: {skipped_count} 个"
return msg
async def trigger_schedule_now(self, schedule: ScheduledJob) -> str:
"""立即触发一个任务"""
success, message = await scheduler_manager.trigger_now(schedule.id)
return (
presenters.format_trigger_success(schedule)
if success
else f"❌ 触发失败: {message}"
)
async def update_schedule(
self, schedule: ScheduledJob, trigger_info: tuple | None, kwargs_str: str | None
) -> str:
"""更新一个任务的配置"""
trigger_type = trigger_info[0] if trigger_info else None
trigger_config = trigger_info[1] if trigger_info else None
job_kwargs = await self._parse_and_validate_kwargs_for_update(
schedule.plugin_name, kwargs_str
)
success, message = await scheduler_manager.update_schedule(
schedule.id, trigger_type, trigger_config, job_kwargs
)
if success:
updated_schedule = await scheduler_manager.get_schedule_by_id(schedule.id)
return (
presenters.format_update_success(updated_schedule)
if updated_schedule
else "✅ 更新成功，但无法获取更新后的任务详情。"
)
return f"❌ 更新失败: {message}"
async def get_schedule_status(self, schedule_id: int) -> str:
"""获取单个任务的状态"""
status = await scheduler_manager.get_schedule_status(schedule_id)
if not status:
return f"未找到ID为 {schedule_id} 的任务。"
return presenters.format_single_status_message(status)
async def get_plugins_list(self) -> str:
"""获取可定时执行的插件列表"""
return await presenters.format_plugins_list()
async def _filter_schedules_for_user(
self, schedules: list[ScheduledJob], user_id: str, group_id: str | None
) -> tuple[list[ScheduledJob], int]:
user_level = await LevelUser.get_user_level(user_id, group_id)
permitted = [s for s in schedules if user_level >= s.required_permission]
skipped_count = len(schedules) - len(permitted)
return permitted, skipped_count
def _generate_view_title(self, filters: dict) -> str:
title = "定时任务"
if filters.get("target_type") == "ALL_GROUPS":
title = "全局定时任务"
elif "target_identifier" in filters:
title = f"群 {filters['target_identifier']} 的定时任务"
if "plugin_name" in filters:
title += f" [插件: {filters['plugin_name']}]"
return title
def _resolve_target_descriptor(self, target_desc: str) -> tuple[str, str]:
if target_desc == scheduler_manager.ALL_GROUPS:
return "ALL_GROUPS", scheduler_manager.ALL_GROUPS
if target_desc.startswith("tag:"):
return "TAG", target_desc[4:]
if target_desc.isdigit():
return "GROUP", target_desc
return "USER", target_desc
def _format_set_result_message(
self, targets: list, success: list, failed: list, permission_denied: list
) -> str:
msg = f"为 {len(targets)} 个目标设置/更新任务完成：\n"
if success:
msg += f"- 成功: {len(success)} 个"
ids_str = ", ".join(str(s[1]) for s in success)
msg += f"\n - ID列表: {ids_str}"
else:
msg += "- 成功: 0 个"
if permission_denied:
msg += f"\n- 因权限不足跳过: {len(permission_denied)} 个"
for target, reason in permission_denied:
msg += f"\n - 目标 {target}: {reason}"
if failed:
msg += f"\n- 失败: {len(failed)} 个"
for target, reason in failed:
msg += f"\n - 目标 {target}: {reason}"
return msg.strip()
async def _parse_and_validate_kwargs_for_update(
self, plugin_name: str, kwargs_str: str | None
) -> dict:
if not kwargs_str:
return {}
task_meta = scheduler_manager._registered_tasks.get(plugin_name)
if not task_meta:
raise ValueError(f"插件 '{plugin_name}' 未注册。")
params_model = task_meta.get("model")
if not (
params_model
and isinstance(params_model, type)
and issubclass(params_model, BaseModel)
):
raise ValueError(f"插件 '{plugin_name}' 不支持或配置了无效的参数模型。")
try:
raw_kwargs = dict(
item.strip().split("=", 1) for item in kwargs_str.split(";")
)
validated_model = model_validate(params_model, raw_kwargs)
return model_dump(validated_model)
except ValidationError as e:
errors = [f" - {err['loc'][0]}: {err['msg']}" for err in e.errors()]
raise ValueError("参数验证失败:\n" + "\n".join(errors))
except Exception as e:
raise ValueError(f"参数格式错误: {e}")
scheduler_admin_service = SchedulerAdminService()
zhenxun/builtin_plugins/scheduler_admin/dependencies.py
+370
View File
@@ -0,0 +1,370 @@
from datetime import datetime
import re
from typing import Any
from arclet.alconna import Alconna
from nonebot.adapters import Bot, Event
from nonebot.params import Depends
from nonebot.permission import SUPERUSER
from nonebot_plugin_alconna import (
AlconnaMatch,
AlconnaMatcher,
AlconnaMatches,
AlconnaQuery,
Arparma,
Match,
Query,
)
from nonebot_plugin_session import EventSession
from nonebot_plugin_uninfo import Uninfo
from zhenxun.configs.config import Config
from zhenxun.models.level_user import LevelUser
from zhenxun.models.scheduled_job import ScheduledJob
from zhenxun.services import scheduler_manager
from zhenxun.utils.time_utils import TimeUtils
async def GetCreatorPermissionLevel(
bot: Bot,
event: Event,
session: Uninfo,
) -> int:
"""
依赖注入函数：获取执行命令的用户的权限等级。
"""
is_superuser = await SUPERUSER(bot, event)
if is_superuser:
return 999
current_group_id = session.group.id if session.group else None
return await LevelUser.get_user_level(session.user.id, current_group_id)
async def RequireTaskPermission(
matcher: AlconnaMatcher,
bot: Bot,
event: Event,
session: EventSession,
schedule_id_match: Match[int] = AlconnaMatch("schedule_id"),
) -> ScheduledJob:
"""
依赖注入函数：获取并验证用户对特定任务的操作权限。
"""
if not schedule_id_match.available:
await matcher.finish("此操作需要一个有效的任务ID。")
schedule_id = schedule_id_match.result
schedule = await scheduler_manager.get_schedule_by_id(schedule_id)
if not schedule:
await matcher.finish(f"未找到ID为 {schedule_id} 的任务。")
is_superuser = await SUPERUSER(bot, event)
if is_superuser:
return schedule
user_id = session.id1
if not user_id:
await matcher.finish("无法获取用户信息，权限检查失败。")
group_id = session.id3 or session.id2
user_level = await LevelUser.get_user_level(user_id, group_id)
if user_level < schedule.required_permission:
await matcher.finish(
f"权限不足！操作此任务需要 {schedule.required_permission} 级权限，"
f"您当前为 {user_level} 级。"
)
return schedule
def parse_daily_time(time_str: str) -> dict:
"""解析每日时间字符串为 cron 配置字典"""
if match := re.match(r"^(\d{1,2}):(\d{1,2})(?::(\d{1,2}))?$", time_str):
hour, minute, second = match.groups()
hour, minute = int(hour), int(minute)
if not (0 <= hour <= 23 and 0 <= minute <= 59):
raise ValueError("小时或分钟数值超出范围。")
cron_config = {
"minute": str(minute),
"hour": str(hour),
"day": "*",
"month": "*",
"day_of_week": "*",
"timezone": Config.get_config("SchedulerManager", "SCHEDULER_TIMEZONE"),
}
if second is not None:
if not (0 <= int(second) <= 59):
raise ValueError("秒数值超出范围。")
cron_config["second"] = str(second)
return cron_config
else:
raise ValueError("时间格式错误，请使用 'HH:MM' 或 'HH:MM:SS' 格式。")
def _parse_trigger_from_arparma(arp: Arparma) -> tuple[str, dict] | None:
"""从 Arparma 中解析时间触发器配置"""
subcommand_name = next(iter(arp.subcommands.keys()), None)
if not subcommand_name:
return None
try:
if cron_expr := arp.query[str](f"{subcommand_name}.cron.cron_expr", None):
return "cron", dict(
zip(
["minute", "hour", "day", "month", "day_of_week"], cron_expr.split()
)
)
if interval_expr := arp.query[str](
f"{subcommand_name}.interval.interval_expr", None
):
return "interval", TimeUtils.parse_interval_to_dict(interval_expr)
if date_expr := arp.query[str](f"{subcommand_name}.date.date_expr", None):
return "date", {"run_date": datetime.fromisoformat(date_expr)}
if daily_expr := arp.query[str](f"{subcommand_name}.daily.daily_expr", None):
return "cron", parse_daily_time(daily_expr)
except ValueError as e:
raise ValueError(f"时间参数解析错误: {e}") from e
return None
async def GetTriggerInfo(
matcher: AlconnaMatcher,
arp: Arparma = AlconnaMatches(),
) -> tuple[str, dict]:
"""依赖注入函数：解析并验证时间触发器"""
try:
trigger_info = _parse_trigger_from_arparma(arp)
if trigger_info:
return trigger_info
except ValueError as e:
await matcher.finish(f"时间参数解析错误: {e}")
await matcher.finish(
"必须提供一种时间选项: --cron, --interval, --date, 或 --daily。"
)
async def GetBotId(bot: Bot, bot_id_match: Match[str] = AlconnaMatch("bot_id")) -> str:
"""依赖注入函数：获取要操作的Bot ID"""
if bot_id_match.available:
return bot_id_match.result
return bot.self_id
async def GetTargeter(
matcher: AlconnaMatcher,
event: Event,
bot: Bot,
arp: Arparma = AlconnaMatches(),
schedule_id: Match[int] = AlconnaMatch("schedule_id"),
plugin_name: Match[str] = AlconnaMatch("plugin_name"),
group_ids: Match[list[str]] = AlconnaMatch("group_ids"),
user_id: Match[str] = AlconnaMatch("user_id"),
tag_name: Match[str] = AlconnaMatch("tag_name"),
bot_id_to_operate: str = Depends(GetBotId),
) -> Any:
"""
依赖注入函数，用于解析命令参数并返回一个配置好的 ScheduleTargeter 实例。
"""
subcommand = next(iter(arp.subcommands.keys()), None)
if not subcommand:
await matcher.finish("内部错误：无法解析子命令。")
if schedule_id.available:
return scheduler_manager.target(id=schedule_id.result)
all_enabled = arp.query(f"{subcommand}.all.value", False)
global_flag = arp.query(f"{subcommand}.global.value", False)
if not any(
[
plugin_name.available,
all_enabled,
global_flag,
user_id.available,
group_ids.available,
tag_name.available,
getattr(event, "group_id", None),
]
):
await matcher.finish(
f"'{subcommand}'操作失败：请提供任务ID，"
f"或通过 -p <插件名> / --global / --all 指定要操作的任务。"
)
filters: dict[str, Any] = {"bot_id": bot_id_to_operate}
if plugin_name.available:
filters["plugin_name"] = plugin_name.result
if global_flag:
filters["target_type"] = "ALL_GROUPS"
filters["target_identifier"] = scheduler_manager.ALL_GROUPS
elif user_id.available:
filters["target_type"] = "USER"
filters["target_identifier"] = user_id.result
elif all_enabled:
pass
elif tag_name.available:
filters["target_type"] = "TAG"
filters["target_identifier"] = tag_name.result
elif group_ids.available:
gids = [str(gid) for gid in group_ids.result]
filters["target_type"] = "GROUP"
filters["target_identifier__in"] = gids
else:
current_group_id = getattr(event, "group_id", None)
if current_group_id:
filters["target_type"] = "GROUP"
filters["target_identifier"] = str(current_group_id)
return scheduler_manager.target(**filters)
async def GetValidatedJobKwargs(
matcher: AlconnaMatcher,
plugin_name: Match[str] = AlconnaMatch("plugin_name"),
cli_string: Match[str] = AlconnaMatch("cli_string"),
kwargs_str: Match[str] = AlconnaMatch("kwargs_str"),
) -> dict:
"""依赖注入函数：解析、合并和验证任务的关键字参数"""
p_name = plugin_name.result
task_meta = scheduler_manager._registered_tasks.get(p_name)
if not task_meta:
await matcher.finish(f"插件 '{p_name}' 未注册可定时执行的任务。")
cli_kwargs = {}
if cli_string.available and cli_string.result.strip():
if not (cli_parser := task_meta.get("cli_parser")):
await matcher.finish(
f"插件 '{p_name}' 不支持通过 --params-cli 设置参数，"
f"因为它没有注册解析器。"
)
try:
temp_parser = Alconna("_", cli_parser.args, *cli_parser.options) # type: ignore
parsed_cli = temp_parser.parse(f"_ {cli_string.result.strip()}")
if not parsed_cli.matched:
raise ValueError(f"参数无法匹配: {parsed_cli.error_info or '未知错误'}")
cli_kwargs = parsed_cli.all_matched_args
except Exception as e:
await matcher.finish(
f"使用 --params-cli 解析参数失败: {e}\n\n请确保参数格式与插件命令一致。"
)
explicit_kwargs = {}
if kwargs_str.available and kwargs_str.result.strip():
try:
explicit_kwargs = dict(
item.strip().split("=", 1)
for item in kwargs_str.result.split(";")
if item.strip()
)
except ValueError:
await matcher.finish(
"参数格式错误，--kwargs 请使用 'key=value;key2=value2' 格式。"
)
final_job_kwargs = {**cli_kwargs, **explicit_kwargs}
is_valid, result = scheduler_manager._validate_and_prepare_kwargs(
p_name, final_job_kwargs
)
if not is_valid:
await matcher.finish(f"任务参数校验失败:\n{result}")
return result if isinstance(result, dict) else {}
async def GetFinalPermission(
matcher: AlconnaMatcher,
bot: Bot,
event: Event,
session: Uninfo,
plugin_name: Match[str] = AlconnaMatch("plugin_name"),
perm_level: Match[int] = AlconnaMatch("perm_level"),
) -> int:
"""依赖注入函数：计算任务的最终权限等级"""
is_superuser = await SUPERUSER(bot, event)
current_group_id = session.group.id if session.group else None
if is_superuser:
effective_user_level = 9
else:
effective_user_level = await LevelUser.get_user_level(
session.user.id, current_group_id
)
if perm_level.available:
requested_perm_level = perm_level.result
if not is_superuser and requested_perm_level > effective_user_level:
await matcher.send(
f"⚠️ 警告：您指定的权限等级 ({requested_perm_level}) "
f"高于自身权限 ({effective_user_level})。\n"
f"任务的管理权限已被自动设置为 {effective_user_level} 级。"
)
return effective_user_level
return requested_perm_level
else:
base_permission = effective_user_level
task_meta = scheduler_manager._registered_tasks.get(plugin_name.result)
if task_meta and "default_permission" in task_meta:
default_perm = task_meta.get("default_permission")
if isinstance(default_perm, int):
base_permission = default_perm
return min(base_permission, effective_user_level)
async def ResolveTargets(
matcher: AlconnaMatcher,
bot: Bot,
event: Event,
session: Uninfo,
group_ids: Match[list[str]] = AlconnaMatch("group_ids"),
tag_name: Match[str] = AlconnaMatch("tag_name"),
user_id: Match[str] = AlconnaMatch("user_id"),
all_flag: Query[bool] = AlconnaQuery("设置.all.value", False),
global_flag: Query[bool] = AlconnaQuery("设置.global.value", False),
) -> list[str]:
"""依赖注入函数，用于解析和计算最终的目标描述符列表，并进行权限检查"""
is_superuser = await SUPERUSER(bot, event)
current_group_id = session.group.id if session.group else None
if not is_superuser:
permission_denied = False
if (
global_flag.result
or all_flag.result
or tag_name.available
or user_id.available
):
permission_denied = True
elif group_ids.available and any(
str(gid) != str(current_group_id) for gid in group_ids.result
):
permission_denied = True
if permission_denied:
await matcher.finish(
"权限不足，只有超级用户才能为其他群组、所有群组或通过标签设置任务。"
)
if user_id.available:
return [user_id.result]
if all_flag.result or global_flag.result:
return [scheduler_manager.ALL_GROUPS]
if tag_name.available:
return [f"tag:{tag_name.result}"]
if group_ids.available:
return group_ids.result
if current_group_id:
return [str(current_group_id)]
await matcher.finish(
"私聊中设置任务必须使用 -u, -g, --all, --global 或 -t 选项指定目标。"
)
zhenxun/builtin_plugins/scheduler_admin/handlers.py
+164 -319
View File
@@ -1,382 +1,227 @@
from datetime import datetime
from typing import cast
from nonebot.adapters import Event
from nonebot.adapters.onebot.v11 import Bot
from nonebot.adapters import Bot, Event
from nonebot.params import Depends
from nonebot.permission import SUPERUSER
from nonebot_plugin_alconna import AlconnaMatch, Arparma, Match, Query
from pydantic import BaseModel, ValidationError
from zhenxun.models.scheduled_job import ScheduledJob
from zhenxun.services.scheduler import scheduler_manager
from zhenxun.services.scheduler.targeter import ScheduleTargeter
from zhenxun.utils.message import MessageUtils
from zhenxun.utils.pydantic_compat import model_dump
from . import presenters
from .commands import (
GetBotId,
GetTargeter,
parse_daily_time,
parse_interval,
schedule_cmd,
from nonebot_plugin_alconna import (
AlconnaMatch,
AlconnaMatches,
AlconnaQuery,
Arparma,
Match,
Query,
)
from nonebot_plugin_uninfo import Uninfo
from zhenxun.configs.config import Config
from zhenxun.models.scheduled_job import ScheduledJob
from zhenxun.services import scheduler_manager
from zhenxun.utils.message import MessageUtils
@schedule_cmd.handle()
async def _handle_time_options_mutex(arp: Arparma):
time_options = ["cron", "interval", "date", "daily"]
provided_options = [opt for opt in time_options if arp.query(opt) is not None]
if len(provided_options) > 1:
await schedule_cmd.finish(
f"时间选项 --{', --'.join(provided_options)} 不能同时使用，请只选择一个。"
)
from .commands import schedule_cmd
from .data_source import scheduler_admin_service
from .dependencies import (
GetBotId,
GetCreatorPermissionLevel,
GetFinalPermission,
GetTargeter,
GetTriggerInfo,
GetValidatedJobKwargs,
RequireTaskPermission,
ResolveTargets,
_parse_trigger_from_arparma,
)
@schedule_cmd.assign("查看")
async def handle_view(
bot: Bot,
event: Event,
target_group_id: Match[str] = AlconnaMatch("target_group_id"),
all_groups: Query[bool] = Query("查看.all"),
plugin_name: Match[str] = AlconnaMatch("plugin_name"),
session: Uninfo,
page: Match[int] = AlconnaMatch("page"),
targeter=Depends(GetTargeter),
):
"""处理 '查看' 子命令"""
is_superuser = await SUPERUSER(bot, event)
title = ""
gid_filter = None
current_page = page.result if page.available else 1
current_group_id = getattr(event, "group_id", None)
if not (all_groups.available or target_group_id.available) and not current_group_id:
await schedule_cmd.finish("私聊中查看任务必须使用 -g <群号> 或 -all 选项。")
if all_groups.available:
if not is_superuser:
await schedule_cmd.finish("需要超级用户权限才能查看所有群组的定时任务。")
title = "所有群组的定时任务"
elif target_group_id.available:
if not is_superuser:
await schedule_cmd.finish("需要超级用户权限才能查看指定群组的定时任务。")
gid_filter = target_group_id.result
title = f"群 {gid_filter} 的定时任务"
else:
gid_filter = str(current_group_id)
title = "本群的定时任务"
p_name_filter = plugin_name.result if plugin_name.available else None
schedules = await scheduler_manager.get_schedules(
plugin_name=p_name_filter, group_id=gid_filter
result = await scheduler_admin_service.get_schedules_view(
user_id=session.user.id,
group_id=session.group.id if session.group else None,
is_superuser=is_superuser,
filters=targeter._filters,
page=current_page,
)
if p_name_filter:
title += f" [插件: {p_name_filter}]"
if not schedules:
await schedule_cmd.finish("没有找到任何相关的定时任务。")
img = await presenters.format_schedule_list_as_image(
schedules=schedules,
title=title,
current_page=page.result if page.available else 1,
)
await MessageUtils.build_message(img).send(reply_to=True)
await MessageUtils.build_message(result).send(reply_to=True)
@schedule_cmd.assign("设置")
async def handle_set(
event: Event,
session: Uninfo,
target_groups: list[str] = Depends(ResolveTargets),
plugin_name: Match[str] = AlconnaMatch("plugin_name"),
cron_expr: Match[str] = AlconnaMatch("cron_expr"),
interval_expr: Match[str] = AlconnaMatch("interval_expr"),
date_expr: Match[str] = AlconnaMatch("date_expr"),
daily_expr: Match[str] = AlconnaMatch("daily_expr"),
group_id: Match[str] = AlconnaMatch("group_id"),
kwargs_str: Match[str] = AlconnaMatch("kwargs_str"),
all_enabled: Query[bool] = Query("设置.all"),
tag_name: Match[str] = AlconnaMatch("tag_name"),
jitter: Match[int] = AlconnaMatch("jitter_seconds"),
spread: Match[int] = AlconnaMatch("spread_seconds"),
job_name: Match[str] = AlconnaMatch("job_name"),
bot_id_to_operate: str = Depends(GetBotId),
trigger_info: tuple[str, dict] = Depends(GetTriggerInfo),
job_kwargs: dict = Depends(GetValidatedJobKwargs),
creator_permission_level: int = Depends(GetCreatorPermissionLevel),
final_permission: int = Depends(GetFinalPermission),
):
if not plugin_name.available:
await schedule_cmd.finish("设置任务时必须提供插件名称。")
has_time_option = any(
[
cron_expr.available,
interval_expr.available,
date_expr.available,
daily_expr.available,
]
)
if not has_time_option:
await schedule_cmd.finish(
"必须提供一种时间选项: --cron, --interval, --date, 或 --daily。"
)
"""处理 '设置' 子命令"""
p_name = plugin_name.result
if p_name not in scheduler_manager.get_registered_plugins():
await schedule_cmd.finish(
f"插件 '{p_name}' 没有注册可用的定时任务。\n"
f"可用插件: {list(scheduler_manager.get_registered_plugins())}"
jitter_val: int | None = jitter.result if jitter.available else None
spread_val: int | None = spread.result if spread.available else None
is_multi_target = (
len(target_groups) > 1
or (
len(target_groups) == 1 and target_groups[0] == scheduler_manager.ALL_GROUPS
)
or tag_name.available
)
trigger_type, trigger_config = "", {}
try:
if cron_expr.available:
trigger_type, trigger_config = (
"cron",
dict(
zip(
["minute", "hour", "day", "month", "day_of_week"],
cron_expr.result.split(),
)
),
)
elif interval_expr.available:
trigger_type, trigger_config = (
"interval",
parse_interval(interval_expr.result),
)
elif date_expr.available:
trigger_type, trigger_config = (
"date",
{"run_date": datetime.fromisoformat(date_expr.result)},
)
elif daily_expr.available:
trigger_type, trigger_config = "cron", parse_daily_time(daily_expr.result)
else:
await schedule_cmd.finish(
"必须提供一种时间选项: --cron, --interval, --date, 或 --daily。"
)
except ValueError as e:
await schedule_cmd.finish(f"时间参数解析错误: {e}")
job_kwargs = {}
if kwargs_str.available:
if is_multi_target:
task_meta = scheduler_manager._registered_tasks.get(p_name)
if not task_meta:
await schedule_cmd.finish(f"插件 '{p_name}' 未注册。")
if jitter_val is None:
if task_meta and task_meta.get("default_jitter") is not None:
jitter_val = cast(int | None, task_meta["default_jitter"])
else:
jitter_val = Config.get_config(
"SchedulerManager", "DEFAULT_JITTER_SECONDS"
)
if spread_val is None:
if task_meta and task_meta.get("default_spread") is not None:
spread_val = cast(int | None, task_meta["default_spread"])
else:
spread_val = Config.get_config(
"SchedulerManager", "DEFAULT_SPREAD_SECONDS"
)
params_model = task_meta.get("model")
if not (
params_model
and isinstance(params_model, type)
and issubclass(params_model, BaseModel)
):
await schedule_cmd.finish(f"插件 '{p_name}' 不支持或配置了无效的参数模型。")
try:
raw_kwargs = dict(
item.strip().split("=", 1) for item in kwargs_str.result.split(",")
)
model_validate = getattr(params_model, "model_validate", None)
if not model_validate:
await schedule_cmd.finish(f"插件 '{p_name}' 的参数模型不支持验证")
validated_model = model_validate(raw_kwargs)
job_kwargs = model_dump(validated_model)
except ValidationError as e:
errors = [f" - {err['loc'][0]}: {err['msg']}" for err in e.errors()]
await schedule_cmd.finish(
f"插件 '{p_name}' 的任务参数验证失败:\n" + "\n".join(errors)
)
except Exception as e:
await schedule_cmd.finish(
f"参数格式错误，请使用 'key=value,key2=value2' 格式。错误: {e}"
)
gid_str = group_id.result if group_id.available else None
target_group_id = (
scheduler_manager.ALL_GROUPS
if (gid_str and gid_str.lower() == "all") or all_enabled.available
else gid_str or getattr(event, "group_id", None)
)
if not target_group_id:
await schedule_cmd.finish(
"私聊中设置定时任务时，必须使用 -g <群号> 或 --all 选项指定目标。"
)
schedule = await scheduler_manager.add_schedule(
p_name,
str(target_group_id),
trigger_type,
trigger_config,
job_kwargs,
result_message = await scheduler_admin_service.set_schedule(
targets=target_groups,
creator_permission_level=creator_permission_level,
plugin_name=p_name,
trigger_info=trigger_info,
job_kwargs=job_kwargs,
permission=final_permission,
bot_id=bot_id_to_operate,
job_name=job_name.result if job_name.available else None,
jitter=jitter_val,
spread=spread_val,
created_by=session.user.id,
)
target_desc = (
f"所有群组 (Bot: {bot_id_to_operate})"
if target_group_id == scheduler_manager.ALL_GROUPS
else f"群组 {target_group_id}"
)
if schedule:
await schedule_cmd.finish(
f"为 [{target_desc}] 已成功设置插件 '{p_name}' 的定时任务 "
f"(ID: {schedule.id})。"
)
else:
await schedule_cmd.finish(f"为 [{target_desc}] 设置任务失败。")
await MessageUtils.build_message(result_message).send()
@schedule_cmd.assign("删除")
async def handle_delete(targeter: ScheduleTargeter = GetTargeter("删除")):
schedules_to_remove: list[ScheduledJob] = await targeter._get_schedules()
if not schedules_to_remove:
await schedule_cmd.finish("没有找到可删除的任务。")
count, _ = await targeter.remove()
if count > 0 and schedules_to_remove:
if len(schedules_to_remove) == 1:
message = presenters.format_remove_success(schedules_to_remove[0])
else:
target_desc = targeter._generate_target_description()
message = f"✅ 成功移除了{target_desc} {count} 个任务。"
else:
message = "没有任务被移除。"
await schedule_cmd.finish(message)
async def handle_delete(
bot: Bot,
event: Event,
session: Uninfo,
targeter=Depends(GetTargeter),
all_flag: Query[bool] = AlconnaQuery("删除.all.value", False),
global_flag: Query[bool] = AlconnaQuery("删除.global.value", False),
):
"""处理 '删除' 子命令"""
is_superuser = await SUPERUSER(bot, event)
result_message = await scheduler_admin_service.perform_bulk_operation(
operation_name="删除",
user_id=session.user.id,
group_id=session.group.id if session.group else None,
is_superuser=is_superuser,
targeter=targeter,
all_flag=all_flag.result,
global_flag=global_flag.result,
)
await schedule_cmd.finish(result_message)
@schedule_cmd.assign("暂停")
async def handle_pause(targeter: ScheduleTargeter = GetTargeter("暂停")):
schedules_to_pause: list[ScheduledJob] = await targeter._get_schedules()
if not schedules_to_pause:
await schedule_cmd.finish("没有找到可暂停的任务。")
count, _ = await targeter.pause()
if count > 0 and schedules_to_pause:
if len(schedules_to_pause) == 1:
message = presenters.format_pause_success(schedules_to_pause[0])
else:
target_desc = targeter._generate_target_description()
message = f"✅ 成功暂停了{target_desc} {count} 个任务。"
else:
message = "没有任务被暂停。"
await schedule_cmd.finish(message)
async def handle_pause(
bot: Bot,
event: Event,
session: Uninfo,
targeter=Depends(GetTargeter),
all_flag: Query[bool] = AlconnaQuery("暂停.all.value", False),
global_flag: Query[bool] = AlconnaQuery("暂停.global.value", False),
):
"""处理 '暂停' 子命令"""
is_superuser = await SUPERUSER(bot, event)
result_message = await scheduler_admin_service.perform_bulk_operation(
operation_name="暂停",
user_id=session.user.id,
group_id=session.group.id if session.group else None,
is_superuser=is_superuser,
targeter=targeter,
all_flag=all_flag.result,
global_flag=global_flag.result,
)
await schedule_cmd.finish(result_message)
@schedule_cmd.assign("恢复")
async def handle_resume(targeter: ScheduleTargeter = GetTargeter("恢复")):
schedules_to_resume: list[ScheduledJob] = await targeter._get_schedules()
if not schedules_to_resume:
await schedule_cmd.finish("没有找到可恢复的任务。")
count, _ = await targeter.resume()
if count > 0 and schedules_to_resume:
if len(schedules_to_resume) == 1:
message = presenters.format_resume_success(schedules_to_resume[0])
else:
target_desc = targeter._generate_target_description()
message = f"✅ 成功恢复了{target_desc} {count} 个任务。"
else:
message = "没有任务被恢复。"
await schedule_cmd.finish(message)
async def handle_resume(
bot: Bot,
event: Event,
session: Uninfo,
targeter=Depends(GetTargeter),
all_flag: Query[bool] = AlconnaQuery("恢复.all.value", False),
global_flag: Query[bool] = AlconnaQuery("恢复.global.value", False),
):
"""处理 '恢复' 子命令"""
is_superuser = await SUPERUSER(bot, event)
result_message = await scheduler_admin_service.perform_bulk_operation(
operation_name="恢复",
user_id=session.user.id,
group_id=session.group.id if session.group else None,
is_superuser=is_superuser,
targeter=targeter,
all_flag=all_flag.result,
global_flag=global_flag.result,
)
await schedule_cmd.finish(result_message)
@schedule_cmd.assign("执行")
async def handle_trigger(schedule_id: Match[int] = AlconnaMatch("schedule_id")):
from zhenxun.services.scheduler.repository import ScheduleRepository
schedule_info = await ScheduleRepository.get_by_id(schedule_id.result)
if not schedule_info:
await schedule_cmd.finish(f"未找到 ID 为 {schedule_id.result} 的任务。")
success, message = await scheduler_manager.trigger_now(schedule_id.result)
if success:
final_message = presenters.format_trigger_success(schedule_info)
else:
final_message = f"❌ 手动触发失败: {message}"
await schedule_cmd.finish(final_message)
async def handle_trigger(schedule: ScheduledJob = Depends(RequireTaskPermission)):
"""处理 '执行' 子命令"""
result_message = await scheduler_admin_service.trigger_schedule_now(schedule)
await schedule_cmd.finish(result_message)
@schedule_cmd.assign("更新")
async def handle_update(
schedule_id: Match[int] = AlconnaMatch("schedule_id"),
cron_expr: Match[str] = AlconnaMatch("cron_expr"),
interval_expr: Match[str] = AlconnaMatch("interval_expr"),
date_expr: Match[str] = AlconnaMatch("date_expr"),
daily_expr: Match[str] = AlconnaMatch("daily_expr"),
schedule: ScheduledJob = Depends(RequireTaskPermission),
arp: Arparma = AlconnaMatches(),
kwargs_str: Match[str] = AlconnaMatch("kwargs_str"),
):
if not any(
[
cron_expr.available,
interval_expr.available,
date_expr.available,
daily_expr.available,
kwargs_str.available,
]
):
"""处理 '更新' 子命令"""
trigger_info = _parse_trigger_from_arparma(arp)
if not trigger_info and not kwargs_str.available:
await schedule_cmd.finish(
"请提供需要更新的时间 (--cron/--interval/--date/--daily) 或参数 (--kwargs)"
)
trigger_type, trigger_config, job_kwargs = None, None, None
try:
if cron_expr.available:
trigger_type, trigger_config = (
"cron",
dict(
zip(
["minute", "hour", "day", "month", "day_of_week"],
cron_expr.result.split(),
)
),
)
elif interval_expr.available:
trigger_type, trigger_config = (
"interval",
parse_interval(interval_expr.result),
)
elif date_expr.available:
trigger_type, trigger_config = (
"date",
{"run_date": datetime.fromisoformat(date_expr.result)},
)
elif daily_expr.available:
trigger_type, trigger_config = "cron", parse_daily_time(daily_expr.result)
except ValueError as e:
await schedule_cmd.finish(f"时间参数解析错误: {e}")
if kwargs_str.available:
job_kwargs = dict(
item.strip().split("=", 1) for item in kwargs_str.result.split(",")
)
success, message = await scheduler_manager.update_schedule(
schedule_id.result, trigger_type, trigger_config, job_kwargs
result_message = await scheduler_admin_service.update_schedule(
schedule, trigger_info, kwargs_str.result if kwargs_str.available else None
)
if success:
from zhenxun.services.scheduler.repository import ScheduleRepository
updated_schedule = await ScheduleRepository.get_by_id(schedule_id.result)
if updated_schedule:
final_message = presenters.format_update_success(updated_schedule)
else:
final_message = "✅ 更新成功，但无法获取更新后的任务详情。"
else:
final_message = f"❌ 更新失败: {message}"
await schedule_cmd.finish(final_message)
await schedule_cmd.finish(result_message)
@schedule_cmd.assign("插件列表")
async def handle_plugins_list():
message = await presenters.format_plugins_list()
"""处理 '插件列表' 子命令"""
message = await scheduler_admin_service.get_plugins_list()
await schedule_cmd.finish(message)
@schedule_cmd.assign("状态")
async def handle_status(schedule_id: Match[int] = AlconnaMatch("schedule_id")):
status = await scheduler_manager.get_schedule_status(schedule_id.result)
if not status:
await schedule_cmd.finish(f"未找到ID为 {schedule_id.result} 的定时任务。")
message = presenters.format_single_status_message(status)
async def handle_status(
schedule: ScheduledJob = Depends(RequireTaskPermission),
):
"""处理 '状态' 子命令"""
message = await scheduler_admin_service.get_schedule_status(schedule.id)
await schedule_cmd.finish(message)
zhenxun/builtin_plugins/scheduler_admin/presenters.py
+34 -34
View File
@@ -1,24 +1,13 @@
import asyncio
from typing import Any
from zhenxun import ui
from zhenxun.models.scheduled_job import ScheduledJob
from zhenxun.services.scheduler import scheduler_manager
from zhenxun.services import scheduler_manager
from zhenxun.ui.builders import TableBuilder
from zhenxun.ui.models import StatusBadgeCell, TextCell
from zhenxun.utils.pydantic_compat import model_json_schema
def _get_type_name(annotation) -> str:
"""获取类型注解的名称"""
if hasattr(annotation, "__name__"):
return annotation.__name__
elif hasattr(annotation, "_name"):
return annotation._name
else:
return str(annotation)
def _get_schedule_attr(schedule: ScheduledJob | dict, attr_name: str) -> Any:
"""兼容地从字典或对象获取属性"""
if isinstance(schedule, dict):
@@ -73,13 +62,8 @@ def _format_operation_result_card(
schedule_info: 相关的 ScheduledJob 对象
extra_info: (可选) 额外的补充信息行
"""
target_desc = (
f"群组 {schedule_info.group_id}"
if schedule_info.group_id
and schedule_info.group_id != scheduler_manager.ALL_GROUPS
else "所有群组"
if schedule_info.group_id == scheduler_manager.ALL_GROUPS
else "全局"
target_desc = format_target_info(
schedule_info.target_type, schedule_info.target_identifier
)
info_lines = [
@@ -128,26 +112,22 @@ def _format_params(schedule_status: dict) -> str:
async def format_schedule_list_as_image(
schedules: list[ScheduledJob], title: str, current_page: int
schedules: list[ScheduledJob], title: str, current_page: int, total_items: int
):
"""将任务列表格式化为图片"""
page_size = 15
total_items = len(schedules)
page_size = 30
total_pages = (total_items + page_size - 1) // page_size
start_index = (current_page - 1) * page_size
end_index = start_index + page_size
paginated_schedules = schedules[start_index:end_index]
if not paginated_schedules:
if not schedules:
return "这一页没有内容了哦~"
status_tasks = [
scheduler_manager.get_schedule_status(s.id) for s in paginated_schedules
]
all_statuses = await asyncio.gather(*status_tasks)
schedule_ids = [s.id for s in schedules]
all_statuses_list = await scheduler_manager.get_schedules_status_bulk(schedule_ids)
all_statuses_map = {status["id"]: status for status in all_statuses_list}
data_list = []
for s in all_statuses:
for schedule_db in schedules:
s = all_statuses_map.get(schedule_db.id)
if not s:
continue
@@ -166,7 +146,9 @@ async def format_schedule_list_as_image(
TextCell(content=str(s["id"])),
TextCell(content=s["plugin_name"]),
TextCell(content=s.get("bot_id") or "N/A"),
TextCell(content=s["group_id"] or "全局"),
TextCell(
content=format_target_info(s["target_type"], s["target_identifier"])
),
TextCell(content=s["next_run_time"]),
TextCell(content=_format_trigger_info(s)),
TextCell(content=_format_params(s)),
@@ -190,17 +172,35 @@ async def format_schedule_list_as_image(
)
def format_target_info(target_type: str, target_identifier: str) -> str:
"""格式化目标信息以供显示"""
if target_type == "GLOBAL":
return "全局"
elif target_type == "ALL_GROUPS":
return "所有群组"
elif target_type == "TAG":
return f"标签: {target_identifier}"
elif target_type == "GROUP":
return f"群: {target_identifier}"
elif target_type == "USER":
return f"用户: {target_identifier}"
else:
return f"{target_type}: {target_identifier}"
def format_single_status_message(status: dict) -> str:
"""格式化单个任务状态为文本消息"""
target_info = format_target_info(status["target_type"], status["target_identifier"])
trigger_info = status.get("trigger_info_str", _format_trigger_info(status))
info_lines = [
f"📋 定时任务详细信息 (ID: {status['id']})",
"--------------------",
f"▫️ 插件: {status['plugin_name']}",
f"▫️ Bot ID: {status.get('bot_id') or '默认'}",
f"▫️ 目标: {status['group_id'] or '全局'}",
f"▫️ 目标: {target_info}",
f"▫️ 状态: {'✔️ 已启用' if status['is_enabled'] else '⏸️ 已暂停'}",
f"▫️ 下次运行: {status['next_run_time']}",
f"▫️ 触发规则: {_format_trigger_info(status)}",
f"▫️ 触发规则: {trigger_info}",
f"▫️ 任务参数: {_format_params(status)}",
]
return "\n".join(info_lines)
zhenxun/builtin_plugins/superuser/tag_manage.py
+399
View File
@@ -0,0 +1,399 @@
from nonebot.adapters import Bot
from nonebot.permission import SUPERUSER
from nonebot.plugin import PluginMetadata
from nonebot_plugin_alconna import (
Alconna,
AlconnaMatch,
AlconnaQuery,
Args,
Match,
MultiVar,
Option,
Query,
Subcommand,
on_alconna,
store_true,
)
from nonebot_plugin_waiter import prompt_until
from tortoise.exceptions import IntegrityError
from zhenxun.configs.utils import PluginExtraData
from zhenxun.services.tags import tag_manager
from zhenxun.utils.enum import PluginType
from zhenxun.utils.message import MessageUtils
__plugin_meta__ = PluginMetadata(
name="群组标签管理",
description="用于管理和操作群组标签",
usage="""### 🏷️ 群组标签管理
用于创建和管理群组标签，以实现对群组的批量操作和筛选。
---
#### **✨ 核心命令**
- **`tag list`** (别名: `ls`)
- 查看所有标签及其基本信息。
- **`tag info <标签名>`**
- 查看指定标签的详细信息，包括关联群组或动态规则的匹配结果。
- **`tag create <标签名> [选项...]`**
- 创建一个新标签。
- **选项**:
- `--type <static|dynamic>`: 标签类型，默认为 `static`。
- `static`: 静态标签，需手动关联群组。
- `dynamic`: 动态标签，根据规则自动匹配。
- `-g <群号...>`: **(静态)** 初始关联的群组ID。
- `--rule "<规则>"`: **(动态)** 定义动态规则，**规则必须用引号包裹**。
- `--desc "<描述>"`: 为标签添加描述。
- `--blacklist`: **(静态)** 将标签设为黑名单（排除）模式。
- **`tag edit <标签名> [操作...]`**
- 编辑一个已存在的标签。
- **通用操作**:
- `--rename <新名>`: 重命名标签。
- `--desc "<描述>"`: 更新描述。
- `--mode <white|black>`: 切换为白名单/黑名单模式。
- **静态标签操作**:
- `--add <群号...>`: 添加群组。
- `--remove <群号...>`: 移除群组。
- `--set <群号...>`: **[覆盖]** 重新设置所有关联群组。
- **动态标签操作**:
- `--rule "<新规则>"`: 更新动态规则。
- **`tag delete <名1> [名2] ...`**
- 删除一个或多个标签。
- **`tag clear`**
- **[⚠️ 危险]** 删除所有标签，操作前会请求确认。
---
#### **🔧 动态规则速查**
规则支持 `and` 和 `or` 组合（`and` 优先）。
**包含空格或特殊字符的规则值建议用英文引号包裹**。
- `member_count > 100`
按 **群成员数** 筛选 (`>`, `>=`, `<`, `<=`, `=`)。
- `level >= 5`
按 **群权限等级** 筛选。
- `status = true`
按 **群是否休眠** 筛选 (`true` / `false`)。
- `is_super = false`
按 **群是否为白名单** 筛选 (`true` / `false`)。
- `group_name contains "模式"`
按 **群名模糊/正则匹配**。
例: `contains "测试.*群$"` 匹配以“测试”开头、“群”结尾的群名。
- `group_name in "群1,群2"`
按 **群名多值精确匹配** (英文逗号分隔)。
---
#### **💡 使用示例**
##### 静态标签示例
```bash
# 创建一个名为“核心群”的静态标签，并关联两个群组
tag create 核心群 -g 12345 67890 --desc "核心业务群"
# 向“核心群”中添加一个新群组
tag edit 核心群 --add 98765
# 创建一个用于排除的黑名单标签
tag create 排除群 --blacklist -g 11111
```
##### 动态标签示例
```bash
# 创建一个动态标签，匹配所有成员数大于200的群
tag create 大群 --type dynamic --rule "member_count > 200"
# 创建一个匹配高权限且未休眠的群的标签
tag create 活跃管理群 --type dynamic --rule "level > 5 and status = true"
# 创建一个匹配群名包含“核心”或“测试”的标签
tag create 业务群 --type dynamic --rule "group_name contains 核心 or group_name contains 测试"
```
""".strip(), # noqa: E501
extra=PluginExtraData(
author="HibiKier",
version="1.0.0",
plugin_type=PluginType.SUPERUSER,
).to_dict(),
)
tag_cmd = on_alconna(
Alconna(
"tag",
Subcommand("list", alias=["ls"], help_text="查看所有标签"),
Subcommand("info", Args["name", str], help_text="查看标签详情"),
Subcommand(
"create",
Args["name", str],
Option(
"--rule",
Args["rule", str],
help_text="动态标签规则 (例如: min_members=100)",
),
Option(
"--type",
Args["tag_type", ["static", "dynamic"]],
help_text="标签类型 (默认: static)",
),
Option(
"--blacklist", action=store_true, help_text="设为黑名单模式(仅静态标签)"
),
Option("--desc", Args["description", str], help_text="标签描述"),
Option(
"-g", Args["group_ids", MultiVar(str)], help_text="创建时要关联的群组ID"
),
),
Subcommand(
"edit",
Args["name", str],
Option(
"--rule",
Args["rule", str],
help_text="更新动态标签规则",
),
Option("--add", Args["add_groups", MultiVar(str)]),
Option("--remove", Args["remove_groups", MultiVar(str)]),
Option("--set", Args["set_groups", MultiVar(str)]),
Option("--rename", Args["new_name", str]),
Option("--desc", Args["description", str]),
Option("--mode", Args["mode", ["black", "white"]]),
help_text="编辑标签",
),
Subcommand(
"delete",
Args["names", MultiVar(str)],
alias=["del", "rm"],
help_text="删除标签",
),
Subcommand("clear", help_text="清空所有标签"),
),
permission=SUPERUSER,
priority=5,
block=True,
)
@tag_cmd.assign("list")
async def handle_list():
tags = await tag_manager.list_tags_with_counts()
if not tags:
await MessageUtils.build_message("当前没有已创建的标签。").finish()
msg = "已创建的群组标签:\n"
for tag in tags:
mode = "黑名单(排除)" if tag["is_blacklist"] else "白名单(包含)"
tag_type = "动态" if tag["tag_type"] == "DYNAMIC" else "静态"
count_desc = (
f"含 {tag['group_count']} 个群组" if tag_type == "静态" else "动态计算"
)
msg += f"- {tag['name']} (类型: {tag_type}, 模式: {mode}): {count_desc}\n"
await MessageUtils.build_message(msg).finish()
@tag_cmd.assign("info")
async def handle_info(name: Match[str], bot: Bot):
details = await tag_manager.get_tag_details(name.result, bot=bot)
if not details:
await MessageUtils.build_message(f"标签 '{name.result}' 不存在。").finish()
mode = "黑名单(排除)" if details["is_blacklist"] else "白名单(包含)"
tag_type_str = "动态" if details["tag_type"] == "DYNAMIC" else "静态"
msg = f"标签详情: {details['name']}\n"
msg += f"类型: {tag_type_str}\n"
msg += f"模式: {mode}\n"
msg += f"描述: {details['description'] or '无'}\n"
if details["tag_type"] == "DYNAMIC":
msg += f"动态规则: {details['dynamic_rule']}\n"
if details["resolved_groups"] is not None:
msg += f"当前匹配群组 ({len(details['resolved_groups'])}个):\n"
if details["resolved_groups"]:
msg += "\n".join(
f"- {g_name} ({g_id})"
for g_id, g_name in details["resolved_groups"]
)
else:
msg += "无"
else:
msg += f"关联群组 ({len(details['groups'])}个):\n"
if details["groups"]:
msg += "\n".join(f"- {gid}" for gid in details["groups"])
else:
msg += "无"
await MessageUtils.build_message(msg).finish()
@tag_cmd.assign("create")
async def handle_create(
name: Match[str],
description: Match[str],
group_ids: Match[list[str]],
rule: Match[str] = AlconnaMatch("rule"),
tag_type: Match[str] = AlconnaMatch("tag_type"),
blacklist: Query[bool] = AlconnaQuery("create.blacklist.value", False),
):
ttype = (
tag_type.result.upper()
if tag_type.available
else ("DYNAMIC" if rule.available else "STATIC")
)
if ttype == "DYNAMIC" and not rule.available:
await MessageUtils.build_message(
"创建失败: 动态标签必须提供至少一个规则。"
).finish()
try:
tag = await tag_manager.create_tag(
name=name.result,
is_blacklist=blacklist.result,
description=description.result if description.available else None,
group_ids=group_ids.result if group_ids.available else None,
tag_type=ttype,
dynamic_rule=rule.result if rule.available else None,
)
msg = f"标签 '{tag.name}' 创建成功！"
if group_ids.available:
msg += f"\n已同时关联 {len(group_ids.result)} 个群组。"
await MessageUtils.build_message(msg).finish()
except IntegrityError:
await MessageUtils.build_message(
f"创建失败: 标签 '{name.result}' 已存在。"
).finish()
except ValueError as e:
await MessageUtils.build_message(f"创建失败: {e}").finish()
@tag_cmd.assign("edit")
async def handle_edit(
name: Match[str],
add_groups: Match[list[str]],
remove_groups: Match[list[str]],
set_groups: Match[list[str]],
new_name: Match[str],
description: Match[str],
mode: Match[str],
rule: Match[str] = AlconnaMatch("rule"),
):
tag_name = name.result
tag_details = await tag_manager.get_tag_details(tag_name)
if not tag_details:
await MessageUtils.build_message(f"标签 '{tag_name}' 不存在。").finish()
group_actions = [
add_groups.available,
remove_groups.available,
set_groups.available,
]
if sum(group_actions) > 1:
await MessageUtils.build_message(
"`--add`, `--remove`, `--set` 选项不能同时使用。"
).finish()
is_dynamic = tag_details.get("tag_type") == "DYNAMIC"
if is_dynamic and any(group_actions):
await MessageUtils.build_message(
"编辑失败: 不能对动态标签执行 --add, --remove, 或 --set 操作。"
).finish()
if not is_dynamic and rule.available:
await MessageUtils.build_message(
"编辑失败: 不能为静态标签设置动态规则。"
).finish()
results = []
try:
rule_str = rule.result if rule.available else None
if add_groups.available:
count = await tag_manager.add_groups_to_tag(tag_name, add_groups.result)
results.append(f"添加了 {count} 个群组。")
if remove_groups.available:
count = await tag_manager.remove_groups_from_tag(
tag_name, remove_groups.result
)
results.append(f"移除了 {count} 个群组。")
if set_groups.available:
count = await tag_manager.set_groups_for_tag(tag_name, set_groups.result)
results.append(f"关联群组已覆盖为 {count} 个。")
if description.available or mode.available or rule_str is not None:
is_blacklist = None
if mode.available:
is_blacklist = mode.result == "black"
await tag_manager.update_tag_attributes(
tag_name,
description.result if description.available else None,
is_blacklist,
rule_str,
)
if rule_str is not None:
results.append(f"动态规则已更新为 '{rule_str}'。")
if description.available:
results.append("描述已更新。")
if mode.available:
results.append(
f"模式已更新为 {'黑名单' if is_blacklist else '白名单'}。"
)
if new_name.available:
await tag_manager.rename_tag(tag_name, new_name.result)
results.append(f"已重命名为 '{new_name.result}'。")
tag_name = new_name.result
except (ValueError, IntegrityError) as e:
await MessageUtils.build_message(f"操作失败: {e}").finish()
if not results:
await MessageUtils.build_message(
"未执行任何操作，请提供至少一个编辑选项。"
).finish()
final_msg = f"对标签 '{tag_name}' 的操作已完成：\n" + "\n".join(
f"- {r}" for r in results
)
await MessageUtils.build_message(final_msg).finish()
@tag_cmd.assign("delete")
async def handle_delete(names: Match[list[str]]):
success, failed = [], []
for name in names.result:
if await tag_manager.delete_tag(name):
success.append(name)
else:
failed.append(name)
msg = ""
if success:
msg += f"成功删除标签: {', '.join(success)}\n"
if failed:
msg += f"标签不存在，删除失败: {', '.join(failed)}"
await MessageUtils.build_message(msg.strip()).finish()
@tag_cmd.assign("clear")
async def handle_clear():
confirm = await prompt_until(
"【警告】此操作将删除所有群组标签，是否继续？\n请输入 `是` 或 `确定` 确认操作",
lambda msg: msg.extract_plain_text().lower()
in ["是", "确定", "yes", "confirm"],
timeout=30,
retry=1,
)
if confirm:
count = await tag_manager.clear_all_tags()
await MessageUtils.build_message(f"操作完成，已清空 {count} 个标签。").finish()
else:
await MessageUtils.build_message("操作已取消。").finish()