diff --git a/plugins/robot_menu/main.py b/plugins/robot_menu/main.py index e3f4141..3d52b6e 100644 --- a/plugins/robot_menu/main.py +++ b/plugins/robot_menu/main.py @@ -1,8 +1,3 @@ -import asyncio -import html -import os -import time -from pathlib import Path from typing import Dict, Any, List, Optional, Tuple from loguru import logger @@ -11,9 +6,9 @@ import xml.etree.ElementTree as ET from base.plugin_common.message_plugin_interface import MessagePluginInterface from base.plugin_common.plugin_interface import PluginStatus +from .menu_render_tool import RobotMenuRenderTool from utils.decorator.plugin_decorators import plugin_stats_decorator from utils.decorator.points_decorator import plugin_points_cost -from utils.markdown_to_image import convert_md_str_to_image, html_to_image from utils.revoke.message_auto_revoke import MessageAutoRevoke from utils.robot_cmd.robot_command import Feature, PermissionStatus, GroupBotManager from utils.wechat.contact_manager import ContactManager @@ -77,11 +72,8 @@ class RobotMenuPlugin(MessagePluginInterface): self.command_format = self._config.get("RobotMenu", {}).get("command-format", "菜单 功能名") self.enable = self._config.get("RobotMenu", {}).get("enable", True) # 输出模式配置：支持 text / image 两种模式。 - # 业务诉求是“菜单默认发图片”，因此配置文件默认会写 image； - # 为兼容历史配置，这里做兜底归一化，非法值统一降级到 text。 - self.menu_output_mode = self._normalize_output_mode( - self._config.get("RobotMenu", {}).get("output_mode", "text") - ) + # 这里不在主插件内实现渲染细节，而是交给独立工具模块处理。 + output_mode = self._config.get("RobotMenu", {}).get("output_mode", "text") # 图片生成失败时是否回退到文本： # - True：优先保证消息可达； # - False：严格按“只发图片”执行（失败时仅提示失败原因）。 @@ -95,6 +87,14 @@ class RobotMenuPlugin(MessagePluginInterface): self.image_render_retries = int( self._config.get("RobotMenu", {}).get("image_render_retries", 1) ) + # 初始化“菜单渲染工具”，后续菜单图片与文本发送统一由该工具负责。 + self.menu_renderer = RobotMenuRenderTool( + output_mode=output_mode, + image_fallback_to_text=self.image_fallback_to_text, + image_render_timeout_seconds=self.image_render_timeout_seconds, + image_render_retries=self.image_render_retries, + log=self.LOG, + ) self.LOG.debug(f"[{self.name}] 插件初始化完成，指令：{self._commands}") return True @@ -176,363 +176,6 @@ class RobotMenuPlugin(MessagePluginInterface): return "当前没有启用机器人的群组" return "\n".join(group_list) - @staticmethod - def _normalize_output_mode(raw_mode: Any) -> str: - """将配置中的输出模式标准化为 text 或 image。""" - mode = str(raw_mode or "").strip().lower() - if mode in {"image", "img", "picture", "pic"}: - return "image" - return "text" - - @staticmethod - def _split_feature_description(feature_desc: str) -> Tuple[str, str]: - """拆分功能描述为“功能说明”和“指令说明”。 - - 约定： - 1. `Feature.description` 中使用 `[...]` 包裹指令或触发方式； - 2. 若未配置方括号，则指令说明使用“无”占位，保证菜单结构稳定。 - """ - desc = str(feature_desc or "").strip() - match = re.match(r"^(.*?)(?:\[(.*?)\])?$", desc) - if not match: - return desc or "未命名功能", "无" - title = (match.group(1) or "").strip() or "未命名功能" - cmd_hint = (match.group(2) or "").strip() or "无" - return title, cmd_hint - - def _build_feature_status_markdown(self, group_id: str) -> str: - """构建菜单图片使用的 Markdown 文本。 - - 输出目标： - 1. 逐项列出每个插件（Feature）当前状态； - 2. 明确展示每个插件的功能描述与指令/触发方式； - 3. 让用户在一张图片里即可完成“查看状态 + 查指令”。 - """ - lines = [ - "# 机器人功能菜单", - "", - f"- 目标：`{group_id}`", - f"- 生成时间：`{time.strftime('%Y-%m-%d %H:%M:%S', time.localtime())}`", - "", - "| 序号 | 功能键 | 状态 | 功能说明 | 指令/触发方式 |", - "| --- | --- | --- | --- | --- |", - ] - for feature in Feature: - status = GroupBotManager.get_group_permission(group_id, feature) - status_text = "启用 ✅" if status == PermissionStatus.ENABLED else "关闭 ❌" - title, cmd_hint = self._split_feature_description(feature.description) - lines.append( - f"| {feature.value} | `{feature.name}` | {status_text} | {title} | `{cmd_hint}` |" - ) - return "\n".join(lines) - - def _get_feature_command_examples(self, feature: Feature) -> str: - """返回功能的详细指令示例文本。 - - 设计说明： - 1. `Feature.description` 中的 `[]` 内容通常是“触发指令”或“触发方式”； - 2. 当某些功能没有显式指令时，给出“自动触发/定时触发”提示，避免用户误以为缺失； - 3. 所有功能统一补充“启用/关闭管理命令”，让用户知道如何控制开关。 - """ - _, cmd_hint = self._split_feature_description(feature.description) - # 统一的功能开关命令模板：管理员可通过序号或功能键控制开关。 - manage_hint = ( - f"管理：菜单 启用 {feature.value} ｜ 菜单 关闭 {feature.value} ｜ " - f"菜单 启用 {feature.name} ｜ 菜单 关闭 {feature.name}" - ) - if cmd_hint == "无": - return f"触发：自动/定时触发（无直接聊天指令）
{manage_hint}" - return f"触发：{html.escape(cmd_hint)}
{manage_hint}" - - def _build_feature_status_html(self, group_id: str) -> str: - """构建菜单图片使用的 HTML 文本（自定义视觉，不复用 md2image 默认样式）。 - - 目标： - 1. 清晰分区：基础命令、管理员命令、功能明细； - 2. 每个功能独立卡片，展示状态、用途、触发指令与管理命令； - 3. 页面在移动端宽度下也可完整展示，减少截图后拥挤感。 - """ - feature_cards: List[str] = [] - for feature in Feature: - status = GroupBotManager.get_group_permission(group_id, feature) - enabled = status == PermissionStatus.ENABLED - status_class = "badge-on" if enabled else "badge-off" - status_text = "已启用" if enabled else "已关闭" - title, _ = self._split_feature_description(feature.description) - command_examples = self._get_feature_command_examples(feature) - feature_cards.append( - f""" -
-
-
#{feature.value}
-
-

{html.escape(title)}

-

功能键：{html.escape(feature.name)}

-
- {status_text} -
-
-

{command_examples}

-
-
- """ - ) - - now_text = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime()) - return f""" - - - - - - - -
-
-

机器人功能菜单

-

目标群/会话：{html.escape(group_id)}

-

生成时间：{now_text}

-
-
-
-

基础命令

-
    -
  • 菜单：查看完整功能菜单（状态 + 详细指令）
    • -
  • 菜单 状态：查看所有功能当前启用状态
    • -
  • 菜单 群列表：查看已启用群机器人的群组清单
    • -
-
-
-

管理员命令

-
    -
  • 菜单 启用 序号 / 菜单 关闭 序号
    • -
  • 菜单 启用 功能键 / 菜单 关闭 功能键
    • -
  • 菜单 管理员 添加 wxid/昵称、菜单 管理员 删除 wxid/昵称、菜单 管理员 列表
    • -
-
-
-

功能明细

-
- {''.join(feature_cards)} -
-
-
-
- - - """ - - async def _send_menu_content( - self, - bot: WechatAPIClient, - target: str, - sender: str, - revoke: MessageAutoRevoke, - text_content: str, - markdown_content: Optional[str] = None, - html_content: Optional[str] = None, - revoke_seconds: int = 90, - ) -> None: - """按配置发送菜单内容（文本或图片）。 - - 发送策略： - 1. `output_mode=text`：走历史文本发送逻辑，并登记自动撤回； - 2. `output_mode=image`：先用 md2image 渲染图片后发送； - 3. 图片失败时根据 `image_fallback_to_text` 决定是否回退文本。 - """ - # 文本模式：保持原有行为，避免影响已有群聊使用习惯。 - if self.menu_output_mode != "image": - client_msg_id, create_time, new_msg_id = await bot.send_text_message(target, text_content, sender) - revoke.add_message_to_revoke(target, client_msg_id, create_time, new_msg_id, revoke_seconds) - return - - # 图片模式：优先使用“自定义 HTML 模板”渲染图片，不使用 md2image 默认 Markdown 样式。 - # 兼容策略： - # 1. 若调用方传了 html_content，则按该模板直接截图； - # 2. 若未传 html_content，才回退到 Markdown 转图（作为兜底，不影响稳定性）。 - md_content = (markdown_content or "").strip() or f"```text\n{text_content}\n```" - output_image = f"robot_menu_{int(time.time() * 1000)}.png" - try: - # 增加总超时保护，防止图片渲染阻塞主流程过久。 - total_timeout = max(30, self.image_render_timeout_seconds * max(1, self.image_render_retries) + 10) - output_dir = Path(os.getcwd()) / "temp" / "md2image" - output_dir.mkdir(parents=True, exist_ok=True) - output_path = output_dir / output_image - if html_content and html_content.strip(): - await asyncio.wait_for( - html_to_image(html_content, str(output_path)), - timeout=total_timeout, - ) - image_path = str(output_path.resolve()) - else: - image_path = await asyncio.wait_for( - convert_md_str_to_image( - md_content, - output_image, - max_retries=max(1, self.image_render_retries), - render_timeout_seconds=max(10, self.image_render_timeout_seconds), - html_timeout_seconds=min(30, max(10, self.image_render_timeout_seconds)), - ), - timeout=total_timeout, - ) - await bot.send_image_message(target, Path(image_path)) - except Exception as e: - self.LOG.error(f"[{self.name}] 菜单图片发送失败: {e}") - if self.image_fallback_to_text: - client_msg_id, create_time, new_msg_id = await bot.send_text_message(target, text_content, sender) - revoke.add_message_to_revoke(target, client_msg_id, create_time, new_msg_id, revoke_seconds) - else: - # 严格图片模式下不回退完整菜单文本，仅发送简短失败提示。 - client_msg_id, create_time, new_msg_id = await bot.send_text_message( - target, - "❌ 菜单图片生成失败，请稍后重试", - sender - ) - revoke.add_message_to_revoke(target, client_msg_id, create_time, new_msg_id, 30) - def at_list(self, xml): try: root = ET.fromstring(xml) @@ -575,9 +218,9 @@ class RobotMenuPlugin(MessagePluginInterface): if len(parts) == 1: # 显示功能菜单 menu_text = self.get_enabled_features(roomid if roomid else sender) - menu_markdown = self._build_feature_status_markdown(roomid if roomid else sender) - menu_html = self._build_feature_status_html(roomid if roomid else sender) - await self._send_menu_content( + menu_markdown = self.menu_renderer.build_feature_status_markdown(roomid if roomid else sender) + menu_html = self.menu_renderer.build_feature_status_html(roomid if roomid else sender) + await self.menu_renderer.send_menu_content( bot=bot, target=target, sender=sender, @@ -597,9 +240,9 @@ class RobotMenuPlugin(MessagePluginInterface): if cmd_name.upper() == "状态": # 显示所有功能状态 status_text = self.display_menu_status(roomid if roomid else sender) - status_markdown = self._build_feature_status_markdown(roomid if roomid else sender) - status_html = self._build_feature_status_html(roomid if roomid else sender) - await self._send_menu_content( + status_markdown = self.menu_renderer.build_feature_status_markdown(roomid if roomid else sender) + status_html = self.menu_renderer.build_feature_status_html(roomid if roomid else sender) + await self.menu_renderer.send_menu_content( bot=bot, target=target, sender=sender, diff --git a/plugins/robot_menu/menu_render_tool.py b/plugins/robot_menu/menu_render_tool.py new file mode 100644 index 0000000..bc1ae34 --- /dev/null +++ b/plugins/robot_menu/menu_render_tool.py @@ -0,0 +1,323 @@ +import asyncio +import html +import os +import re +import time +from pathlib import Path +from typing import Any, Optional, Tuple + +from loguru import logger as default_logger + +from utils.markdown_to_image import convert_md_str_to_image, html_to_image +from utils.revoke.message_auto_revoke import MessageAutoRevoke +from utils.robot_cmd.robot_command import Feature, GroupBotManager, PermissionStatus +from wechat_ipad import WechatAPIClient + + +class RobotMenuRenderTool: + """机器人菜单渲染工具。 + + 设计目标： + 1. 将“菜单排版 + 图片渲染 + 发送策略”从主插件拆出，降低 main.py 维护成本； + 2. 统一菜单文本/图片输出行为，保证菜单与状态页展示一致； + 3. 对外只暴露清晰方法，主插件只负责“取业务数据 + 调用工具”。 + """ + + def __init__( + self, + output_mode: Any, + image_fallback_to_text: bool, + image_render_timeout_seconds: int, + image_render_retries: int, + log=default_logger, + ): + # 输出模式：支持 text / image，非法值自动归一化为 text。 + self.output_mode = self.normalize_output_mode(output_mode) + # 图片失败时是否回退文本。 + self.image_fallback_to_text = bool(image_fallback_to_text) + # 渲染超时与重试参数，统一集中在工具层处理。 + self.image_render_timeout_seconds = int(image_render_timeout_seconds) + self.image_render_retries = int(image_render_retries) + # 注入日志对象，便于主插件统一控制日志风格与输出目标。 + self.log = log or default_logger + + @staticmethod + def normalize_output_mode(raw_mode: Any) -> str: + """将配置中的输出模式标准化为 `text` 或 `image`。""" + mode = str(raw_mode or "").strip().lower() + if mode in {"image", "img", "picture", "pic"}: + return "image" + return "text" + + @staticmethod + def _split_feature_description(feature_desc: str) -> Tuple[str, str]: + """拆分功能描述为“功能说明”和“触发指令提示”。 + + 约定： + 1. 描述中 `[]` 内的文本作为“触发说明”； + 2. 无 `[]` 时返回“无”，保证渲染模板字段完整。 + """ + desc = str(feature_desc or "").strip() + match = re.match(r"^(.*?)(?:\[(.*?)\])?$", desc) + if not match: + return desc or "未命名功能", "无" + title = (match.group(1) or "").strip() or "未命名功能" + cmd_hint = (match.group(2) or "").strip() or "无" + return title, cmd_hint + + def build_feature_status_markdown(self, group_id: str) -> str: + """构建菜单 Markdown（作为图片渲染兜底或文本调试来源）。""" + lines = [ + "# 机器人功能菜单", + "", + f"- 目标：`{group_id}`", + f"- 生成时间：`{time.strftime('%Y-%m-%d %H:%M:%S', time.localtime())}`", + "", + "| 序号 | 功能键 | 状态 | 功能说明 | 指令/触发方式 |", + "| --- | --- | --- | --- | --- |", + ] + for feature in Feature: + status = GroupBotManager.get_group_permission(group_id, feature) + status_text = "启用 ✅" if status == PermissionStatus.ENABLED else "关闭 ❌" + title, cmd_hint = self._split_feature_description(feature.description) + lines.append( + f"| {feature.value} | `{feature.name}` | {status_text} | {title} | `{cmd_hint}` |" + ) + return "\n".join(lines) + + def _get_feature_command_examples(self, feature: Feature) -> str: + """返回单个功能在菜单卡片中的详细指令文案。""" + _, cmd_hint = self._split_feature_description(feature.description) + manage_hint = ( + f"管理：菜单 启用 {feature.value} ｜ 菜单 关闭 {feature.value} ｜ " + f"菜单 启用 {feature.name} ｜ 菜单 关闭 {feature.name}" + ) + if cmd_hint == "无": + return f"触发：自动/定时触发（无直接聊天指令）
{manage_hint}" + return f"触发：{html.escape(cmd_hint)}
{manage_hint}" + + def build_feature_status_html(self, group_id: str) -> str: + """构建菜单 HTML（自定义样式，不复用 md2image 默认样式）。""" + feature_cards = [] + for feature in Feature: + status = GroupBotManager.get_group_permission(group_id, feature) + enabled = status == PermissionStatus.ENABLED + status_class = "badge-on" if enabled else "badge-off" + status_text = "已启用" if enabled else "已关闭" + title, _ = self._split_feature_description(feature.description) + command_examples = self._get_feature_command_examples(feature) + feature_cards.append( + f""" +
+
+
#{feature.value}
+
+

{html.escape(title)}

+

功能键：{html.escape(feature.name)}

+
+ {status_text} +
+
+

{command_examples}

+
+
+ """ + ) + + now_text = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime()) + return f""" + + + + + + + +
+
+

机器人功能菜单

+

目标群/会话：{html.escape(group_id)}

+

生成时间：{now_text}

+
+
+
+

基础命令

+
    +
  • 菜单：查看完整功能菜单（状态 + 详细指令）
    • +
  • 菜单 状态：查看所有功能当前启用状态
    • +
  • 菜单 群列表：查看已启用群机器人的群组清单
    • +
+
+
+

管理员命令

+
    +
  • 菜单 启用 序号 / 菜单 关闭 序号
    • +
  • 菜单 启用 功能键 / 菜单 关闭 功能键
    • +
  • 菜单 管理员 添加 wxid/昵称、菜单 管理员 删除 wxid/昵称、菜单 管理员 列表
    • +
+
+
+

功能明细

+
+ {''.join(feature_cards)} +
+
+
+
+ + + """ + + async def send_menu_content( + self, + bot: WechatAPIClient, + target: str, + sender: str, + revoke: MessageAutoRevoke, + text_content: str, + markdown_content: Optional[str] = None, + html_content: Optional[str] = None, + revoke_seconds: int = 90, + ) -> None: + """按配置发送菜单内容（文本或图片）。""" + if self.output_mode != "image": + client_msg_id, create_time, new_msg_id = await bot.send_text_message(target, text_content, sender) + revoke.add_message_to_revoke(target, client_msg_id, create_time, new_msg_id, revoke_seconds) + return + + md_content = (markdown_content or "").strip() or f"```text\n{text_content}\n```" + output_image = f"robot_menu_{int(time.time() * 1000)}.png" + try: + total_timeout = max(30, self.image_render_timeout_seconds * max(1, self.image_render_retries) + 10) + output_dir = Path(os.getcwd()) / "temp" / "md2image" + output_dir.mkdir(parents=True, exist_ok=True) + output_path = output_dir / output_image + if html_content and html_content.strip(): + await asyncio.wait_for( + html_to_image(html_content, str(output_path)), + timeout=total_timeout, + ) + image_path = str(output_path.resolve()) + else: + image_path = await asyncio.wait_for( + convert_md_str_to_image( + md_content, + output_image, + max_retries=max(1, self.image_render_retries), + render_timeout_seconds=max(10, self.image_render_timeout_seconds), + html_timeout_seconds=min(30, max(10, self.image_render_timeout_seconds)), + ), + timeout=total_timeout, + ) + await bot.send_image_message(target, Path(image_path)) + except Exception as e: + self.log.error(f"[机器人菜单渲染工具] 菜单图片发送失败: {e}") + if self.image_fallback_to_text: + client_msg_id, create_time, new_msg_id = await bot.send_text_message(target, text_content, sender) + revoke.add_message_to_revoke(target, client_msg_id, create_time, new_msg_id, revoke_seconds) + else: + client_msg_id, create_time, new_msg_id = await bot.send_text_message( + target, + "❌ 菜单图片生成失败，请稍后重试", + sender + ) + revoke.add_message_to_revoke(target, client_msg_id, create_time, new_msg_id, 30)