shirobana-bot/qq_channel.py

# qq_channel.py — QQ 通道适配（NapCat / OneBot 11）
# 与 tg_channel.py 同级：本模块只负责「统一 content ↔ OneBot 消息段」的翻译，
# 不含任何业务逻辑；收事件的入站传输层在 utility/bridges/napcat.py（反向 WS :9089）。
# 通道命名: "qq" = NapCat/OneBot11 反向WS
import asyncio
import json
import re
import uuid

from channel import Channel

BOT_QQ = 1295798840

# 发送回执等待上限（秒）：NapCat 需读 OSS 挂载文件 + 上传 QQ，大图可能数十秒
RECEIPT_TIMEOUT = 60
# 发送失败后的重试前等待（秒）：OSS 挂载瞬断（ENOTCONN）等通常数秒内自愈
RETRY_DELAY = 3


class NapcatChannel(Channel):
    name = "qq"

    def __init__(self, admin_qq=None):
        self.ws = None
        self._admin_qq = set(int(x) for x in (admin_qq or []))
        self._pending = {}          # echo -> asyncio.Future（send_msg 的 API 回执）

    @property
    def connected(self):
        return self.ws is not None

    def attach(self, ws):
        # 旧连接残留的回执 Future 不能静默丢弃：置失败让等待方立即返回，
        # 否则会悬挂到 RECEIPT_TIMEOUT 才按"结果未知"放弃（消息静默丢失）
        for fut in self._pending.values():
            if not fut.done():
                fut.set_result({"status": "failed", "message": "NapCat 连接已重建"})
        self._pending.clear()       # 重连后旧回执全部作废
        self.ws = ws

    def detach(self, ws=None):
        # 传入 ws 时只清理"自己这条连接"的状态：重连重叠（新连接已 attach、
        # 旧连接 handler 才退出）时不得误清新连接的 ws 与回执；
        # 无参调用保持旧行为（无条件清理，向后兼容）
        if ws is not None and self.ws is not ws:
            return
        # 断开时把所有等待中的回执置为失败，避免 send() 悬挂到超时
        for fut in self._pending.values():
            if not fut.done():
                fut.set_result({"status": "failed", "message": "NapCat 连接已断开"})
        self._pending.clear()
        self.ws = None

    def resolve_receipt(self, data):
        """供 WS 接收循环调用：若是指令回执（带 echo）则核销并返回 True。"""
        echo = data.get("echo")
        if not echo:
            return False
        fut = self._pending.pop(echo, None)
        if fut is None:
            return False
        if not fut.done():
            fut.set_result(data)
        return True

    def is_admin(self, user_id) -> bool:
        try:
            if int(user_id) in self._admin_qq:
                return True
        except (TypeError, ValueError):
            pass
        return super().is_admin(user_id)

    async def send(self, msg_type, target_id, content):
        """返回 bool: True=NapCat 回执发送成功；False=失败（未连接/重试后仍失败/结果未知）。

        回执语义（OneBot 11 echo 机制）：send_msg 携带 echo，等待 NapCat API 响应，
        status=='failed' 视为失败并自动重试一次；回执超时按"结果未知"处理——
        此时消息可能已送达，为免重复发送不再重试，仅记日志。
        不抛异常（reply 路径依赖现有行为）。
        """
        if not self.ws:
            print("[qq] NapCat 未连接，跳过发送")
            return False
        outcome = await self._attempt_send(msg_type, target_id, content)
        if outcome == "ok":
            return True
        if outcome == "unknown":
            return False
        # 明确失败（NapCat 报告 failed / 发送异常）：等几秒让挂载/网络自愈后重试一次
        await asyncio.sleep(RETRY_DELAY)
        if not self.ws:
            print("[qq] 重试前 NapCat 已断开，放弃")
            return False
        print(f"[qq] {RETRY_DELAY} 秒后重试发送 ({msg_type}/{target_id})...")
        return await self._attempt_send(msg_type, target_id, content) == "ok"

    async def _attempt_send(self, msg_type, target_id, content):
        """单次发送并等待回执。返回 "ok" / "failed" / "unknown"（超时，结果不明）。"""
        echo = uuid.uuid4().hex
        fut = asyncio.get_running_loop().create_future()
        self._pending[echo] = fut
        action = {
            "action": "send_msg",
            "params": {
                "message_type": msg_type,
                "message": content,
            },
            "echo": echo,
        }
        if msg_type == "group":
            action["params"]["group_id"] = target_id
        else:
            action["params"]["user_id"] = target_id
        try:
            await self.ws.send(json.dumps(action))
            resp = await asyncio.wait_for(fut, timeout=RECEIPT_TIMEOUT)
        except asyncio.TimeoutError:
            self._pending.pop(echo, None)
            print(f"[qq] 发送回执超时(>{RECEIPT_TIMEOUT}s, 结果未知不重发): {msg_type}/{target_id}")
            return "unknown"
        except Exception as e:
            self._pending.pop(echo, None)
            print(f"[qq] 发送异常: {type(e).__name__}: {str(e)[:100]}")
            return "failed"
        if resp.get("status") == "failed":
            detail = resp.get("message") or resp.get("wording") or resp
            print(f"[qq] NapCat 报告发送失败: {str(detail)[:120]}")
            return "failed"
        return "ok"


# ---------------- OneBot 11 事件解析（QQ 侧协议） ----------------

def get_plain_text(segments):
    text_parts = []
    for seg in segments:
        if seg.get("type") == "text":
            text_parts.append(seg.get("data", {}).get("text", ""))
    return "".join(text_parts).strip()


def get_image_infos(segments):
    infos = []
    for seg in segments:
        if seg.get("type") == "image" and str(seg.get("data", {}).get("sub_type", "0")) != "1":
            url = seg.get("data", {}).get("url", "")
            if url:
                infos.append({
                    "url": url,
                    "file": seg.get("data", {}).get("file", ""),
                    "file_size": seg.get("data", {}).get("file_size", "0"),
                })
    return infos


def is_at_me(segments, self_id):
    for seg in segments:
        if seg.get("type") == "at" and str(seg.get("data", {}).get("qq", "")) == str(self_id):
            return True
    return False


def is_reply_to_me(segments, self_id):
    for seg in segments:
        if seg.get("type") == "reply" and str(seg.get("data", {}).get("qq", "")) == str(self_id):
            return True
    return False


def re_sub_at(raw_text):
    return re.sub(r'@[0-9]+\s*', '', raw_text).strip()