能否不真正登录QQ使用？

[clacknya]

调试插件的时候不想反复重启服务器的框架影响 BOT 使用，能否在本地就能模拟 QQ 行为来调试？

[Ink-33]

？

[clacknya]

就是提供一个虚拟的 QQ 环境，可以模拟收发消息

[Ink-33]

那直接写一个简单的ob服务端实现当压测就可以了，自定义一些消息

[clacknya]

有无样例提供参考？

[synodriver]

# -*- coding: utf-8 -*-
"""
Copyright (c) 2008-2021 synodriver <synodriver@gmail.com>
"""
import asyncio
import time

import aiohttp


def get_post_data() -> dict:
    test_groupmsg_data = {
        "time": int(time.time()),
        "self_id": 114514,
        "post_type": "message",
        "message_type": "group",
        "sub_type": "normal",
        "message_id": 1919810,
        "group_id": 1414114,
        "user_id": 341919,
        "anonymous": None,
        "message": "哼哼 嗯啊 啊啊啊啊",
        "raw_message": "哼哼 嗯啊 啊啊啊啊",
        "font": 0,
        "sender":
            {
                "user_id": 341919,
                "nickname": "rc",
                "card": "rc",
                "sex": "female",
                "age": 24,
                "area": "下北泽",
                "level": "是学生",
                "role": "owner",
                "title": "global rc"
            }
    }
    return test_groupmsg_data


headers = {
    "Host": "127.0.0.1:8080",
    "Connection": "Upgrade",
    "Upgrade": "websocket",
    "X-Self-ID": "114514",
    "X-Client-Role": "Universal"
}

enable = {"time": time.time(),
          "self_id": 114514,
          "post_type": "meta_event",
          "meta_event_type": " lifecycle",
          "sub_type": "enable"
          }


async def send(ws):
    while True:
        await ws.send_json(get_post_data())
        print("send")

    # print(await ws.receive_json())


async def receive(ws: aiohttp.ClientWebSocketResponse):
    while not ws.closed:
        await ws.receive_json()


async def send_group_msg(session: aiohttp.ClientSession):
    async with session.ws_connect("http://127.0.0.1:8080/cqhttp/ws", headers=headers) as ws:
        await ws.send_json(enable)
        start = time.time()
        asyncio.create_task(receive())
        tasks = [asyncio.create_task(send(ws)) for i in range(10000)]
        await asyncio.gather(*tasks)

    pass


async def main():
    async with aiohttp.ClientSession() as session:
        await send_group_msg(session)


if __name__ == "__main__":
    asyncio.run(main())

一个压测客户端 用于反向ws

[clacknya]

# -*- coding: utf-8 -*-
"""
Copyright (c) 2008-2021 synodriver <synodriver@gmail.com>
"""
import asyncio
import time

import aiohttp


def get_post_data() -> dict:
    test_groupmsg_data = {
        "time": int(time.time()),
        "self_id": 114514,
        "post_type": "message",
        "message_type": "group",
        "sub_type": "normal",
        "message_id": 1919810,
        "group_id": 1414114,
        "user_id": 341919,
        "anonymous": None,
        "message": "哼哼 嗯啊 啊啊啊啊",
        "raw_message": "哼哼 嗯啊 啊啊啊啊",
        "font": 0,
        "sender":
            {
                "user_id": 341919,
                "nickname": "rc",
                "card": "rc",
                "sex": "female",
                "age": 24,
                "area": "下北泽",
                "level": "是学生",
                "role": "owner",
                "title": "global rc"
            }
    }
    return test_groupmsg_data


headers = {
    "Host": "127.0.0.1:8080",
    "Connection": "Upgrade",
    "Upgrade": "websocket",
    "X-Self-ID": "114514",
    "X-Client-Role": "Universal"
}

enable = {"time": time.time(),
          "self_id": 114514,
          "post_type": "meta_event",
          "meta_event_type": " lifecycle",
          "sub_type": "enable"
          }


async def send(ws):
    while True:
        await ws.send_json(get_post_data())
        print("send")

    # print(await ws.receive_json())


async def receive(ws: aiohttp.ClientWebSocketResponse):
    while not ws.closed:
        await ws.receive_json()


async def send_group_msg(session: aiohttp.ClientSession):
    async with session.ws_connect("http://127.0.0.1:8080/cqhttp/ws", headers=headers) as ws:
        await ws.send_json(enable)
        start = time.time()
        asyncio.create_task(receive())
        tasks = [asyncio.create_task(send(ws)) for i in range(10000)]
        await asyncio.gather(*tasks)

    pass


async def main():
    async with aiohttp.ClientSession() as session:
        await send_group_msg(session)


if __name__ == "__main__":
    asyncio.run(main())

一个压测客户端 用于反向ws

感谢

[clacknya]

@synodriver 你好，请问下在插件接收到 send 的消息并作出响应后，是不是还需要 send 某个数据来表示插件的消息发送成功了？方便告知数据的结构吗

[clacknya]

翻了半天 CQ 源码找到了（

[synodriver]

翻了半天 CQ 源码找到了（

onebot规范 就是