mirror of
https://github.com/zhenxun-org/zhenxun_bot.git
synced 2025-12-15 14:22:55 +08:00
299 lines
11 KiB
Python
299 lines
11 KiB
Python
|
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 nonebot_plugin_alconna import (
|
|||
|
|
Alconna,
|
|||
|
|
AlconnaMatch,
|
|||
|
|
Args,
|
|||
|
|
Match,
|
|||
|
|
Option,
|
|||
|
|
Query,
|
|||
|
|
Subcommand,
|
|||
|
|
on_alconna,
|
|||
|
|
)
|
|||
|
|
|
|||
|
|
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
|
|||
|
|
|
|||
|
|
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="按插件名筛选"),
|
|||
|
|
Option("--page", Args["page", int, 1], help_text="指定页码"),
|
|||
|
|
alias=["ls", "list"],
|
|||
|
|
help_text="查看定时任务",
|
|||
|
|
),
|
|||
|
|
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="设置特定执行日期"),
|
|||
|
|
Option(
|
|||
|
|
"--daily",
|
|||
|
|
Args["daily_expr", str],
|
|||
|
|
help_text="设置每天执行的时间 (如 08:20)",
|
|||
|
|
),
|
|||
|
|
Option("-g", Args["group_id", str], help_text="指定群组ID或'all'"),
|
|||
|
|
Option("-all", help_text="对所有群生效 (等同于 -g all)"),
|
|||
|
|
Option("--kwargs", Args["kwargs_str", str], help_text="设置任务参数"),
|
|||
|
|
Option(
|
|||
|
|
"--bot", Args["bot_id", str], help_text="指定操作的Bot ID (SUPERUSER)"
|
|||
|
|
),
|
|||
|
|
alias=["add", "开启"],
|
|||
|
|
help_text="设置/开启一个定时任务",
|
|||
|
|
),
|
|||
|
|
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)"
|
|||
|
|
),
|
|||
|
|
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)"
|
|||
|
|
),
|
|||
|
|
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)"
|
|||
|
|
),
|
|||
|
|
alias=["resume"],
|
|||
|
|
help_text="恢复一个或多个定时任务",
|
|||
|
|
),
|
|||
|
|
Subcommand(
|
|||
|
|
"执行",
|
|||
|
|
Args["schedule_id", int],
|
|||
|
|
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)",
|
|||
|
|
),
|
|||
|
|
Option("--kwargs", Args["kwargs_str", str], help_text="更新参数"),
|
|||
|
|
alias=["update", "modify", "修改"],
|
|||
|
|
help_text="更新任务配置",
|
|||
|
|
),
|
|||
|
|
Subcommand(
|
|||
|
|
"状态",
|
|||
|
|
Args["schedule_id", int],
|
|||
|
|
alias=["status", "info"],
|
|||
|
|
help_text="查看单个任务的详细状态",
|
|||
|
|
),
|
|||
|
|
Subcommand(
|
|||
|
|
"插件列表",
|
|||
|
|
alias=["plugins"],
|
|||
|
|
help_text="列出所有可用的插件",
|
|||
|
|
),
|
|||
|
|
),
|
|||
|
|
priority=5,
|
|||
|
|
block=True,
|
|||
|
|
rule=admin_check(1),
|
|||
|
|
)
|
|||
|
|
|
|||
|
|
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)
|