HeXi/hexi/core/message_utils.py

import mimetypes
from pathlib import Path

from nonebot import get_bot
from nonebot import require
from nonebot.adapters import Event
from nonebot.adapters.onebot.v11 import Bot, Message, MessageEvent, MessageSegment
from nonebot.log import logger
from nonebot_plugin_alconna import UniMessage
from nonebot_plugin_alconna.uniseg import Receipt, Target

require("nonebot_plugin_htmlrender")
from nonebot_plugin_htmlrender import md_to_pic

# 「处理中」占位动图(任务驱动通用回复用)
THINKING_GIF = Path(__file__).parents[1] / "resource" / "imgs" / "thinking.gif"


async def send_markdown(
        bot: Bot,
        event: MessageEvent,
        markdown: str,
        fallback_text: str | None = None,
) -> None:
    """发送 Markdown

    优先尝试 OneBot 后端的原生 markdown 消息段;后端不支持/失败时
    自动降级为 htmlrender 把 Markdown 渲染成图片发送,保证任何后端都能看到。
    """
    # 1. 尝试原生 markdown 消息段(不同后端字段可能不同,失败自动降级)
    try:
        seg = MessageSegment("markdown", {"content": markdown})
        await bot.send(event=event, message=Message(seg))
        return
    except Exception as e:
        logger.warning(f"原生 markdown 发送失败,尝试降级为图片: {e}")

    # 2. 降级:htmlrender 渲染成图片
    try:
        img = await md_to_pic(md=markdown, width=540, device_scale_factor=4)
        await bot.send(event=event, message=MessageSegment.image(img))
    except Exception as e:
        logger.error(f"Markdown 渲染图片失败,发送纯文本: {e}")
        await bot.send(event=event, message=fallback_text or markdown)


async def send_poke(group_id: int | None = None, user_id: int = 0):
    """发送戳一戳(OneBot V11 send_poke API)

    group_id 为空时向私聊发送戳一戳;失败返回 None
    """
    bot = get_bot()
    params = {"user_id": int(user_id)}
    if group_id:
        params["group_id"] = int(group_id)
    logger.debug(f"send_poke: {params}")
    try:
        return await bot.call_api("send_poke", **params)
    except Exception as e:
        logger.error(f"send_poke 失败: {e}")
        return None


def get_reply_message(event: MessageEvent) -> Message:
    """获取消息内容:若为回复消息则返回被回复的消息,否则返回当前消息"""
    if event.reply is not None:
        logger.info(f"此条消息为回复消息: {event.reply.message}")
        return event.reply.message
    logger.info(f"此条消息为普通消息: {event.message}")
    return event.message


def build_forward_nodes(
        self_id: int,
        entries: list[tuple[str, str]],
        sender_name: str | None = None,
) -> list[dict]:
    """构造合并转发节点列表

    entries: [(发送者名, 消息内容), ...]
    sender_name: 非空时所有节点统一使用该昵称(用于把插件名等放进 content,
    而不是作为转发节点的发送者昵称)
    """
    return [
        {
            "type": "node",
            "data": {
                "name": sender_name if sender_name is not None else name,
                "uin": str(self_id),
                "content": content,
            },
        }
        for name, content in entries
    ]


async def send_forward_msg(
        bot: Bot,
        event: MessageEvent,
        entries: list[tuple[str, str]],
        fallback: str,
        sender_name: str | None = None,
):
    """群聊发送合并转发消息,私聊或发送失败时回退纯文本

    entries: [(发送者名, 消息内容), ...]
    sender_name: 非空时统一用于节点发送者昵称(插件名等应写入 content)
    """
    group_id = getattr(event, "group_id", None)
    if group_id:
        try:
            await bot.send_group_forward_msg(
                group_id=group_id,
                messages=build_forward_nodes(event.self_id, entries, sender_name),
            )
            return
        except Exception as e:
            logger.warning(f"合并转发发送失败,回退文本: {e}")
    await bot.send(event, fallback)


async def common_proc_reply(
    message_id: str | None = None,
    *,
    target: Event | Target | None = None,
    image: str | Path = THINKING_GIF,
    text: str | None = None,
) -> Receipt | None:
    """任务驱动通用回复:引用指定消息并回一张「处理中」占位图(UniMessage 版)

    引用哪条消息 / 发到哪:
    - 事件处理中(有上下文):不传参数即引用当前触发消息,发回当前会话;
    - 传 message_id 则引用指定消息(仍需事件上下文或用 target 指定会话);
    - 后台任务(无事件上下文):必须显式传 target(Event 或 alconna Target),
      如 `Target("872490448")`(群)/`Target("2931589710", private=True)`(私聊);
      此时不传 message_id 即为不带引用的普通发送。

    Args:
        message_id: 要引用的消息 id,默认当前事件消息
        target: 发送目标(Event/Target);仅后台任务需要
        image: 占位图路径(或 http(s) 链接),默认 hexi/resource/imgs/thinking.gif
        text: 附加文本,为空只发图

    Returns:
        alconna Receipt: 可 `.recall()` 撤回占位图、`.msg_ids` 取消息 id
        (是否支持编辑见 `.editable`);发送失败返回 None
    """

    def build() -> UniMessage:
        # 每次重建:alconna 的 reply_to 会把引用段插入消息自身,降级重发需干净副本
        # 本地文件转 base64 内联发送:后端(NapCat)读的是它自己 CWD 下的路径,
        # file:/// 形式不可靠;base64:// 由 bot 直接携带字节,OneBot 实现通用支持
        if isinstance(image, str) and image.startswith(("http://", "https://")):
            img = UniMessage.image(url=image)
        else:
            img = UniMessage.image(
                raw=Path(image).read_bytes(),
                mimetype=mimetypes.guess_type(str(image))[0],
            )
        return UniMessage.text(f"{text} ") + img if text else img

    if message_id:
        reply_to: str | bool = message_id
    elif target is None:
        reply_to = True  # 从当前事件上下文取被引用消息
    else:
        reply_to = False  # 后台任务且未指定消息 id → 普通发送

    try:
        return await build().send(target=target, reply_to=reply_to)
    except Exception as e:
        if not reply_to:
            logger.error(f"占位图发送失败: {e}")
            return None
        logger.warning(f"占位回复引用发送失败,降级为不带引用: {e}")

    try:
        return await build().send(target=target)
    except Exception as e:
        logger.error(f"占位图发送失败: {e}")
        return None