Compare commits

...
Author SHA1 Message Date
pre-commit-ci[bot] 5a5e9ae5c7 🚨 auto fix by pre-commit hooks 2025-11-26 02:09:25 +00:00
webjoin111 a327f1adaa 🐛 fix(codeql): 移除对 JavaScript 和 TypeScript 的分析支持 2025-11-26 10:08:02 +08:00
webjoin111 ddbe75704d 🐛 fix(gemini): 增加响应验证以处理内容过滤(promptFeedback) 2025-11-16 00:52:55 +08:00
webjoin111 50c04f3625 ✨ feat(tag): 实现群组标签自动清理及手动清理功能 2025-11-15 12:43:16 +08:00
webjoin111 c36c161b0e ✨ feat(renderer): 支持组件变体样式收集 2025-11-13 17:28:07 +08:00
webjoin111 b4df0c0fe9 ✨ feat(broadcast): 实现标签定向广播、强制发送及并发控制
- 【新功能】
  - 新增标签定向广播功能,支持通过 `-t <标签名>` 或 `广播到 <标签名>` 命令向指定标签的群组发送消息
  - 引入广播强制发送模式,允许绕过群组的任务阻断设置
  - 实现广播并发控制,通过配置限制同时发送任务数量,避免API速率限制
  - 优化视频消息处理,支持从URL下载视频内容并作为原始数据发送,提高跨平台兼容性
- 【配置】
  - 添加 `DEFAULT_BROADCAST` 配置项,用于设置群组进群时广播功能的默认开关状态
  - 添加 `BROADCAST_CONCURRENCY_LIMIT` 配置项,用于控制广播时的最大并发任务数
2025-11-12 16:12:37 +08:00
webjoin111 412654d165 ✨ feat(tag): 添加标签克隆功能
- 新增 `tag clone <源标签名> <新标签名>` 命令,用于复制现有标签。
- 【优化】在 `tag create`, `tag edit --add`, `tag edit --set` 命令中,自动去重传入的群组ID,避免重复关联。
2025-11-12 16:07:59 +08:00
webjoin111 2581e335af ✨ feat(renderer): 添加 Jinja2 inline_asset 全局函数
- 新增 `RendererService._inline_asset_global` 方法,并注册为 Jinja2 全局函数 `inline_asset`。
- 允许模板通过 `{{ inline_asset('@namespace/path/to/asset.svg') }}` 直接内联已注册命名空间下的资源文件内容。
- 主要用于解决内联 SVG 时可能遇到的跨域安全问题。
- 【重构】优化 `ResourceResolver.resolve_asset_uri` 中对命名空间资源 (以 `@` 开头) 的解析逻辑,确保能够正确获取文件绝对路径并返回 URI。
- 改进 `RenderableComponent.get_extra_css`,使其在组件定义 `component_css` 时自动返回该 CSS 内容。
- 清理 `Renderable` 协议和 `RenderableComponent` 基类中已存在方法的 `[新增]` 标记。
2025-11-12 16:07:58 +08:00
webjoin111 3ee0f6f2b1 ✨ feat(llm): 增强 LLM 管理功能,支持纯文本列表输出,优化模型能力识别并新增提供商
- 【LLM 管理器】为 `llm list` 命令添加 `--text` 选项,支持以纯文本格式输出模型列表。
- 【LLM 配置】新增 `OpenRouter` LLM 提供商的默认配置。
- 【模型能力】增强 `get_model_capabilities` 函数的查找逻辑,支持模型名称分段匹配和更灵活的通配符匹配。
- 【模型能力】为 `Gemini` 模型能力注册表使用更通用的通配符模式。
- 【模型能力】新增 `GPT` 系列模型的详细能力定义,包括多模态输入输出和工具调用支持。
2025-11-12 16:07:58 +08:00
webjoin111 0bd1d6c59c ⚡️ perf(image_utils): 优化图片哈希获取避免阻塞异步 2025-11-12 16:07:58 +08:00
ThelevenFDandHibiKier c839b44256 转换specify_probability为float 增加鉴权配置 (#2067)
检查bot是否运行正常 / bot check (push) Has been cancelled
CodeQL Code Security Analysis / Analyze (${{ matrix.language }}) (none, javascript-typescript) (push) Has been cancelled
CodeQL Code Security Analysis / Analyze (${{ matrix.language }}) (none, python) (push) Has been cancelled
Sequential Lint and Type Check / ruff-call (push) Has been cancelled
Release Drafter / Update Release Draft (push) Has been cancelled
Force Sync to Aliyun / sync (push) Has been cancelled
Update Version / update-version (push) Has been cancelled
Sequential Lint and Type Check / pyright-call (push) Has been cancelled
* 转换specify_probability为float

* 解决#2045 添加密钥配置

* Revise access token configuration in .env.example

Updated comments and modified access token configuration.

---------

Co-authored-by: HibiKier <45528451+HibiKier@users.noreply.github.com>
2025-11-03 16:36:43 +08:00
70bde00757 ✨ feat(core): 增强定时任务与群组标签管理,重构调度核心 (#2068)
* ✨ feat(core): 更新群组信息、Markdown 样式与 Pydantic 兼容层

- 【group】添加更新所有群组信息指令,并同步群组控制台数据
- 【markdown】支持合并 Markdown 的 CSS 来源
- 【pydantic-compat】提供 model_validate 兼容函数

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

✨ 新功能

* **标签 (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` 解耦业务逻辑与依赖注入。
    * 适配新的任务目标类型展示。

* 🐛 fix(tag): 修复黑名单标签解析逻辑并优化标签详情展示

* ✨ feat(scheduler): 为多目标定时任务添加固定间隔串行执行选项

* ✨ feat(schedulerAdmin): 允许定时任务删除、暂停、恢复命令支持多ID操作

* 🚨 auto fix by pre-commit hooks

---------

Co-authored-by: webjoin111 <455457521@qq.com>
Co-authored-by: pre-commit-ci[bot] <66853113+pre-commit-ci[bot]@users.noreply.github.com>
2025-11-03 10:53:40 +08:00
molanp eb6d90ae88 docs(data-source): 更新插件安装函数的参数文档说明 (#2069)
检查bot是否运行正常 / bot check (push) Has been cancelled
CodeQL Code Security Analysis / Analyze (${{ matrix.language }}) (none, javascript-typescript) (push) Has been cancelled
CodeQL Code Security Analysis / Analyze (${{ matrix.language }}) (none, python) (push) Has been cancelled
Sequential Lint and Type Check / ruff-call (push) Has been cancelled
Release Drafter / Update Release Draft (push) Has been cancelled
Force Sync to Aliyun / sync (push) Has been cancelled
Update Version / update-version (push) Has been cancelled
Sequential Lint and Type Check / pyright-call (push) Has been cancelled
修改 StoreManager 类中安装插件函数的文档字符串,更新参数列表说
明。将原有的 github_url、module_path、is_dir 参数说明替换为
plugin_info 和 source 参数说明,保持文档与实际函数签名一致。
2025-10-22 20:57:07 +08:00
molanp 4b8013d2d6 Feat: Add spaces (#2064)
检查bot是否运行正常 / bot check (push) Has been cancelled
CodeQL Code Security Analysis / Analyze (${{ matrix.language }}) (none, javascript-typescript) (push) Has been cancelled
CodeQL Code Security Analysis / Analyze (${{ matrix.language }}) (none, python) (push) Has been cancelled
Sequential Lint and Type Check / ruff-call (push) Has been cancelled
Release Drafter / Update Release Draft (push) Has been cancelled
Force Sync to Aliyun / sync (push) Has been cancelled
Update Version / update-version (push) Has been cancelled
Sequential Lint and Type Check / pyright-call (push) Has been cancelled
2025-10-17 09:22:18 +08:00
HibiKier d528711641 🐛 fix(http_utils): 增强错误处理,记录请求失败的详细信息 (#2065)
检查bot是否运行正常 / bot check (push) Waiting to run
CodeQL Code Security Analysis / Analyze (${{ matrix.language }}) (none, javascript-typescript) (push) Waiting to run
CodeQL Code Security Analysis / Analyze (${{ matrix.language }}) (none, python) (push) Waiting to run
Sequential Lint and Type Check / ruff-call (push) Waiting to run
Sequential Lint and Type Check / pyright-call (push) Blocked by required conditions
Release Drafter / Update Release Draft (push) Waiting to run
Force Sync to Aliyun / sync (push) Waiting to run
Update Version / update-version (push) Waiting to run
* 🐛 fix(http_utils): 增强错误处理,记录请求失败的详细信息

* 🐛 fix(http_utils): 改进HTTP错误处理,记录请求失败的状态码和响应内容
2025-10-16 17:31:08 +08:00
molanp 1cc18bb195 fix(shop): 修改道具不存在时的提示信息 (#2061)
检查bot是否运行正常 / bot check (push) Has been cancelled
CodeQL Code Security Analysis / Analyze (${{ matrix.language }}) (none, javascript-typescript) (push) Has been cancelled
CodeQL Code Security Analysis / Analyze (${{ matrix.language }}) (none, python) (push) Has been cancelled
Sequential Lint and Type Check / ruff-call (push) Has been cancelled
Release Drafter / Update Release Draft (push) Has been cancelled
Force Sync to Aliyun / sync (push) Has been cancelled
Update Version / update-version (push) Has been cancelled
Sequential Lint and Type Check / pyright-call (push) Has been cancelled
- 将道具不存在时的提示信息从具体的道具名称改为通用提示,避免暴露内部实现细节,
提升用户体验和安全性。
- resolve Bug: 使用道具功能优化
Fixes #2060
2025-10-09 09:01:20 +08:00
Rumioandwebjoin111 74a9f3a843 ✨ feat(core): 支持LLM多图片响应,增强UI主题皮肤系统及优化JSON/Markdown处理 (#2062)
- 【LLM服务】
  - `LLMResponse` 模型现在支持 `images: list[bytes]`,允许模型返回多张图片。
  - LLM适配器 (`base.py`, `gemini.py`) 和 API 层 (`api.py`, `service.py`) 已更新以处理多图片响应。
  - 响应验证逻辑已调整,以检查 `images` 列表而非单个 `image_bytes`。
- 【UI渲染服务】
  - 引入组件“皮肤”(variant)概念,允许为同一组件提供不同视觉风格。
  - 改进了 `manifest.json` 的加载、合并和缓存机制,支持基础清单与皮肤清单的递归合并。
  - `ThemeManager` 现在会缓存已加载的清单,并在主题重载时清除缓存。
  - 增强了资源解析器 (`ResourceResolver`),支持 `@` 命名空间路径和更健壮的相对路径处理。
  - 独立模板现在会继承主 Jinja 环境的过滤器。
- 【工具函数】
  - 引入 `dump_json_safely` 工具函数,用于更安全地序列化包含 Pydantic 模型、枚举等复杂类型的对象为 JSON。
  - LLM 服务中的请求体和缓存键生成已改用 `dump_json_safely`。
  - 优化了 `format_usage_for_markdown` 函数,改进了 Markdown 文本的格式化,确保块级元素前有正确换行,并正确处理段落内硬换行。

Co-authored-by: webjoin111 <455457521@qq.com>
2025-10-09 08:50:40 +08:00
HibiKierandpre-commit-ci[bot] e7f3c210df 修复并发时数据库超时 (#2063)
* 🔧 修复和优化:调整超时设置,重构检查逻辑,简化代码结构

- 在 `chkdsk_hook.py` 中重构 `check` 方法,提取公共逻辑
- 更新 `CacheManager` 中的超时设置,使用新的 `CACHE_TIMEOUT`
- 在 `utils.py` 中添加缓存逻辑,记录数据库操作的执行情况

* ✨ feat(auth): 添加并发控制,优化权限检查逻辑

* Update utils.py

* 🚨 auto fix by pre-commit hooks

---------

Co-authored-by: pre-commit-ci[bot] <66853113+pre-commit-ci[bot]@users.noreply.github.com>
2025-10-09 08:46:08 +08:00
molanp f94121080f fix(check): 修复自检插件在ARM设备下的CPU频率获取逻辑 (#2057)
检查bot是否运行正常 / bot check (push) Has been cancelled
CodeQL Code Security Analysis / Analyze (${{ matrix.language }}) (none, javascript-typescript) (push) Has been cancelled
CodeQL Code Security Analysis / Analyze (${{ matrix.language }}) (none, python) (push) Has been cancelled
Sequential Lint and Type Check / ruff-call (push) Has been cancelled
Release Drafter / Update Release Draft (push) Has been cancelled
Force Sync to Aliyun / sync (push) Has been cancelled
Update Version / update-version (push) Has been cancelled
Sequential Lint and Type Check / pyright-call (push) Has been cancelled
- 将插件版本从0.1更新至0.2
- 新增安全获取ARM设备CPU频率的函数get_arm_cpu_freq_safe
- 优化CPU信息采集逻辑,提高在ARM架构下的兼容性
2025-10-01 18:42:47 +08:00
HibiKier 761c8daac4 ✨ feat(configs): 优化 ConfigsManager 中的键值获取逻辑,确保未定义键时自动创建 ConfigGroup 实例 (#2058) 2025-10-01 18:42:19 +08:00
Rumioandwebjoin111 c667fc215e ✨ feat(llm): 增强LLM服务,支持图片生成、响应验证与OpenRouter集成 (#2054)
* ✨ feat(llm): 增强LLM服务,支持图片生成、响应验证与OpenRouter集成

- 【新功能】统一图片生成与编辑API `create_image`,支持文生图、图生图及多图输入
- 【新功能】引入LLM响应验证机制,通过 `validation_policy` 和 `response_validator` 确保响应内容符合预期,例如强制返回图片
- 【新功能】适配OpenRouter API,扩展LLM服务提供商支持,并添加OpenRouter特定请求头
- 【重构】将日志净化逻辑重构至 `log_sanitizer` 模块,提供统一的净化入口,并应用于NoneBot消息、LLM请求/响应日志
- 【修复】优化Gemini适配器,正确解析图片生成响应中的Base64图片数据,并更新模型能力注册表

* ✨ feat(image): 优化图片生成响应并返回完整LLMResponse

* ✨ feat(llm): 为 OpenAI 兼容请求体添加日志净化

* 🐛 fix(ui): 截断UI调试HTML日志中的长base64图片数据

---------

Co-authored-by: webjoin111 <455457521@qq.com>
2025-10-01 18:41:46 +08:00
76 changed files with 5012 additions and 1808 deletions
+5 -1
View File
@@ -10,6 +10,9 @@ SESSION_EXPIRE_TIMEOUT=00:00:30
ALCONNA_USE_COMMAND_START=True
# ws连接密钥,若bot能被公网访问则建议打开该注释并设置该配置项
# ONEBOT_ACCESS_TOKEN=""
# 全局图片统一使用bytes发送,当真寻与协议端不在同一服务器上时为True
IMAGE_TO_BYTES = True
@@ -29,6 +32,7 @@ DB_URL = ""
# NONE: 不使用缓存, MEMORY: 使用内存缓存, REDIS: 使用Redis缓存
CACHE_MODE = NONE
# REDIS配置,使用REDIS替换Cache内存缓存
# REDIS地址
# REDIS_HOST = "127.0.0.1"
@@ -86,4 +90,4 @@ PORT = 8080
# '
# application_commands的{"*": ["*"]}代表将全部应用命令注册为全局应用命令
# {"admin": ["123", "456"]}则代表将admin命令注册为id是123、456服务器的局部命令,其余命令不注册
# {"admin": ["123", "456"]}则代表将admin命令注册为id是123、456服务器的局部命令,其余命令不注册
-3
View File
@@ -45,12 +45,9 @@ jobs:
include:
- language: python
build-mode: none
- language: javascript-typescript
build-mode: none
# CodeQL supports the following values keywords for 'language': 'c-cpp', 'csharp', 'go', 'java-kotlin', 'javascript-typescript', 'python', 'ruby', 'swift'
# Use `c-cpp` to analyze code written in C, C++ or both
# Use 'java-kotlin' to analyze code written in Java, Kotlin or both
# Use 'javascript-typescript' to analyze code written in JavaScript, TypeScript or both
# To learn more about changing the languages that are analyzed or customizing the build mode for your analysis,
# see https://docs.github.com/en/code-security/code-scanning/creating-an-advanced-setup-for-code-scanning/customizing-your-advanced-setup-for-code-scanning.
# If you are analyzing a compiled language, you can modify the 'build-mode' for that language to customize how
@@ -1,7 +1,11 @@
import asyncio
import random
import nonebot
from nonebot import on_notice
from nonebot.adapters import Bot
from nonebot.adapters.onebot.v11 import GroupIncreaseNoticeEvent
from nonebot.permission import SUPERUSER
from nonebot.plugin import PluginMetadata
from nonebot_plugin_alconna import Alconna, Arparma, on_alconna
from nonebot_plugin_apscheduler import scheduler
@@ -10,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
@@ -45,12 +50,79 @@ _matcher = on_alconna(
_notice = on_notice(priority=1, block=False, rule=notice_rule(GroupIncreaseNoticeEvent))
_update_all_matcher = on_alconna(
Alconna("更新所有群组信息"),
permission=SUPERUSER,
priority=1,
block=True,
)
async def _update_all_groups_task(bot: Bot, session: EventSession):
"""
在后台执行所有群组的更新任务,并向超级用户发送最终报告。
"""
success_count = 0
fail_count = 0
total_count = 0
bot_id = bot.self_id
logger.info(f"Bot {bot_id}: 开始执行所有群组信息更新任务...", "更新所有群组")
try:
group_list, _ = await PlatformUtils.get_group_list(bot)
total_count = len(group_list)
for i, group in enumerate(group_list):
try:
logger.debug(
f"Bot {bot_id}: 正在更新第 {i + 1}/{total_count} 个群组: "
f"{group.group_id}",
"更新所有群组",
)
await MemberUpdateManage.update_group_member(bot, group.group_id)
success_count += 1
except Exception as e:
fail_count += 1
logger.error(
f"Bot {bot_id}: 更新群组 {group.group_id} 信息失败",
"更新所有群组",
e=e,
)
await asyncio.sleep(random.uniform(1.5, 3.0))
except Exception as e:
logger.error(f"Bot {bot_id}: 获取群组列表失败,任务中断", "更新所有群组", e=e)
await PlatformUtils.send_superuser(
bot,
f"Bot {bot_id} 更新所有群组信息任务失败:无法获取群组列表。",
session.id1,
)
return
await tag_manager._invalidate_cache()
summary_message = (
f"🤖 Bot {bot_id} 所有群组信息更新任务完成!\n"
f"总计群组: {total_count}\n"
f"✅ 成功: {success_count}\n"
f"❌ 失败: {fail_count}"
)
logger.info(summary_message.replace("\n", " | "), "更新所有群组")
await PlatformUtils.send_superuser(bot, summary_message, session.id1)
@_update_all_matcher.handle()
async def _(bot: Bot, session: EventSession):
await MessageUtils.build_message(
"已开始在后台更新所有群组信息,过程可能需要几分钟到几十分钟,完成后将私聊通知您。"
).send(reply_to=True)
asyncio.create_task(_update_all_groups_task(bot, session)) # noqa: RUF006
@_matcher.handle()
async def _(bot: Bot, session: EventSession, arparma: Arparma):
if gid := session.id3 or session.id2:
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()
@@ -64,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(
@@ -91,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()
@@ -6,6 +6,7 @@ from nonebot.adapters import Bot
from nonebot_plugin_uninfo import Member, SceneType, get_interface
from zhenxun.configs.config import Config
from zhenxun.models.group_console import GroupConsole
from zhenxun.models.group_member_info import GroupInfoUser
from zhenxun.models.level_user import LevelUser
from zhenxun.services.log import logger
@@ -94,6 +95,25 @@ class MemberUpdateManage:
)
return "更新群组失败,群组不存在..."
members = await interface.get_members(SceneType.GROUP, group_list[0].id)
try:
group_console, _ = await GroupConsole.get_or_create(
group_id=group_id, defaults={"platform": platform}
)
group_console.member_count = len(members)
group_console.group_name = group_list[0].name or ""
await group_console.save(update_fields=["member_count", "group_name"])
logger.debug(
f"已更新群组 {group_id} 的成员总数为 {len(members)}",
"更新群组成员信息",
)
except Exception as e:
logger.error(
f"更新群组 {group_id} 的 GroupConsole 信息失败",
"更新群组成员信息",
e=e,
)
db_user = await GroupInfoUser.filter(group_id=group_id).all()
db_user_uid = [u.user_id for u in db_user]
data_list = ([], [], [])
+1 -1
View File
@@ -26,7 +26,7 @@ __plugin_meta__ = PluginMetadata(
""".strip(),
extra=PluginExtraData(
author="HibiKier",
version="0.1",
version="0.2",
plugin_type=PluginType.SUPERUSER,
configs=[
RegisterConfig(
+45 -35
View File
@@ -1,3 +1,4 @@
import contextlib
from dataclasses import dataclass
import os
from pathlib import Path
@@ -18,7 +19,47 @@ BAIDU_URL = "https://www.baidu.com/"
GOOGLE_URL = "https://www.google.com/"
VERSION_FILE = Path() / "__version__"
ARM_KEY = "aarch64"
def get_arm_cpu_freq_safe():
"""获取ARM设备CPU频率"""
# 方法1: 优先从系统频率文件读取
freq_files = [
"/sys/devices/system/cpu/cpu0/cpufreq/cpuinfo_max_freq",
"/sys/devices/system/cpu/cpu0/cpufreq/scaling_max_freq",
"/sys/devices/system/cpu/cpu0/cpufreq/cpuinfo_cur_freq",
"/sys/devices/system/cpu/cpu0/cpufreq/scaling_cur_freq",
]
for freq_file in freq_files:
try:
with open(freq_file) as f:
frequency = int(f.read().strip())
return round(frequency / 1000000, 2) # 转换为GHz
except (OSError, ValueError):
continue
# 方法2: 解析/proc/cpuinfo
with contextlib.suppress(OSError, FileNotFoundError, ValueError, PermissionError):
with open("/proc/cpuinfo") as f:
for line in f:
if "CPU MHz" in line:
freq = float(line.split(":")[1].strip())
return round(freq / 1000, 2) # 转换为GHz
# 方法3: 使用lscpu命令
with contextlib.suppress(OSError, subprocess.SubprocessError, ValueError):
env = os.environ.copy()
env["LC_ALL"] = "C"
result = subprocess.run(
["lscpu"], capture_output=True, text=True, env=env, timeout=10
)
if result.returncode == 0:
for line in result.stdout.split("\n"):
if "CPU max MHz" in line or "CPU MHz" in line:
freq = float(line.split(":")[1].strip())
return round(freq / 1000, 2) # 转换为GHz
return 0 # 如果所有方法都失败,返回0
@dataclass
@@ -37,7 +78,7 @@ class CPUInfo:
if _cpu_freq := psutil.cpu_freq():
cpu_freq = round(_cpu_freq.current / 1000, 2)
else:
cpu_freq = 0
cpu_freq = get_arm_cpu_freq_safe()
return CPUInfo(core=cpu_core, usage=cpu_usage, freq=cpu_freq)
@@ -160,44 +201,13 @@ def __get_version() -> str | None:
return None
def __get_arm_cpu():
env = os.environ.copy()
env["LC_ALL"] = "en_US.UTF-8"
cpu_info = subprocess.check_output(["lscpu"], env=env).decode()
model_name = ""
cpu_freq = 0
for line in cpu_info.splitlines():
if "Model name" in line:
model_name = line.split(":")[1].strip()
if "CPU MHz" in line:
cpu_freq = float(line.split(":")[1].strip())
return model_name, cpu_freq
def __get_arm_oracle_cpu_freq():
cpu_freq = subprocess.check_output(
["dmidecode", "-s", "processor-frequency"]
).decode()
return round(float(cpu_freq.split()[0]) / 1000, 2)
async def get_status_info() -> dict:
"""获取信息"""
data = await __build_status()
system = platform.uname()
if system.machine == ARM_KEY and not (
cpuinfo.get_cpu_info().get("brand_raw") and data.cpu.freq
):
model_name, cpu_freq = __get_arm_cpu()
if not data.cpu.freq:
data.cpu.freq = cpu_freq or __get_arm_oracle_cpu_freq()
data = data.get_system_info()
data["brand_raw"] = model_name
else:
data = data.get_system_info()
data["brand_raw"] = cpuinfo.get_cpu_info().get("brand_raw", "Unknown")
data = data.get_system_info()
data["brand_raw"] = cpuinfo.get_cpu_info().get("brand_raw", "Unknown")
baidu, google = await __get_network_info()
data["baidu"] = "#8CC265" if baidu else "red"
data["google"] = "#8CC265" if google else "red"
+2 -2
View File
@@ -74,8 +74,8 @@ async def _(matcher: Matcher, message: UniMsg, session: EventSession):
message_list.append(image)
message_list.append(
"桀桀桀,预判到会有 '笨蛋' 把功能名称当命令用,特地前来嘲笑!"
f"但还是好心来帮帮你啦!\n请at我发送 '帮助{plugin.name}' 或者"
f" '帮助{plugin.id}' 来获取该功能帮助!"
f"但还是好心来帮帮你啦!\n请at我发送 '帮助 {plugin.name}' 或者"
f" '帮助 {plugin.id}' 来获取该功能帮助!"
)
logger.info("检测到功能名称当命令使用,已发送帮助信息", "功能帮助", session=session)
await MessageUtils.build_message(message_list).send(reply_to=True)
@@ -58,5 +58,14 @@ Config.add_plugin_config(
type=bool,
)
Config.add_plugin_config(
"hook",
"AUTH_HOOKS_CONCURRENCY_LIMIT",
5,
help="同步进入权限钩子最大并发数",
default_value=5,
type=int,
)
nonebot.load_plugins(str(Path(__file__).parent.resolve()))
+7 -16
View File
@@ -96,7 +96,6 @@ async def is_ban(user_id: str | None, group_id: str | None) -> int:
f"查询ban记录超时: user_id={user_id}, group_id={group_id}",
LOGGER_COMMAND,
)
# 超时时返回0,避免阻塞
return 0
# 检查记录并计算ban时间
@@ -199,7 +198,7 @@ async def group_handle(group_id: str) -> None:
)
async def user_handle(module: str, entity: EntityIDs, session: Uninfo) -> None:
async def user_handle(plugin: PluginInfo, entity: EntityIDs, session: Uninfo) -> None:
"""用户ban检查
参数:
@@ -217,22 +216,12 @@ async def user_handle(module: str, entity: EntityIDs, session: Uninfo) -> None:
if not time_val:
return
time_str = format_time(time_val)
plugin_dao = DataAccess(PluginInfo)
try:
db_plugin = await asyncio.wait_for(
plugin_dao.safe_get_or_none(module=module), timeout=DB_TIMEOUT_SECONDS
)
except asyncio.TimeoutError:
logger.error(f"查询插件信息超时: {module}", LOGGER_COMMAND)
# 超时时不阻塞,继续执行
raise SkipPluginException("用户处于黑名单中...")
if (
db_plugin
and not db_plugin.ignore_prompt
plugin
and time_val != -1
and ban_result
and freq.is_send_limit_message(db_plugin, entity.user_id, False)
and freq.is_send_limit_message(plugin, entity.user_id, False)
):
try:
await asyncio.wait_for(
@@ -260,7 +249,9 @@ async def user_handle(module: str, entity: EntityIDs, session: Uninfo) -> None:
)
async def auth_ban(matcher: Matcher, bot: Bot, session: Uninfo) -> None:
async def auth_ban(
matcher: Matcher, bot: Bot, session: Uninfo, plugin: PluginInfo
) -> None:
"""权限检查 - ban 检查
参数:
@@ -289,7 +280,7 @@ async def auth_ban(matcher: Matcher, bot: Bot, session: Uninfo) -> None:
if entity.user_id:
try:
await asyncio.wait_for(
user_handle(matcher.plugin_name, entity, session),
user_handle(plugin, entity, session),
timeout=DB_TIMEOUT_SECONDS,
)
except asyncio.TimeoutError:
@@ -1,50 +1,36 @@
import asyncio
import time
from nonebot_plugin_alconna import UniMsg
from zhenxun.models.group_console import GroupConsole
from zhenxun.models.plugin_info import PluginInfo
from zhenxun.services.data_access import DataAccess
from zhenxun.services.db_context import DB_TIMEOUT_SECONDS
from zhenxun.services.log import logger
from zhenxun.utils.utils import EntityIDs
from .config import LOGGER_COMMAND, WARNING_THRESHOLD, SwitchEnum
from .exception import SkipPluginException
async def auth_group(plugin: PluginInfo, entity: EntityIDs, message: UniMsg):
async def auth_group(
plugin: PluginInfo,
group: GroupConsole | None,
message: UniMsg,
group_id: str | None,
):
"""群黑名单检测 群总开关检测
参数:
plugin: PluginInfo
entity: EntityIDs
group: GroupConsole
message: UniMsg
"""
start_time = time.time()
if not entity.group_id:
if not group_id:
return
start_time = time.time()
try:
text = message.extract_plain_text()
# 从数据库或缓存中获取群组信息
group_dao = DataAccess(GroupConsole)
try:
group: GroupConsole | None = await asyncio.wait_for(
group_dao.safe_get_or_none(
group_id=entity.group_id, channel_id__isnull=True
),
timeout=DB_TIMEOUT_SECONDS,
)
except asyncio.TimeoutError:
logger.error("查询群组信息超时", LOGGER_COMMAND, session=entity.user_id)
# 超时时不阻塞,继续执行
return
if not group:
raise SkipPluginException("群组信息不存在...")
if group.level < 0:
@@ -63,6 +49,5 @@ async def auth_group(plugin: PluginInfo, entity: EntityIDs, message: UniMsg):
logger.warning(
f"auth_group 耗时: {elapsed:.3f}s, plugin={plugin.module}",
LOGGER_COMMAND,
session=entity.user_id,
group_id=entity.group_id,
group_id=group_id,
)
@@ -6,12 +6,10 @@ from nonebot_plugin_uninfo import Uninfo
from zhenxun.models.group_console import GroupConsole
from zhenxun.models.plugin_info import PluginInfo
from zhenxun.services.data_access import DataAccess
from zhenxun.services.db_context import DB_TIMEOUT_SECONDS
from zhenxun.services.log import logger
from zhenxun.utils.common_utils import CommonUtils
from zhenxun.utils.enum import BlockType
from zhenxun.utils.utils import get_entity_ids
from .config import LOGGER_COMMAND, WARNING_THRESHOLD
from .exception import IsSuperuserException, SkipPluginException
@@ -20,30 +18,17 @@ from .utils import freq, is_poke, send_message
class GroupCheck:
def __init__(
self, plugin: PluginInfo, group_id: str, session: Uninfo, is_poke: bool
self, plugin: PluginInfo, group: GroupConsole, session: Uninfo, is_poke: bool
) -> None:
self.group_id = group_id
self.session = session
self.is_poke = is_poke
self.plugin = plugin
self.group_dao = DataAccess(GroupConsole)
self.group_data = None
self.group_data = group
self.group_id = group.group_id
async def check(self):
start_time = time.time()
try:
# 只查询一次数据库,使用 DataAccess 的缓存机制
try:
self.group_data = await asyncio.wait_for(
self.group_dao.safe_get_or_none(
group_id=self.group_id, channel_id__isnull=True
),
timeout=DB_TIMEOUT_SECONDS,
)
except asyncio.TimeoutError:
logger.error(f"查询群组数据超时: {self.group_id}", LOGGER_COMMAND)
return # 超时时不阻塞,继续执行
# 检查超级用户禁用
if (
self.group_data
@@ -113,12 +98,13 @@ class GroupCheck:
class PluginCheck:
def __init__(self, group_id: str | None, session: Uninfo, is_poke: bool):
def __init__(self, group: GroupConsole | None, session: Uninfo, is_poke: bool):
self.session = session
self.is_poke = is_poke
self.group_id = group_id
self.group_dao = DataAccess(GroupConsole)
self.group_data = None
self.group_data = group
self.group_id = None
if group:
self.group_id = group.group_id
async def check_user(self, plugin: PluginInfo):
"""全局私聊禁用检测
@@ -156,21 +142,8 @@ class PluginCheck:
if plugin.status or plugin.block_type != BlockType.ALL:
return
"""全局状态"""
if self.group_id:
# 使用 DataAccess 的缓存机制
try:
self.group_data = await asyncio.wait_for(
self.group_dao.safe_get_or_none(
group_id=self.group_id, channel_id__isnull=True
),
timeout=DB_TIMEOUT_SECONDS,
)
except asyncio.TimeoutError:
logger.error(f"查询群组数据超时: {self.group_id}", LOGGER_COMMAND)
return # 超时时不阻塞,继续执行
if self.group_data and self.group_data.is_super:
raise IsSuperuserException()
if self.group_data and self.group_data.is_super:
raise IsSuperuserException()
sid = self.group_id or self.session.user.id
if freq.is_send_limit_message(plugin, sid, self.is_poke):
@@ -193,7 +166,9 @@ class PluginCheck:
)
async def auth_plugin(plugin: PluginInfo, session: Uninfo, event: Event):
async def auth_plugin(
plugin: PluginInfo, group: GroupConsole | None, session: Uninfo, event: Event
):
"""插件状态
参数:
@@ -203,35 +178,23 @@ async def auth_plugin(plugin: PluginInfo, session: Uninfo, event: Event):
"""
start_time = time.time()
try:
entity = get_entity_ids(session)
is_poke_event = is_poke(event)
user_check = PluginCheck(entity.group_id, session, is_poke_event)
user_check = PluginCheck(group, session, is_poke_event)
if entity.group_id:
group_check = GroupCheck(plugin, entity.group_id, session, is_poke_event)
try:
await asyncio.wait_for(
group_check.check(), timeout=DB_TIMEOUT_SECONDS * 2
)
except asyncio.TimeoutError:
logger.error(f"群组检查超时: {entity.group_id}", LOGGER_COMMAND)
# 超时时不阻塞,继续执行
tasks = []
if group:
tasks.append(GroupCheck(plugin, group, session, is_poke_event).check())
else:
try:
await asyncio.wait_for(
user_check.check_user(plugin), timeout=DB_TIMEOUT_SECONDS
)
except asyncio.TimeoutError:
logger.error("用户检查超时", LOGGER_COMMAND)
# 超时时不阻塞,继续执行
tasks.append(user_check.check_user(plugin))
tasks.append(user_check.check_global(plugin))
try:
await asyncio.wait_for(
user_check.check_global(plugin), timeout=DB_TIMEOUT_SECONDS
asyncio.gather(*tasks), timeout=DB_TIMEOUT_SECONDS * 2
)
except asyncio.TimeoutError:
logger.error("全局检查超时", LOGGER_COMMAND)
# 超时时不阻塞,继续执行
logger.error("插件用户/群组/全局检查超时...", LOGGER_COMMAND)
finally:
# 记录总执行时间
elapsed = time.time() - start_time
+1 -1
View File
@@ -85,7 +85,7 @@ class FreqUtils:
return False
if plugin.plugin_type == PluginType.DEPENDANT:
return False
return plugin.module != "ai" if self._flmt_s.check(sid) else False
return False if plugin.ignore_prompt else self._flmt_s.check(sid)
freq = FreqUtils()
+73 -4
View File
@@ -8,6 +8,7 @@ from nonebot_plugin_alconna import UniMsg
from nonebot_plugin_uninfo import Uninfo
from tortoise.exceptions import IntegrityError
from zhenxun.models.group_console import GroupConsole
from zhenxun.models.plugin_info import PluginInfo
from zhenxun.models.user_console import UserConsole
from zhenxun.services.data_access import DataAccess
@@ -31,6 +32,7 @@ from .auth.exception import (
PermissionExemption,
SkipPluginException,
)
from .auth.utils import base_config
# 超时设置(秒)
TIMEOUT_SECONDS = 5.0
@@ -46,6 +48,16 @@ CIRCUIT_BREAKERS = {
# 熔断重置时间(秒)
CIRCUIT_RESET_TIME = 300 # 5分钟
# 并发控制:限制同时进入 hooks 并行检查的协程数
# 默认为 6,可通过环境变量 AUTH_HOOKS_CONCURRENCY_LIMIT 调整
HOOKS_CONCURRENCY_LIMIT = base_config.get("AUTH_HOOKS_CONCURRENCY_LIMIT")
# 全局信号量与计数器
HOOKS_SEMAPHORE = asyncio.Semaphore(HOOKS_CONCURRENCY_LIMIT)
HOOKS_ACTIVE_COUNT = 0
HOOKS_ACTIVE_LOCK = asyncio.Lock()
# 超时装饰器
async def with_timeout(coro, timeout=TIMEOUT_SECONDS, name=None):
@@ -259,6 +271,30 @@ async def time_hook(coro, name, time_dict):
time_dict[name] = f"{time.time() - start:.3f}s"
async def _enter_hooks_section():
"""尝试获取全局信号量并更新计数器,超时则抛出 PermissionExemption。"""
global HOOKS_ACTIVE_COUNT
# 队列模式:如果达到上限,协程将排队等待直到获取到信号量
await HOOKS_SEMAPHORE.acquire()
async with HOOKS_ACTIVE_LOCK:
HOOKS_ACTIVE_COUNT += 1
logger.debug(f"当前并发权限检查数量: {HOOKS_ACTIVE_COUNT}", LOGGER_COMMAND)
async def _leave_hooks_section():
"""释放信号量并更新计数器。"""
global HOOKS_ACTIVE_COUNT
from contextlib import suppress
with suppress(Exception):
HOOKS_SEMAPHORE.release()
async with HOOKS_ACTIVE_LOCK:
HOOKS_ACTIVE_COUNT -= 1
# 保证计数不为负
HOOKS_ACTIVE_COUNT = max(HOOKS_ACTIVE_COUNT, 0)
logger.debug(f"当前并发权限检查数量: {HOOKS_ACTIVE_COUNT}", LOGGER_COMMAND)
async def auth(
matcher: Matcher,
event: Event,
@@ -285,6 +321,9 @@ async def auth(
hook_times = {}
hooks_time = 0 # 初始化 hooks_time 变量
# 记录是否已进入 hooks 区域(用于 finally 中释放)
entered_hooks = False
try:
if not module:
raise PermissionExemption("Matcher插件名称不存在...")
@@ -304,6 +343,10 @@ async def auth(
)
raise PermissionExemption("获取插件和用户数据超时,请稍后再试...")
# 进入 hooks 并行检查区域(会在高并发时排队)
await _enter_hooks_section()
entered_hooks = True
# 获取插件费用
cost_start = time.time()
try:
@@ -320,16 +363,32 @@ async def auth(
# 执行 bot_filter
bot_filter(session)
group = None
if entity.group_id:
group_dao = DataAccess(GroupConsole)
group = await with_timeout(
group_dao.safe_get_or_none(
group_id=entity.group_id, channel_id__isnull=True
),
name="get_group",
)
# 并行执行所有 hook 检查,并记录执行时间
hooks_start = time.time()
# 创建所有 hook 任务
hook_tasks = [
time_hook(auth_ban(matcher, bot, session), "auth_ban", hook_times),
time_hook(auth_ban(matcher, bot, session, plugin), "auth_ban", hook_times),
time_hook(auth_bot(plugin, bot.self_id), "auth_bot", hook_times),
time_hook(auth_group(plugin, entity, message), "auth_group", hook_times),
time_hook(
auth_group(plugin, group, message, entity.group_id),
"auth_group",
hook_times,
),
time_hook(auth_admin(plugin, session), "auth_admin", hook_times),
time_hook(auth_plugin(plugin, session, event), "auth_plugin", hook_times),
time_hook(
auth_plugin(plugin, group, session, event), "auth_plugin", hook_times
),
time_hook(auth_limit(plugin, session), "auth_limit", hook_times),
]
@@ -358,7 +417,17 @@ async def auth(
logger.debug("超级用户跳过权限检测...", LOGGER_COMMAND, session=session)
except PermissionExemption as e:
logger.info(str(e), LOGGER_COMMAND, session=session)
finally:
# 如果进入过 hooks 区域,确保释放信号量(即使上层处理抛出了异常)
if entered_hooks:
try:
await _leave_hooks_section()
except Exception:
logger.error(
"释放 hooks 信号量时出错",
LOGGER_COMMAND,
session=session,
)
# 扣除金币
if not ignore_flag and cost_gold > 0:
gold_start = time.time()
+3 -32
View File
@@ -1,12 +1,12 @@
from typing import Any
from nonebot.adapters import Bot, Message
from nonebot.adapters.onebot.v11 import MessageSegment
from zhenxun.configs.config import Config
from zhenxun.models.bot_message_store import BotMessageStore
from zhenxun.services.log import logger
from zhenxun.utils.enum import BotSentType
from zhenxun.utils.log_sanitizer import sanitize_for_logging
from zhenxun.utils.manager.message_manager import MessageManager
from zhenxun.utils.platform import PlatformUtils
@@ -41,35 +41,6 @@ def replace_message(message: Message) -> str:
return result
def format_message_for_log(message: Message) -> str:
"""
将消息对象转换为适合日志记录的字符串,对base64等长内容进行摘要处理。
"""
if not isinstance(message, Message):
return str(message)
log_parts = []
for seg in message:
seg: MessageSegment
if seg.type == "text":
log_parts.append(seg.data.get("text", ""))
elif seg.type in ("image", "record", "video"):
file_info = seg.data.get("file", "")
if isinstance(file_info, str) and file_info.startswith("base64://"):
b64_data = file_info[9:]
data_size_bytes = (len(b64_data) * 3) / 4 - b64_data.count("=", -2)
log_parts.append(
f"[{seg.type}: base64, size={data_size_bytes / 1024:.2f}KB]"
)
else:
log_parts.append(f"[{seg.type}]")
elif seg.type == "at":
log_parts.append(f"[@{seg.data.get('qq', 'unknown')}]")
else:
log_parts.append(f"[{seg.type}]")
return "".join(log_parts)
@Bot.on_called_api
async def handle_api_result(
bot: Bot, exception: Exception | None, api: str, data: dict[str, Any], result: Any
@@ -82,7 +53,6 @@ async def handle_api_result(
message: Message = data.get("message", "")
message_type = data.get("message_type")
try:
# 记录消息id
if user_id and message_id:
MessageManager.add(str(user_id), str(message_id))
logger.debug(
@@ -108,7 +78,8 @@ async def handle_api_result(
else replace_message(message),
platform=PlatformUtils.get_platform(bot),
)
logger.debug(f"消息发送记录,message: {format_message_for_log(message)}")
sanitized_message = sanitize_for_logging(message, context="nonebot_message")
logger.debug(f"消息发送记录,message: {sanitized_message}")
except Exception as e:
logger.warning(
f"消息发送记录发生错误...data: {data}, result: {result}",
+45 -45
View File
@@ -43,18 +43,20 @@ class BanCheckLimiter:
def check(self, key: str | float) -> bool:
if time.time() - self.mtime[key] > self.default_check_time:
self.mtime[key] = time.time()
self.mint[key] = 0
return False
return self._extracted_from_check_3(key, False)
if (
self.mint[key] >= self.default_count
and time.time() - self.mtime[key] < self.default_check_time
):
self.mtime[key] = time.time()
self.mint[key] = 0
return True
return self._extracted_from_check_3(key, True)
return False
# TODO Rename this here and in `check`
def _extracted_from_check_3(self, key, arg1):
self.mtime[key] = time.time()
self.mint[key] = 0
return arg1
_blmt = BanCheckLimiter(
malicious_check_time,
@@ -70,16 +72,15 @@ async def _(
module = None
if plugin := matcher.plugin:
module = plugin.module_name
if metadata := plugin.metadata:
extra = metadata.extra
if extra.get("plugin_type") in [
PluginType.HIDDEN,
PluginType.DEPENDANT,
PluginType.ADMIN,
PluginType.SUPERUSER,
]:
return
else:
if not (metadata := plugin.metadata):
return
extra = metadata.extra
if extra.get("plugin_type") in [
PluginType.HIDDEN,
PluginType.DEPENDANT,
PluginType.ADMIN,
PluginType.SUPERUSER,
]:
return
if matcher.type == "notice":
return
@@ -88,32 +89,31 @@ async def _(
malicious_ban_time = Config.get_config("hook", "MALICIOUS_BAN_TIME")
if not malicious_ban_time:
raise ValueError("模块: [hook], 配置项: [MALICIOUS_BAN_TIME] 为空或小于0")
if user_id:
if module:
if _blmt.check(f"{user_id}__{module}"):
await BanConsole.ban(
user_id,
group_id,
9,
"恶意触发命令检测",
malicious_ban_time * 60,
bot.self_id,
)
logger.info(
f"触发了恶意触发检测: {matcher.plugin_name}",
"HOOK",
session=session,
)
await MessageUtils.build_message(
[
At(flag="user", target=user_id),
"检测到恶意触发命令,您将被封禁 30 分钟",
]
).send()
logger.debug(
f"触发了恶意触发检测: {matcher.plugin_name}",
"HOOK",
session=session,
)
raise IgnoredException("检测到恶意触发命令")
_blmt.add(f"{user_id}__{module}")
if user_id and module:
if _blmt.check(f"{user_id}__{module}"):
await BanConsole.ban(
user_id,
group_id,
9,
"恶意触发命令检测",
malicious_ban_time * 60,
bot.self_id,
)
logger.info(
f"触发了恶意触发检测: {matcher.plugin_name}",
"HOOK",
session=session,
)
await MessageUtils.build_message(
[
At(flag="user", target=user_id),
"检测到恶意触发命令,您将被封禁 30 分钟",
]
).send()
logger.debug(
f"触发了恶意触发检测: {matcher.plugin_name}",
"HOOK",
session=session,
)
raise IgnoredException("检测到恶意触发命令")
_blmt.add(f"{user_id}__{module}")
@@ -1,3 +1,5 @@
from collections import defaultdict
from nonebot.permission import SUPERUSER
from nonebot.plugin import PluginMetadata
from nonebot_plugin_alconna import (
@@ -58,7 +60,12 @@ __plugin_meta__ = PluginMetadata(
llm_cmd = on_alconna(
Alconna(
"llm",
Subcommand("list", alias=["ls"], help_text="查看模型列表"),
Subcommand(
"list",
Option("--text", action=store_true, help_text="以纯文本格式输出模型列表"),
alias=["ls"],
help_text="查看模型列表",
),
Subcommand("info", Args["model_name", str], help_text="查看模型详情"),
Subcommand("default", Args["model_name?", str], help_text="查看或设置默认模型"),
Subcommand(
@@ -80,13 +87,36 @@ llm_cmd = on_alconna(
@llm_cmd.assign("list")
async def handle_list(arp: Arparma, show_all: Query[bool] = Query("all")):
async def handle_list(
arp: Arparma,
show_all: Query[bool] = Query("all"),
text_mode: Query[bool] = Query("list.text.value", False),
):
"""处理 'llm list' 命令"""
logger.info("获取LLM模型列表", command="LLM Manage", session=arp.header_result)
models = await DataSource.get_model_list(show_all=show_all.result)
image = await Presenters.format_model_list_as_image(models, show_all.result)
await llm_cmd.finish(MessageUtils.build_message(image))
if text_mode.result:
if not models:
await llm_cmd.finish("当前没有配置任何LLM模型。")
grouped_models = defaultdict(list)
for model in models:
grouped_models[model["provider_name"]].append(model)
response_parts = ["可用的LLM模型列表:"]
for provider, model_list in grouped_models.items():
response_parts.append(f"\n{provider}:")
for model in model_list:
response_parts.append(
f" {model['provider_name']}/{model['model_name']}"
)
response_text = "\n".join(response_parts)
await llm_cmd.finish(response_text)
else:
image = await Presenters.format_model_list_as_image(models, show_all.result)
await llm_cmd.finish(MessageUtils.build_message(image))
@llm_cmd.assign("info")
@@ -114,7 +144,7 @@ async def handle_default(arp: Arparma, model_name: Match[str]):
command="LLM Manage",
session=arp.header_result,
)
success, message = await DataSource.set_default_model(model_name.result)
_success, message = await DataSource.set_default_model(model_name.result)
await llm_cmd.finish(message)
else:
logger.info("查看默认模型", command="LLM Manage", session=arp.header_result)
@@ -132,7 +162,7 @@ async def handle_test(arp: Arparma, model_name: Match[str]):
)
await llm_cmd.send(f"正在测试模型 '{model_name.result}',请稍候...")
success, message = await DataSource.test_model_connectivity(model_name.result)
_success, message = await DataSource.test_model_connectivity(model_name.result)
await llm_cmd.finish(message)
@@ -167,5 +197,5 @@ async def handle_reset_key(
)
logger.info(log_msg, command="LLM Manage", session=arp.header_result)
success, message = await DataSource.reset_key(provider_name.result, key_to_reset)
_success, message = await DataSource.reset_key(provider_name.result, key_to_reset)
await llm_cmd.finish(message)
@@ -17,6 +17,8 @@ from zhenxun.configs.utils import PluginExtraData, RegisterConfig, Task
from zhenxun.models.event_log import EventLog
from zhenxun.models.group_console import GroupConsole
from zhenxun.services.cache import CacheRoot
from zhenxun.services.log import logger
from zhenxun.services.tags import tag_manager
from zhenxun.utils.common_utils import CommonUtils
from zhenxun.utils.enum import EventLogType, PluginType
from zhenxun.utils.platform import PlatformUtils
@@ -135,6 +137,11 @@ async def _(
await EventLog.create(
user_id=user_id, group_id=group_id, event_type=EventLogType.KICK_BOT
)
await tag_manager.remove_group_from_all_tags(group_id)
logger.info(
f"机器人被移出群聊,已自动从所有静态标签中移除群组 {group_id}",
"群组标签管理",
)
elif event.sub_type in ["leave", "kick"]:
if event.sub_type == "leave":
"""主动退群"""
@@ -263,10 +263,9 @@ class StoreManager:
"""安装插件
参数:
github_url: 仓库地址
module_path: 模块路径
is_dir: 是否是文件夹
plugin_info: 插件信息
is_external: 是否是外部仓库
source: 源
"""
repo_type = RepoType.GITHUB if is_external else None
if source == "ali":
@@ -2,6 +2,7 @@ import nonebot
from nonebot_plugin_apscheduler import scheduler
from zhenxun.services.log import logger
from zhenxun.services.tags import tag_manager
from zhenxun.utils.platform import PlatformUtils
@@ -37,3 +38,20 @@ async def _():
f"Bot: {bot.self_id} 自动更新好友信息错误", "自动更新好友", e=e
)
logger.info("自动更新好友信息成功...")
# 自动清理静态标签中的无效群组
@scheduler.scheduled_job(
"cron",
hour=23,
minute=30,
)
async def _prune_stale_tags():
deleted_count = await tag_manager.prune_stale_group_links()
if deleted_count > 0:
logger.info(
f"定时任务:成功清理了 {deleted_count} 个无效的群组标签" f"关联。",
"群组标签管理",
)
else:
logger.debug("定时任务:未发现无效的群组标签关联。", "群组标签管理")
@@ -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,38 @@ __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,
),
RegisterConfig(
module="SchedulerManager",
key="DEFAULT_INTERVAL_SECONDS",
value=0,
help="为多目标定时任务设置的默认串行执行间隔秒数(大于0时生效),用于控制任务间的固定时间间隔。",
default_value=0,
type=int,
),
],
).to_dict(),
)
@@ -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,41 @@ 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(
"--fixed-interval",
Args["interval_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 +146,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_ids?",
MultiVar(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_ids?",
MultiVar(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_ids?",
MultiVar(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 +223,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)
@@ -0,0 +1,314 @@
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,
interval: 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
if interval is not None:
execution_options["interval"] = interval
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()
@@ -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_ids: Match[list[int]] = AlconnaMatch("schedule_ids"),
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_ids.available:
return scheduler_manager.target(id__in=schedule_ids.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 选项指定目标。"
)
@@ -1,382 +1,238 @@
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"),
interval: Match[int] = AlconnaMatch("interval_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
interval_val: int | None = interval.result if interval.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(",")
)
if interval_val is None:
if task_meta and task_meta.get("default_interval") is not None:
interval_val = cast(int | None, task_meta["default_interval"])
else:
interval_val = Config.get_config(
"SchedulerManager", "DEFAULT_INTERVAL_SECONDS"
)
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,
interval=interval_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)
@@ -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)
+1 -1
View File
@@ -367,7 +367,7 @@ class ShopManage:
else:
goods_info = await GoodsInfo.get_or_none(goods_name=goods_name)
if not goods_info:
return f"{goods_name} 不存在..."
return "对应的道具不存在..."
if goods_info.is_passive:
return f"{goods_info.goods_name} 是被动道具, 无法使用..."
goods = cls.uuid2goods.get(goods_info.uuid)
@@ -175,7 +175,7 @@ class SignManage:
impression_added = (secrets.randbelow(99) + 1) / 100
rand = random.random()
add_probability = float(user.add_probability)
specify_probability = user.specify_probability
specify_probability = float(user.specify_probability)
if rand + add_probability > 0.97 or rand < specify_probability:
impression_added *= 2
await SignUser.sign(user, impression_added, session.self_id, platform)
@@ -28,7 +28,8 @@ from nonebot_plugin_alconna.uniseg.segment import (
)
from nonebot_plugin_session import EventSession
from zhenxun.configs.utils import PluginExtraData, Task
from zhenxun.configs.utils import PluginExtraData, RegisterConfig, Task
from zhenxun.services.log import logger
from zhenxun.utils.enum import PluginType
from zhenxun.utils.message import MessageUtils
@@ -45,34 +46,52 @@ __plugin_meta__ = PluginMetadata(
name="广播",
description="昭告天下!",
usage="""
广播 [消息内容]
- 直接发送消息到除当前群组外的所有群组
- 支持文本、图片、@、表情、视频等多种消息类型
- 示例:广播 你们好!
- 示例:广播 [图片] 新活动开始啦!
向所有群组或指定标签的群组发送广播消息。
广播 + 引用消息
- 将引用的消息作为广播内容发送
- 支持引用普通消息或合并转发消息
- 示例:(引用一条消息) 广播
**基础用法**
- `广播 [消息内容]`:向所有群组发送广播。
- `广播` (并引用一条消息):将引用的消息作为内容进行广播。
广播撤回
- 撤回最近一次由您触发的广播消息
- 仅能撤回短时间内的消息
- 示例:广播撤回
**高级定向广播**
- `广播 -t <标签名> [消息内容]`:向指定标签下的所有群组广播。
- `广播到 <标签名> [消息内容]`:与 `-t` 等效的快捷方式。
特性:
- 在群组中使用广播时,不会将消息发送到当前群组
- 在私聊中使用广播时,会发送到所有群组
**标签可以是静态的,也可以是动态的,例如:**
- `广播到 核心群 通知:...`
- `广播到 成员数>500的群 通知:...`
别名:
- bc (广播的简写)
- recall (广播撤回的别名)
**其他命令**
- `广播撤回` (别名: `recall`):撤回最近一次发送的广播。
特性:
- 在群组中使用广播时,不会将消息发送到当前群组
- 在私聊中使用广播时,会发送到所有群组
别名:
- bc (广播的简写)
- recall (广播撤回的别名)
""".strip(),
extra=PluginExtraData(
author="HibiKier",
version="1.2",
version="1.3",
plugin_type=PluginType.SUPERUSER,
configs=[
RegisterConfig(
module="_task",
key="DEFAULT_BROADCAST",
value=True,
help="被动 广播 进群默认开关状态",
default_value=True,
type=bool,
),
RegisterConfig(
module="_task",
key="BROADCAST_CONCURRENCY_LIMIT",
value=10,
help="广播时的最大并发任务数,以避免API速率限制",
default_value=10,
),
],
tasks=[Task(module="broadcast", name="广播")],
).to_dict(),
)
@@ -103,6 +122,9 @@ _matcher = on_alconna(
Alconna(
"广播",
Args["content?", AllParam],
alc.Option(
"-t|--tag", Args["tag_name_bc", str], help_text="向指定标签的群组广播"
),
),
aliases={"bc"},
priority=1,
@@ -112,6 +134,8 @@ _matcher = on_alconna(
use_origin=False,
)
_matcher.shortcut("广播到 {tag}", command="广播 -t {tag} {%*}")
_recall_matcher = on_alconna(
Alconna("广播撤回"),
aliases={"recall"},
@@ -128,23 +152,59 @@ async def handle_broadcast(
event: Event,
session: EventSession,
arp: alc.Arparma,
tag_name_match: alc.Match[str] = alc.AlconnaMatch("tag_name_bc"),
):
broadcast_content_msg = await _extract_broadcast_content(bot, event, arp, session)
if not broadcast_content_msg:
return
target_groups, enabled_groups = await get_broadcast_target_groups(bot, session)
if not target_groups or not enabled_groups:
tag_name_to_broadcast = None
force_send = False
if tag_name_match.available:
tag_name_to_broadcast = tag_name_match.result
force_send = True
mode_desc = "强制发送到标签" if force_send else "普通发送"
logger.debug(
f"广播模式: {mode_desc}, 标签名: {tag_name_to_broadcast}",
"广播",
)
target_groups_console, groups_to_actually_send = await get_broadcast_target_groups(
bot, session, tag_name_to_broadcast, force_send
)
if not target_groups_console:
if tag_name_to_broadcast:
await MessageUtils.build_message(
f"标签 '{tag_name_to_broadcast}' 中没有群组或标签不存在。"
).send(reply_to=True)
return
if not groups_to_actually_send:
if not force_send and target_groups_console:
await MessageUtils.build_message(
"没有启用了广播功能的目标群组可供立即发送。"
).send(reply_to=True)
return
try:
await send_broadcast_and_notify(
bot, event, broadcast_content_msg, enabled_groups, target_groups, session
bot,
event,
broadcast_content_msg,
groups_to_actually_send,
target_groups_console,
session,
force_send,
)
except Exception as e:
error_msg = "发送广播失败"
BroadcastManager.log_error(error_msg, e, session)
await MessageUtils.build_message(f"{error_msg}。").send(reply_to=True)
await bot.send_private_msg(
user_id=str(event.get_user_id()), message=f"{error_msg}。"
)
@_recall_matcher.handle()
@@ -178,5 +238,6 @@ async def handle_broadcast_recall(
except Exception as e:
error_msg = "撤回广播消息失败"
BroadcastManager.log_error(error_msg, e, session)
user_id = str(event.get_user_id())
await bot.send_private_msg(user_id=user_id, message=f"{error_msg}。")
await bot.send_private_msg(
user_id=str(event.get_user_id()), message=f"{error_msg}。"
)
@@ -5,11 +5,12 @@ from typing import ClassVar
from nonebot.adapters import Bot
from nonebot.adapters.onebot.v11 import Bot as V11Bot
from nonebot.exception import ActionFailed
from nonebot.exception import ActionFailed, AdapterException
from nonebot_plugin_alconna import UniMessage
from nonebot_plugin_alconna.uniseg import Receipt, Reference
from nonebot_plugin_session import EventSession
from zhenxun.configs.config import Config
from zhenxun.models.group_console import GroupConsole
from zhenxun.services.log import logger
from zhenxun.utils.common_utils import CommonUtils
@@ -18,6 +19,8 @@ from zhenxun.utils.platform import PlatformUtils
from .models import BroadcastDetailResult, BroadcastResult
from .utils import custom_nodes_to_v11_nodes, uni_message_to_v11_list_of_dicts
BROADCAST_SEND_DELAY_RANGE = (1, 3)
class BroadcastManager:
"""广播管理器"""
@@ -92,8 +95,16 @@ class BroadcastManager:
logger.debug("清空上一次的广播消息ID记录", "广播", session=session)
cls.clear_last_broadcast_msg_ids()
concurrency_limit = Config.get_config(
"_task",
"BROADCAST_CONCURRENCY_LIMIT",
10,
)
all_groups, _ = await cls.get_all_groups(bot)
return await cls.send_to_specific_groups(bot, message, all_groups, session)
return await cls.send_to_specific_groups(
bot, message, all_groups, session, concurrency_limit=concurrency_limit
)
@classmethod
async def send_to_specific_groups(
@@ -102,14 +113,17 @@ class BroadcastManager:
message: UniMessage,
target_groups: list[GroupConsole],
session_info: EventSession | str | None = None,
force_send: bool = False,
concurrency_limit: int = 10,
) -> BroadcastResult:
"""发送广播到指定群组"""
log_session = session_info or bot.self_id
logger.debug(
f"开始广播,目标 {len(target_groups)} 个群组,Bot ID: {bot.self_id}",
"广播",
session=log_session,
target_count = len(target_groups)
log_message = (
f"开始广播,目标 {target_count} 个群组 (并发数: {concurrency_limit}),"
f"Bot ID: {bot.self_id}, ForceSend: {force_send}"
)
logger.info(log_message, "广播", session=log_session)
if not target_groups:
logger.debug("目标群组列表为空,广播结束", "广播", session=log_session)
@@ -165,7 +179,12 @@ class BroadcastManager:
)
return 0, len(target_groups)
success_count, error_count, skip_count = await cls._broadcast_forward(
bot, log_session, target_groups, v11_nodes
bot,
log_session,
target_groups,
v11_nodes,
force_send,
concurrency_limit,
)
else:
if is_forward_broadcast:
@@ -175,7 +194,12 @@ class BroadcastManager:
session=log_session,
)
success_count, error_count, skip_count = await cls._broadcast_normal(
bot, log_session, target_groups, message
bot,
log_session,
target_groups,
message,
force_send,
concurrency_limit,
)
total = len(target_groups)
@@ -287,11 +311,16 @@ class BroadcastManager:
)
@classmethod
async def _check_group_availability(cls, bot: Bot, group: GroupConsole) -> bool:
async def _check_group_availability(
cls, bot: Bot, group: GroupConsole, force_send: bool = False
) -> bool:
"""检查群组是否可用"""
if not group.group_id:
return False
if force_send:
return True
if await CommonUtils.task_is_block(bot, "broadcast", group.group_id):
return False
@@ -304,54 +333,69 @@ class BroadcastManager:
session_info: EventSession | str,
group_list: list[GroupConsole],
v11_nodes: list[dict],
force_send: bool = False,
concurrency_limit: int = 10,
) -> BroadcastDetailResult:
"""发送合并转发"""
success_count = 0
error_count = 0
skip_count = 0
semaphore = asyncio.Semaphore(concurrency_limit)
msg_id_lock = asyncio.Lock()
for _, group in enumerate(group_list):
async def send_to_group(group: GroupConsole) -> GroupConsole:
group_key = group.group_id or group.channel_id
async with semaphore:
try:
result = await bot.send_group_forward_msg(
group_id=int(group.group_id), messages=v11_nodes
)
async with msg_id_lock:
await cls._extract_message_id_from_result(
result, group_key, session_info, "合并转发"
)
await asyncio.sleep(random.uniform(*BROADCAST_SEND_DELAY_RANGE))
return group
except (ActionFailed, AdapterException) as ae:
logger.error(
f"发送失败(合并转发) to {group_key}: {ae}",
"广播",
session=session_info,
e=ae,
)
raise
except Exception as e:
logger.error(
f"发送失败(合并转发) to {group_key}: {e}",
"广播",
session=session_info,
e=e,
)
raise
if not await cls._check_group_availability(bot, group):
skip_count += 1
continue
tasks: list[asyncio.Task] = []
skipped_groups: list[GroupConsole] = []
for group in group_list:
if await cls._check_group_availability(bot, group, force_send):
tasks.append(asyncio.create_task(send_to_group(group)))
else:
skipped_groups.append(group)
try:
result = await bot.send_group_forward_msg(
group_id=int(group.group_id), messages=v11_nodes
)
if skipped_groups:
logger.info(
f"跳过 {len(skipped_groups)} 个不符合条件的群组",
"广播",
session=session_info,
)
logger.debug(
f"合并转发消息发送结果: {result}, 类型: {type(result)}",
"广播",
session=session_info,
)
if not tasks:
return 0, 0, len(skipped_groups)
await cls._extract_message_id_from_result(
result, group_key, session_info, "合并转发"
)
results = await asyncio.gather(*tasks, return_exceptions=True)
success_count += 1
await asyncio.sleep(random.randint(1, 3))
except ActionFailed as af_e:
error_count += 1
logger.error(
f"发送失败(合并转发) to {group_key}: {af_e}",
"广播",
session=session_info,
e=af_e,
)
except Exception as e:
error_count += 1
logger.error(
f"发送失败(合并转发) to {group_key}: {e}",
"广播",
session=session_info,
e=e,
)
success_count = sum(
1 for result in results if not isinstance(result, Exception)
)
error_count = len(results) - success_count
return success_count, error_count, skip_count
return success_count, error_count, len(skipped_groups)
@classmethod
async def _broadcast_normal(
@@ -360,58 +404,83 @@ class BroadcastManager:
session_info: EventSession | str,
group_list: list[GroupConsole],
message: UniMessage,
force_send: bool = False,
concurrency_limit: int = 10,
) -> BroadcastDetailResult:
"""发送普通消息"""
success_count = 0
error_count = 0
skip_count = 0
semaphore = asyncio.Semaphore(concurrency_limit)
msg_id_lock = asyncio.Lock()
for _, group in enumerate(group_list):
async def send_to_group(group: GroupConsole) -> GroupConsole:
group_key = (
f"{group.group_id}:{group.channel_id}"
if group.channel_id
else str(group.group_id)
)
if not await cls._check_group_availability(bot, group):
skip_count += 1
continue
try:
target = PlatformUtils.get_target(
group_id=group.group_id, channel_id=group.channel_id
)
if target:
receipt: Receipt = await message.send(target, bot=bot)
logger.debug(
f"广播消息发送结果: {receipt}, 类型: {type(receipt)}",
"广播",
session=session_info,
)
await cls._extract_message_id_from_result(
receipt, group_key, session_info
)
success_count += 1
await asyncio.sleep(random.randint(1, 3))
else:
logger.warning(
"target为空", "广播", session=session_info, target=group_key
)
skip_count += 1
except Exception as e:
error_count += 1
logger.error(
f"发送失败(普通) to {group_key}: {e}",
target = PlatformUtils.get_target(
group_id=group.group_id, channel_id=group.channel_id
)
if not target:
logger.warning(
"target为空",
"广播",
session=session_info,
e=e,
target=group_key,
)
raise ValueError(f"无法为群组 {group_key} 创建发送目标")
return success_count, error_count, skip_count
async with semaphore:
try:
receipt: Receipt = await message.send(target, bot=bot)
async with msg_id_lock:
await cls._extract_message_id_from_result(
receipt, group_key, session_info
)
await asyncio.sleep(random.uniform(*BROADCAST_SEND_DELAY_RANGE))
return group
except (ActionFailed, AdapterException) as ae:
logger.error(
f"发送失败(普通) to {group_key}: {ae}",
"广播",
session=session_info,
e=ae,
)
raise
except Exception as e:
logger.error(
f"发送失败(普通) to {group_key}: {e}",
"广播",
session=session_info,
e=e,
)
raise
tasks: list[asyncio.Task] = []
skipped_groups: list[GroupConsole] = []
for group in group_list:
if await cls._check_group_availability(bot, group, force_send):
tasks.append(asyncio.create_task(send_to_group(group)))
else:
skipped_groups.append(group)
if skipped_groups:
logger.info(
f"跳过 {len(skipped_groups)} 个不符合条件的群组",
"广播",
session=session_info,
)
if not tasks:
return 0, 0, len(skipped_groups)
results = await asyncio.gather(*tasks, return_exceptions=True)
success_count = sum(
1 for result in results if not isinstance(result, Exception)
)
error_count = len(results) - success_count
return success_count, error_count, len(skipped_groups)
@classmethod
async def recall_last_broadcast(
@@ -21,8 +21,11 @@ from nonebot_plugin_alconna.uniseg.segment import (
from nonebot_plugin_alconna.uniseg.tools import reply_fetch
from nonebot_plugin_session import EventSession
from zhenxun.models.group_console import GroupConsole
from zhenxun.services.log import logger
from zhenxun.services.tags import tag_manager as TagManager
from zhenxun.utils.common_utils import CommonUtils
from zhenxun.utils.http_utils import AsyncHttpx
from zhenxun.utils.message import MessageUtils
from .broadcast_manager import BroadcastManager
@@ -399,22 +402,29 @@ async def _process_v11_segment(
elif target_qq:
result.append(At(flag="user", target=target_qq))
elif seg_type == "video":
video_seg = None
if data_dict.get("url"):
video_seg = Video(url=data_dict["url"])
elif data_dict.get("file"):
file_val = data_dict["file"]
if url := data_dict.get("url"):
try:
logger.debug(f"[D{depth}] 正在下载视频用于广播: {url}", "广播")
video_bytes = await AsyncHttpx.get_content(url)
video_seg = Video(raw=video_bytes)
logger.debug(
f"[D{depth}] 视频下载成功, 大小: {len(video_bytes)} bytes",
"广播",
)
result.append(video_seg)
except Exception as e:
logger.error(f"[D{depth}] 广播时下载视频失败: {url}", "广播", e=e)
result.append(Text(f"[视频下载失败: {url}]"))
elif file_val := data_dict.get("file"):
if isinstance(file_val, str) and file_val.startswith("base64://"):
b64_data = file_val[9:]
raw_bytes = base64.b64decode(b64_data)
video_seg = Video(raw=raw_bytes)
result.append(video_seg)
else:
video_seg = Video(path=file_val)
if video_seg:
result.append(video_seg)
logger.debug(f"[Depth {depth}] 处理视频消息成功", "广播")
else:
logger.warning(f"[Depth {depth}] V11 视频 {index} 缺少URL/文件", "广播")
result.append(video_seg)
return result
elif seg_type == "forward":
nested_forward_id = data_dict.get("id") or data_dict.get("resid")
nested_forward_content = data_dict.get("content")
@@ -515,70 +525,129 @@ async def _extract_content_from_message(
async def get_broadcast_target_groups(
bot: Bot, session: EventSession
bot: Bot,
session: EventSession,
tag_name: str | None = None,
force_send: bool = False,
) -> tuple[list, list]:
"""获取广播目标群组和启用了广播功能的群组"""
target_groups = []
all_groups, _ = await BroadcastManager.get_all_groups(bot)
target_groups_console: list[GroupConsole] = []
current_group_id = None
if hasattr(session, "id2") and session.id2:
current_group_id = session.id2
current_group_raw = getattr(session, "id2", None) or getattr(
session, "group_id", None
)
current_group_id = str(current_group_raw) if current_group_raw else None
if current_group_id:
target_groups = [
group for group in all_groups if group.group_id != current_group_id
]
logger.info(
f"向除当前群组({current_group_id})外的所有群组广播", "广播", session=session
)
logger.debug(f"当前群组ID: {current_group_id}", "广播")
if tag_name:
tagged_group_ids = await TagManager.resolve_tag_to_group_ids(tag_name, bot=bot)
if not tagged_group_ids:
return [], []
valid_groups = await GroupConsole.filter(group_id__in=tagged_group_ids)
if current_group_id:
target_groups_console = [
group
for group in valid_groups
if str(group.group_id) != current_group_id
]
excluded_msg = (
f",已排除当前群组({current_group_id})"
if any(
str(group.group_id) == current_group_id for group in valid_groups
)
else ""
)
broadcast_msg = (
f"向标签 '{tag_name}' 中的 {len(target_groups_console)} 个群组广播 "
f"(ForceSend: {force_send}){excluded_msg}"
)
logger.info(broadcast_msg, "广播", session=session)
else:
target_groups_console = valid_groups
broadcast_msg = (
f"向标签 '{tag_name}' 中的 {len(target_groups_console)} 个群组广播 "
f"(ForceSend: {force_send})"
)
logger.info(broadcast_msg, "广播", session=session)
else:
target_groups = all_groups
logger.info("向所有群组广播", "广播", session=session)
all_groups, _ = await BroadcastManager.get_all_groups(bot)
if not target_groups:
await MessageUtils.build_message("没有找到符合条件的广播目标群组。").send(
reply_to=True
)
if current_group_id:
target_groups_console = [
group for group in all_groups if str(group.group_id) != current_group_id
]
logger.info(
(
f"向除当前群组({current_group_id})外的所有群组广播 "
f"(ForceSend: {force_send})"
),
"广播",
session=session,
)
else:
target_groups_console = all_groups
logger.info(
f"向所有群组广播 (ForceSend: {force_send})", "广播", session=session
)
if not target_groups_console:
if not tag_name:
await MessageUtils.build_message("没有找到符合条件的广播目标群组。").send(
reply_to=True
)
return [], []
enabled_groups = []
for group in target_groups:
if not await CommonUtils.task_is_block(bot, "broadcast", group.group_id):
enabled_groups.append(group)
groups_to_actually_send = []
if force_send:
groups_to_actually_send = target_groups_console
logger.debug(
f"强制发送模式,将向 {len(groups_to_actually_send)} 个目标群组尝试发送。",
"广播",
)
else:
for group in target_groups_console:
if not await CommonUtils.task_is_block(bot, "broadcast", group.group_id):
groups_to_actually_send.append(group)
logger.debug(
f"普通发送模式,筛选后将向 {len(groups_to_actually_send)} "
f"个目标群组尝试发送",
"广播",
)
if not enabled_groups:
await MessageUtils.build_message(
"没有启用了广播功能的目标群组可供立即发送。"
).send(reply_to=True)
return target_groups, []
return target_groups, enabled_groups
return target_groups_console, groups_to_actually_send
async def send_broadcast_and_notify(
bot: Bot,
event: Event,
message: UniMessage,
enabled_groups: list,
target_groups: list,
groups_to_send: list,
all_target_groups_for_stats: list,
session: EventSession,
force_send: bool = False,
) -> None:
"""发送广播并通知结果"""
BroadcastManager.clear_last_broadcast_msg_ids()
count, error_count = await BroadcastManager.send_to_specific_groups(
bot, message, enabled_groups, session
bot, message, groups_to_send, session, force_send
)
result = f"成功广播 {count} 个群组"
if error_count:
result += f"\n发送失败 {error_count} 个群组"
result += f"\n有效: {len(enabled_groups)} / 总计: {len(target_groups)}"
effective_sent_count = len(groups_to_send)
total_considered_count = len(all_target_groups_for_stats)
result += f"\n有效: {effective_sent_count} / 总计目标: {total_considered_count}"
user_id = str(event.get_user_id())
await bot.send_private_msg(user_id=user_id, message=f"发送广播完成!\n{result}")
BroadcastManager.log_info(
f"广播完成,有效/总计: {len(enabled_groups)}/{len(target_groups)}",
f"广播完成,有效/总计目标: {effective_sent_count}/{total_considered_count}",
session,
)
@@ -59,7 +59,7 @@ def uni_segment_to_v11_segment_dict(
logger.warning(f"无法处理 Video.raw 的类型: {type(raw_data)}", "广播")
elif getattr(seg, "path", None):
logger.warning(
f"在合并转发中使用了本地视频路径,可能无法显示: {seg.path}", "广播"
f"在合并转发中使用了本地视频路径,可能无法发送: {seg.path}", "广播"
)
return {"type": "video", "data": {"file": f"file:///{seg.path}"}}
else:
@@ -0,0 +1,483 @@
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="清空所有标签"),
Subcommand("prune", alias=["check", "清理"], help_text="清理无效的群组关联"),
Subcommand(
"clone",
Args["source_name", str]["new_name", str],
Option("--add", Args["add_groups", MultiVar(str)]),
Option("--remove", Args["remove_groups", MultiVar(str)]),
Option("--as-dynamic", action=store_true),
Option("--desc", Args["description", str]),
Option("--mode", Args["mode", ["black", "white"]]),
help_text="克隆标签",
),
),
permission=SUPERUSER,
priority=5,
block=True,
)
tag_cmd.shortcut(
"清理标签",
command="tag",
arguments=["prune"],
prefix=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"] == "STATIC" and details["is_blacklist"]:
msg += f"排除群组 ({len(details['groups'])}个):\n"
if details["groups"]:
msg += "\n".join(f"- {gid}" for gid in details["groups"])
else:
msg += "无"
msg += "\n\n"
if details["tag_type"] == "DYNAMIC" and details.get("dynamic_rule"):
msg += f"动态规则: {details['dynamic_rule']}\n"
title = (
"当前生效群组"
if details["tag_type"] == "DYNAMIC" or details["is_blacklist"]
else "关联群组"
)
if details["resolved_groups"] is not None:
msg += f"{title} ({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:
gids_to_create = None
unique_gids_count = 0
if group_ids.available:
unique_gids = list(dict.fromkeys(group_ids.result))
gids_to_create = unique_gids
unique_gids_count = len(unique_gids)
tag = await tag_manager.create_tag(
name=name.result,
is_blacklist=blacklist.result,
description=description.result if description.available else None,
group_ids=gids_to_create,
tag_type=ttype,
dynamic_rule=rule.result if rule.available else None,
)
msg = f"标签 '{tag.name}' 创建成功!"
if group_ids.available:
msg += f"\n已同时关联 {unique_gids_count} 个群组。"
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()
@tag_cmd.assign("clone")
async def handle_clone(
bot: Bot,
source_name: Match[str],
new_name: Match[str],
add_groups: Query[list[str] | None] = AlconnaQuery("clone.add.add_groups", None),
remove_groups: Query[list[str] | None] = AlconnaQuery(
"clone.remove.remove_groups", None
),
as_dynamic: Query[bool] = AlconnaQuery("clone.as-dynamic.value", False),
description: Query[str | None] = AlconnaQuery("clone.desc.description", None),
mode: Query[str | None] = AlconnaQuery("clone.mode.mode", None),
):
try:
new_tag = await tag_manager.clone_tag(
source_name=source_name.result,
new_name=new_name.result,
bot=bot,
add_groups=add_groups.result,
remove_groups=remove_groups.result,
as_dynamic=as_dynamic.result,
description=description.result,
mode=mode.result,
)
tag_type_str = "动态" if new_tag.tag_type == "DYNAMIC" else "静态"
group_count = 0
if new_tag.tag_type == "STATIC":
group_count = await new_tag.groups.all().count()
msg = f"✅ 成功克隆标签!\n- 新标签: {new_tag.name}\n- 类型: {tag_type_str}"
if new_tag.tag_type == "STATIC":
msg += f" (含 {group_count} 个群组)"
await MessageUtils.build_message(msg).finish()
except (ValueError, IntegrityError) as e:
await MessageUtils.build_message(f"克隆失败: {e}").finish()
@tag_cmd.assign("prune")
async def handle_prune():
deleted_count = await tag_manager.prune_stale_group_links()
msg = f"清理完成!共移除了 {deleted_count} 个无效的群组关联。"
await MessageUtils.build_message(msg).finish()
+3 -1
View File
@@ -344,7 +344,9 @@ class ConfigsManager:
返回:
ConfigGroup: ConfigGroup
"""
return self._data.get(key) or ConfigGroup(module="")
if key not in self._data:
self._data[key] = ConfigGroup(module=key)
return self._data[key]
def save(self, path: str | Path | None = None, save_simple_data: bool = False):
"""保存数据
+54
View File
@@ -0,0 +1,54 @@
from tortoise import fields
from zhenxun.services.db_context import Model
class GroupTag(Model):
"""群组标签模型"""
id = fields.IntField(pk=True, generated=True, auto_increment=True)
"""自增ID"""
name = fields.CharField(max_length=255, unique=True, description="标签名称")
"""标签名称"""
description = fields.TextField(null=True, description="标签描述")
"""标签描述"""
owner_id = fields.CharField(
max_length=255, null=True, description="创建者ID, null为系统级"
)
"""创建此标签的用户ID"""
bot_id = fields.CharField(
max_length=255, null=True, description="所属Bot ID, null为全局通用"
)
"""此标签所属的Bot ID"""
tag_type = fields.CharField(
max_length=20, default="STATIC", description="标签类型 (STATIC, DYNAMIC)"
)
"""标签类型"""
dynamic_rule = fields.TextField(null=True, description="动态标签的计算规则")
"""动态标签的计算规则"""
is_blacklist = fields.BooleanField(default=False, description="是否为黑名单模式")
"""是否为黑名单模式 (True: 排除模式, False: 包含模式)"""
groups: fields.ReverseRelation["GroupTagLink"]
class Meta: # type: ignore
table = "group_tags"
table_description = "群组标签表"
class GroupTagLink(Model):
"""群组与标签的多对多关联模型"""
id = fields.IntField(pk=True, generated=True, auto_increment=True)
"""自增ID"""
tag = fields.ForeignKeyField(
"models.GroupTag", related_name="groups", on_delete=fields.CASCADE
)
"""关联的标签"""
group_id = fields.CharField(max_length=255, description="群组ID")
"""群组ID"""
class Meta: # type: ignore
table = "group_tag_links"
table_description = "群组标签关联表"
unique_together = ("tag", "group_id")
+37 -19
View File
@@ -5,34 +5,52 @@ from zhenxun.services.db_context import Model
class ScheduledJob(Model):
id = fields.IntField(pk=True, generated=True, auto_increment=True)
"""自增id"""
name = fields.CharField(
max_length=255, null=True, description="任务别名,方便用户辨识"
)
created_by = fields.CharField(
max_length=255, null=True, description="创建任务的用户ID"
)
required_permission = fields.IntField(
default=5, description="管理此任务所需的最低权限等级"
)
source = fields.CharField(
max_length=50, default="USER", description="任务来源 (USER, PLUGIN_DEFAULT)"
)
bot_id = fields.CharField(
255, null=True, default=None, description="任务关联的Bot ID"
255, null=True, description="执行任务的Bot约束 (具体Bot ID或平台)"
)
"""任务关联的Bot ID"""
plugin_name = fields.CharField(255, description="插件模块名")
"""插件模块名"""
group_id = fields.CharField(
255,
null=True,
description="群组ID, '__ALL_GROUPS__' 表示所有群, 为空表示全局任务",
target_type = fields.CharField(
max_length=50, description="目标类型 (GROUP, USER, TAG, ALL_GROUPS, GLOBAL)"
)
"""群组ID, 为空表示全局任务"""
target_identifier = fields.CharField(
max_length=255, description="目标标识符 (群号, 标签名等)"
)
trigger_type = fields.CharField(
max_length=20, default="cron", description="触发器类型 (cron, interval, date)"
)
"""触发器类型 (cron, interval, date)"""
trigger_config = fields.JSONField(description="触发器具体配置")
"""触发器具体配置"""
job_kwargs = fields.JSONField(
default=dict, description="传递给任务函数的额外关键字参数"
)
"""传递给任务函数的额外关键字参数"""
is_enabled = fields.BooleanField(default=True, description="是否启用")
"""是否启用"""
create_time = fields.DatetimeField(auto_now_add=True)
"""创建时间"""
class Meta: # pyright: ignore [reportIncompatibleVariableOverride]
table = "scheduled_jobs"
table_description = "通用定时任务表"
is_enabled = fields.BooleanField(default=True, description="是否启用")
is_one_off = fields.BooleanField(default=False, description="是否为一次性任务")
last_run_at = fields.DatetimeField(null=True, description="上次执行完成时间")
last_run_status = fields.CharField(
max_length=20, null=True, description="上次执行状态 (SUCCESS, FAILURE)"
)
consecutive_failures = fields.IntField(default=0, description="连续失败次数")
execution_options = fields.JSONField(
null=True,
description="任务执行的额外选项 (例如: jitter, spread, "
"interval, concurrency_policy)",
)
create_time = fields.DatetimeField(auto_now_add=True)
class Meta: # type: ignore
table = "scheduled_tasks"
table_description = "通用定时任务定义表"
+9 -1
View File
@@ -45,12 +45,18 @@ from .llm import (
from .log import logger
from .plugin_init import PluginInit, PluginInitManager
from .renderer import renderer_service
from .scheduler import scheduler_manager
from .scheduler import (
ExecutionPolicy,
ScheduleContext,
Trigger,
scheduler_manager,
)
__all__ = [
"AI",
"AIConfig",
"CommonOverrides",
"ExecutionPolicy",
"LLMContentPart",
"LLMException",
"LLMGenerationConfig",
@@ -58,6 +64,8 @@ __all__ = [
"Model",
"PluginInit",
"PluginInitManager",
"ScheduleContext",
"Trigger",
"avatar_service",
"chat",
"clear_model_cache",
+2 -2
View File
@@ -98,6 +98,7 @@ from .cache_containers import CacheDict, CacheList
from .config import (
CACHE_KEY_PREFIX,
CACHE_KEY_SEPARATOR,
CACHE_TIMEOUT,
DEFAULT_EXPIRE,
LOG_COMMAND,
SPECIAL_KEY_FORMATS,
@@ -551,7 +552,6 @@ class CacheManager:
返回:
Any: 缓存数据,如果不存在返回默认值
"""
from zhenxun.services.db_context import DB_TIMEOUT_SECONDS
# 如果缓存被禁用或缓存模式为NONE,直接返回默认值
if not self.enabled or cache_config.cache_mode == CacheMode.NONE:
@@ -561,7 +561,7 @@ class CacheManager:
cache_key = self._build_key(cache_type, key)
data = await asyncio.wait_for(
self.cache_backend.get(cache_key), # type: ignore
timeout=DB_TIMEOUT_SECONDS,
timeout=CACHE_TIMEOUT,
)
if data is None:
+3
View File
@@ -5,6 +5,9 @@
# 日志标识
LOG_COMMAND = "CacheRoot"
# 缓存获取超时时间(秒)
CACHE_TIMEOUT = 10
# 默认缓存过期时间(秒)
DEFAULT_EXPIRE = 600
+4 -1
View File
@@ -27,5 +27,8 @@ async def with_db_timeout(
return result
except asyncio.TimeoutError:
if operation:
logger.error(f"数据库操作超时: {operation} (>{timeout}s)", LOG_COMMAND)
logger.error(
f"数据库操作超时: {operation} (>{timeout}s) 来源: {source}",
LOG_COMMAND,
)
raise
+2
View File
@@ -7,6 +7,7 @@ LLM 服务模块 - 公共 API 入口
from .api import (
chat,
code,
create_image,
embed,
generate,
generate_structured,
@@ -74,6 +75,7 @@ __all__ = [
"chat",
"clear_model_cache",
"code",
"create_image",
"create_multimodal_message",
"embed",
"function_tool",
+44
View File
@@ -3,6 +3,9 @@ LLM 适配器基类和通用数据结构
"""
from abc import ABC, abstractmethod
import base64
import binascii
import json
from typing import TYPE_CHECKING, Any
from pydantic import BaseModel
@@ -32,6 +35,7 @@ class ResponseData(BaseModel):
"""响应数据封装 - 支持所有高级功能"""
text: str
images: list[bytes] | None = None
usage_info: dict[str, Any] | None = None
raw_response: dict[str, Any] | None = None
tool_calls: list[LLMToolCall] | None = None
@@ -242,6 +246,38 @@ class BaseAdapter(ABC):
if content:
content = content.strip()
images_bytes: list[bytes] = []
if content and content.startswith("{") and content.endswith("}"):
try:
content_json = json.loads(content)
if "b64_json" in content_json:
images_bytes.append(base64.b64decode(content_json["b64_json"]))
content = "[图片已生成]"
elif "data" in content_json and isinstance(
content_json["data"], str
):
images_bytes.append(base64.b64decode(content_json["data"]))
content = "[图片已生成]"
except (json.JSONDecodeError, KeyError, binascii.Error):
pass
elif (
"images" in message
and isinstance(message["images"], list)
and message["images"]
):
image_info = message["images"][0]
if image_info.get("type") == "image_url":
image_url_obj = image_info.get("image_url", {})
url_str = image_url_obj.get("url", "")
if url_str.startswith("data:image/png;base64,"):
try:
b64_data = url_str.split(",", 1)[1]
images_bytes.append(base64.b64decode(b64_data))
content = content if content else "[图片已生成]"
except (IndexError, binascii.Error) as e:
logger.warning(f"解析OpenRouter Base64图片数据失败: {e}")
parsed_tool_calls: list[LLMToolCall] | None = None
if message_tool_calls := message.get("tool_calls"):
from ..types.models import LLMToolFunction
@@ -280,6 +316,7 @@ class BaseAdapter(ABC):
text=final_text,
tool_calls=parsed_tool_calls,
usage_info=usage_info,
images=images_bytes if images_bytes else None,
raw_response=response_json,
)
@@ -450,6 +487,13 @@ class OpenAICompatAdapter(BaseAdapter):
"""准备高级请求 - OpenAI兼容格式"""
url = self.get_api_url(model, self.get_chat_endpoint(model))
headers = self.get_base_headers(api_key)
if model.api_type == "openrouter":
headers.update(
{
"HTTP-Referer": "https://github.com/zhenxun-org/zhenxun_bot",
"X-Title": "Zhenxun Bot",
}
)
openai_messages = self.convert_messages_to_openai_format(messages)
body = {
+36 -1
View File
@@ -2,6 +2,7 @@
Gemini API 适配器
"""
import base64
from typing import TYPE_CHECKING, Any
from zhenxun.services.log import logger
@@ -353,6 +354,24 @@ class GeminiAdapter(BaseAdapter):
return safety_settings if safety_settings else None
def validate_response(self, response_json: dict[str, Any]) -> None:
"""验证 Gemini API 响应,增加对 promptFeedback 的检查"""
super().validate_response(response_json)
if prompt_feedback := response_json.get("promptFeedback"):
if block_reason := prompt_feedback.get("blockReason"):
logger.warning(
f"Gemini 内容因 promptFeedback 被安全过滤: {block_reason}"
)
raise LLMException(
f"内容被安全过滤: {block_reason}",
code=LLMErrorCode.CONTENT_FILTERED,
details={
"block_reason": block_reason,
"safety_ratings": prompt_feedback.get("safetyRatings"),
},
)
def parse_response(
self,
model: "LLMModel",
@@ -373,7 +392,16 @@ class GeminiAdapter(BaseAdapter):
self.validate_response(response_json)
try:
candidates = response_json.get("candidates", [])
if "image_generation" in response_json and isinstance(
response_json["image_generation"], dict
):
candidates_source = response_json["image_generation"]
else:
candidates_source = response_json
candidates = candidates_source.get("candidates", [])
usage_info = response_json.get("usageMetadata")
if not candidates:
logger.debug("Gemini响应中没有candidates。")
return ResponseData(text="", raw_response=response_json)
@@ -398,6 +426,7 @@ class GeminiAdapter(BaseAdapter):
parts = content_data.get("parts", [])
text_content = ""
images_bytes: list[bytes] = []
parsed_tool_calls: list["LLMToolCall"] | None = None
thought_summary_parts = []
answer_parts = []
@@ -409,6 +438,11 @@ class GeminiAdapter(BaseAdapter):
thought_summary_parts.append(part["thought"])
elif "thoughtSummary" in part:
thought_summary_parts.append(part["thoughtSummary"])
elif "inlineData" in part:
inline_data = part["inlineData"]
if "data" in inline_data:
images_bytes.append(base64.b64decode(inline_data["data"]))
elif "functionCall" in part:
if parsed_tool_calls is None:
parsed_tool_calls = []
@@ -475,6 +509,7 @@ class GeminiAdapter(BaseAdapter):
return ResponseData(
text=text_content,
tool_calls=parsed_tool_calls,
images=images_bytes if images_bytes else None,
usage_info=usage_info,
raw_response=response_json,
grounding_metadata=grounding_metadata_obj,
+8 -1
View File
@@ -21,7 +21,14 @@ class OpenAIAdapter(OpenAICompatAdapter):
@property
def supported_api_types(self) -> list[str]:
return ["openai", "deepseek", "zhipu", "general_openai_compat", "ark"]
return [
"openai",
"deepseek",
"zhipu",
"general_openai_compat",
"ark",
"openrouter",
]
def get_chat_endpoint(self, model: "LLMModel") -> str:
"""返回聊天完成端点"""
+100 -2
View File
@@ -2,7 +2,8 @@
LLM 服务的高级 API 接口 - 便捷函数入口 (无状态)
"""
from typing import Any, TypeVar
from pathlib import Path
from typing import Any, TypeVar, overload
from nonebot_plugin_alconna.uniseg import UniMessage
from pydantic import BaseModel
@@ -10,7 +11,7 @@ from pydantic import BaseModel
from zhenxun.services.log import logger
from .config import CommonOverrides
from .config.generation import create_generation_config_from_kwargs
from .config.generation import LLMGenerationConfig, create_generation_config_from_kwargs
from .manager import get_model_instance
from .session import AI
from .tools.manager import tool_provider_manager
@@ -23,6 +24,7 @@ from .types import (
LLMResponse,
ModelName,
)
from .utils import create_multimodal_message
T = TypeVar("T", bound=BaseModel)
@@ -303,3 +305,99 @@ async def run_with_tools(
raise LLMException(
"带工具的执行循环未能产生有效的助手回复。", code=LLMErrorCode.GENERATION_FAILED
)
async def _generate_image_from_message(
message: UniMessage,
model: ModelName = None,
**kwargs: Any,
) -> LLMResponse:
"""
[内部] 从 UniMessage 生成图片的核心辅助函数。
"""
from .utils import normalize_to_llm_messages
config = (
create_generation_config_from_kwargs(**kwargs)
if kwargs
else LLMGenerationConfig()
)
config.validation_policy = {"require_image": True}
config.response_modalities = ["IMAGE", "TEXT"]
try:
messages = await normalize_to_llm_messages(message)
async with await get_model_instance(model) as model_instance:
if not model_instance.can_generate_images():
raise LLMException(
f"模型 '{model_instance.provider_name}/{model_instance.model_name}'"
f"不支持图片生成",
code=LLMErrorCode.CONFIGURATION_ERROR,
)
response = await model_instance.generate_response(messages, config=config)
if not response.images:
error_text = response.text or "模型未返回图片数据。"
logger.warning(f"图片生成调用未返回图片,返回文本内容: {error_text}")
return response
except LLMException:
raise
except Exception as e:
logger.error(f"执行图片生成时发生未知错误: {e}", e=e)
raise LLMException(f"图片生成失败: {e}", cause=e)
@overload
async def create_image(
prompt: str | UniMessage,
*,
images: None = None,
model: ModelName = None,
**kwargs: Any,
) -> LLMResponse:
"""根据文本提示生成一张新图片。"""
...
@overload
async def create_image(
prompt: str | UniMessage,
*,
images: list[Path | bytes | str] | Path | bytes | str,
model: ModelName = None,
**kwargs: Any,
) -> LLMResponse:
"""在给定图片的基础上,根据文本提示进行编辑或重新生成。"""
...
async def create_image(
prompt: str | UniMessage,
*,
images: list[Path | bytes | str] | Path | bytes | str | None = None,
model: ModelName = None,
**kwargs: Any,
) -> LLMResponse:
"""
智能图片生成/编辑函数。
- 如果 `images` 为 None,执行文生图。
- 如果提供了 `images`,执行图+文生图,支持多张图片输入。
"""
text_prompt = (
prompt.extract_plain_text() if isinstance(prompt, UniMessage) else str(prompt)
)
image_list = []
if images:
if isinstance(images, list):
image_list.extend(images)
else:
image_list.append(images)
message = create_multimodal_message(text=text_prompt, images=image_list)
return await _generate_image_from_message(message, model=model, **kwargs)
+12 -1
View File
@@ -2,13 +2,15 @@
LLM 生成配置相关类和函数
"""
from collections.abc import Callable
from typing import Any
from pydantic import BaseModel, Field
from pydantic import BaseModel, ConfigDict, Field
from zhenxun.services.log import logger
from zhenxun.utils.pydantic_compat import model_dump
from ..types import LLMResponse
from ..types.enums import ResponseFormat
from ..types.exceptions import LLMErrorCode, LLMException
@@ -64,6 +66,15 @@ class ModelConfigOverride(BaseModel):
custom_params: dict[str, Any] | None = Field(default=None, description="自定义参数")
validation_policy: dict[str, Any] | None = Field(
default=None, description="声明式的响应验证策略 (例如: {'require_image': True})"
)
response_validator: Callable[[LLMResponse], None] | None = Field(
default=None, description="一个高级回调函数,用于验证响应,验证失败时应抛出异常"
)
model_config = ConfigDict(arbitrary_types_allowed=True)
def to_dict(self) -> dict[str, Any]:
"""转换为字典,排除None值"""
+12 -2
View File
@@ -192,10 +192,20 @@ def get_default_providers() -> list[dict[str, Any]]:
"api_base": "https://generativelanguage.googleapis.com",
"api_type": "gemini",
"models": [
{"model_name": "gemini-2.0-flash"},
{"model_name": "gemini-2.5-flash"},
{"model_name": "gemini-2.5-pro"},
{"model_name": "gemini-2.5-flash-lite-preview-06-17"},
{"model_name": "gemini-2.5-flash-lite"},
],
},
{
"name": "OpenRouter",
"api_key": "YOUR_OPENROUTER_API_KEY",
"api_base": "https://openrouter.ai/api",
"api_type": "openrouter",
"models": [
{"model_name": "google/gemini-2.5-pro"},
{"model_name": "google/gemini-2.5-flash"},
{"model_name": "x-ai/grok-4"},
],
},
]
+14 -18
View File
@@ -50,8 +50,8 @@ class LLMHttpClient:
async with self._lock:
if self._client is None or self._client.is_closed:
logger.debug(
f"LLMHttpClient: Initializing new httpx.AsyncClient "
f"with config: {self.config}"
f"LLMHttpClient: 正在初始化新的 httpx.AsyncClient "
f"配置: {self.config}"
)
headers = get_user_agent()
limits = httpx.Limits(
@@ -92,7 +92,7 @@ class LLMHttpClient:
)
if self._client is None:
raise LLMException(
"HTTP client failed to initialize.", LLMErrorCode.CONFIGURATION_ERROR
"HTTP 客户端初始化失败。", LLMErrorCode.CONFIGURATION_ERROR
)
return self._client
@@ -110,17 +110,17 @@ class LLMHttpClient:
async with self._lock:
if self._client and not self._client.is_closed:
logger.debug(
f"LLMHttpClient: Closing with config: {self.config}. "
f"Active requests: {self._active_requests}"
f"LLMHttpClient: 正在关闭,配置: {self.config}. "
f"活跃请求数: {self._active_requests}"
)
if self._active_requests > 0:
logger.warning(
f"LLMHttpClient: Closing while {self._active_requests} "
f"requests are still active."
f"LLMHttpClient: 关闭时仍有 {self._active_requests} "
f"个请求处于活跃状态。"
)
await self._client.aclose()
self._client = None
logger.debug(f"LLMHttpClient for config {self.config} definitively closed.")
logger.debug(f"配置为 {self.config} 的 LLMHttpClient 已完全关闭。")
@property
def is_closed(self) -> bool:
@@ -145,20 +145,17 @@ class LLMHttpClientManager:
client = self._clients.get(key)
if client and not client.is_closed:
logger.debug(
f"LLMHttpClientManager: Reusing existing LLMHttpClient "
f"for key: {key}"
f"LLMHttpClientManager: 复用现有的 LLMHttpClient 密钥: {key}"
)
return client
if client and client.is_closed:
logger.debug(
f"LLMHttpClientManager: Found a closed client for key {key}. "
f"Creating a new one."
f"LLMHttpClientManager: 发现密钥 {key} 对应的客户端已关闭。"
f"正在创建新的客户端。"
)
logger.debug(
f"LLMHttpClientManager: Creating new LLMHttpClient for key: {key}"
)
logger.debug(f"LLMHttpClientManager: 为密钥 {key} 创建新的 LLMHttpClient")
http_client_config = HttpClientConfig(
timeout=provider_config.timeout, proxy=provider_config.proxy
)
@@ -169,8 +166,7 @@ class LLMHttpClientManager:
async def shutdown(self):
async with self._lock:
logger.info(
f"LLMHttpClientManager: Shutting down. "
f"Closing {len(self._clients)} client(s)."
f"LLMHttpClientManager: 正在关闭。关闭 {len(self._clients)} 个客户端。"
)
close_tasks = [
client.close()
@@ -180,7 +176,7 @@ class LLMHttpClientManager:
if close_tasks:
await asyncio.gather(*close_tasks, return_exceptions=True)
self._clients.clear()
logger.info("LLMHttpClientManager: Shutdown complete.")
logger.info("LLMHttpClientManager: 关闭完成。")
http_client_manager = LLMHttpClientManager()
+3 -2
View File
@@ -5,12 +5,12 @@ LLM 模型管理器
"""
import hashlib
import json
import time
from typing import Any
from zhenxun.configs.config import Config
from zhenxun.services.log import logger
from zhenxun.utils.pydantic_compat import dump_json_safely
from .config import validate_override_params
from .config.providers import AI_CONFIG_GROUP, PROVIDERS_CONFIG_KEY, get_ai_config
@@ -43,7 +43,7 @@ def _make_cache_key(
) -> str:
"""生成缓存键"""
config_str = (
json.dumps(override_config, sort_keys=True) if override_config else "None"
dump_json_safely(override_config, sort_keys=True) if override_config else "None"
)
key_data = f"{provider_model_name}:{config_str}"
return hashlib.md5(key_data.encode()).hexdigest()
@@ -118,6 +118,7 @@ def get_default_api_base_for_type(api_type: str) -> str | None:
"deepseek": "https://api.deepseek.com",
"zhipu": "https://open.bigmodel.cn",
"gemini": "https://generativelanguage.googleapis.com",
"openrouter": "https://openrouter.ai/api",
"general_openai_compat": None,
}
+81 -9
View File
@@ -12,6 +12,8 @@ from typing import Any, TypeVar
from pydantic import BaseModel
from zhenxun.services.log import logger
from zhenxun.utils.log_sanitizer import sanitize_for_logging
from zhenxun.utils.pydantic_compat import dump_json_safely
from .adapters.base import RequestData
from .config import LLMGenerationConfig
@@ -34,7 +36,6 @@ from .types import (
ToolExecutable,
)
from .types.capabilities import ModelCapabilities, ModelModality
from .utils import _sanitize_request_body_for_logging
T = TypeVar("T", bound=BaseModel)
@@ -187,21 +188,32 @@ class LLMModel(LLMModelBase):
logger.debug(f"🔑 API密钥: {masked_key}")
logger.debug(f"📋 请求头: {dict(request_data.headers)}")
sanitized_body = _sanitize_request_body_for_logging(request_data.body)
request_body_str = json.dumps(sanitized_body, ensure_ascii=False, indent=2)
sanitizer_req_context_map = {"gemini": "gemini_request"}
sanitizer_req_context = sanitizer_req_context_map.get(
self.api_type, "openai_request"
)
sanitized_body = sanitize_for_logging(
request_data.body, context=sanitizer_req_context
)
request_body_str = dump_json_safely(
sanitized_body, ensure_ascii=False, indent=2
)
logger.debug(f"📦 请求体: {request_body_str}")
http_response = await http_client.post(
request_data.url,
headers=request_data.headers,
json=request_data.body,
content=dump_json_safely(request_data.body, ensure_ascii=False),
)
logger.debug(f"📥 响应状态码: {http_response.status_code}")
logger.debug(f"📄 响应头: {dict(http_response.headers)}")
response_bytes = await http_response.aread()
logger.debug(f"📦 响应体已完整读取 ({len(response_bytes)} bytes)")
if http_response.status_code != 200:
error_text = http_response.text
error_text = response_bytes.decode("utf-8", errors="ignore")
logger.error(
f"❌ HTTP请求失败: {http_response.status_code} - {error_text} "
f"[{log_context}]"
@@ -232,13 +244,22 @@ class LLMModel(LLMModelBase):
)
try:
response_json = http_response.json()
response_json = json.loads(response_bytes)
sanitizer_context_map = {"gemini": "gemini_response"}
sanitizer_context = sanitizer_context_map.get(
self.api_type, "openai_response"
)
sanitized_for_log = sanitize_for_logging(
response_json, context=sanitizer_context
)
response_json_str = json.dumps(
response_json, ensure_ascii=False, indent=2
sanitized_for_log, ensure_ascii=False, indent=2
)
logger.debug(f"📋 响应JSON: {response_json_str}")
parsed_data = parse_response_func(response_json)
except Exception as e:
logger.error(f"解析 {log_context} 响应失败: {e}", e=e)
await self.key_store.record_failure(api_key, None, str(e))
@@ -290,7 +311,7 @@ class LLMModel(LLMModelBase):
adapter.validate_embedding_response(response_json)
return adapter.parse_embedding_response(response_json)
parsed_data, api_key_used = await self._perform_api_call(
parsed_data, _api_key_used = await self._perform_api_call(
prepare_request_func=prepare_request,
parse_response_func=parse_response,
http_client=http_client,
@@ -376,6 +397,7 @@ class LLMModel(LLMModelBase):
return LLMResponse(
text=response_data.text,
usage_info=response_data.usage_info,
images=response_data.images,
raw_response=response_data.raw_response,
tool_calls=response_tool_calls if response_tool_calls else None,
code_executions=response_data.code_executions,
@@ -390,6 +412,56 @@ class LLMModel(LLMModelBase):
failed_keys=failed_keys,
log_context="Generation",
)
if config:
if config.response_validator:
try:
config.response_validator(parsed_data)
except Exception as e:
raise LLMException(
f"响应内容未通过自定义验证器: {e}",
code=LLMErrorCode.API_RESPONSE_INVALID,
details={"validator_error": str(e)},
cause=e,
) from e
policy = config.validation_policy
if policy:
if policy.get("require_image") and not parsed_data.images:
if self.api_type == "gemini" and parsed_data.raw_response:
usage_metadata = parsed_data.raw_response.get(
"usageMetadata", {}
)
prompt_token_details = usage_metadata.get(
"promptTokensDetails", []
)
prompt_had_image = any(
detail.get("modality") == "IMAGE"
for detail in prompt_token_details
)
if prompt_had_image:
raise LLMException(
"响应验证失败:模型接收了图片输入但未生成图片。",
code=LLMErrorCode.API_RESPONSE_INVALID,
details={
"policy": policy,
"text_response": parsed_data.text,
"raw_response": parsed_data.raw_response,
},
)
else:
logger.debug("Gemini提示词中未包含图片,跳过图片要求重试。")
else:
raise LLMException(
"响应验证失败:要求返回图片但未找到图片数据。",
code=LLMErrorCode.API_RESPONSE_INVALID,
details={
"policy": policy,
"text_response": parsed_data.text,
},
)
return parsed_data, api_key_used
async def close(self):
+79 -11
View File
@@ -9,6 +9,8 @@ import fnmatch
from pydantic import BaseModel, Field
from zhenxun.services.log import logger
class ModelModality(str, Enum):
TEXT = "text"
@@ -44,6 +46,53 @@ GEMINI_CAPABILITIES = ModelCapabilities(
supports_tool_calling=True,
)
GEMINI_IMAGE_GEN_CAPABILITIES = ModelCapabilities(
input_modalities={ModelModality.TEXT, ModelModality.IMAGE},
output_modalities={ModelModality.TEXT, ModelModality.IMAGE},
supports_tool_calling=True,
)
GPT_ADVANCED_TEXT_IMAGE_CAPABILITIES = ModelCapabilities(
input_modalities={ModelModality.TEXT, ModelModality.IMAGE},
output_modalities={ModelModality.TEXT},
supports_tool_calling=True,
)
GPT_MULTIMODAL_IO_CAPABILITIES = ModelCapabilities(
input_modalities={ModelModality.TEXT, ModelModality.AUDIO, ModelModality.IMAGE},
output_modalities={ModelModality.TEXT, ModelModality.AUDIO},
supports_tool_calling=True,
)
GPT_IMAGE_GENERATION_CAPABILITIES = ModelCapabilities(
input_modalities={ModelModality.TEXT, ModelModality.IMAGE},
output_modalities={ModelModality.IMAGE},
supports_tool_calling=True,
)
GPT_VIDEO_GENERATION_CAPABILITIES = ModelCapabilities(
input_modalities={ModelModality.TEXT, ModelModality.IMAGE, ModelModality.VIDEO},
output_modalities={ModelModality.VIDEO},
supports_tool_calling=True,
)
DEFAULT_PERMISSIVE_CAPABILITIES = ModelCapabilities(
input_modalities={
ModelModality.TEXT,
ModelModality.IMAGE,
ModelModality.AUDIO,
ModelModality.VIDEO,
},
output_modalities={
ModelModality.TEXT,
ModelModality.IMAGE,
ModelModality.AUDIO,
ModelModality.VIDEO,
},
supports_tool_calling=True,
)
DOUBAO_ADVANCED_MULTIMODAL_CAPABILITIES = ModelCapabilities(
input_modalities={ModelModality.TEXT, ModelModality.IMAGE, ModelModality.VIDEO},
output_modalities={ModelModality.TEXT},
@@ -83,11 +132,9 @@ MODEL_CAPABILITIES_REGISTRY: dict[str, ModelCapabilities] = {
output_modalities={ModelModality.EMBEDDING},
is_embedding_model=True,
),
"gemini-2.5-pro*": GEMINI_CAPABILITIES,
"gemini-1.5-pro*": GEMINI_CAPABILITIES,
"gemini-2.5-flash*": GEMINI_CAPABILITIES,
"gemini-2.0-flash*": GEMINI_CAPABILITIES,
"gemini-1.5-flash*": GEMINI_CAPABILITIES,
"*gemini-*-image-preview*": GEMINI_IMAGE_GEN_CAPABILITIES,
"gemini-*-pro*": GEMINI_CAPABILITIES,
"gemini-*-flash*": GEMINI_CAPABILITIES,
"GLM-4V-Flash": ModelCapabilities(
input_modalities={ModelModality.TEXT, ModelModality.IMAGE},
output_modalities={ModelModality.TEXT},
@@ -104,6 +151,13 @@ MODEL_CAPABILITIES_REGISTRY: dict[str, ModelCapabilities] = {
"doubao-1-5-thinking-vision-pro": DOUBAO_ADVANCED_MULTIMODAL_CAPABILITIES,
"deepseek-chat": STANDARD_TEXT_TOOL_CAPABILITIES,
"deepseek-reasoner": STANDARD_TEXT_TOOL_CAPABILITIES,
"gpt-5*": GPT_ADVANCED_TEXT_IMAGE_CAPABILITIES,
"gpt-4.1*": GPT_ADVANCED_TEXT_IMAGE_CAPABILITIES,
"gpt-4o*": GPT_MULTIMODAL_IO_CAPABILITIES,
"o3*": GPT_ADVANCED_TEXT_IMAGE_CAPABILITIES,
"o4-mini*": GPT_ADVANCED_TEXT_IMAGE_CAPABILITIES,
"gpt image*": GPT_IMAGE_GENERATION_CAPABILITIES,
"sora*": GPT_VIDEO_GENERATION_CAPABILITIES,
}
@@ -118,11 +172,25 @@ def get_model_capabilities(model_name: str) -> ModelCapabilities:
canonical_name = c_name
break
if canonical_name in MODEL_CAPABILITIES_REGISTRY:
return MODEL_CAPABILITIES_REGISTRY[canonical_name]
parts = canonical_name.split("/")
names_to_check = ["/".join(parts[i:]) for i in range(len(parts))]
for pattern, capabilities in MODEL_CAPABILITIES_REGISTRY.items():
if "*" in pattern and fnmatch.fnmatch(model_name, pattern):
return capabilities
logger.trace(f"为 '{model_name}' 生成的检查列表: {names_to_check}")
return ModelCapabilities()
for name in names_to_check:
if name in MODEL_CAPABILITIES_REGISTRY:
logger.debug(f"模型 '{model_name}' 通过精确匹配 '{name}' 找到能力定义。")
return MODEL_CAPABILITIES_REGISTRY[name]
for pattern, capabilities in MODEL_CAPABILITIES_REGISTRY.items():
if "*" in pattern and fnmatch.fnmatch(name, pattern):
logger.debug(
f"模型 '{model_name}' 通过通配符匹配 '{name}'(pattern: '{pattern}')"
f"找到能力定义。"
)
return capabilities
logger.warning(
f"模型 '{model_name}' 的能力定义未在注册表中找到,将使用默认的'全功能'回退配置"
)
return DEFAULT_PERMISSIVE_CAPABILITIES
+1
View File
@@ -425,6 +425,7 @@ class LLMResponse(BaseModel):
"""LLM 响应"""
text: str
images: list[bytes] | None = None
usage_info: dict[str, Any] | None = None
raw_response: dict[str, Any] | None = None
tool_calls: list[Any] | None = None
-48
View File
@@ -273,54 +273,6 @@ def message_to_unimessage(message: PlatformMessage) -> UniMessage:
return UniMessage(uni_segments)
def _sanitize_request_body_for_logging(body: dict) -> dict:
"""
净化请求体用于日志记录,移除大数据字段并添加摘要信息
参数:
body: 原始请求体字典。
返回:
dict: 净化后的请求体字典。
"""
try:
sanitized_body = copy.deepcopy(body)
if "contents" in sanitized_body and isinstance(
sanitized_body["contents"], list
):
for content_item in sanitized_body["contents"]:
if "parts" in content_item and isinstance(content_item["parts"], list):
media_summary = []
new_parts = []
for part in content_item["parts"]:
if "inlineData" in part and isinstance(
part["inlineData"], dict
):
data = part["inlineData"].get("data")
if isinstance(data, str):
mime_type = part["inlineData"].get(
"mimeType", "unknown"
)
media_summary.append(f"{mime_type} ({len(data)} chars)")
continue
new_parts.append(part)
if media_summary:
summary_text = (
f"[多模态内容: {len(media_summary)}个文件 - "
f"{', '.join(media_summary)}]"
)
new_parts.insert(0, {"text": summary_text})
content_item["parts"] = new_parts
return sanitized_body
except Exception as e:
logger.warning(f"日志净化失败: {e},将记录原始请求体。")
return body
def sanitize_schema_for_llm(schema: Any, api_type: str) -> Any:
"""
递归地净化 JSON Schema,移除特定 LLM API 不支持的关键字。
+1 -1
View File
@@ -40,7 +40,7 @@ class Renderable(ABC):
@abstractmethod
def get_children(self) -> Iterable["Renderable"]:
"""
[新增] 返回一个包含所有直接子组件的可迭代对象。
返回一个包含所有直接子组件的可迭代对象。
这使得渲染服务能够递归地遍历整个组件树,以执行依赖收集(CSS、JS)等任务。
非容器组件应返回一个空列表。
+54 -23
View File
@@ -22,6 +22,7 @@ from zhenxun.configs.config import Config
from zhenxun.configs.path_config import THEMES_PATH, UI_CACHE_PATH
from zhenxun.services.log import logger
from zhenxun.utils.exception import RenderingError
from zhenxun.utils.log_sanitizer import sanitize_for_logging
from zhenxun.utils.pydantic_compat import _dump_pydantic_obj
from .config import RESERVED_TEMPLATE_KEYS
@@ -74,6 +75,7 @@ class RendererService:
self._custom_globals: dict[str, Callable] = {}
self.filter("dump_json")(self._pydantic_tojson_filter)
self.global_function("inline_asset")(self._inline_asset_global)
def _create_jinja_env(self) -> Environment:
"""
@@ -175,9 +177,24 @@ class RendererService:
return decorator
async def _inline_asset_global(self, namespaced_path: str) -> str:
"""
一个Jinja2全局函数,用于读取并内联一个已注册命名空间下的资源文件内容。
主要用于内联SVG,以解决浏览器的跨域安全问题。
"""
if not self._jinja_env or not self._jinja_env.loader:
return f"<!-- Error: Jinja env not ready for {namespaced_path} -->"
try:
source, _, _ = self._jinja_env.loader.get_source(
self._jinja_env, namespaced_path
)
return source
except TemplateNotFound:
return f"<!-- Asset not found: {namespaced_path} -->"
async def initialize(self):
"""
[新增] 延迟初始化方法,在 on_startup 钩子中调用。
延迟初始化方法,在 on_startup 钩子中调用。
负责初始化截图引擎和主题管理器,确保在首次渲染前所有依赖都已准备就绪。
使用锁来防止并发初始化。
@@ -216,32 +233,42 @@ class RendererService:
context.processed_components.add(component_id)
component_path_base = str(component.template_name)
variant = getattr(component, "variant", None)
manifest = await context.theme_manager.get_template_manifest(
component_path_base
component_path_base, skin=variant
)
style_paths_to_load = []
if manifest and manifest.styles:
styles = (
[manifest.styles]
if isinstance(manifest.styles, str)
else manifest.styles
)
for style_path in styles:
full_style_path = str(Path(component_path_base) / style_path).replace(
"\\", "/"
if manifest and manifest.get("styles"):
styles = manifest["styles"]
styles = [styles] if isinstance(styles, str) else styles
resolution_base_path = Path(component_path_base)
if variant:
skin_manifest_path = str(Path(component_path_base) / "skins" / variant)
skin_manifest = await context.theme_manager._load_single_manifest(
skin_manifest_path
)
style_paths_to_load.append(full_style_path)
if skin_manifest and "styles" in skin_manifest:
resolution_base_path = Path(skin_manifest_path)
style_paths_to_load.extend(
str(resolution_base_path / style).replace("\\", "/") for style in styles
)
else:
resolved_template_name = (
base_template_path = (
await context.theme_manager._resolve_component_template(
component, context
)
)
conventional_style_path = str(
Path(resolved_template_name).with_name("style.css")
base_style_path = str(
Path(base_template_path).with_name("style.css")
).replace("\\", "/")
style_paths_to_load.append(conventional_style_path)
style_paths_to_load.append(base_style_path)
if variant:
skin_style_path = f"{component_path_base}/skins/{variant}/style.css"
style_paths_to_load.append(skin_style_path)
for css_template_path in style_paths_to_load:
try:
@@ -382,6 +409,7 @@ class RendererService:
)
temp_env.globals.update(context.theme_manager.jinja_env.globals)
temp_env.filters.update(context.theme_manager.jinja_env.filters)
temp_env.globals["asset"] = (
context.theme_manager._create_standalone_asset_loader(template_dir)
)
@@ -430,10 +458,11 @@ class RendererService:
component_render_options = {}
manifest_options = {}
variant = getattr(component, "variant", None)
if manifest := await context.theme_manager.get_template_manifest(
component.template_name
component.template_name, skin=variant
):
manifest_options = manifest.render_options or {}
manifest_options = manifest.get("render_options", {})
final_render_options = component_render_options.copy()
final_render_options.update(manifest_options)
@@ -470,10 +499,7 @@ class RendererService:
) from e
async def render(
self,
component: Renderable,
use_cache: bool = False,
**render_options,
self, component: Renderable, use_cache: bool = False, **render_options
) -> bytes:
"""
统一的、多态的渲染入口,直接返回图片字节。
@@ -504,9 +530,12 @@ class RendererService:
)
result = await self._render_component(context)
if Config.get_config("UI", "DEBUG_MODE") and result.html_content:
sanitized_html = sanitize_for_logging(
result.html_content, context="ui_html"
)
logger.info(
f"--- [UI DEBUG] HTML for {component.__class__.__name__} ---\n"
f"{result.html_content}\n"
f"{sanitized_html}\n"
f"--- [UI DEBUG] End of HTML ---"
)
if result.image_bytes is None:
@@ -556,6 +585,8 @@ class RendererService:
await self.initialize()
assert self._theme_manager is not None, "ThemeManager 未初始化"
self._theme_manager._manifest_cache.clear()
logger.debug("已清除UI清单缓存 (manifest cache)。")
current_theme_name = Config.get_config("UI", "THEME", "default")
await self._theme_manager.load_theme(current_theme_name)
logger.info(f"主题 '{current_theme_name}' 已成功重载。")
+159 -24
View File
@@ -1,11 +1,11 @@
from __future__ import annotations
import asyncio
from collections.abc import Callable
import os
from pathlib import Path
from typing import TYPE_CHECKING, Any
import aiofiles
from jinja2 import (
ChoiceLoader,
Environment,
@@ -21,7 +21,6 @@ import ujson as json
from zhenxun.configs.path_config import THEMES_PATH
from zhenxun.services.log import logger
from zhenxun.services.renderer.models import TemplateManifest
from zhenxun.services.renderer.protocols import Renderable
from zhenxun.services.renderer.registry import asset_registry
from zhenxun.utils.pydantic_compat import model_dump
@@ -32,6 +31,20 @@ if TYPE_CHECKING:
from .config import RESERVED_TEMPLATE_KEYS
def deep_merge_dict(base: dict, new: dict) -> dict:
"""
递归地将 new 字典合并到 base 字典中。
new 字典中的值会覆盖 base 字典中的值。
"""
result = base.copy()
for key, value in new.items():
if isinstance(value, dict) and key in result and isinstance(result[key], dict):
result[key] = deep_merge_dict(result[key], value)
else:
result[key] = value
return result
class RelativePathEnvironment(Environment):
"""
一个自定义的 Jinja2 环境,重写了 join_path 方法以支持模板间的相对路径引用。
@@ -151,14 +164,63 @@ class ResourceResolver:
def resolve_asset_uri(self, asset_path: str, current_template_name: str) -> str:
"""解析资源路径,实现完整的回退逻辑,并返回可用的URI。"""
if not self.theme_manager.current_theme:
if (
not self.theme_manager.current_theme
or not self.theme_manager.jinja_env.loader
):
return ""
if asset_path.startswith("@"):
try:
if "/" not in asset_path:
raise TemplateNotFound(f"无效的命名空间路径: {asset_path}")
namespace, rel_path = asset_path.split("/", 1)
loader = self.theme_manager.jinja_env.loader
if (
isinstance(loader, ChoiceLoader)
and loader.loaders
and isinstance(loader.loaders[0], PrefixLoader)
):
prefix_loader = loader.loaders[0]
if namespace in prefix_loader.mapping:
loader_for_namespace = prefix_loader.mapping[namespace]
if isinstance(loader_for_namespace, FileSystemLoader):
base_path = Path(loader_for_namespace.searchpath[0])
file_abs_path = (base_path / rel_path).resolve()
if file_abs_path.is_file():
logger.debug(
f"Resolved namespaced asset"
f" '{asset_path}' -> '{file_abs_path}'"
)
return file_abs_path.as_uri()
else:
raise TemplateNotFound(asset_path)
else:
raise TemplateNotFound(
f"Unsupported loader type for namespace '{namespace}'."
)
else:
raise TemplateNotFound(f"Namespace '{namespace}' not found.")
else:
raise TemplateNotFound(
f"无法解析命名空间资源 '{asset_path}',加载器结构不符合预期。"
)
except TemplateNotFound:
logger.warning(f"资源文件在命名空间中未找到: '{asset_path}'")
return ""
search_paths: list[tuple[str, Path]] = []
if asset_path.startswith("./"):
if asset_path.startswith("./") or asset_path.startswith("../"):
relative_part = (
asset_path[2:] if asset_path.startswith("./") else asset_path
)
search_paths.extend(
self._search_paths_for_relative_asset(
asset_path[2:], current_template_name
relative_part, current_template_name
)
)
else:
@@ -209,6 +271,9 @@ class ThemeManager:
self.jinja_env.filters["md"] = self._markdown_filter
self._manifest_cache: dict[str, Any] = {}
self._manifest_cache_lock = asyncio.Lock()
def list_available_themes(self) -> list[str]:
"""扫描主题目录并返回所有可用的主题名称。"""
if not THEMES_PATH.is_dir():
@@ -377,16 +442,26 @@ class ThemeManager:
logger.error(f"指定的模板文件路径不存在: '{component_path_base}'", e=e)
raise e
entrypoint_filename = "main.html"
manifest = await self.get_template_manifest(component_path_base)
if manifest and manifest.entrypoint:
entrypoint_filename = manifest.entrypoint
base_manifest = await self.get_template_manifest(component_path_base)
skin_to_use = variant or (base_manifest.get("skin") if base_manifest else None)
final_manifest = await self.get_template_manifest(
component_path_base, skin=skin_to_use
)
logger.debug(f"final_manifest: {final_manifest}")
entrypoint_filename = (
final_manifest.get("entrypoint", "main.html")
if final_manifest
else "main.html"
)
potential_paths = []
if variant:
if skin_to_use:
potential_paths.append(
f"{component_path_base}/skins/{variant}/{entrypoint_filename}"
f"{component_path_base}/skins/{skin_to_use}/{entrypoint_filename}"
)
potential_paths.append(f"{component_path_base}/{entrypoint_filename}")
@@ -410,28 +485,88 @@ class ThemeManager:
logger.error(err_msg)
raise TemplateNotFound(err_msg)
async def get_template_manifest(
self, component_path: str
) -> TemplateManifest | None:
"""
查找并解析组件的 manifest.json 文件。
"""
manifest_path_str = f"{component_path}/manifest.json"
async def _load_single_manifest(self, path_str: str) -> dict[str, Any] | None:
"""从指定路径加载单个 manifest.json 文件。"""
normalized_path = path_str.replace("\\", "/")
manifest_path_str = f"{normalized_path}/manifest.json"
if not self.jinja_env.loader:
return None
try:
_, full_path, _ = self.jinja_env.loader.get_source(
source, filepath, _ = self.jinja_env.loader.get_source(
self.jinja_env, manifest_path_str
)
if full_path and Path(full_path).exists():
async with aiofiles.open(full_path, encoding="utf-8") as f:
manifest_data = json.loads(await f.read())
return TemplateManifest(**manifest_data)
logger.debug(f"找到清单文件: '{manifest_path_str}' (从 '{filepath}' 加载)")
return json.loads(source)
except TemplateNotFound:
logger.trace(f"未找到清单文件: '{manifest_path_str}'")
return None
return None
except json.JSONDecodeError:
logger.warning(f"清单文件 '{manifest_path_str}' 解析失败")
return None
async def _load_and_merge_manifests(
self, component_path: Path | str, skin: str | None = None
) -> dict[str, Any] | None:
"""加载基础和皮肤清单并进行合并。"""
logger.debug(f"开始加载清单: component_path='{component_path}', skin='{skin}'")
base_manifest = await self._load_single_manifest(str(component_path))
if skin:
skin_path = Path(component_path) / "skins" / skin
skin_manifest = await self._load_single_manifest(str(skin_path))
if skin_manifest:
if base_manifest:
merged = deep_merge_dict(base_manifest, skin_manifest)
logger.debug(
f"已合并基础清单和皮肤清单: '{component_path}' + skin '{skin}'"
)
return merged
else:
logger.debug(f"只找到皮肤清单: '{skin_path}'")
return skin_manifest
if base_manifest:
logger.debug(f"只找到基础清单: '{component_path}'")
else:
logger.debug(f"未找到任何清单: '{component_path}'")
return base_manifest
async def get_template_manifest(
self, component_path: str, skin: str | None = None
) -> dict[str, Any] | None:
"""
查找并解析组件的 manifest.json 文件。
支持皮肤清单的继承与合并,并带有缓存。
Args:
component_path: 组件路径
skin: 皮肤名称(可选)
Returns:
合并后的清单字典,如果不存在则返回 None
"""
cache_key = f"{component_path}:{skin or 'base'}"
if cache_key in self._manifest_cache:
logger.debug(f"清单缓存命中: '{cache_key}'")
return self._manifest_cache[cache_key]
async with self._manifest_cache_lock:
if cache_key in self._manifest_cache:
logger.debug(f"清单缓存命中(锁内): '{cache_key}'")
return self._manifest_cache[cache_key]
manifest = await self._load_and_merge_manifests(component_path, skin)
self._manifest_cache[cache_key] = manifest
logger.debug(f"清单已缓存: '{cache_key}'")
return manifest
async def resolve_markdown_style_path(
self, style_name: str, context: "RenderContext"
+4 -5
View File
@@ -4,11 +4,10 @@
提供一个统一的、持久化的定时任务管理器,供所有插件使用。
"""
from .job import ScheduleContext
from .lifecycle import _load_schedules_from_db
from .service import ExecutionPolicy, scheduler_manager
from .triggers import Trigger
from . import lifecycle
from .manager import scheduler_manager
from .types import ExecutionPolicy, ScheduleContext, Trigger
_ = _load_schedules_from_db
_ = lifecycle
__all__ = ["ExecutionPolicy", "ScheduleContext", "Trigger", "scheduler_manager"]
-174
View File
@@ -1,174 +0,0 @@
"""
引擎适配层 (Adapter)
封装所有对具体调度器引擎 (APScheduler) 的操作,
使上层服务与调度器实现解耦。
"""
from collections.abc import Callable
from nonebot_plugin_apscheduler import scheduler
from zhenxun.models.scheduled_job import ScheduledJob
from zhenxun.services.log import logger
from .job import ScheduleContext, _execute_job
JOB_PREFIX = "zhenxun_schedule_"
class APSchedulerAdapter:
"""封装对 APScheduler 的操作"""
@staticmethod
def _get_job_id(schedule_id: int) -> str:
"""
生成 APScheduler 的 Job ID
参数:
schedule_id: 定时任务的ID。
返回:
str: APScheduler 使用的 Job ID。
"""
return f"{JOB_PREFIX}{schedule_id}"
@staticmethod
def add_or_reschedule_job(schedule: ScheduledJob):
"""
根据 ScheduledJob 添加或重新调度一个 APScheduler 任务
参数:
schedule: 定时任务对象,包含任务的所有配置信息。
"""
job_id = APSchedulerAdapter._get_job_id(schedule.id)
if not isinstance(schedule.trigger_config, dict):
logger.error(
f"任务 {schedule.id} 的 trigger_config 不是字典类型: "
f"{type(schedule.trigger_config)}"
)
return
job = scheduler.get_job(job_id)
if job:
scheduler.reschedule_job(
job_id, trigger=schedule.trigger_type, **schedule.trigger_config
)
logger.debug(f"已更新APScheduler任务: {job_id}")
else:
scheduler.add_job(
_execute_job,
trigger=schedule.trigger_type,
id=job_id,
misfire_grace_time=300,
args=[schedule.id],
**schedule.trigger_config,
)
logger.debug(f"已添加新的APScheduler任务: {job_id}")
@staticmethod
def remove_job(schedule_id: int):
"""
移除一个 APScheduler 任务
参数:
schedule_id: 要移除的定时任务ID。
"""
job_id = APSchedulerAdapter._get_job_id(schedule_id)
try:
scheduler.remove_job(job_id)
logger.debug(f"已从APScheduler中移除任务: {job_id}")
except Exception:
pass
@staticmethod
def pause_job(schedule_id: int):
"""
暂停一个 APScheduler 任务
参数:
schedule_id: 要暂停的定时任务ID。
"""
job_id = APSchedulerAdapter._get_job_id(schedule_id)
try:
scheduler.pause_job(job_id)
except Exception:
pass
@staticmethod
def resume_job(schedule_id: int):
"""
恢复一个 APScheduler 任务
参数:
schedule_id: 要恢复的定时任务ID。
"""
job_id = APSchedulerAdapter._get_job_id(schedule_id)
try:
scheduler.resume_job(job_id)
except Exception:
import asyncio
from .repository import ScheduleRepository
async def _re_add_job():
schedule = await ScheduleRepository.get_by_id(schedule_id)
if schedule:
APSchedulerAdapter.add_or_reschedule_job(schedule)
asyncio.create_task(_re_add_job()) # noqa: RUF006
@staticmethod
def get_job_status(schedule_id: int) -> dict:
"""
获取 APScheduler Job 的状态
参数:
schedule_id: 定时任务的ID。
返回:
dict: 包含任务状态信息的字典,包含next_run_time等字段。
"""
job_id = APSchedulerAdapter._get_job_id(schedule_id)
job = scheduler.get_job(job_id)
return {
"next_run_time": job.next_run_time.strftime("%Y-%m-%d %H:%M:%S")
if job and job.next_run_time
else "N/A",
"is_paused_in_scheduler": not bool(job.next_run_time) if job else "N/A",
}
@staticmethod
def add_ephemeral_job(
job_id: str,
func: Callable,
trigger_type: str,
trigger_config: dict,
context: ScheduleContext,
):
"""
直接向 APScheduler 添加一个临时的、非持久化的任务
参数:
job_id: 临时任务的唯一ID。
func: 要执行的函数。
trigger_type: 触发器类型。
trigger_config: 触发器配置字典。
context: 任务执行上下文。
"""
job = scheduler.get_job(job_id)
if job:
logger.warning(f"尝试添加一个已存在的临时任务ID: {job_id},操作被忽略。")
return
scheduler.add_job(
_execute_job,
trigger=trigger_type,
id=job_id,
misfire_grace_time=60,
args=[None],
kwargs={"context_override": context},
**trigger_config,
)
logger.debug(f"已添加新的临时APScheduler任务: {job_id}")
+482
View File
@@ -0,0 +1,482 @@
"""
引擎适配层 (Adapter) 与 任务执行逻辑 (Job)
封装所有对具体调度器引擎 (APScheduler) 的操作,
以及被 APScheduler 实际调度的函数。
"""
import asyncio
from collections.abc import Callable
from datetime import datetime
from functools import partial
import random
import nonebot
from nonebot.adapters import Bot
from nonebot.dependencies import Dependent
from nonebot.exception import FinishedException, PausedException, SkippedException
from nonebot.matcher import Matcher
from nonebot.typing import T_State
from nonebot_plugin_apscheduler import scheduler
from pydantic import BaseModel
from zhenxun.configs.config import Config
from zhenxun.models.scheduled_job import ScheduledJob
from zhenxun.services.log import logger
from zhenxun.utils.common_utils import CommonUtils
from zhenxun.utils.decorator.retry import Retry
from zhenxun.utils.pydantic_compat import parse_as
from .repository import ScheduleRepository
from .types import ExecutionPolicy, ScheduleContext
JOB_PREFIX = "zhenxun_schedule_"
SCHEDULE_CONCURRENCY_KEY = "all_groups_concurrency_limit"
class APSchedulerAdapter:
"""封装对 APScheduler 的操作"""
@staticmethod
def _get_job_id(schedule_id: int) -> str:
"""
生成 APScheduler 的 Job ID
参数:
schedule_id: 定时任务的ID。
返回:
str: APScheduler 使用的 Job ID。
"""
return f"{JOB_PREFIX}{schedule_id}"
@staticmethod
def add_or_reschedule_job(schedule: ScheduledJob):
"""
根据 ScheduledJob 添加或重新调度一个 APScheduler 任务
参数:
schedule: 定时任务对象,包含任务的所有配置信息。
"""
job_id = APSchedulerAdapter._get_job_id(schedule.id)
try:
scheduler.remove_job(job_id)
except Exception:
pass
if not isinstance(schedule.trigger_config, dict):
logger.error(
f"任务 {schedule.id} 的 trigger_config 不是字典类型: "
f"{type(schedule.trigger_config)}"
)
return
trigger_params = schedule.trigger_config.copy()
execution_options = (
schedule.execution_options
if isinstance(schedule.execution_options, dict)
else {}
)
if jitter := execution_options.get("jitter"):
if isinstance(jitter, int) and jitter > 0:
trigger_params["jitter"] = jitter
concurrency_policy = execution_options.get("concurrency_policy", "ALLOW")
job_params = {
"id": job_id,
"misfire_grace_time": 300,
"args": [schedule.id],
}
if concurrency_policy == "SKIP":
job_params["max_instances"] = 1
job_params["coalesce"] = True
elif concurrency_policy == "QUEUE":
job_params["max_instances"] = 1
job_params["coalesce"] = False
scheduler.add_job(
_execute_job,
trigger=schedule.trigger_type,
**job_params,
**trigger_params,
)
logger.debug(
f"已添加或更新APScheduler任务: {job_id} | 并发策略: {concurrency_policy}, "
f"抖动: {trigger_params.get('jitter', '无')}"
)
@staticmethod
def remove_job(schedule_id: int):
"""
移除一个 APScheduler 任务
参数:
schedule_id: 要移除的定时任务ID。
"""
job_id = APSchedulerAdapter._get_job_id(schedule_id)
try:
scheduler.remove_job(job_id)
logger.debug(f"已从APScheduler中移除任务: {job_id}")
except Exception:
pass
@staticmethod
def pause_job(schedule_id: int):
"""
暂停一个 APScheduler 任务
参数:
schedule_id: 要暂停的定时任务ID。
"""
job_id = APSchedulerAdapter._get_job_id(schedule_id)
try:
scheduler.pause_job(job_id)
except Exception:
pass
@staticmethod
def resume_job(schedule_id: int):
"""
恢复一个 APScheduler 任务
参数:
schedule_id: 要恢复的定时任务ID。
"""
job_id = APSchedulerAdapter._get_job_id(schedule_id)
try:
scheduler.resume_job(job_id)
except Exception:
import asyncio
async def _re_add_job():
schedule = await ScheduleRepository.get_by_id(schedule_id)
if schedule:
APSchedulerAdapter.add_or_reschedule_job(schedule)
asyncio.create_task(_re_add_job()) # noqa: RUF006
@staticmethod
def get_job_status(schedule_id: int) -> dict:
"""
获取 APScheduler Job 的状态
参数:
schedule_id: 定时任务的ID。
返回:
dict: 包含任务状态信息的字典,包含next_run_time等字段。
"""
job_id = APSchedulerAdapter._get_job_id(schedule_id)
job = scheduler.get_job(job_id)
return {
"next_run_time": job.next_run_time.strftime("%Y-%m-%d %H:%M:%S")
if job and job.next_run_time
else "N/A",
"is_paused_in_scheduler": not bool(job.next_run_time) if job else "N/A",
}
@staticmethod
def add_ephemeral_job(
job_id: str,
func: Callable,
trigger_type: str,
trigger_config: dict,
context: ScheduleContext,
):
"""
直接向 APScheduler 添加一个临时的、非持久化的任务
参数:
job_id: 临时任务的唯一ID。
func: 要执行的函数。
trigger_type: 触发器类型。
trigger_config: 触发器配置字典。
context: 任务执行上下文。
"""
job = scheduler.get_job(job_id)
if job:
logger.warning(f"尝试添加一个已存在的临时任务ID: {job_id},操作被忽略。")
return
scheduler.add_job(
_execute_job,
trigger=trigger_type,
id=job_id,
misfire_grace_time=60,
args=[None],
kwargs={"context_override": context},
**trigger_config,
)
logger.debug(f"已添加新的临时APScheduler任务: {job_id}")
async def _execute_single_job_instance(
schedule: ScheduledJob, bot, group_id: str | None = None
):
"""
负责执行一个具体目标的任务实例。
"""
from .manager import scheduler_manager
plugin_name = schedule.plugin_name
if group_id is None and schedule.target_type == "GROUP":
group_id = schedule.target_identifier
task_meta = scheduler_manager._registered_tasks.get(plugin_name)
if not task_meta:
logger.error(f"无法执行任务:插件 '{plugin_name}' 在执行期间变得不可用。")
return
is_blocked = await CommonUtils.task_is_block(bot, plugin_name, group_id)
if is_blocked:
target_desc = f"群 {group_id}" if group_id else "全局"
logger.info(
f"插件 '{plugin_name}' 的定时任务在目标 [{target_desc}] "
f"因功能被禁用而跳过执行。"
)
return
context = ScheduleContext(
schedule_id=schedule.id,
plugin_name=plugin_name,
bot_id=bot.self_id,
group_id=group_id,
job_kwargs=schedule.job_kwargs if isinstance(schedule.job_kwargs, dict) else {},
)
state: T_State = {ScheduleContext: context}
policy_data = context.job_kwargs.pop("execution_policy", {})
policy = ExecutionPolicy(**policy_data)
async def task_execution_coro():
injected_params = {"context": context}
params_model = task_meta.get("model")
if params_model and isinstance(context.job_kwargs, dict):
try:
if isinstance(params_model, type) and issubclass(
params_model, BaseModel
):
params_instance = parse_as(params_model, context.job_kwargs)
injected_params["params"] = params_instance # type: ignore
except Exception as e:
logger.error(
f"任务 {schedule.id} (目标: {group_id}) 参数验证失败: {e}", e=e
)
raise
async def wrapper(bot: Bot):
return await task_meta["func"](bot=bot, **injected_params) # type: ignore
dependent = Dependent.parse(
call=wrapper,
allow_types=Matcher.HANDLER_PARAM_TYPES,
)
return await dependent(bot=bot, state=state)
try:
if policy.retries > 0:
on_success_handler = None
if policy.on_success_callback:
on_success_handler = partial(policy.on_success_callback, context)
on_failure_handler = None
if policy.on_failure_callback:
on_failure_handler = partial(policy.on_failure_callback, context)
retry_exceptions = tuple(policy.retry_on_exceptions or [])
retry_decorator = Retry.api(
stop_max_attempt=policy.retries + 1,
strategy="exponential" if policy.retry_backoff else "fixed",
wait_fixed_seconds=policy.retry_delay_seconds,
exception=retry_exceptions,
on_success=on_success_handler,
on_failure=on_failure_handler,
log_name=f"ScheduledJob-{schedule.id}-{group_id or 'global'}",
)
decorated_executor = retry_decorator(task_execution_coro)
await decorated_executor()
else:
logger.info(
f"插件 '{plugin_name}' 开始为目标 [{group_id or '全局'}] "
f"执行定时任务 (ID: {schedule.id})。"
)
await task_execution_coro()
except (PausedException, FinishedException, SkippedException) as e:
logger.warning(
f"定时任务 {schedule.id} (目标: {group_id}) 被中断: {type(e).__name__}"
)
except Exception as e:
logger.error(
f"执行定时任务 {schedule.id} (目标: {group_id}) "
f"时发生未被策略处理的最终错误",
e=e,
)
async def _execute_job(
schedule_id: int | None,
force: bool = False,
context_override: ScheduleContext | None = None,
):
"""
APScheduler 调度的入口函数,现在作为分发器。
"""
from .manager import scheduler_manager
schedule = None
if context_override:
plugin_name = context_override.plugin_name
task_meta = scheduler_manager._registered_tasks.get(plugin_name)
if not task_meta or not task_meta["func"]:
logger.error(f"无法执行临时任务:函数 '{plugin_name}' 未注册。")
return
try:
bot = nonebot.get_bot()
logger.info(f"开始执行临时任务: {plugin_name}")
injected_params = {"context": context_override}
state: T_State = {ScheduleContext: context_override}
async def wrapper(bot: Bot):
return await task_meta["func"](bot=bot, **injected_params) # type: ignore
dependent = Dependent.parse(
call=wrapper,
allow_types=Matcher.HANDLER_PARAM_TYPES,
)
await dependent(bot=bot, state=state)
logger.info(f"临时任务 '{plugin_name}' 执行完成。")
except Exception as e:
logger.error(f"执行临时任务 '{plugin_name}' 时发生错误", e=e)
return
if schedule_id is None:
logger.error("执行持久化任务时 schedule_id 不能为空。")
return
scheduler_manager._running_tasks.add(schedule_id)
try:
schedule = await ScheduleRepository.get_by_id(schedule_id)
if not schedule or (not schedule.is_enabled and not force):
logger.warning(f"定时任务 {schedule_id} 不存在或已禁用,跳过执行。")
return
try:
bot = (
nonebot.get_bot(schedule.bot_id)
if schedule.bot_id
else nonebot.get_bot()
)
except (KeyError, ValueError):
logger.warning(
f"任务 {schedule_id} 需要的 Bot {schedule.bot_id} "
f"不在线,本次执行跳过。"
)
raise
resolver = scheduler_manager._target_resolvers.get(schedule.target_type)
if not resolver:
logger.error(
f"任务 {schedule.id} 的目标类型 '{schedule.target_type}' "
f"没有注册解析器,执行跳过。"
)
raise ValueError(f"未知的目标类型: {schedule.target_type}")
try:
resolved_targets = await resolver(schedule.target_identifier, bot)
except Exception as e:
logger.error(f"为任务 {schedule.id} 解析目标失败", e=e)
raise
logger.info(
f"任务 {schedule.id} ({schedule.name or schedule.plugin_name}) 开始执行, "
f"目标类型: {schedule.target_type}, "
f"解析出 {len(resolved_targets)} 个目标"
)
concurrency_limit = Config.get_config(
"SchedulerManager", SCHEDULE_CONCURRENCY_KEY, 5
)
semaphore = asyncio.Semaphore(concurrency_limit if concurrency_limit > 0 else 5)
spread_config = (
schedule.execution_options
if isinstance(schedule.execution_options, dict)
else {}
)
interval_seconds = spread_config.get("interval")
if interval_seconds is not None and interval_seconds > 0:
logger.debug(
f"任务 {schedule.id}: 使用串行模式执行 {len(resolved_targets)} "
f"个目标,固定间隔 {interval_seconds} 秒。"
)
for i, target_id in enumerate(resolved_targets):
if i > 0:
logger.debug(
f"任务 {schedule.id} 目标 [{target_id or '全局'}]: "
f"等待 {interval_seconds} 秒后执行。"
)
await asyncio.sleep(interval_seconds)
await _execute_single_job_instance(schedule, bot, group_id=target_id)
else:
spread_seconds = spread_config.get("spread", 1.0)
logger.debug(
f"任务 {schedule.id}: 将在 {spread_seconds:.2f} 秒内分散执行 "
f"{len(resolved_targets)} 个目标。"
)
async def worker(target_id: str | None):
delay = random.uniform(0.1, spread_seconds)
logger.debug(
f"任务 {schedule.id} 目标 [{target_id or '全局'}]: "
f"随机延迟 {delay:.2f} 秒后执行。"
)
await asyncio.sleep(delay)
async with semaphore:
await _execute_single_job_instance(
schedule, bot, group_id=target_id
)
tasks_to_run = [worker(target_id) for target_id in resolved_targets]
if tasks_to_run:
await asyncio.gather(*tasks_to_run, return_exceptions=True)
schedule.last_run_at = datetime.now()
schedule.last_run_status = "SUCCESS"
schedule.consecutive_failures = 0
await schedule.save(
update_fields=["last_run_at", "last_run_status", "consecutive_failures"]
)
if schedule.is_one_off:
logger.info(f"一次性任务 {schedule.id} 执行成功,将被删除。")
await ScheduledJob.filter(id=schedule.id).delete()
APSchedulerAdapter.remove_job(schedule.id)
if schedule.plugin_name.startswith("runtime_one_off__"):
scheduler_manager._registered_tasks.pop(schedule.plugin_name, None)
logger.debug(f"已注销一次性运行时任务: {schedule.plugin_name}")
except Exception as e:
logger.error(f"执行任务 {schedule_id} 期间发生严重错误", e=e)
if schedule:
schedule.last_run_at = datetime.now()
schedule.last_run_status = "FAILURE"
schedule.consecutive_failures = (schedule.consecutive_failures or 0) + 1
await schedule.save(
update_fields=["last_run_at", "last_run_status", "consecutive_failures"]
)
finally:
if schedule_id is not None:
scheduler_manager._running_tasks.discard(schedule_id)
-239
View File
@@ -1,239 +0,0 @@
"""
定时任务的执行逻辑
包含被 APScheduler 实际调度的函数,以及处理不同目标(单个、所有群组)的执行策略。
"""
import asyncio
import copy
from functools import partial
import random
import nonebot
from nonebot.adapters import Bot
from nonebot.dependencies import Dependent
from nonebot.exception import FinishedException, PausedException, SkippedException
from nonebot.matcher import Matcher
from nonebot.typing import T_State
from pydantic import BaseModel, Field
from zhenxun.configs.config import Config
from zhenxun.models.scheduled_job import ScheduledJob
from zhenxun.services.log import logger
from zhenxun.utils.common_utils import CommonUtils
from zhenxun.utils.decorator.retry import Retry
from zhenxun.utils.platform import PlatformUtils
from zhenxun.utils.pydantic_compat import parse_as
SCHEDULE_CONCURRENCY_KEY = "all_groups_concurrency_limit"
class ScheduleContext(BaseModel):
"""
定时任务执行上下文,可通过依赖注入获取。
"""
schedule_id: int = Field(..., description="数据库中的任务ID")
plugin_name: str = Field(..., description="任务所属的插件名称")
bot_id: str | None = Field(None, description="执行任务的Bot ID")
group_id: str | None = Field(None, description="任务目标群组ID")
job_kwargs: dict = Field(default_factory=dict, description="任务配置的参数")
async def _execute_single_job_instance(schedule: ScheduledJob, bot):
"""
负责执行一个具体目标的任务实例。
"""
plugin_name = schedule.plugin_name
group_id = schedule.group_id
from .service import ExecutionPolicy, scheduler_manager
task_meta = scheduler_manager._registered_tasks.get(plugin_name)
if not task_meta:
logger.error(f"无法执行任务:插件 '{plugin_name}' 在执行期间变得不可用。")
return
is_blocked = await CommonUtils.task_is_block(bot, plugin_name, group_id)
if is_blocked:
target_desc = f"群 {group_id}" if group_id else "全局"
logger.info(
f"插件 '{plugin_name}' 的定时任务在目标 [{target_desc}] "
f"因功能被禁用而跳过执行。"
)
return
context = ScheduleContext(
schedule_id=schedule.id,
plugin_name=schedule.plugin_name,
bot_id=bot.self_id,
group_id=schedule.group_id,
job_kwargs=schedule.job_kwargs if isinstance(schedule.job_kwargs, dict) else {},
)
state: T_State = {ScheduleContext: context}
policy_data = context.job_kwargs.pop("execution_policy", {})
policy = ExecutionPolicy(**policy_data)
async def task_execution_coro():
injected_params = {"context": context}
params_model = task_meta.get("model")
if params_model and isinstance(context.job_kwargs, dict):
try:
if isinstance(params_model, type) and issubclass(
params_model, BaseModel
):
params_instance = parse_as(params_model, context.job_kwargs)
injected_params["params"] = params_instance # type: ignore
except Exception as e:
logger.error(
f"任务 {schedule.id} (目标: {group_id}) 参数验证失败: {e}", e=e
)
raise
async def wrapper(bot: Bot):
return await task_meta["func"](bot=bot, **injected_params) # type: ignore
dependent = Dependent.parse(
call=wrapper,
allow_types=Matcher.HANDLER_PARAM_TYPES,
)
return await dependent(bot=bot, state=state)
try:
if policy.retries > 0:
on_success_handler = None
if policy.on_success_callback:
on_success_handler = partial(policy.on_success_callback, context)
on_failure_handler = None
if policy.on_failure_callback:
on_failure_handler = partial(policy.on_failure_callback, context)
retry_exceptions = tuple(policy.retry_on_exceptions or [])
retry_decorator = Retry.api(
stop_max_attempt=policy.retries + 1,
strategy="exponential" if policy.retry_backoff else "fixed",
wait_fixed_seconds=policy.retry_delay_seconds,
exception=retry_exceptions,
on_success=on_success_handler,
on_failure=on_failure_handler,
log_name=f"ScheduledJob-{schedule.id}-{schedule.group_id or 'global'}",
)
decorated_executor = retry_decorator(task_execution_coro)
await decorated_executor()
else:
logger.info(
f"插件 '{plugin_name}' 开始为目标 [{group_id or '全局'}] "
f"执行定时任务 (ID: {schedule.id})。"
)
await task_execution_coro()
except (PausedException, FinishedException, SkippedException) as e:
logger.warning(
f"定时任务 {schedule.id} (目标: {group_id}) 被中断: {type(e).__name__}"
)
except Exception as e:
logger.error(
f"执行定时任务 {schedule.id} (目标: {group_id}) "
f"时发生未被策略处理的最终错误",
e=e,
)
async def _execute_job(schedule_id: int):
"""
APScheduler 调度的入口函数,现在作为分发器。
"""
from .repository import ScheduleRepository
from .service import scheduler_manager
scheduler_manager._running_tasks.add(schedule_id)
try:
schedule = await ScheduleRepository.get_by_id(schedule_id)
if not schedule or not schedule.is_enabled:
logger.warning(f"定时任务 {schedule_id} 不存在或已禁用,跳过执行。")
return
if schedule.plugin_name not in scheduler_manager._registered_tasks:
logger.error(
f"无法执行定时任务:插件 '{schedule.plugin_name}' "
f"未注册或已卸载。将禁用该任务。"
)
schedule.is_enabled = False
await ScheduleRepository.save(schedule, update_fields=["is_enabled"])
from .adapter import APSchedulerAdapter
APSchedulerAdapter.remove_job(schedule.id)
return
try:
bot = (
nonebot.get_bot(schedule.bot_id)
if schedule.bot_id
else nonebot.get_bot()
)
except (KeyError, ValueError):
logger.warning(
f"定时任务 {schedule_id} 需要的 Bot {schedule.bot_id} "
f"不在线,本次执行跳过。"
)
return
if schedule.group_id == scheduler_manager.ALL_GROUPS:
concurrency_limit = Config.get_config(
"SchedulerManager", SCHEDULE_CONCURRENCY_KEY, 5
)
if not isinstance(concurrency_limit, int) or concurrency_limit <= 0:
concurrency_limit = 5
logger.info(
f"开始执行针对 [所有群组] 的任务 (ID: {schedule.id}, "
f"插件: {schedule.plugin_name}, Bot: {bot.self_id}),"
f"并发限制: {concurrency_limit}"
)
try:
group_list, _ = await PlatformUtils.get_group_list(bot)
all_gids = {
g.group_id for g in group_list if g.group_id and not g.channel_id
}
except Exception as e:
logger.error(f"为 'all' 任务获取 Bot {bot.self_id} 的群列表失败", e=e)
return
specific_tasks_gids = set(
await ScheduledJob.filter(
plugin_name=schedule.plugin_name, group_id__in=list(all_gids)
).values_list("group_id", flat=True)
)
semaphore = asyncio.Semaphore(concurrency_limit)
async def worker(gid: str):
await asyncio.sleep(random.uniform(0.1, 1.0))
async with semaphore:
temp_schedule = copy.deepcopy(schedule)
temp_schedule.group_id = gid
await _execute_single_job_instance(temp_schedule, bot)
tasks_to_run = [
worker(gid) for gid in all_gids if gid not in specific_tasks_gids
]
if tasks_to_run:
await asyncio.gather(*tasks_to_run)
logger.info(
f"针对 [所有群组] 的任务 (ID: {schedule.id}) 执行完毕,"
f"共处理 {len(tasks_to_run)} 个群组。"
)
else:
await _execute_single_job_instance(schedule, bot)
finally:
scheduler_manager._running_tasks.discard(schedule_id)
+9 -5
View File
@@ -8,10 +8,10 @@ from zhenxun.services.log import logger
from zhenxun.utils.manager.priority_manager import PriorityLifecycle
from zhenxun.utils.pydantic_compat import model_dump
from .adapter import APSchedulerAdapter
from .job import ScheduleContext
from .engine import APSchedulerAdapter
from .manager import scheduler_manager
from .repository import ScheduleRepository
from .service import scheduler_manager
from .types import ScheduleContext
@PriorityLifecycle.on_startup(priority=90)
@@ -37,7 +37,7 @@ async def _load_schedules_from_db():
query_kwargs = {
"plugin_name": plugin_name,
"group_id": group_id,
"target_identifier": group_id or "",
"bot_id": bot_id,
}
exists = await ScheduleRepository.exists(**query_kwargs)
@@ -49,9 +49,13 @@ async def _load_schedules_from_db():
task_info.trigger, exclude={"trigger_type"}
)
target_type = "GROUP" if group_id else "GLOBAL"
target_identifier = group_id or ""
schedule = await scheduler_manager.add_schedule(
plugin_name=plugin_name,
group_id=group_id,
target_type=target_type,
target_identifier=target_identifier,
trigger_type=task_info.trigger.trigger_type,
trigger_config=trigger_config_dict,
job_kwargs=task_info.job_kwargs,
@@ -1,81 +1,92 @@
"""
服务层 (Service)
服务层 (Service Manager)
定义 SchedulerManager 类作为定时任务服务的公共 API 入口。
它负责编排业务逻辑,并调用 Repository 和 Adapter 层来完成具体工作。
"""
from __future__ import annotations
from collections.abc import Awaitable, Callable, Coroutine
from datetime import datetime
import inspect
from typing import Any, ClassVar
import uuid
from arclet.alconna import Alconna, Option
import nonebot
from nonebot.adapters import Bot
from pydantic import BaseModel
from zhenxun.configs.config import Config
from zhenxun.models.scheduled_job import ScheduledJob
from zhenxun.services.log import logger
from zhenxun.utils.pydantic_compat import model_dump
from zhenxun.utils.pydantic_compat import model_dump, model_validate
from .adapter import APSchedulerAdapter
from .job import ScheduleContext, _execute_job
from .engine import APSchedulerAdapter
from .repository import ScheduleRepository
from .targeter import ScheduleTargeter
from .triggers import BaseTrigger
class ExecutionPolicy(BaseModel):
"""
封装定时任务的执行策略,包括重试和回调。
"""
retries: int = 0
retry_delay_seconds: int = 30
retry_backoff: bool = False
retry_on_exceptions: list[type[Exception]] | None = None
on_success_callback: Callable[[ScheduleContext, Any], Awaitable[None]] | None = None
on_failure_callback: (
Callable[[ScheduleContext, Exception], Awaitable[None]] | None
) = None
class Config:
arbitrary_types_allowed = True
class ScheduledJobDeclaration(BaseModel):
"""用于在启动时声明默认定时任务的内部数据模型"""
plugin_name: str
group_id: str | None
bot_id: str | None
trigger: BaseTrigger
job_kwargs: dict[str, Any]
class Config:
arbitrary_types_allowed = True
class EphemeralJobDeclaration(BaseModel):
"""用于在启动时声明临时任务的内部数据模型"""
plugin_name: str
func: Callable[..., Coroutine]
trigger: BaseTrigger
class Config:
arbitrary_types_allowed = True
from .targeting import (
ScheduleTargeter,
)
from .types import (
BaseTrigger,
EphemeralJobDeclaration,
ExecutionOptions,
ExecutionPolicy,
ScheduleContext,
ScheduledJobDeclaration,
)
class SchedulerManager:
ALL_GROUPS: ClassVar[str] = "__ALL_GROUPS__"
_registered_tasks: ClassVar[
dict[str, dict[str, Callable | type[BaseModel] | None]]
dict[
str,
dict[str, Callable | type[BaseModel] | int | list[Option] | Alconna | None],
]
] = {}
_declared_tasks: ClassVar[list[ScheduledJobDeclaration]] = []
_ephemeral_declared_tasks: ClassVar[list[EphemeralJobDeclaration]] = []
_running_tasks: ClassVar[set] = set()
_target_resolvers: ClassVar[
dict[str, Callable[[str, Bot], Awaitable[list[str | None]]]]
] = {}
def __init__(self):
self._register_builtin_resolvers()
def _register_builtin_resolvers(self):
"""在管理器初始化时注册所有内置的目标解析器。"""
from .targeting import (
_resolve_all_groups,
_resolve_global_or_user,
_resolve_group,
_resolve_tag,
_resolve_user,
)
if "GROUP" in self._target_resolvers:
return
self.register_target_resolver("GROUP", _resolve_group)
self.register_target_resolver("TAG", _resolve_tag)
self.register_target_resolver("ALL_GROUPS", _resolve_all_groups)
self.register_target_resolver("GLOBAL", _resolve_global_or_user)
self.register_target_resolver("USER", _resolve_user)
logger.debug("已注册所有内置的定时任务目标解析器。")
def register_target_resolver(
self,
target_type: str,
resolver_func: Callable[[str, Bot], Awaitable[list[str | None]]],
):
"""
注册一个新的目标类型解析器。
"""
if target_type in self._target_resolvers:
logger.warning(f"目标解析器 '{target_type}' 已存在,将被覆盖。")
self._target_resolvers[target_type.upper()] = resolver_func
logger.info(f"已注册新的定时任务目标解析器: '{target_type}'")
def target(self, **filters: Any) -> ScheduleTargeter:
"""
@@ -96,22 +107,12 @@ class SchedulerManager:
bot_id: str | None = None,
default_params: BaseModel | None = None,
policy: ExecutionPolicy | None = None,
default_jitter: int | None = None,
default_spread: int | None = None,
default_interval: int | None = None,
):
"""
声明式定时任务的统一装饰器。
此装饰器用于将一个异步函数注册为一个可调度的定时任务,
并为其创建一个默认的调度计划。
参数:
trigger: 一个由 `Trigger` 工厂类创建的触发器配置对象
(例如 `Trigger.cron(hour=8)`)。
group_id: 默认的目标群组ID。`None` 表示全局任务,
`SchedulerManager.ALL_GROUPS` 表示所有群组。
bot_id: 默认的目标Bot ID,`None` 表示使用任意可用Bot。
default_params: (可选) 一个Pydantic模型实例,为任务提供默认参数。
任务函数需要有对应的Pydantic模型类型注解。
policy: (可选) 一个ExecutionPolicy实例,定义任务的执行策略。
"""
def decorator(func: Callable[..., Coroutine]) -> Callable[..., Coroutine]:
@@ -122,7 +123,7 @@ class SchedulerManager:
plugin_name = plugin.name
params_model = None
from .job import ScheduleContext
from .types import ScheduleContext
for param in inspect.signature(func).parameters.values():
if (
@@ -138,6 +139,9 @@ class SchedulerManager:
self._registered_tasks[plugin_name] = {
"func": func,
"model": params_model,
"default_jitter": default_jitter,
"default_spread": default_spread,
"default_interval": default_interval,
}
job_kwargs = model_dump(default_params) if default_params else {}
@@ -165,13 +169,6 @@ class SchedulerManager:
def runtime_job(self, trigger: BaseTrigger):
"""
声明一个临时的、非持久化的定时任务。
这个任务只存在于内存中,随程序重启而消失。
它非常适合用于插件内部的、固定的、无需用户配置的系统级定时任务。
被此装饰器修饰的函数依然可以享受完整的依赖注入功能。
参数:
trigger: 一个由 `Trigger` 工厂类创建的触发器配置对象。
"""
def decorator(func: Callable[..., Coroutine]) -> Callable[..., Coroutine]:
@@ -203,17 +200,17 @@ class SchedulerManager:
return decorator
def register(
self, plugin_name: str, params_model: type[BaseModel] | None = None
self,
plugin_name: str,
params_model: type[BaseModel] | None = None,
cli_parser: Alconna | None = None,
default_permission: int = 5,
default_jitter: int | None = None,
default_spread: int | None = None,
default_interval: int | None = None,
) -> Callable:
"""
注册可调度的任务函数
参数:
plugin_name: 插件名称,用于标识任务。
params_model: 参数验证模型,继承自BaseModel的类。
返回:
Callable: 装饰器函数。
"""
def decorator(func: Callable[..., Coroutine]) -> Callable[..., Coroutine]:
@@ -222,6 +219,11 @@ class SchedulerManager:
self._registered_tasks[plugin_name] = {
"func": func,
"model": params_model,
"cli_parser": cli_parser,
"default_permission": default_permission,
"default_jitter": default_jitter,
"default_spread": default_spread,
"default_interval": default_interval,
}
model_name = params_model.__name__ if params_model else "无"
logger.debug(
@@ -234,25 +236,14 @@ class SchedulerManager:
def get_registered_plugins(self) -> list[str]:
"""
获取已注册插件列表
返回:
list[str]: 已注册的插件名称列表。
"""
return list(self._registered_tasks.keys())
async def run_at(self, func: Callable[..., Coroutine], trigger: BaseTrigger) -> str:
"""
【新增】在未来的某个时间点,运行一个一次性的临时任务。
这是一个编程式API,用于动态调度一个非持久化的任务。
参数:
func: 要执行的异步函数。
trigger: 一个由 `Trigger` 工廠類創建的觸發器配置對象。
返回:
str: 临时任务的唯一ID,可用于未来的管理(如取消)。
在未来的某个时间点,运行一个一次性的临时任务。
"""
job_id = f"ephemeral_runtime_{uuid.uuid4()}"
context = ScheduleContext(
@@ -273,6 +264,47 @@ class SchedulerManager:
logger.info(f"已动态调度一个临时任务 (ID: {job_id}),将在 {trigger} 触发。")
return job_id
async def schedule_once(
self,
func: Callable[..., Coroutine],
trigger: BaseTrigger,
*,
user_id: str | None = None,
group_id: str | None = None,
bot_id: str | None = None,
job_kwargs: dict | None = None,
name: str | None = None,
created_by: str | None = None,
required_permission: int = 5,
) -> "ScheduledJob | None":
"""
编程式API,用于动态调度一个持久化的、一次性的任务。
"""
if user_id and group_id:
raise ValueError("user_id 和 group_id 不能同时提供。")
temp_plugin_name = f"runtime_one_off__{func.__module__}.{func.__name__}__{uuid.uuid4().hex[:8]}" # noqa: E501
self._registered_tasks[temp_plugin_name] = {"func": func, "model": None}
logger.debug(f"为一次性任务动态注册临时插件: '{temp_plugin_name}'")
target_type = "USER" if user_id else ("GROUP" if group_id else "GLOBAL")
target_identifier = user_id or group_id or ""
return await self.add_schedule(
plugin_name=temp_plugin_name,
target_type=target_type,
target_identifier=target_identifier,
trigger_type=trigger.trigger_type,
trigger_config=model_dump(trigger, exclude={"trigger_type"}),
job_kwargs=job_kwargs,
bot_id=bot_id,
name=name,
created_by=created_by,
required_permission=required_permission,
is_one_off=True,
)
async def add_daily_task(
self,
plugin_name: str,
@@ -285,18 +317,6 @@ class SchedulerManager:
) -> "ScheduledJob | None":
"""
添加每日定时任务
参数:
plugin_name: 插件名称。
group_id: 目标群组ID,None表示全局任务。
hour: 执行小时(0-23)。
minute: 执行分钟(0-59)。
second: 执行秒数(0-59),默认为0。
job_kwargs: 任务参数字典。
bot_id: 目标Bot ID,None表示使用默认Bot。
返回:
ScheduledJob | None: 创建的任务信息,失败时返回None。
"""
trigger_config = {
"hour": hour,
@@ -306,9 +326,10 @@ class SchedulerManager:
}
return await self.add_schedule(
plugin_name,
group_id,
"cron",
trigger_config,
target_type="GROUP" if group_id else "GLOBAL",
target_identifier=group_id or "",
trigger_type="cron",
trigger_config=trigger_config,
job_kwargs=job_kwargs,
bot_id=bot_id,
)
@@ -329,17 +350,6 @@ class SchedulerManager:
) -> "ScheduledJob | None":
"""
添加间隔性定时任务
参数:
plugin_name: 插件名称。
group_id: 目标群组ID,None表示全局任务。
weeks/days/hours/minutes/seconds: 间隔时间,至少指定一个。
start_date: 开始时间,None表示立即开始。
job_kwargs: 任务参数字典。
bot_id: 目标Bot ID。
返回:
ScheduledJob | None: 创建的任务信息,失败时返回None。
"""
trigger_config = {
"weeks": weeks,
@@ -352,9 +362,10 @@ class SchedulerManager:
trigger_config = {k: v for k, v in trigger_config.items() if v}
return await self.add_schedule(
plugin_name,
group_id,
"interval",
trigger_config,
target_type="GROUP" if group_id else "GLOBAL",
target_identifier=group_id or "",
trigger_type="interval",
trigger_config=trigger_config,
job_kwargs=job_kwargs,
bot_id=bot_id,
)
@@ -384,11 +395,7 @@ class SchedulerManager:
return False, f"插件 '{plugin_name}' 的参数模型配置错误"
try:
model_validate = getattr(params_model, "model_validate", None)
if not model_validate:
return False, f"插件 '{plugin_name}' 的参数模型不支持验证"
validated_model = model_validate(job_kwargs)
validated_model = model_validate(params_model, job_kwargs)
return True, model_dump(validated_model)
except ValidationError as e:
@@ -400,22 +407,37 @@ class SchedulerManager:
async def add_schedule(
self,
plugin_name: str,
group_id: str | None,
target_type: str,
target_identifier: str,
trigger_type: str,
trigger_config: dict,
job_kwargs: dict | None = None,
bot_id: str | None = None,
*,
name: str | None = None,
created_by: str | None = None,
required_permission: int = 5,
source: str = "USER",
is_one_off: bool = False,
execution_options: dict | None = None,
) -> "ScheduledJob | None":
"""
添加定时任务(通用方法)
参数:
plugin_name: 插件名称。
group_id: 目标群组ID,None表示全局任务。
trigger_type: 触发器类型,如'cron'、'interval'等。
target_type: 目标类型 (GROUP, USER, TAG, ALL_GROUPS, GLOBAL)。
target_identifier: 目标标识符。
trigger_type: 触发器类型 (cron, interval, date)。
trigger_config: 触发器配置字典。
job_kwargs: 任务参数字典。
bot_id: 目标Bot ID,None表示使用默认Bot。
job_kwargs: 传递给任务函数的额外参数。
bot_id: Bot ID约束。
name: 任务别名。
created_by: 创建者ID。
required_permission: 管理此任务所需的权限。
source: 任务来源 (USER, PLUGIN_DEFAULT)。
is_one_off: 是否为一次性任务。
execution_options: 任务执行的额外选项 (例如: jitter, spread)。
返回:
ScheduledJob | None: 创建的任务信息,失败时返回None。
@@ -429,51 +451,84 @@ class SchedulerManager:
logger.error(f"任务参数校验失败: {result}")
return None
search_kwargs = {"plugin_name": plugin_name, "group_id": group_id}
if bot_id and group_id == self.ALL_GROUPS:
options_dict = execution_options or {}
validated_options = ExecutionOptions(**options_dict)
search_kwargs = {
"plugin_name": plugin_name,
"target_type": target_type,
"target_identifier": target_identifier,
}
if bot_id:
search_kwargs["bot_id"] = bot_id
else:
search_kwargs["bot_id__isnull"] = True
defaults = {
"name": name,
"trigger_type": trigger_type,
"trigger_config": trigger_config,
"job_kwargs": result,
"is_enabled": True,
"created_by": created_by,
"required_permission": required_permission,
"source": source,
"is_one_off": is_one_off,
"execution_options": model_dump(validated_options, exclude_none=True),
}
defaults = {k: v for k, v in defaults.items() if v is not None}
schedule, created = await ScheduleRepository.update_or_create(
defaults, **search_kwargs
)
APSchedulerAdapter.add_or_reschedule_job(schedule)
action = "设置" if created else "更新"
action_str = "创建" if created else "更新"
logger.info(
f"已成功{action}插件 '{plugin_name}' 的定时任务 (ID: {schedule.id})。"
f"已成功{action_str}任务 '{name or plugin_name}' (ID: {schedule.id})"
)
return schedule
async def get_schedules(
self,
plugin_name: str | None = None,
group_id: str | None = None,
bot_id: str | None = None,
) -> list[ScheduledJob]:
self, page: int | None = None, page_size: int | None = None, **filters: Any
) -> tuple[list[ScheduledJob], int]:
"""
根据条件获取定时任务列表
参数:
plugin_name: 插件名称,None表示不限制。
group_id: 群组ID,None表示不限制。
bot_id: Bot ID,None表示不限制。
返回:
list[ScheduledJob]: 符合条件的任务信息列表。
"""
cleaned_filters = {k: v for k, v in filters.items() if v is not None}
return await ScheduleRepository.query_schedules(
plugin_name=plugin_name, group_id=group_id, bot_id=bot_id
page=page, page_size=page_size, **cleaned_filters
)
async def get_schedules_status_bulk(
self, schedule_ids: list[int]
) -> list[dict[str, Any]]:
"""
批量获取多个定时任务的详细状态信息
"""
if not schedule_ids:
return []
schedules = await ScheduleRepository.filter(id__in=schedule_ids).all()
schedule_map = {s.id: s for s in schedules}
statuses = []
for schedule_id in schedule_ids:
if schedule := schedule_map.get(schedule_id):
status_from_scheduler = APSchedulerAdapter.get_job_status(schedule.id)
status_dict = {
field: getattr(schedule, field)
for field in schedule._meta.fields_map
}
status_dict.update(status_from_scheduler)
status_dict["is_enabled"] = (
"运行中"
if schedule_id in self._running_tasks
else ("启用" if schedule.is_enabled else "暂停")
)
statuses.append(status_dict)
return statuses
async def update_schedule(
self,
schedule_id: int,
@@ -483,15 +538,6 @@ class SchedulerManager:
) -> tuple[bool, str]:
"""
更新定时任务配置
参数:
schedule_id: 任务ID。
trigger_type: 新的触发器类型,None表示不更新。
trigger_config: 新的触发器配置,None表示不更新。
job_kwargs: 新的任务参数,None表示不更新。
返回:
tuple[bool, str]: (是否成功, 结果消息)。
"""
schedule = await ScheduleRepository.get_by_id(schedule_id)
if not schedule:
@@ -533,12 +579,6 @@ class SchedulerManager:
async def get_schedule_status(self, schedule_id: int) -> dict | None:
"""
获取定时任务的详细状态信息
参数:
schedule_id: 定时任务的ID。
返回:
dict | None: 任务详细信息字典,不存在时返回None。
"""
schedule = await ScheduleRepository.get_by_id(schedule_id)
if not schedule:
@@ -556,7 +596,8 @@ class SchedulerManager:
"id": schedule.id,
"bot_id": schedule.bot_id,
"plugin_name": schedule.plugin_name,
"group_id": schedule.group_id,
"target_type": schedule.target_type,
"target_identifier": schedule.target_identifier,
"is_enabled": status_text,
"trigger_type": schedule.trigger_type,
"trigger_config": schedule.trigger_config,
@@ -567,12 +608,6 @@ class SchedulerManager:
async def pause_schedule(self, schedule_id: int) -> tuple[bool, str]:
"""
暂停指定的定时任务
参数:
schedule_id: 要暂停的定时任务ID。
返回:
tuple[bool, str]: (是否成功, 操作结果消息)。
"""
schedule = await ScheduleRepository.get_by_id(schedule_id)
if not schedule or not schedule.is_enabled:
@@ -586,12 +621,6 @@ class SchedulerManager:
async def resume_schedule(self, schedule_id: int) -> tuple[bool, str]:
"""
恢复指定的定时任务
参数:
schedule_id: 要恢复的定时任务ID。
返回:
tuple[bool, str]: (是否成功, 操作结果消息)。
"""
schedule = await ScheduleRepository.get_by_id(schedule_id)
if not schedule or schedule.is_enabled:
@@ -605,13 +634,9 @@ class SchedulerManager:
async def trigger_now(self, schedule_id: int) -> tuple[bool, str]:
"""
立即手动触发指定的定时任务
参数:
schedule_id: 要触发的定时任务ID。
返回:
tuple[bool, str]: (是否成功, 操作结果消息)。
"""
from .engine import _execute_job
schedule = await ScheduleRepository.get_by_id(schedule_id)
if not schedule:
return False, f"未找到 ID 为 {schedule_id} 的定时任务。"
@@ -619,12 +644,23 @@ class SchedulerManager:
return False, f"插件 '{schedule.plugin_name}' 没有注册可用的定时任务。"
try:
await _execute_job(schedule.id)
await _execute_job(schedule.id, force=True)
return True, f"已手动触发任务 (ID: {schedule.id})。"
except Exception as e:
logger.error(f"手动触发任务失败: {e}")
return False, f"手动触发任务失败: {e}"
async def get_schedule_by_id(self, schedule_id: int) -> "ScheduledJob | None":
"""
通过ID获取任务对象的公共方法。
参数:
schedule_id: 任务ID。
返回:
ScheduledJob | None: 任务对象,不存在时返回None。
"""
return await ScheduleRepository.get_by_id(schedule_id)
scheduler_manager = SchedulerManager()
scheduler = scheduler_manager
+17 -7
View File
@@ -64,9 +64,9 @@ class ScheduleRepository:
async def get_by_plugin_and_group(
plugin_name: str, group_ids: list[str]
) -> list[ScheduledJob]:
"""根据插件和群组ID列表获取任务"""
"""[DEPRECATED] 根据插件和群组ID列表获取任务"""
return await ScheduledJob.filter(
plugin_name=plugin_name, group_id__in=group_ids
plugin_name=plugin_name, target_descriptor__in=group_ids
).all()
@staticmethod
@@ -77,20 +77,30 @@ class ScheduleRepository:
return await ScheduledJob.update_or_create(defaults=defaults, **kwargs)
@staticmethod
async def query_schedules(**filters: Any) -> list[ScheduledJob]:
async def query_schedules(
page: int | None = None, page_size: int | None = None, **filters: Any
) -> tuple[list[ScheduledJob], int]:
"""
根据任意条件查询任务列表
参数:
page: 页码(从1开始)
page_size: 每页数量
**filters: 过滤条件,如 group_id="123", plugin_name="abc"
返回:
list[ScheduledJob]: 任务列表
tuple[list[ScheduledJob], int]: (任务列表, 总数)
"""
cleaned_filters = {k: v for k, v in filters.items() if v is not None}
if not cleaned_filters:
return await ScheduledJob.all()
return await ScheduledJob.filter(**cleaned_filters).all()
query = ScheduledJob.filter(**cleaned_filters)
total_count = await query.count()
if page is not None and page_size is not None:
offset = (page - 1) * page_size
query = query.offset(offset).limit(page_size)
return await query.all(), total_count
@staticmethod
def filter(**kwargs: Any) -> QuerySet[ScheduledJob]:
@@ -1,14 +1,46 @@
"""
目标选择器 (Targeter)
目标解析与选择器 (Targeting)
提供链式API,用于构建和执行对多个定时任务的批量操作。
提供用于解析任务目标和批量操作目标的 ScheduleTargeter 类。
"""
from collections.abc import Callable, Coroutine
from typing import Any
from .adapter import APSchedulerAdapter
from .repository import ScheduleRepository
from nonebot.adapters import Bot
from zhenxun.services.tags import tag_manager
__all__ = [
"ScheduleTargeter",
"_resolve_all_groups",
"_resolve_global_or_user",
"_resolve_group",
"_resolve_tag",
"_resolve_user",
]
async def _resolve_group(target_identifier: str, bot: Bot) -> list[str | None]:
return [target_identifier]
async def _resolve_tag(target_identifier: str, bot: Bot) -> list[str | None]:
result = await tag_manager.resolve_tag_to_group_ids(target_identifier)
return result # type: ignore
async def _resolve_user(target_identifier: str, bot: Bot) -> list[str | None]:
return [target_identifier]
async def _resolve_all_groups(target_identifier: str, bot: Bot) -> list[str | None]:
result = await tag_manager.resolve_tag_to_group_ids("@all", bot=bot)
return result
async def _resolve_global_or_user(target_identifier: str, bot: Bot) -> list[str | None]:
return [None]
class ScheduleTargeter:
@@ -34,6 +66,8 @@ class ScheduleTargeter:
返回:
list[ScheduledJob]: 符合过滤条件的任务列表。
"""
from .repository import ScheduleRepository
query = ScheduleRepository.filter(**self._filters)
return await query.all()
@@ -48,12 +82,14 @@ class ScheduleTargeter:
return f"任务 ID {self._filters['id']} 的"
parts = []
if "group_id" in self._filters:
group_id = self._filters["group_id"]
if group_id == self._manager.ALL_GROUPS:
if "target_descriptor" in self._filters:
descriptor = self._filters["target_descriptor"]
if descriptor == self._manager.ALL_GROUPS:
parts.append("所有群组中")
elif descriptor.startswith("tag:"):
parts.append(f"标签 '{descriptor[4:]}' 的")
else:
parts.append(f"群 {group_id} 中")
parts.append(f"群 {descriptor} 中")
if "plugin_name" in self._filters:
parts.append(f"插件 '{self._filters['plugin_name']}' 的")
@@ -111,6 +147,9 @@ class ScheduleTargeter:
返回:
tuple[int, str]: (成功移除的任务数量, 操作结果消息)。
"""
from .engine import APSchedulerAdapter
from .repository import ScheduleRepository
schedules = await self._get_schedules()
if not schedules:
target_desc = self._generate_target_description()
+151
View File
@@ -0,0 +1,151 @@
"""
定时任务服务的数据模型与类型定义
"""
from collections.abc import Awaitable, Callable
from datetime import datetime
from typing import Any, Literal
from pydantic import BaseModel, Field
class BaseTrigger(BaseModel):
"""触发器配置的基类"""
trigger_type: str = Field(..., exclude=True)
class CronTrigger(BaseTrigger):
"""Cron 触发器配置"""
trigger_type: Literal["cron"] = "cron" # type: ignore
year: int | str | None = None
month: int | str | None = None
day: int | str | None = None
week: int | str | None = None
day_of_week: int | str | None = None
hour: int | str | None = None
minute: int | str | None = None
second: int | str | None = None
start_date: datetime | str | None = None
end_date: datetime | str | None = None
timezone: str | None = None
jitter: int | None = None
class IntervalTrigger(BaseTrigger):
"""Interval 触发器配置"""
trigger_type: Literal["interval"] = "interval" # type: ignore
weeks: int = 0
days: int = 0
hours: int = 0
minutes: int = 0
seconds: int = 0
start_date: datetime | str | None = None
end_date: datetime | str | None = None
timezone: str | None = None
jitter: int | None = None
class DateTrigger(BaseTrigger):
"""Date 触发器配置"""
trigger_type: Literal["date"] = "date" # type: ignore
run_date: datetime | str
timezone: str | None = None
class Trigger:
"""
一个用于创建类型安全触发器配置的工厂类。
提供了流畅的、具备IDE自动补全功能的API。
"""
@staticmethod
def cron(**kwargs) -> CronTrigger:
"""创建一个 Cron 触发器配置。"""
return CronTrigger(**kwargs)
@staticmethod
def interval(**kwargs) -> IntervalTrigger:
"""创建一个 Interval 触发器配置。"""
return IntervalTrigger(**kwargs)
@staticmethod
def date(**kwargs) -> DateTrigger:
"""创建一个 Date 触发器配置。"""
return DateTrigger(**kwargs)
class ExecutionOptions(BaseModel):
"""
封装定时任务的执行策略,包括重试和回调。
"""
jitter: int | None = Field(None, description="触发时间抖动(秒)")
spread: int | None = Field(
None, description="(并发模式)多目标执行的最大分散延迟(秒)"
)
interval: int | None = Field(
None, description="多目标执行的固定间隔(秒),设置后将强制串行执行"
)
concurrency_policy: Literal["ALLOW", "SKIP", "QUEUE"] = Field(
"ALLOW", description="并发策略"
)
retries: int = 0
retry_delay_seconds: int = 30
class ScheduleContext(BaseModel):
"""
定时任务执行上下文,可通过依赖注入获取。
"""
schedule_id: int = Field(..., description="数据库中的任务ID")
plugin_name: str = Field(..., description="任务所属的插件名称")
bot_id: str | None = Field(None, description="执行任务的Bot ID")
group_id: str | None = Field(None, description="当前执行实例的目标群组ID")
job_kwargs: dict = Field(default_factory=dict, description="任务配置的参数")
class ExecutionPolicy(BaseModel):
"""
封装定时任务的执行策略,包括重试和回调。
"""
retries: int = 0
retry_delay_seconds: int = 30
retry_backoff: bool = False
retry_on_exceptions: list[type[Exception]] | None = None
on_success_callback: Callable[[ScheduleContext, Any], Awaitable[None]] | None = None
on_failure_callback: (
Callable[[ScheduleContext, Exception], Awaitable[None]] | None
) = None
class Config:
arbitrary_types_allowed = True
class ScheduledJobDeclaration(BaseModel):
"""用于在启动时声明默认定时任务的内部数据模型"""
plugin_name: str
group_id: str | None
bot_id: str | None
trigger: BaseTrigger
job_kwargs: dict[str, Any]
class Config:
arbitrary_types_allowed = True
class EphemeralJobDeclaration(BaseModel):
"""用于在启动时声明临时任务的内部数据模型"""
plugin_name: str
func: Callable[..., Awaitable[Any]]
trigger: BaseTrigger
class Config:
arbitrary_types_allowed = True
+11
View File
@@ -0,0 +1,11 @@
"""
标签服务入口,提供 ``TagManager`` 实例并加载内置规则。
"""
from .manager import TagManager
tag_manager = TagManager()
from . import filters # noqa: F401
__all__ = ["tag_manager"]
+11
View File
@@ -0,0 +1,11 @@
"""
动态标签的内置过滤器集合,可通过装饰器注册到标签管理器。
"""
from . import tag_manager
tag_manager.add_field_rule("member_count", db_field="member_count", value_type=int)
tag_manager.add_field_rule("level", db_field="level", value_type=int)
tag_manager.add_field_rule("status", db_field="status", value_type=bool)
tag_manager.add_field_rule("is_super", db_field="is_super", value_type=bool)
tag_manager.add_field_rule("group_name", db_field="group_name", value_type=str)
+641
View File
@@ -0,0 +1,641 @@
"""
标签服务的核心实现,负责标签的增删改查与动态规则解析。
"""
from collections.abc import Callable, Coroutine
from dataclasses import dataclass
from functools import partial
from typing import Any, ClassVar
from aiocache import Cache, cached
from arclet.alconna import Alconna, Args
import nonebot
from nonebot.adapters import Bot
from tortoise.exceptions import IntegrityError
from tortoise.expressions import Q
from tortoise.transactions import in_transaction
from zhenxun.models.group_console import GroupConsole
from zhenxun.models.group_tag import GroupTag, GroupTagLink
from zhenxun.services.log import logger
from zhenxun.utils.platform import PlatformUtils
from .models import (
ErrorResult,
IDSetResult,
QueryResult,
RuleExecutionError,
RuleExecutionResult,
)
@dataclass
class HandlerInfo:
"""存储已注册处理器的元信息。"""
func: Callable[..., Coroutine[Any, Any, RuleExecutionResult]]
alconna: Alconna
def invalidate_on_change(func: Callable) -> Callable:
"""装饰器: 在方法成功执行后自动使标签缓存失效。"""
async def wrapper(self: "TagManager", *args, **kwargs):
result = await func(self, *args, **kwargs)
await self._invalidate_cache()
return result
return wrapper
class TagManager:
"""群组标签管理服务。提供对群组标签的注册、解析与维护等操作。"""
_dynamic_handlers: ClassVar[dict[str, HandlerInfo]] = {}
def add_field_rule(self, name: str, db_field: str, value_type: type):
"""
一个便捷的快捷方式,用于快速创建一个基于 `GroupConsole` 模型字段的规则。
它在内部使用 `register_rule`。
"""
from arclet.alconna import CommandMeta
alc = Alconna(
name,
Args["op", str]["value", value_type],
meta=CommandMeta(
fuzzy_match=True,
compact=False,
),
)
handler = partial(self._generic_field_handler, db_field=db_field)
self.register_rule(alc)(handler)
logger.debug(f"已添加字段规则: '{name}' -> {db_field} ({value_type.__name__})")
async def _generic_field_handler(
self, db_field: str, op: str, value: Any
) -> QueryResult:
"""所有通过 add_field_rule 添加的规则共享的处理器。"""
op_map = {">": "__gt", ">=": "__gte", "<": "__lt", "<=": "__lte", "=": ""}
op_lower = op.lower()
if op_lower == "contains":
op_suffix = "__iposix_regex"
elif op_lower == "in":
op_suffix = "__in"
value = [v.strip() for v in str(value).split(",")]
elif op == "!=":
return QueryResult(q_object=~Q(**{db_field: value}))
elif op in op_map:
op_suffix = op_map[op]
else:
raise RuleExecutionError(f"字段 '{db_field}' 不支持操作符: {op}")
q_kwargs: dict[str, Any] = {
f"{db_field}{op_suffix}" if op_suffix else db_field: value
}
return QueryResult(q_object=Q(**q_kwargs))
def register_rule(self, alconna: Alconna):
"""
装饰器:注册一个完全自定义的规则处理器及其语法定义(Alconna)。
"""
def decorator(handler: Callable[..., Coroutine[Any, Any, RuleExecutionResult]]):
name = alconna.command
if name in self._dynamic_handlers:
logger.warning(f"动态标签规则 '{name}' 已被注册,将被覆盖。")
self._dynamic_handlers[name] = HandlerInfo(func=handler, alconna=alconna)
logger.debug(f"已注册动态标签规则: '{name}'")
return handler
return decorator
async def _invalidate_cache(self):
"""辅助函数,用于清除标签相关的缓存,确保数据一致性。"""
cache = Cache(Cache.MEMORY, namespace="tag_service")
await cache.clear()
logger.debug("已清除所有群组标签缓存。")
@invalidate_on_change
async def create_tag(
self,
name: str,
is_blacklist: bool = False,
description: str | None = None,
group_ids: list[str] | None = None,
tag_type: str = "STATIC",
dynamic_rule: dict | str | None = None,
) -> GroupTag:
"""
创建新的群组标签。
参数:
name: 标签名称。
is_blacklist: 是否为黑名单标签,黑名单标签会在最终结果中剔除关联群组。
description: 标签描述信息。
group_ids: 需要关联的静态群组 ID 列表,动态标签必须留空。
tag_type: 标签类型,支持 ``STATIC`` 或 ``DYNAMIC``。
dynamic_rule: 动态标签所使用的规则配置。
返回:
新创建的 ``GroupTag`` 实例。
"""
if tag_type == "DYNAMIC" and group_ids:
raise ValueError("动态标签不能在创建时关联静态群组。")
if tag_type == "STATIC" and dynamic_rule:
raise ValueError("静态标签不能设置动态规则。")
async with in_transaction():
tag = await GroupTag.create(
name=name,
is_blacklist=is_blacklist,
description=description,
tag_type=tag_type,
dynamic_rule=dynamic_rule,
)
if group_ids:
unique_group_ids = list(dict.fromkeys(group_ids))
await GroupTagLink.bulk_create(
[GroupTagLink(tag=tag, group_id=gid) for gid in unique_group_ids]
)
return tag
@invalidate_on_change
async def delete_tag(self, name: str) -> bool:
"""
删除指定标签。
参数:
name: 标签名称。
返回:
``True`` 表示删除成功,``False`` 表示标签不存在。
"""
deleted_count = await GroupTag.filter(name=name).delete()
return deleted_count > 0
@invalidate_on_change
async def remove_group_from_all_tags(self, group_id: str) -> int:
"""
从所有静态标签中移除一个指定的群组ID。
主要用于机器人退群时的实时清理。
参数:
group_id: 要移除的群组ID。
返回:
被删除的关联数量。
"""
deleted_count = await GroupTagLink.filter(group_id=group_id).delete()
if deleted_count > 0:
logger.info(f"已从 {deleted_count} 个标签中移除群组 {group_id} 的关联。")
return deleted_count
@invalidate_on_change
async def prune_stale_group_links(self) -> int:
"""
清理所有静态标签中无效的群组关联。
无效指的是机器人已不再任何一个已连接的Bot的群组列表中。
返回:
被清理的无效关联的总数。
"""
all_bot_group_ids = set()
for bot in nonebot.get_bots().values():
groups, _ = await PlatformUtils.get_group_list(bot)
all_bot_group_ids.update(g.group_id for g in groups if g.group_id)
all_static_links = await GroupTagLink.filter(tag__tag_type="STATIC").all()
stale_link_ids = [
link.id
for link in all_static_links
if link.group_id not in all_bot_group_ids
]
if stale_link_ids:
return await GroupTagLink.filter(id__in=stale_link_ids).delete()
return 0
@invalidate_on_change
async def add_groups_to_tag(self, name: str, group_ids: list[str]) -> int: # type: ignore
"""
向静态标签追加群组关联。
"""
tag = await GroupTag.get_or_none(name=name)
if not tag:
raise ValueError(f"标签 '{name}' 不存在。")
if tag.tag_type == "DYNAMIC":
raise ValueError("不能向动态标签手动添加群组。")
unique_group_ids = list(dict.fromkeys(group_ids))
await GroupTagLink.bulk_create(
[GroupTagLink(tag=tag, group_id=gid) for gid in unique_group_ids],
ignore_conflicts=True,
)
return len(unique_group_ids)
@invalidate_on_change
async def remove_groups_from_tag(self, name: str, group_ids: list[str]) -> int:
"""从静态标签移除指定群组。"""
tag = await GroupTag.get_or_none(name=name)
if not tag:
return 0
if tag.tag_type == "DYNAMIC":
raise ValueError("不能从动态标签手动移除群组。")
deleted_count = await GroupTagLink.filter(
tag=tag, group_id__in=group_ids
).delete()
return deleted_count
@invalidate_on_change
async def clone_tag(
self,
source_name: str,
new_name: str,
bot: Bot,
add_groups: list[str] | None = None,
remove_groups: list[str] | None = None,
as_dynamic: bool = False,
description: str | None = None,
mode: str | None = None,
) -> GroupTag:
"""
克隆一个标签,支持动态转静态、修改群组等。
"""
source_tag = await GroupTag.get_or_none(name=source_name)
if not source_tag:
raise ValueError(f"源标签 '{source_name}' 不存在。")
if await GroupTag.exists(name=new_name):
raise IntegrityError(f"目标标签 '{new_name}' 已存在。")
tag_type = "STATIC"
group_ids_to_set: list[str] | None = None
dynamic_rule: str | dict | None = None
if source_tag.tag_type == "STATIC":
if as_dynamic:
raise ValueError("不能将静态标签克隆为动态标签。")
group_ids_to_set = await GroupTagLink.filter(tag=source_tag).values_list( # type: ignore
"group_id", flat=True
)
else:
if as_dynamic:
tag_type = "DYNAMIC"
dynamic_rule = source_tag.dynamic_rule
if add_groups or remove_groups:
raise ValueError(
"克隆为动态标签时,不支持 --add 或 --remove 操作。"
)
else:
group_ids_to_set = await self.resolve_tag_to_group_ids(
source_name, bot=bot
)
if group_ids_to_set is not None:
final_group_set = set(group_ids_to_set)
if add_groups:
final_group_set.update(add_groups)
if remove_groups:
final_group_set.difference_update(remove_groups)
group_ids_to_set = list(final_group_set)
is_blacklist = (
(mode == "black") if mode is not None else source_tag.is_blacklist
)
return await self.create_tag(
name=new_name,
is_blacklist=is_blacklist,
description=description,
group_ids=group_ids_to_set,
tag_type=tag_type,
dynamic_rule=dynamic_rule,
)
async def list_tags_with_counts(self) -> list[dict]:
"""列出所有标签及其关联的群组数量。"""
tags = await GroupTag.all().prefetch_related("groups")
return [
{
"name": tag.name,
"description": tag.description,
"is_blacklist": tag.is_blacklist,
"tag_type": tag.tag_type,
"group_count": len(tag.groups),
}
for tag in tags
]
async def get_tag_details(self, name: str, bot: Bot | None = None) -> dict | None:
"""
获取标签的完整信息,包括基础属性、静态群组与动态解析结果。
参数:
name: 标签名称。
bot: 可选的 ``Bot`` 实例,用于在动态标签下获取实时群组信息。
返回:
包含标签详情的字典;若标签不存在则返回 ``None``。
"""
tag = await GroupTag.get_or_none(name=name).prefetch_related("groups")
if not tag:
return None
final_group_ids = await self.resolve_tag_to_group_ids(name, bot=bot)
resolved_groups: list[tuple[str, str]] = []
if final_group_ids:
groups_from_db = await GroupConsole.filter(
group_id__in=final_group_ids
).all()
resolved_groups = [(g.group_id, g.group_name) for g in groups_from_db]
return {
"name": tag.name,
"description": tag.description,
"is_blacklist": tag.is_blacklist,
"tag_type": tag.tag_type,
"dynamic_rule": tag.dynamic_rule,
"groups": [link.group_id for link in tag.groups],
"resolved_groups": resolved_groups,
}
async def _execute_rule(
self, rule_str: str, bot: Bot | None
) -> RuleExecutionResult:
"""使用Alconna解析并执行单个规则。"""
rule_str = " ".join(rule_str.split())
parts = rule_str.strip().split(maxsplit=1)
if not parts:
raise RuleExecutionError("规则字符串不能为空")
rule_name = parts[0]
handler_info = self._dynamic_handlers.get(rule_name)
if not handler_info:
available_rules = ", ".join(sorted(self._dynamic_handlers.keys()))
raise RuleExecutionError(
f"未知的规则名称: '{rule_name}'\n可用规则: {available_rules}"
)
try:
arparma = handler_info.alconna.parse(rule_str)
if not arparma.matched:
error_msg = (
str(arparma.error_info) if arparma.error_info else "未知语法错误"
)
args_info = []
if handler_info.alconna.args:
for arg in handler_info.alconna.args.argument:
arg_name = arg.name
arg_type = getattr(arg.value, "origin", arg.value)
type_name = getattr(arg_type, "__name__", str(arg_type))
args_info.append(f"<{arg_name}:{type_name}>")
expected_format = (
f"{rule_name} {' '.join(args_info)}" if args_info else rule_name
)
example = ""
if rule_name in ["member_count", "level"]:
example = f"\n示例: {rule_name} > 100"
elif rule_name in ["status", "is_super"]:
example = f"\n示例: {rule_name} = true"
elif rule_name == "group_name":
example = f"\n示例: {rule_name} contains 测试"
raise RuleExecutionError(
f"规则 '{rule_name}' 参数错误: {error_msg}\n"
f"期望格式: {expected_format}{example}"
)
func_to_check = (
handler_info.func.func
if isinstance(handler_info.func, partial)
else handler_info.func
)
extra_kwargs = {}
if "bot" in getattr(func_to_check, "__annotations__", {}):
extra_kwargs["bot"] = bot
result = await arparma.call(handler_info.func, **extra_kwargs)
if not isinstance(result, RuleExecutionResult):
raise TypeError(
f"处理器 '{rule_name}' 返回了不支持的类型 '{type(result)}'。 "
"必须返回 QueryResult, IDSetResult 或 ErrorResult。"
)
return result
except RuleExecutionError:
raise
except Exception as e:
raise RuleExecutionError(f"执行规则 '{rule_name}' 时发生内部错误: {e}")
async def _resolve_dynamic_tag(
self, rule: dict | str, bot: Bot | None = None
) -> set[str]:
"""根据动态规则解析符合条件的群组 ID 集合。"""
if isinstance(rule, dict):
raise RuleExecutionError("动态规则必须是字符串格式。")
final_ids: set[str] = set()
or_clauses = [part.strip() for part in rule.split(" or ")]
for or_clause in or_clauses:
current_and_q = Q()
current_and_ids: set[str] | None = None
and_rules = [part.strip() for part in or_clause.split(" and ")]
for simple_rule in and_rules:
try:
result = await self._execute_rule(simple_rule, bot)
if isinstance(result, QueryResult):
current_and_q &= result.q_object
elif isinstance(result, IDSetResult):
if current_and_ids is None:
current_and_ids = result.group_ids
else:
current_and_ids.intersection_update(result.group_ids)
elif isinstance(result, ErrorResult):
raise RuleExecutionError(result.message)
except Exception as e:
raise RuleExecutionError(
f"解析规则 '{simple_rule}' 时失败: {e}"
) from e
ids_from_q: set[str] | None = None
if current_and_q.children:
q_filtered_groups = await GroupConsole.filter(
current_and_q
).values_list("group_id", flat=True)
ids_from_q = {str(gid) for gid in q_filtered_groups}
if ids_from_q is not None:
if current_and_ids is None:
clause_result_ids = ids_from_q
else:
clause_result_ids = current_and_ids.intersection(ids_from_q)
else:
if current_and_ids is None:
clause_result_ids = set()
else:
clause_result_ids = current_and_ids
final_ids.update(clause_result_ids)
if bot:
bot_groups, _ = await PlatformUtils.get_group_list(bot)
bot_group_ids = {g.group_id for g in bot_groups if g.group_id}
final_ids.intersection_update(bot_group_ids)
return final_ids
@cached(ttl=300, namespace="tag_service")
async def resolve_tag_to_group_ids(
self, name: str, bot: Bot | None = None
) -> list[str]:
"""
核心解析方法:根据标签名解析出最终的群组ID列表
参数:
name: 需要解析的标签名称,特殊值 ``@all`` 表示所有群。
bot: 可选的 ``Bot`` 实例,用于拉取最新的群信息。
返回:
标签对应的群组 ID 列表。当标签不存在或无法解析时返回空列表。
"""
if name == "@all":
if bot:
all_groups, _ = await PlatformUtils.get_group_list(bot)
return [str(g.group_id) for g in all_groups if g.group_id]
else:
all_group_ids = await GroupConsole.all().values_list(
"group_id", flat=True
)
return [str(gid) for gid in all_group_ids]
tag = await GroupTag.get_or_none(name=name).prefetch_related("groups")
if not tag:
return []
associated_groups: set[str] = set()
if tag.tag_type == "STATIC":
associated_groups = {str(link.group_id) for link in tag.groups}
elif tag.tag_type == "DYNAMIC":
if not tag.dynamic_rule or not isinstance(tag.dynamic_rule, dict | str):
return []
dynamic_ids = await self._resolve_dynamic_tag(tag.dynamic_rule, bot)
associated_groups = {str(gid) for gid in dynamic_ids}
else:
associated_groups = {str(link.group_id) for link in tag.groups}
if tag.is_blacklist:
all_groups_query = GroupConsole.all()
if bot:
bot_groups, _ = await PlatformUtils.get_group_list(bot)
bot_group_ids = {str(g.group_id) for g in bot_groups if g.group_id}
if bot_group_ids:
all_groups_query = all_groups_query.filter(
group_id__in=bot_group_ids
)
else:
return []
all_relevant_group_ids_from_db = await all_groups_query.values_list(
"group_id", flat=True
)
all_relevant_group_ids = {
str(gid) for gid in all_relevant_group_ids_from_db
}
return list(all_relevant_group_ids - associated_groups)
else:
return list(associated_groups)
@invalidate_on_change
async def rename_tag(self, old_name: str, new_name: str) -> GroupTag:
"""重命名已有标签"""
if await GroupTag.exists(name=new_name):
raise IntegrityError(f"标签 '{new_name}' 已存在。")
tag = await GroupTag.get(name=old_name)
tag.name = new_name
await tag.save(update_fields=["name"])
return tag
@invalidate_on_change
async def update_tag_attributes(
self,
name: str,
description: str | None = None,
is_blacklist: bool | None = None,
dynamic_rule: dict | str | None = None,
) -> GroupTag:
"""
局部更新标签属性。
参数:
name: 标签名称。
description: 可选的新描述。
is_blacklist: 可选的新黑名单标记。
dynamic_rule: 可选的新动态规则配置。
返回:
更新后的 ``GroupTag`` 实例。
"""
tag = await GroupTag.get(name=name)
update_fields = []
if dynamic_rule is not None:
if tag.tag_type != "DYNAMIC":
raise ValueError("只能为动态标签更新规则。")
tag.dynamic_rule = dynamic_rule # type: ignore
update_fields.append("dynamic_rule")
if description is not None:
tag.description = description
update_fields.append("description")
if is_blacklist is not None:
tag.is_blacklist = is_blacklist
update_fields.append("is_blacklist")
if update_fields:
await tag.save(update_fields=update_fields)
return tag
@invalidate_on_change
async def set_groups_for_tag(self, name: str, group_ids: list[str]) -> int:
"""
覆盖设置静态标签的群组列表。
参数:
name: 标签名称。
group_ids: 需要绑定的群组 ID 列表。
返回:
设置成功后的群组数量。
"""
tag = await GroupTag.get(name=name)
if tag.tag_type == "DYNAMIC":
raise ValueError("不能为动态标签设置静态群组列表。")
async with in_transaction():
await GroupTagLink.filter(tag=tag).delete()
unique_group_ids = list(dict.fromkeys(group_ids))
if unique_group_ids:
await GroupTagLink.bulk_create(
[GroupTagLink(tag=tag, group_id=gid) for gid in unique_group_ids],
ignore_conflicts=True,
)
return len(unique_group_ids)
@invalidate_on_change
async def clear_all_tags(self) -> int:
"""删除所有标签,并清空缓存。"""
deleted_count = await GroupTag.all().delete()
return deleted_count
+41
View File
@@ -0,0 +1,41 @@
"""
动态标签的规则执行结果模型。
"""
from abc import ABC
from pydantic import BaseModel
from tortoise.expressions import Q
class RuleExecutionError(ValueError):
"""在规则执行期间,由处理器返回的、可向用户展示的错误。"""
pass
class RuleExecutionResult(BaseModel, ABC):
"""规则执行结果的抽象基类。"""
pass
class QueryResult(RuleExecutionResult):
"""表示数据库查询条件的结果。"""
q_object: Q
class Config:
arbitrary_types_allowed = True
class IDSetResult(RuleExecutionResult):
"""表示一组群组ID的结果。"""
group_ids: set[str]
class ErrorResult(RuleExecutionResult):
"""表示一个可向用户显示的错误。"""
message: str
+4 -4
View File
@@ -64,13 +64,13 @@ class RenderableComponent(BaseModel, Renderable):
@compat_computed_field
def inline_style_str(self) -> str:
"""[新增] 一个辅助属性,将内联样式字典转换为CSS字符串"""
"""一个辅助属性,将内联样式字典转换为CSS字符串"""
if not self.inline_style:
return ""
return "; ".join(f"{k}: {v}" for k, v in self.inline_style.items())
def get_extra_css(self, context: Any) -> str | Awaitable[str]:
return ""
return self.component_css or ""
class ContainerComponent(RenderableComponent, ABC):
@@ -86,7 +86,7 @@ class ContainerComponent(RenderableComponent, ABC):
raise NotImplementedError
def get_required_scripts(self) -> list[str]:
"""[新增] 聚合所有子组件的脚本依赖。"""
"""聚合所有子组件的脚本依赖。"""
scripts = set(super().get_required_scripts())
for child in self.get_children():
if child:
@@ -94,7 +94,7 @@ class ContainerComponent(RenderableComponent, ABC):
return list(scripts)
def get_required_styles(self) -> list[str]:
"""[新增] 聚合所有子组件的样式依赖。"""
"""聚合所有子组件的样式依赖。"""
styles = set(super().get_required_styles())
for child in self.get_children():
if child:
+8 -3
View File
@@ -192,11 +192,15 @@ class MarkdownData(ContainerComponent):
yield from find_components_recursive(self.elements)
async def get_extra_css(self, context: Any) -> str:
css_parts = []
if self.component_css:
css_parts.append(self.component_css)
if self.css_path:
css_file = Path(self.css_path)
if css_file.is_file():
async with aiofiles.open(css_file, encoding="utf-8") as f:
return await f.read()
css_parts.append(await f.read())
else:
logger.warning(f"Markdown自定义CSS文件不存在: {self.css_path}")
else:
@@ -206,5 +210,6 @@ class MarkdownData(ContainerComponent):
)
if css_path and css_path.exists():
async with aiofiles.open(css_path, encoding="utf-8") as f:
return await f.read()
return ""
css_parts.append(await f.read())
return "\n".join(css_parts)
+7 -4
View File
@@ -126,12 +126,15 @@ class SqlUtils:
def format_usage_for_markdown(text: str) -> str:
"""
智能地将Python多行字符串转换为适合Markdown渲染的格式。
- 将单个换行符替换为Markdown的硬换行(行尾加两个空格)。
- 在列表、标题等块级元素前自动插入换行,确保正确解析。
- 将段落内的单个换行符替换为Markdown的硬换行(行尾加两个空格)。
- 保留两个或更多的连续换行符,使其成为Markdown的段落分隔。
"""
if not text:
return ""
text = re.sub(r"\n{2,}", "<<PARAGRAPH_BREAK>>", text)
text = text.replace("\n", " \n")
text = text.replace("<<PARAGRAPH_BREAK>>", "\n\n")
text = re.sub(r"([^\n])\n(\s*[-*] |\s*#+\s|\s*>)", r"\1\n\n\2", text)
text = re.sub(r"(?<!\n)\n(?!\n)", " \n", text)
return text
+12
View File
@@ -263,6 +263,18 @@ class AsyncHttpx:
)
return result
except Exception as e:
if isinstance(e, HTTPStatusError):
status = getattr(e.response, "status_code", "?")
try:
body_text = getattr(e.response, "text", None)
if body_text is not None and len(body_text) > 2000:
body_text = body_text[:2000] + "...(truncated)"
except Exception:
body_text = "<unavailable>"
logger.debug(
f"请求失败: {url} {status} {body_text}",
"AsyncHttpx:FallbackExecutor",
)
exceptions.append(e)
if url != url_list[-1]:
logger.warning(
+4 -2
View File
@@ -6,7 +6,7 @@ import random
import re
import imagehash
from nonebot.utils import is_coroutine_callable
from nonebot.utils import is_coroutine_callable, run_sync
from PIL import Image
from zhenxun.configs.path_config import TEMP_PATH
@@ -378,7 +378,9 @@ async def get_download_image_hash(url: str, mark: str, use_proxy: bool = False)
if await AsyncHttpx.download_file(
url, TEMP_PATH / f"compare_download_{mark}_img.jpg", use_proxy=use_proxy
):
img_hash = get_img_hash(TEMP_PATH / f"compare_download_{mark}_img.jpg")
img_hash = await run_sync(get_img_hash)(
TEMP_PATH / f"compare_download_{mark}_img.jpg"
)
return str(img_hash)
except Exception as e:
logger.warning("下载读取图片Hash出错", e=e)
+202
View File
@@ -0,0 +1,202 @@
import copy
import re
from typing import Any
from nonebot.adapters import Message, MessageSegment
def _truncate_base64_string(value: str, threshold: int = 256) -> str:
"""如果字符串是超长的base64或data URI,则截断它。"""
if not isinstance(value, str):
return value
prefixes = ("base64://", "data:image", "data:video", "data:audio")
if value.startswith(prefixes) and len(value) > threshold:
prefix = next((p for p in prefixes if value.startswith(p)), "base64")
return f"[{prefix}_data_omitted_len={len(value)}]"
return value
def _sanitize_ui_html(html_string: str) -> str:
"""
专门用于净化UI渲染调试HTML的函数。
它会查找所有内联的base64数据(如字体、图片)并将其截断。
"""
if not isinstance(html_string, str):
return html_string
pattern = re.compile(r"(data:[^;]+;base64,)[A-Za-z0-9+/=\s]{100,}")
def replacer(match):
prefix = match.group(1)
original_len = len(match.group(0)) - len(prefix)
return f"{prefix}[...base64_omitted_len={original_len}...]"
return pattern.sub(replacer, html_string)
def _sanitize_nonebot_message(message: Message) -> Message:
"""净化nonebot.adapter.Message对象,用于日志记录。"""
sanitized_message = copy.deepcopy(message)
for seg in sanitized_message:
seg: MessageSegment
if seg.type in ("image", "record", "video"):
file_info = seg.data.get("file", "")
if isinstance(file_info, str):
seg.data["file"] = _truncate_base64_string(file_info)
return sanitized_message
def _sanitize_openai_response(response_json: dict) -> dict:
"""净化OpenAI兼容API的响应体。"""
try:
sanitized_json = copy.deepcopy(response_json)
if "choices" in sanitized_json and isinstance(sanitized_json["choices"], list):
for choice in sanitized_json["choices"]:
if "message" in choice and isinstance(choice["message"], dict):
message = choice["message"]
if "images" in message and isinstance(message["images"], list):
for i, image_info in enumerate(message["images"]):
if "image_url" in image_info and isinstance(
image_info["image_url"], dict
):
url = image_info["image_url"].get("url", "")
message["images"][i]["image_url"]["url"] = (
_truncate_base64_string(url)
)
return sanitized_json
except Exception:
return response_json
def _sanitize_openai_request(body: dict) -> dict:
"""净化OpenAI兼容API的请求体,主要截断图片base64。"""
try:
sanitized_json = copy.deepcopy(body)
if "messages" in sanitized_json and isinstance(
sanitized_json["messages"], list
):
for message in sanitized_json["messages"]:
if "content" in message and isinstance(message["content"], list):
for i, part in enumerate(message["content"]):
if part.get("type") == "image_url":
if "image_url" in part and isinstance(
part["image_url"], dict
):
url = part["image_url"].get("url", "")
message["content"][i]["image_url"]["url"] = (
_truncate_base64_string(url)
)
return sanitized_json
except Exception:
return body
def _sanitize_gemini_response(response_json: dict) -> dict:
"""净化Gemini API的响应体,处理文本和图片生成两种格式。"""
try:
sanitized_json = copy.deepcopy(response_json)
def _process_candidates(candidates_list: list):
"""辅助函数,用于处理任何 candidates 列表。"""
if not isinstance(candidates_list, list):
return
for candidate in candidates_list:
if "content" in candidate and isinstance(candidate["content"], dict):
content = candidate["content"]
if "parts" in content and isinstance(content["parts"], list):
for i, part in enumerate(content["parts"]):
if "inlineData" in part and isinstance(
part["inlineData"], dict
):
data = part["inlineData"].get("data", "")
if isinstance(data, str) and len(data) > 256:
content["parts"][i]["inlineData"]["data"] = (
f"[base64_data_omitted_len={len(data)}]"
)
if "candidates" in sanitized_json:
_process_candidates(sanitized_json["candidates"])
if "image_generation" in sanitized_json and isinstance(
sanitized_json["image_generation"], dict
):
if "candidates" in sanitized_json["image_generation"]:
_process_candidates(sanitized_json["image_generation"]["candidates"])
return sanitized_json
except Exception:
return response_json
def _sanitize_gemini_request(body: dict) -> dict:
"""净化Gemini API的请求体,进行结构转换和总结。"""
try:
sanitized_body = copy.deepcopy(body)
if "contents" in sanitized_body and isinstance(
sanitized_body["contents"], list
):
for content_item in sanitized_body["contents"]:
if "parts" in content_item and isinstance(content_item["parts"], list):
media_summary = []
new_parts = []
for part in content_item["parts"]:
if "inlineData" in part and isinstance(
part["inlineData"], dict
):
data = part["inlineData"].get("data")
if isinstance(data, str):
mime_type = part["inlineData"].get(
"mimeType", "unknown"
)
media_summary.append(f"{mime_type} ({len(data)} chars)")
continue
new_parts.append(part)
if media_summary:
summary_text = (
f"[多模态内容: {len(media_summary)}个文件 - "
f"{', '.join(media_summary)}]"
)
new_parts.insert(0, {"text": summary_text})
content_item["parts"] = new_parts
return sanitized_body
except Exception:
return body
def sanitize_for_logging(data: Any, context: str | None = None) -> Any:
"""
统一的日志净化入口。
Args:
data: 需要净化的数据 (dict, Message, etc.).
context: 净化场景的上下文标识,例如 'gemini_request', 'openai_response'.
Returns:
净化后的数据。
"""
if context == "nonebot_message":
if isinstance(data, Message):
return _sanitize_nonebot_message(data)
elif context == "openai_response":
if isinstance(data, dict):
return _sanitize_openai_response(data)
elif context == "gemini_response":
if isinstance(data, dict):
return _sanitize_gemini_response(data)
elif context == "gemini_request":
if isinstance(data, dict):
return _sanitize_gemini_request(data)
elif context == "openai_request":
if isinstance(data, dict):
return _sanitize_openai_request(data)
elif context == "ui_html":
if isinstance(data, str):
return _sanitize_ui_html(data)
else:
if isinstance(data, str):
return _truncate_base64_string(data)
return data
+39
View File
@@ -5,10 +5,14 @@ Pydantic V1 & V2 兼容层模块
包括 model_dump, model_copy, model_json_schema, parse_as 等。
"""
from datetime import datetime
from enum import Enum
from pathlib import Path
from typing import Any, TypeVar, get_args, get_origin
from nonebot.compat import PYDANTIC_V2, model_dump
from pydantic import VERSION, BaseModel
import ujson as json
T = TypeVar("T", bound=BaseModel)
V = TypeVar("V")
@@ -19,9 +23,11 @@ __all__ = [
"_dump_pydantic_obj",
"_is_pydantic_type",
"compat_computed_field",
"dump_json_safely",
"model_copy",
"model_dump",
"model_json_schema",
"model_validate",
"parse_as",
]
@@ -39,6 +45,16 @@ def model_copy(
return model.copy(update=update_dict, deep=deep)
def model_validate(model_class: type[T], obj: Any) -> T:
"""
Pydantic `model_validate` (v2) 与 `parse_obj` (v1) 的兼容函数。
"""
if PYDANTIC_V2:
return model_class.model_validate(obj)
else:
return model_class.parse_obj(obj)
if PYDANTIC_V2:
from pydantic import computed_field as compat_computed_field
else:
@@ -93,3 +109,26 @@ def parse_as(type_: type[V], obj: Any) -> V:
from pydantic import TypeAdapter # type: ignore
return TypeAdapter(type_).validate_python(obj)
def dump_json_safely(obj: Any, **kwargs) -> str:
"""
安全地将可能包含 Pydantic 特定类型 (如 Enum) 的对象序列化为 JSON 字符串。
"""
def default_serializer(o):
if isinstance(o, Enum):
return o.value
if isinstance(o, datetime):
return o.isoformat()
if isinstance(o, Path):
return str(o.as_posix())
if isinstance(o, set):
return list(o)
if isinstance(o, BaseModel):
return model_dump(o)
raise TypeError(
f"Object of type {o.__class__.__name__} is not JSON serializable"
)
return json.dumps(obj, default=default_serializer, **kwargs)
+29 -3
View File
@@ -48,13 +48,13 @@ class TimeUtils:
@classmethod
def parse_time_string(cls, time_str: str) -> int:
"""
将带有单位的时间字符串 (e.g., "10s", "5m", "1h") 解析为总秒数。
将带有单位的时间字符串 (e.g., "10s", "5m", "1h", "1d") 解析为总秒数。
"""
time_str = time_str.lower().strip()
match = re.match(r"^(\d+)([smh])$", time_str)
match = re.match(r"^(\d+)([smhd])$", time_str)
if not match:
raise ValueError(
f"无效的时间格式: '{time_str}'。请使用如 '30s', '10m', '2h' 的格式。"
f"无效的时间格式: '{time_str}'。请使用如 '30s', '10m', '2h', '1d'的格式"
)
value, unit = int(match.group(1)), match.group(2)
@@ -65,8 +65,34 @@ class TimeUtils:
return value * 60
if unit == "h":
return value * 3600
if unit == "d":
return value * 86400
return 0
@classmethod
def parse_interval_to_dict(cls, interval_str: str) -> dict:
"""
将时间间隔字符串解析为 APScheduler 的 interval 触发器所需的字典。
"""
time_str_lower = interval_str.lower().strip()
match = re.match(r"^(\d+)([smhd])$", time_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 {}
@classmethod
def format_duration(cls, seconds: float) -> str:
"""