AmiyaBot

Appearance

On this page

QQ 群机器人 Beta ​

在 QQ 开放平台 完成主体入驻，即可创建可在 QQ 群聊里使用的 QQ 群机器人。

在公网下使用 ​

QQ 群聊适配器默认在本地启动资源服务让腾讯服务器能够访问媒体资源，需要在公网下使用。如果无法在公网下部署，请参考自定义资源服务。

python
from amiyabot.adapters.tencent.qqGroup import qq_group

client_secret = '******' # 密钥

bot = AmiyaBot(appid='******', token='******', adapter=qq_group(client_secret))

qq_group 参数

参数名类型释义默认值
client_secretstr机器人密钥
default_chain_builder默认消息构建器None
default_chain_builder_options默认消息构建器参数QQGroupChainBuilderOptions()
shard_indexint分片下标，从 0 开始0
shardsint分片总数1
  • 在机器人启动时，资源服务也会一同启动。
  • 默认的资源服务是端口单例的，实例化多个 QQ 群聊适配器 AmiyaBot 或使用 多账号 时，同一个端口的资源服务会相互共享。

事件分片 ​

考虑到开发者事件接收时可以实现负载均衡，QQ 提供了分片逻辑，事件通知会落在不同的分片上，可参考官方文档 分片连接LoadBalance 了解分片机制。

python
bot1 = AmiyaBot(
    appid='...',
    token='...',
    adapter=qq_group(client_secret='...', shard_index=0, shards=2),
)
bot2 = AmiyaBot(
    appid='...',
    token='...',
    adapter=qq_group(client_secret='...', shard_index=1, shards=2),
)

注意

每个分片的启动应当按顺序缓慢进行，切勿同时启动，以免 gateway 返回的信息一致造成连接失败。

python
# 仅作示意，实际上每个分片应当是独立的服务。
def start():
    asyncio.create_task(bot1.start())
    time.sleep(2)
    asyncio.create_task(bot2.start())

修改资源服务配置 ​

引入 QQGroupChainBuilderOptions 修改默认的资源服务配置。

参数名类型释义默认值
hoststr资源服务监听地址0.0.0.0
portint资源服务监听端口8086
resource_pathstr临时文件存放目录./resource
http_server_optionsdictHttpServer **kwargs
python
from amiyabot.adapters.tencent.qqGroup import qq_group
from amiyabot.adapters.tencent.qqGroup.builder import QQGroupChainBuilderOptions

bot = AmiyaBot(
    appid='******',
    token='******',
    adapter=qq_group(
        client_secret='******',
        default_chain_builder_options=QQGroupChainBuilderOptions(
            '0.0.0.0',
            8086,
            './resource',
        ),
    ),
)

自定义资源服务 ​

非公网部署下的资源服务难题解决途径非常多，这里列举两个比较常见的解决办法。

使用内网穿透 ​

使用一些内网穿透工具代理本地地址 http://127.0.0.1:8086（视配置而定）后，通常会得到一个新的地址。继承 QQGroupChainBuilder 并覆盖 domain 方法，即可使用内网穿透让腾讯服务器访问资源。

示例：

http://3913rc56vl17.vicp.fun:40229/resource

红色高亮部分即为内网穿透地址，/resource 为固定的路由值。

python
from amiyabot.adapters.tencent.qqGroup import qq_group, QQGroupChainBuilder, QQGroupChainBuilderOptions

class PenetrationChainBuilder(QQGroupChainBuilder):
    @property
    def domain(self):
        return 'http://3913rc56vl17.vicp.fun:40229/resource'


bot = AmiyaBot(
    ...,
    adapter=qq_group(
        ...,
        default_chain_builder=PenetrationChainBuilder(
            QQGroupChainBuilderOptions(),
        ),
    ),
)

继承 ChainBuilder 并实现相关方法使用第三方托管服务。 ​

多数情况下我们推荐使用第三方托管服务来搭建资源服务，如 腾讯云COS 或 阿里云OSS 等。通过自定义默认的 ChainBuilder ，来实现上传文件到托管服务以及返回生成的 url。

可参考 进阶指南 - 介入媒体消息的构建过程

python
from typing import Union
from graiax import silkcoder
from amiyabot import ChainBuilder

class ThirdPartyChainBuilder(ChainBuilder):
    @classmethod
    async def get_image(cls, image: Union[str, bytes]) -> Union[str, bytes]:
        # 上传图片到第三方托管服务
        ...
        return url # 返回访问资源的 URL

    @classmethod
    async def get_voice(cls, voice_file: str) -> str:
        # 上传语音文件到第三方托管服务，语音文件必须是 silk 格式
        voice: bytes = await silkcoder.async_encode(voice_file, ios_adaptive=True)
        ...
        return url # 返回访问资源的 URL

    @classmethod
    async def get_video(cls, video_file: str) -> str:
        # 上传视频文件到第三方托管服务
        ...
        return url # 返回访问资源的 URL


bot = AmiyaBot(
    ...,
    adapter=qq_group(
        ...,
        default_chain_builder=ThirdPartyChainBuilder(),
    ),
)