亿点点魔法让Nonebot2适配OneBot V11的插件运行在RedProtocol
import nonebot
from nonebot import on_command

nonebot.init()

# init_onebot_pretender 必须在 import adapter 之前

from nonebot_adapter_onebot_pretender import (
    init_onebot_pretender,
    create_ob11_adapter_pretender,
)

init_onebot_pretender()

from nonebot.adapters.red import Adapter as RedAdapter
from nonebot.adapters.onebot.v11 import Bot, MessageEvent

driver = nonebot.get_driver()

driver.register_adapter(create_ob11_adapter_pretender(RedAdapter))

nonebot.load_from_toml("pyproject.toml")


@on_command("hello").handle()
async def handle_hello(bot: Bot, event: MessageEvent):
    await bot.send(event, event.message)


if __name__ == "__main__":
    nonebot.run()

bot_example.py
bot-ssttkkl/nonebot-adapter-onebot-pretender
from nonebot import require

from .v11 import create_ob11_adapter_pretender
from .v11.impl import init_ob11_pretender_impl
from .patch_handle_event import init_patch_handle_event

require("nonebot_plugin_datastore")


def init_onebot_pretender():
    init_patch_handle_event()
    init_ob11_pretender_impl()


__all__ = ("init_onebot_pretender", "create_ob11_adapter_pretender")

src/nonebot_adapter_onebot_pretender/__init__.py
bot-ssttkkl/nonebot-adapter-onebot-pretender
from unqlite import UnQLite

from nonebot_adapter_onebot_pretender.store import datastore

db = UnQLite(str(datastore.data_dir / "database.db"))

src/nonebot_adapter_onebot_pretender/data/__init__.py
bot-ssttkkl/nonebot-adapter-onebot-pretender
from asyncio import create_task
from typing import List, Literal, Optional

from nonebot import logger
from pydantic import Field, BaseModel
from nonebot.adapters.onebot.v11 import Message
from nonebot.adapters.onebot.v11.event import Sender
from nonebot.utils import DataclassEncoder, run_sync

from nonebot_adapter_onebot_pretender.data import db


class OB11MsgModel(BaseModel):
    message_id: int
    group: bool = False
    group_id: Optional[int]
    real_id: int = 0
    message_type: Literal["group", "private"]
    sender: Sender = Field(default_factory=Sender)
    time: int = 0
    message: Message = Field(default_factory=Message)
    raw_message: str = ""
    forward: Optional[List[dict]]

    class Config:
        extra = "allow"
        json_encoders = {Message: DataclassEncoder}


@run_sync
def load_ob11_msg(message_id: str) -> Optional[OB11MsgModel]:
    try:
        msg = db[f"ob11_msg_{message_id}"]
        msg = OB11MsgModel.parse_raw(msg, content_type="json")
        return msg
    except KeyError:
        return None


@run_sync
@logger.catch
def _do_save_ob11_msg(message_id: str, message: OB11MsgModel):
    db[f"ob11_msg_{message_id}"] = message.json()
    logger.debug(f"Saved OB11 Msg: {message.dict()}")


def save_ob11_msg(message_id: str, message: OB11MsgModel):
    create_task(_do_save_ob11_msg(message_id, message))

src/nonebot_adapter_onebot_pretender/data/ob11_msg.py
bot-ssttkkl/nonebot-adapter-onebot-pretender
from functools import wraps
from typing import TYPE_CHECKING, List, Callable, Optional, Awaitable

from nonebot import message

if TYPE_CHECKING:
    from nonebot import Bot
    from nonebot.internal.adapter import Event

    T_EventHandler = Callable[[Bot, Event], Awaitable[bool]]

_origin_handle_event: Optional[Callable[["Bot", "Event"], Awaitable[None]]] = None
_event_handler: List["T_EventHandler"] = []


async def origin_handle_event(bot: "Bot", event: "Event") -> None:
    await _origin_handle_event(bot, event)


def patch_event_handle(handler: "T_EventHandler") -> "T_EventHandler":
    _event_handler.append(handler)
    return handler


def init_patch_handle_event():
    global _origin_handle_event
    _origin_handle_event = message.handle_event

    @wraps(_origin_handle_event)
    async def handle_event(bot: "Bot", event: "Event") -> None:
        for handler in _event_handler:
            if await handler(bot, event):
                return
        await _origin_handle_event(bot, event)

    message.handle_event = handle_event


__all__ = ("patch_event_handle", "origin_handle_event", "init_patch_handle_event")

src/nonebot_adapter_onebot_pretender/patch_handle_event.py
bot-ssttkkl/nonebot-adapter-onebot-pretender
from nonebot_plugin_datastore import get_plugin_data

datastore = get_plugin_data("nonebot_adapter_onebot_pretender")

src/nonebot_adapter_onebot_pretender/store.py
bot-ssttkkl/nonebot-adapter-onebot-pretender
from .factory import create_ob11_adapter_pretender  # noqa: F401

src/nonebot_adapter_onebot_pretender/v11/__init__.py
bot-ssttkkl/nonebot-adapter-onebot-pretender
from copy import copy
from abc import ABC, abstractmethod
from typing_extensions import override
from typing import TYPE_CHECKING, Any, Type, Generic, TypeVar, Optional

from nonebot import Driver
from nonebot.adapters import Bot as BaseBot
from nonebot.adapters import Event as BaseEvent
from nonebot.adapters import Adapter as BaseAdapter
from nonebot.adapters.onebot.v11 import Bot as OB11Bot
from nonebot.adapters.onebot.v11 import Adapter as OB11Adapter

from ..patch_handle_event import patch_event_handle, origin_handle_event

if TYPE_CHECKING:
    from .pretender import OB11Pretender

T_ActualAdapter = TypeVar("T_ActualAdapter", bound=BaseAdapter)
T_ActualBot = TypeVar("T_ActualBot", bound=BaseBot)


class Adapter(OB11Adapter, Generic[T_ActualAdapter, T_ActualBot], ABC):
    @override
    def __init__(self, driver: Driver, **kwargs: Any):
        super().__init__(driver, **kwargs)
        self.pretender = self.get_pretender_type()(self)
        self.actual_adapter = self._create_actual_adapter_with_hack(driver, **kwargs)

    def _setup(self) -> None:
        @patch_event_handle
        async def handle_event(bot: BaseBot, event: BaseEvent) -> bool:
            if bot.self_id in self.actual_adapter.bots:
                handled_event = await self.pretender.handle_event(bot, event)
                if handled_event is None:
                    self.pretender.log(
                        "WARNING",
                        f"No event handler for {type(event).__name__} "
                        f"({self.actual_adapter.get_name()}) found, "
                        f"event was ignored",
                    )
                else:
                    await origin_handle_event(self.bots[bot.self_id], handled_event)
                return True
            else:
                return False

    @classmethod
    @abstractmethod
    def get_pretender_type(cls) -> Type["OB11Pretender[T_ActualAdapter, T_ActualBot]"]:
        ...

    def get_actual_bot(self, bot: OB11Bot) -> Optional[T_ActualBot]:
        return self.actual_adapter.bots.get(bot.self_id)

    def _create_actual_adapter_with_hack(
        self, driver: Driver, **kwargs: Any
    ) -> T_ActualAdapter:
        pretender_adapter = self

        hacky_driver = copy(driver)

        def bot_connect(bot: T_ActualBot) -> None:
            ob11bot = OB11Bot(pretender_adapter, bot.self_id)
            pretender_adapter.bot_connect(ob11bot)

        hacky_driver._bot_connect = bot_connect

        def bot_disconnect(bot: T_ActualBot) -> None:
            pretender_bot = pretender_adapter.bots.get(bot.self_id)
            assert (
                pretender_bot is not None
            ), f"cannot find pretender bot {bot.self_id} ({bot.type})"
            pretender_adapter.bot_disconnect(pretender_bot)

        hacky_driver._bot_disconnect = bot_disconnect

        return self.pretender.get_actual_adapter_type()(hacky_driver, **kwargs)

    @override
    async def _call_api(self, bot: OB11Bot, api: str, **data: Any) -> Any:
        return await self.pretender.handle_api_call(bot, api, **data)

src/nonebot_adapter_onebot_pretender/v11/adapter.py
bot-ssttkkl/nonebot-adapter-onebot-pretender
from typing import Any, Dict, Type, Callable

from nonebot import Driver
from nonebot import Adapter as BaseAdapter
from nonebot.adapters.onebot.v11 import Adapter as OB11Adapter

from .pretender import OB11Pretender
from .adapter import Adapter as OB11PretenderAdapter

_pretenders: Dict[str, Type[OB11Pretender]] = {}


def register_ob11_pretender(
    adapter: Type[BaseAdapter],
) -> Callable[[Type[OB11Pretender]], Type[OB11Pretender]]:
    def decorator(pretender: Type[OB11Pretender]) -> Type[OB11Pretender]:
        _pretenders[adapter.get_name()] = pretender

        return pretender

    return decorator


def create_ob11_adapter_pretender(
    *adapter: Type[BaseAdapter],
) -> Type[OB11PretenderAdapter]:
    pretender_adapters = []

    for actual_adapter in adapter:
        pretender = _pretenders.get(actual_adapter.get_name())
        if pretender is None:
            raise RuntimeError(f"未找到 {actual_adapter.get_name()} 的 Pretender 实现")

        class Adapter(OB11PretenderAdapter):
            @classmethod
            def get_pretender_type(cls):
                return pretender

        Adapter.__module__ = OB11Adapter.__module__
        Adapter.__qualname__ = OB11Adapter.__qualname__

        pretender_adapters.append(Adapter)

    class Adapter(OB11Adapter):
        def __init__(self, driver: Driver, **kwargs: Any):
            self.pretender_adapters = []
            for adp in pretender_adapters:
                self.pretender_adapters.append(adp(driver, **kwargs))

    Adapter.__module__ = OB11Adapter.__module__
    Adapter.__qualname__ = OB11Adapter.__qualname__

    return Adapter


__all__ = ("create_ob11_adapter_pretender",)

src/nonebot_adapter_onebot_pretender/v11/factory.py
bot-ssttkkl/nonebot-adapter-onebot-pretender
from nonebot import logger


def init_ob11_pretender_impl():
    try:
        from . import red  # noqa: F401
    except ImportError as e:
        logger.opt(exception=e).debug("Failed to Import OB11 Red Pretender")

src/nonebot_adapter_onebot_pretender/v11/impl/__init__.py
bot-ssttkkl/nonebot-adapter-onebot-pretender
import json
from base64 import b64decode
from datetime import datetime
from pathlib import Path
from typing import Dict, List, Type, Union, Optional
from urllib.parse import urlencode, urlunsplit

from nonebot.adapters.onebot.v11 import ActionFailed
from nonebot.adapters.onebot.v11 import Message as OB11Msg
from nonebot.adapters.onebot.v11 import MessageSegment as OB11MS
from nonebot.adapters.onebot.v11 import event as ob11_event
from nonebot.adapters.onebot.v11.event import Reply, Sender
from nonebot.adapters.red import Adapter as RedAdapter
from nonebot.adapters.red import Bot as RedBot
from nonebot.adapters.red import Message as RedMsg
from nonebot.adapters.red import MessageSegment as RedMS
from nonebot.adapters.red import event as red_event
from nonebot.adapters.red.api.model import ChatType
from nonebot.adapters.red.message import ForwardNode, MediaMessageSegment
from nonebot.utils import DataclassEncoder, logger_wrapper

from ...factory import register_ob11_pretender
from ...pretender import OB11Pretender, event_handler, api_call_handler
from ....data.ob11_msg import OB11MsgModel, load_ob11_msg, save_ob11_msg
from ....webapi import red  # noqa: F401

log = logger_wrapper("OneBot V11 Pretender (RedProtocol)")


@register_ob11_pretender(RedAdapter)
class RedOB11Pretender(OB11Pretender[RedAdapter, RedBot, red_event.Event]):
    @classmethod
    def get_actual_adapter_type(cls) -> Type[RedAdapter]:
        return RedAdapter

    @classmethod
    def log(cls, level: str, content: str, exc: Optional[BaseException] = None):
        log(level, content, exc)

    def construct_media_url(self, bot_self_id: str, seg: MediaMessageSegment) -> str:
        parameters = {
            "botId": bot_self_id,
            "msgId": seg.data["_msg_id"],
            "chatType": seg.data["_chat_type"],
            "target": seg.data["_peer_uin"],
            "elementId": seg.data["id"],
            "thumbSize": 0,
            "downloadType": 2,
        }
        query = urlencode(query=parameters, doseq=True)
        url = urlunsplit(
            (
                "http",
                f"localhost:{self.adapter.driver.config.port}",
                "/ob_pretender/red/media",
                query,
                "",
            )
        )
        return url

    def convert_outgoing_msg(self, outgoing: OB11Msg) -> RedMsg:
        msg = RedMsg()
        for m in outgoing:
            if m.type == "text":
                msg.append(RedMS.text(m.data["text"]))
            elif m.type == "at":
                if m.data["qq"] == "all":
                    msg.append(RedMS.at_all())
                else:
                    msg.append(RedMS.at(m.data["qq"]))
            elif m.type == "face":
                msg.append(RedMS.face(m.data["id"]))
            elif m.type == "image":
                file = m.data["file"]
                if isinstance(file, str):
                    if file.startswith("base64://"):
                        file = b64decode(file.removeprefix("base64://"))
                    else:
                        file = Path(file)
                msg.append(RedMS.image(file))
            elif m.type == "video":
                file = m.data["file"]
                if isinstance(file, str):
                    file = Path(file)
                msg.append(RedMS.video(file))
            elif m.type == "record":
                file = m.data["file"]
                if isinstance(file, str):
                    file = Path(file)
                msg.append(RedMS.voice(file))
            elif m.type == "reply":
                msg.append(
                    RedMS.reply(
                        m.data.get("seq"),
                        m.data.get("id"),
                        m.data.get("qq"),
                    )
                )
            else:
                log("WARNING", f"暂不支持 {m.type} 类型消息转换 (OB11 -> Red)")
        return msg

    def convert_incoming_msg(self, bot: RedBot, incoming: RedMsg) -> OB11Msg:
        msg = OB11Msg()
        for m in incoming:
            if m.type == "text":
                msg.append(OB11MS.text(m.data["text"]))
            elif m.type == "at":
                msg.append(
                    OB11MS(
                        "at",
                        {"qq": str(m.data["user_id"]), "name": m.data.get("user_name")},
                    )
                )
            elif m.type == "at_all":
                msg.append(OB11MS.at("all"))
            elif m.type == "face":
                msg.append(OB11MS("face", {"id": str(m.data["face_id"])}))
            # elif m.type == "max_member_countrket_face":
            #     msg.append(
            #         OB11MS(
            #             "image",
            #             {
            #                 "file": "file://" + m.data["static_path"],
            #                 "url": "file://" + m.data["static_path"],
            #             },
            #         )
            #     )
            elif m.type == "image" and isinstance(m, MediaMessageSegment):
                msg.append(
                    OB11MS(
                        "image",
                        {
                            "file": "file://" + m.data["path"],
                            "url": self.construct_media_url(bot.self_id, m),
                        },
                    )
                )
            elif m.type == "video" and isinstance(m, MediaMessageSegment):
                msg.append(
                    OB11MS(
                        "video",
                        {
                            "file": "file://" + m.data["path"],
                            "url": self.construct_media_url(bot.self_id, m),
                            "cover": "file://" + m.data["thumb_path"],
                        },
                    )
                )
            elif m.type == "voice" and isinstance(m, MediaMessageSegment):
                msg.append(
                    OB11MS(
                        "record",
                        {
                            "file": "file://" + m.data["path"],
                            "url": self.construct_media_url(bot.self_id, m),
                        },
                    )
                )
            elif m.type == "reply":
                msg.append(
                    OB11MS(
                        "reply",
                        {
                            "id": str(m.data["msg_id"]),
                        },
                    )
                )
            else:
                log("WARNING", f"暂不支持 {m.type} 类型消息转换 (Red -> OB11)")
        return msg

    @api_call_handler()
    async def send_msg(
        self,
        bot: RedBot,
        *,
        message_type: Optional[str] = None,
        user_id: Optional[int] = None,
        group_id: Optional[int] = None,
        message: Union[str, OB11Msg],
        **data: Dict,
    ) -> Dict:
        if message_type == "private":
            return await self.send_private_msg(bot, user_id=user_id, message=message)
        elif message_type == "group":
            return await self.send_group_msg(bot, group_id=group_id, message=message)
        elif user_id:
            return await self.send_private_msg(bot, user_id=user_id, message=message)
        elif group_id:
            return await self.send_group_msg(bot, group_id=group_id, message=message)
        else:
            raise ValueError("请传入正确的参数")

    @api_call_handler()
    async def send_group_msg(
        self, bot: RedBot, *, group_id: int, message: Union[str, OB11Msg], **data: Dict
    ) -> Dict:
        if isinstance(message, str):
            message = OB11Msg(message)
        ob11_msg = message

        message = self.convert_outgoing_msg(message)

        res = await bot.send_group_message(group_id, message)
        save_ob11_msg(
            res.msgId,
            OB11MsgModel(
                group=True,
                group_id=group_id,
                message_id=int(res.msgId),
                real_id=int(res.msgId),
                message_type="group",
                sender=Sender(
                    nickname=res.sendNickName or res.sendMemberName,
                    user_id=int(bot.self_id),
                ),
                time=int(res.msgTime or "0"),
                message=ob11_msg,
                raw_message=ob11_msg.extract_plain_text(),
            ),
        )
        return {"message_id": int(res.msgId)}

    @api_call_handler()
    async def send_private_msg(
        self, bot: RedBot, *, user_id: int, message: Union[str, OB11Msg], **data: Dict
    ) -> Dict:
        if isinstance(message, str):
            message = OB11Msg(message)
        ob11_msg = message

        message = self.convert_outgoing_msg(message)

        res = await bot.send_friend_message(user_id, message)
        save_ob11_msg(
            res.msgId,
            OB11MsgModel(
                group=False,
                group_id=None,
                message_id=int(res.msgId),
                real_id=int(res.msgId),
                message_type="private",
                sender=Sender(
                    nickname=res.sendNickName or res.sendMemberName,
                    user_id=int(bot.self_id),
                ),
                time=int(res.msgTime or "0"),
                message=ob11_msg,
                raw_message=ob11_msg.extract_plain_text(),
            ),
        )
        return {"message_id": int(res.msgId)}

    async def _send_forward_msg(
        self,
        bot: RedBot,
        *,
        chat_type: ChatType,
        target: int,
        messages: Union[List[Dict], List[OB11MS], OB11Msg],
    ) -> Dict:
        messages = [
            json.loads(json.dumps(seg, cls=DataclassEncoder))
            if isinstance(seg, OB11MS)
            else seg
            for seg in messages
        ]  # 将Message转换为List

        nodes = []

        for seg in messages:
            if seg.get("type") != "node":
                continue
            data = seg.get("data") or {}

            if "id" in data:
                ob11_msg = await load_ob11_msg(data["id"])
                if ob11_msg is None:
                    continue

                nodes.append(
                    ForwardNode(
                        uin=str(ob11_msg.sender.user_id),
                        name=ob11_msg.sender.nickname,
                        group=ob11_msg.group_id,
                        message=self.convert_outgoing_msg(ob11_msg.message),
                        time=ob11_msg.time,
                    )
                )
            elif "uin" in data or "user_id" in data:
                nodes.append(
                    ForwardNode(
                        uin=str(data.get("uin") or data.get("user_id")),
                        name=data.get("name")
                             or data.get("nickname")
                             or str(data.get("uin") or data.get("user_id")),
                        group=0,
                        message=self.convert_outgoing_msg(
                            OB11Msg(OB11MS(**raw_seg) for raw_seg in data["content"])
                        ),
                    )
                )

        await bot.send_fake_forward(nodes, chat_type, target)
        return {
            "message_id": 0,
            "forward_id": "",
        }

    @api_call_handler()
    async def send_group_forward_msg(
        self, bot: RedBot, *, group_id: int, messages: List[Dict], **data: Dict
    ) -> Dict:
        return await self._send_forward_msg(
            bot, chat_type=ChatType.GROUP, target=group_id, messages=messages
        )

    @api_call_handler()
    async def send_private_forward_msg(
        self, bot: RedBot, *, user_id: int, messages: List[Dict], **data: Dict
    ) -> Dict:
        return await self._send_forward_msg(
            bot, chat_type=ChatType.FRIEND, target=user_id, messages=messages
        )

    @api_call_handler()
    async def get_login_info(self, bot: RedBot, **data: Dict) -> Dict:
        profile = await bot.get_self_profile()
        return {
            "user_id": int(profile.uin or profile.uid or profile.qid),
            "nickname": profile.longNick or profile.nick,
        }

    @api_call_handler()
    async def get_friend_list(self, bot: RedBot, **data: Dict) -> List:
        friends = await bot.get_friends()
        return [
            {
                "user_id": int(profile.uin or profile.uid or profile.qid),
                "nickname": profile.nick,
                "remark": profile.remark,
            }
            for profile in friends
        ]

    @api_call_handler()
    async def get_group_list(self, bot: RedBot, **data: Dict) -> List:
        groups = await bot.get_groups()
        return [
            {
                "group_id": int(group.groupCode),
                "group_name": group.groupName,
                "group_memo": group.remarkName,
                "group_create_time": 0,
                "group_level": 0,
                "member_count": group.memberCount,
                "max_member_count": group.maxMember,
            }
            for group in groups
        ]

    @api_call_handler()
    async def get_group_member_list(
        self, bot: RedBot, *, group_id: int, **data: Dict
    ) -> List:
        members = await bot.get_members(group_id, 2 ** 16 - 1)
        return [
            {
                "group_id": group_id,
                "user_id": int(member.uin or member.uid or member.qid),
                "nickname": member.nick,
                "card": member.cardName,
                "sex": "unknown",
                "age": 0,
                "area": "",
                "join_time": 0,
                "last_sent_time": 0,
                "level": 0,
                "role": "member",  # TODO
                "unfriendly": False,
                "title": "",
                "title_expire_time": 0,
                "card_changeable": False,
                "shut_up_timestamp": member.shutUpTime,
            }
            for member in members
        ]

    @api_call_handler()
    async def delete_msg(self, bot: RedBot, *, message_id: int, **data: Dict) -> None:
        msg = await load_ob11_msg(message_id)
        if msg is None:
            raise ActionFailed(msg="消息不存在")

        if msg.group:
            await bot.recall_group_message(msg.group_id, str(message_id))
        else:
            await bot.recall_friend_message(msg.sender.user_id, str(message_id))

    @api_call_handler()
    async def get_msg(self, bot: RedBot, *, message_id: int, **data: Dict) -> Dict:
        msg = await load_ob11_msg(str(message_id))
        if msg is not None:
            return json.loads(msg.json(exclude_none=True))
        else:
            raise ActionFailed(msg="消息不存在")

    @api_call_handler()
    async def set_group_ban(
        self,
        bot: RedBot,
        *,
        group_id: int,
        user_id: int,
        duration: int = 30 * 60,
        **data: Dict,
    ) -> None:
        if duration > 0:
            await bot.mute_member(group_id, user_id, duration=duration)
        else:
            await bot.unmute_member(group_id, user_id)

    @api_call_handler()
    async def set_group_whole_ban(
        self,
        bot: RedBot,
        *,
        group_id: int,
        enable: bool = True,
        **data: Dict,
    ) -> None:
        if enable:
            await bot.mute_everyone(group_id)
        else:
            await bot.unmute_everyone(group_id)

    @event_handler(red_event.GroupMessageEvent)
    async def handle_group_message_event(
        self, bot: RedBot, event: red_event.GroupMessageEvent
    ) -> ob11_event.GroupMessageEvent:
        msg = self.convert_incoming_msg(bot, event.message)
        ori_msg = self.convert_incoming_msg(bot, event.original_message)

        reply = None
        if event.reply and event.reply.replayMsgId:
            reply = await self.get_msg(bot, int(event.reply.replayMsgId))
            if reply is not None:
                reply = Reply.parse_obj(reply)

        save_ob11_msg(
            event.msgId,
            OB11MsgModel(
                group=True,
                group_id=int(event.peerUin or "0"),
                message_id=int(event.msgId),
                real_id=int(event.msgId),
                message_type="group",
                sender=Sender(
                    nickname=event.sendNickName or event.sendMemberName,
                    user_id=int(event.senderUin or "0"),
                ),
                time=int(event.msgTime or "0"),
                message=msg,
                raw_message=ori_msg.extract_plain_text(),
            ),
        )

        return ob11_event.GroupMessageEvent(
            time=int(event.msgTime or "0"),
            self_id=int(bot.self_id or "0"),
            post_type="message",
            sub_type="normal",
            user_id=int(event.senderUin or "0"),
            message_id=int(event.msgId or "0"),
            message=msg,
            original_message=ori_msg,
            raw_message=ori_msg.extract_plain_text(),
            font=0,
            sender=Sender(
                user_id=int(event.senderUin or "0"),
                nickname=event.sendNickName or event.sendMemberName,
                sex="unknown",
                age=0,
                card=event.sendMemberName,
                role="member",  # todo
                title="",
            ),
            message_type="group",
            group_id=int(event.peerUin),
            to_me=event.to_me,
            reply=reply,
            anonymous=None,
        )

    @event_handler(red_event.PrivateMessageEvent)
    async def handle_private_message_event(
        self, bot: RedBot, event: red_event.PrivateMessageEvent
    ) -> ob11_event.PrivateMessageEvent:
        msg = self.convert_incoming_msg(bot, event.message)
        ori_msg = self.convert_incoming_msg(bot, event.original_message)

        reply = None
        if event.reply and event.reply.replayMsgId:
            reply = Reply.parse_obj(
                await self.get_msg(bot, int(event.reply.replayMsgId))
            )

        save_ob11_msg(
            event.msgId,
            OB11MsgModel(
                group=False,
                message_id=int(event.msgId),
                real_id=int(event.msgId),
                message_type="private",
                sender=Sender(
                    nickname=event.sendNickName or event.sendMemberName,
                    user_id=int(event.senderUin or "0"),
                ),
                time=int(event.msgTime or "0"),
                message=msg,
                raw_message=ori_msg.extract_plain_text(),
            ),
        )

        return ob11_event.PrivateMessageEvent(
            time=int(event.msgTime or "0"),
            self_id=int(bot.self_id or "0"),
            post_type="message",
            sub_type="friend",
            user_id=int(event.senderUin or "0"),
            message_id=int(event.msgId or "0"),
            message=msg,
            original_message=ori_msg,
            raw_message=ori_msg.extract_plain_text(),
            font=0,
            sender=Sender(
                user_id=int(event.senderUin or "0"),
                nickname=event.sendNickName or event.sendMemberName,
                sex="unknown",
                age=0,
            ),
            message_type="private",
            to_me=event.to_me,
            reply=reply,
        )

    @event_handler(red_event.MemberAddEvent)
    async def handle_member_add_event(
        self, bot: RedBot, event: red_event.MemberAddEvent
    ) -> ob11_event.GroupIncreaseNoticeEvent:
        return ob11_event.GroupIncreaseNoticeEvent(
            time=int(event.msgTime or "0"),
            self_id=int(bot.self_id or "0"),
            post_type="notice",
            notice_type="group_increase",
            sub_type="approve",
            user_id=int(event.memberUid or "0"),
            group_id=int(event.peerUin or event.peerUid or "0"),
            operator_id=int(event.operatorUid or "0"),
        )

    @event_handler(red_event.MemberMutedEvent)
    async def handle_member_muted_event(
        self, bot: RedBot, event: red_event.MemberMutedEvent
    ) -> ob11_event.GroupBanNoticeEvent:
        return ob11_event.GroupBanNoticeEvent(
            time=int(datetime.now().timestamp()),
            self_id=int(bot.self_id or "0"),
            post_type="notice",
            notice_type="group_ban",
            sub_type="ban",
            user_id=int(event.member.uin or event.member.uid or "0"),
            group_id=int(event.peerUin or event.peerUid or "0"),
            operator_id=int(event.operator.uin or event.operator.uid or "0"),
            duration=int(event.duration.total_seconds()),
        )

    @event_handler(red_event.MemberUnmuteEvent)
    async def handle_member_unmuted_event(
        self, bot: RedBot, event: red_event.MemberUnmuteEvent
    ) -> ob11_event.GroupBanNoticeEvent:
        return ob11_event.GroupBanNoticeEvent(
            time=int(datetime.now().timestamp()),
            self_id=int(bot.self_id or "0"),
            post_type="notice",
            notice_type="group_ban",
            sub_type="lift_ban",
            user_id=int(event.member.uin or event.member.uid or "0"),
            group_id=int(event.peerUin or event.peerUid or "0"),
            operator_id=int(event.operator.uin or event.operator.uid or "0"),
            duration=int(event.duration.total_seconds()),
        )

src/nonebot_adapter_onebot_pretender/v11/impl/red/__init__.py
bot-ssttkkl/nonebot-adapter-onebot-pretender
from abc import ABCMeta, abstractmethod
from typing import (
    TYPE_CHECKING,
    Any,
    Dict,
    Type,
    Tuple,
    Generic,
    TypeVar,
    Callable,
    Optional,
    ParamSpec,
)

from nonebot import logger
from nonebot.adapters import Bot as BaseBot
from nonebot.adapters import Event as BaseEvent
from nonebot.adapters import Adapter as BaseAdapter
from nonebot.adapters.onebot.v11 import Bot as OB11Bot
from nonebot.adapters.onebot.v11 import ApiNotAvailable
from nonebot.adapters.onebot.v11 import Event as OB11Event

if TYPE_CHECKING:
    from .adapter import Adapter as OB11PretenderAdapter

T = TypeVar("T")
P = ParamSpec("P")

T_ActualAdapter = TypeVar("T_ActualAdapter", bound=BaseAdapter)
T_ActualBot = TypeVar("T_ActualBot", bound=BaseBot)
T_ActualEvent = TypeVar("T_ActualEvent", bound=BaseEvent)

T_ApiHandler = Callable[["OB11Pretender", T_ActualBot, ...], Any]
T_EventHandler = Callable[["OB11Pretender", T_ActualBot, T_ActualEvent], OB11Event]


class OB11PretenderMeta(ABCMeta):
    def __new__(
        mcls, name: str, base: Tuple[type, ...], namespace: dict, *args, **kwargs
    ):
        api_call_handlers = {}

        for item in namespace.values():
            if getattr(item, "__api_call_handler__", False):
                api_call_handlers[item.__api_call_name__] = item

        namespace["_api_call_handler_mapping"] = api_call_handlers

        event_handlers = {}

        for item in namespace.values():
            if getattr(item, "__event_handler__", False):
                event_handlers[item.__event_type__] = item

        namespace["_event_handler_mapping"] = event_handlers

        return super().__new__(mcls, name, base, namespace, **kwargs)


class OB11Pretender(
    Generic[T_ActualAdapter, T_ActualBot, T_ActualEvent], metaclass=OB11PretenderMeta
):
    _api_call_handler_mapping: Dict[str, T_ApiHandler]
    _event_handler_mapping: Dict[Type[BaseEvent], T_EventHandler]

    def __init__(self, adapter: "OB11PretenderAdapter[T_ActualAdapter, T_ActualBot]"):
        self.adapter = adapter

    @classmethod
    def log(cls, level: str, content: str, exc: Optional[BaseException] = None):
        logger.opt(exception=exc).log(level, content)

    @classmethod
    @abstractmethod
    def get_actual_adapter_type(cls) -> Type[T_ActualAdapter]:
        ...

    async def handle_api_call(self, bot: OB11Bot, api: str, **data: Any) -> Any:
        handler = self._api_call_handler_mapping.get(api)
        if handler is None:
            raise ApiNotAvailable
        actual_bot = self.adapter.get_actual_bot(bot)
        return await handler(self, actual_bot, **data)

    async def handle_event(
        self, bot: T_ActualBot, event: T_ActualEvent
    ) -> Optional[OB11Event]:
        self.log(
            "DEBUG",
            f"Receive {self.adapter.actual_adapter.get_name()}"
            f" {type(event).__name__}: " + str(event),
        )
        self.log(
            "TRACE",
            f"{self.adapter.actual_adapter.get_name()}"
            f" {type(event).__name__}: " + str(event.json()),
        )

        handler = None
        for t_event in type(event).mro():
            handler = self._event_handler_mapping.get(t_event)
            if handler is not None:
                break

        if handler is None:
            return None

        return await handler(self, bot, event)


def api_call_handler(api: Optional[str] = None):
    def decorator(func: T_ApiHandler) -> T_ApiHandler:
        func.__api_call_handler__ = True
        func.__api_call_name__ = api or func.__name__

        return func

    return decorator


def event_handler(event_type: Type[BaseEvent]):
    def decorator(func: T_EventHandler) -> T_EventHandler:
        func.__event_handler__ = True
        func.__event_type__ = event_type

        return func

    return decorator

src/nonebot_adapter_onebot_pretender/v11/pretender.py
bot-ssttkkl/nonebot-adapter-onebot-pretender
from nonebot import get_driver
from nonebot.internal.driver import ReverseDriver

if not isinstance(get_driver(), ReverseDriver):
    raise RuntimeError("需要使用反向驱动器")

src/nonebot_adapter_onebot_pretender/webapi/__init__.py
bot-ssttkkl/nonebot-adapter-onebot-pretender
from typing import cast

from fastapi import Response
from starlette import status
from nonebot import get_app, get_bot
from starlette.responses import StreamingResponse
from nonebot.adapters.red.api.model import ChatType
from nonebot.adapters.red import Adapter as RedAdapter

from ..v11.adapter import Adapter as OB11PretenderAdapter

app = get_app()


@app.get("/ob_pretender/red/media")
async def get_red_media(
    response: Response,
    botId: str,
    msgId: str,
    chatType: ChatType,
    target: str,
    elementId: str,
    thumbSize: int = 0,
    downloadType: int = 2,
):
    bot = get_bot(botId)
    if bot is None or not isinstance(bot.adapter, OB11PretenderAdapter):
        response.status_code = status.HTTP_404_NOT_FOUND
        return

    bot = cast(OB11PretenderAdapter, bot.adapter).get_actual_bot(bot)
    if bot.type != RedAdapter.get_name():
        response.status_code = status.HTTP_404_NOT_FOUND
        return

    async def fetch():
        yield await bot.call_api(
            "fetch_media",
            msg_id=msgId,
            chat_type=chatType,
            target=target,
            element_id=elementId,
            thumb_size=thumbSize,
            download_type=downloadType,
        )

    return StreamingResponse(fetch())

src/nonebot_adapter_onebot_pretender/webapi/red.py
bot-ssttkkl/nonebot-adapter-onebot-pretender