shirobana-bot/channel.py

# channel.py — 统一消息通道抽象
# 各业务模块不再直接构造 OneBot action，而是通过 Channel.send() 发送。
# 通道命名: "qq" = NapCat/OneBot11 反向WS, "tg" = Telegram Bot
import contextlib
import json
import os

import identity

# 每个入口进程在自己的目录运行（utility/ 或 chat/），
# settings.json 放在各自入口目录下，按 CWD 读取。
def load_settings() -> dict:
    path = os.path.join(os.getcwd(), "settings.json")
    try:
        with open(path, "r", encoding="utf-8") as f:
            data = json.load(f)
    except (OSError, json.JSONDecodeError):
        data = {}
    return data if isinstance(data, dict) else {}


def admin_list(channel_name: str) -> list:
    """取某通道的管理员 id 列表（settings.json -> admin -> <channel>）"""
    s = load_settings().get("admin", {})
    lst = s.get(channel_name, [])
    return [int(x) for x in lst] if isinstance(lst, list) else []


class Channel:
    """消息通道基类。
    content 约定: str 纯文本, 或 list[dict] 段数组（OneBot 段格式:
    {"type":"text","data":{"text":..}} / {"type":"image","data":{"file":..}}）
    """
    name = "base"

    def is_admin(self, user_id) -> bool:
        try:
            if int(user_id) in admin_list(self.name):
                return True
        except (TypeError, ValueError):
            pass
        # 统一身份：管理员判定以身份库为准（QQ/TG 两侧管理员合并到同一 uid）
        return identity.is_admin(user_id)

    async def send(self, msg_type, target_id, content):
        raise NotImplementedError
    # msg_type: "group" | "private"

    def typing(self, target_id):
        """处理消息期间的「正在输入」提示（异步上下文管理器，默认 no-op）。
        用法: async with channel.typing(user_id): await 耗时操作()
        """
        @contextlib.asynccontextmanager
        async def _noop():
            yield
        return _noop()


CHANNELS = {}  # channel name -> Channel 实例


def register_channel(ch: Channel):
    CHANNELS[ch.name] = ch


def unregister_channel(name: str):
    CHANNELS.pop(name, None)


def get_channel(name: str):
    return CHANNELS.get(name)


def all_channels():
    return list(CHANNELS.values())


async def _mirror_private(channel, target_id, content):
    """私聊回复镜像：按身份推送开关，把消息复制到该身份的其它平台私聊。

    - 仅私聊生效；群聊、含 file 段的内容不镜像；
    - 镜像发送直接走目标通道的 send()（不经过 send_message），不会递归；
    - 开关关闭的一侧由 identity.push_targets() 直接剔除。
    """
    try:
        if isinstance(content, list) and any(
                isinstance(seg, dict) and seg.get("type") == "file" for seg in content):
            print(f"[channel] 镜像跳过（含 file 段）: {getattr(channel, 'name', '?')}/{target_id}")
            return
        uid = identity.resolve(channel.name, target_id)
        if not uid:
            return
        targets = [t for t in identity.push_targets(uid)
                   if not (t[0] == channel.name and str(t[1]) == str(target_id))]
        for name, tid in targets:
            ch = CHANNELS.get(name)
            if ch is None:
                continue
            try:
                ok = await ch.send("private", tid, content)
                print(f"[channel] 镜像 uid={uid} -> {name}/{tid} "
                      f"{'成功' if ok is not False else '失败'}")
            except Exception as e:
                print(f"[channel] 镜像 uid={uid} -> {name}/{tid} 失败: {e}")
    except Exception as e:
        print(f"[channel] 镜像异常: {e}")


async def send_message(channel, msg_type, target_id, content, mirror=True):
    """统一发送入口（与旧 main.send_message 签名兼容: 第一个参数是通道对象）

    返回 bool: 是否确实投递。
      - 通道缺失 / send() 抛异常      -> False
      - send() 显式返回 False          -> False（通道自我报告投递失败）
      - send() 返回 None / True        -> True（兼容旧通道实现，语义不变）
    mirror: 私聊回复是否按身份推送开关镜像到其它平台（主动扇出 push_to_user 传 False 防重复）
    """
    if not channel:
        print("[channel] 通道未连接，跳过发送")
        return False
    try:
        result = await channel.send(msg_type, target_id, content)
    except Exception as e:
        print(f"[channel:{getattr(channel, 'name', '?')}] 发送失败: {e}")
        return False
    if result is False:
        print(f"[channel:{getattr(channel, 'name', '?')}] 通道报告投递失败")
        return False
    if mirror and msg_type == "private":
        await _mirror_private(channel, target_id, content)
    return True


async def push_to_user(uid, content, tag="推送", targets=None):
    """按统一身份扇出「个人私聊推送」：TG 私聊 + 每个已绑定 QQ 私聊（QQ 镜像）。

    参数:
      uid      统一身份 id
      content  消息内容（str 或 OneBot 段数组）
      tag      日志/审计标签（仅用于日志）
      targets  可选：限定本次扇出的目标集合 [(channel, target_id), ...]。
               持久化队列重试时只补投未成功的目标，避免已成功目标被重复推送。
    返回 {(channel_name, target_id): bool}；通道未注册/发送失败一律记 False，不抛异常。
    """
    if targets is None:
        targets = identity.push_targets(uid)
    else:
        targets = list(targets)
    results = {}
    if not targets:
        print(f"[channel] 身份 {uid} 无可用的推送目标，跳过{tag}")
        return results
    for name, target_id in targets:
        ch = CHANNELS.get(name)
        if ch is None:
            print(f"[channel] {tag} uid={uid} 目标 {name}/{target_id} 失败：通道未注册")
            results[(name, target_id)] = False
            continue
        try:
            ok = await send_message(ch, "private", target_id, content, mirror=False)
        except Exception as e:      # send_message 已兜底；此处仅防御
            print(f"[channel] {tag} uid={uid} 目标 {name}/{target_id} 失败: {e}")
            ok = False
        ok = True if ok is None else bool(ok)
        results[(name, target_id)] = ok
        print(f"[channel] {tag} uid={uid} -> {name}/{target_id} {'成功' if ok else '失败'}")
    return results