驱动器对接

驱动器对接包含各类与go-cqhttp交互的函数。

其中最重要的即发送信息的函数 send ，参数信息如下：

参数名称	描述	取值	取值说明	备注
id	保存地址	int型变量	表示要发送的群号或者对象qq号
message	消息	str型变量	表示要发送的消息	如要发送图片、音频、表情、文件等，请使用 cq码转义
type	类型	str型变量	表示要发送的消息类型	群聊：'group'；私聊：'private'

关于其他对接函数即功能（如获取群聊列表，获取群聊历史记录），请参见下方代码分析：

代码分析

def get_avatar_pic(id: int)->Union[None, bytes]:
    """获取QQ头像
    @id: qq号
    @return:
        None if QQ头像获取失败
        bytes if QQ头像获取成功
    """
    url_avatar = requests.get(f'http://q2.qlogo.cn/headimg_dl?dst_uin={id}&spec=100')
    if url_avatar.status_code != requests.codes.ok:
        return None
    else:
        return url_avatar.content

def send(id: int, message: str, type:str='group')->None:
    """发送消息
    id: 群号或者私聊对象qq号
    message: 消息
    type: Union['group', 'private'], 默认 'group'
    """
    url = HTTP_URL+"/send_msg"
    if type=='group':
        params = {
            "message_type": type,
            "group_id": id,
            "message": message
        }
        print(params)
        requests.get(url, params=params)
    elif type=='private':
        params = {
            "message_type": type,
            "user_id": id,
            "message": message
        }
        print(params)
        requests.get(url, params=params)

def get_group_list()->list:
    """获取群聊列表
    @return:
        [
            {
                'group_create_time': int,
                'group_id': int,
                'group_name': str,
                'group_level': int,
                'max_member_count': int,
                'member_count': int,
            },
            ....
        ]
    参考链接： https://docs.go-cqhttp.org/api/#%E8%8E%B7%E5%8F%96%E7%BE%A4%E5%88%97%E8%A1%A8
    """
    url = HTTP_URL+"/get_group_list"
    try:
        groupList = json.loads(requests.get(url).text)
        if groupList['status'] != 'ok':
            warning("get_group_list requests not return ok")
            return []
        return groupList['data']
    except BaseException as e:
        warning("error in get_group_list, error: {}".format(e))
        return []

def get_group_msg_history(group_id: int, message_seq: Union[int, None]=None)->list:
    """获取群消息历史记录
    @message_seq:
        起始消息序号, 可通过 get_msg 获得
        如果是None将默认获取最新的消息
    @group_id: 群号
    @return: 从起始序号开始的前19条消息
    参考链接： https://docs.go-cqhttp.org/api/#%E8%8E%B7%E5%8F%96%E7%BE%A4%E6%B6%88%E6%81%AF%E5%8E%86%E5%8F%B2%E8%AE%B0%E5%BD%95
    """
    url = HTTP_URL+"/get_group_msg_history"
    try:
        params = {
            "group_id": group_id
        }
        if message_seq != None:
            params["message_seq"] = message_seq

        messageHistory = json.loads(requests.get(url, params=params).text)
        if messageHistory['status'] != 'ok':
            if messageHistory['msg'] == 'MESSAGES_API_ERROR':
                print("group {} meet message API error".format(group_id))
            else:
                warning("get_group_msg_history requests not return ok\nmessages = {}\ngroup_id={}\nmessage_seq={}".format(
                messageHistory, group_id, message_seq))
            return []
        return messageHistory['data']['messages']
    except BaseException as e:
        warning('error in get_group_msg_history, error: [}'.format(e))
        return []
def get_essence_msg_list(group_id: int)->list:
    """获取精华消息列表
    @group_id:  群号
    @return:    精华消息列表
    """
    url = HTTP_URL+"/get_essence_msg_list"
    try:
        params = {
            "group_id": group_id
        }
        essenceMsgs = json.loads(requests.get(url, params=params).text)
        if essenceMsgs['status'] != 'ok':
            warning("get_essence_msg_list requests not return ok")
            return []
        return essenceMsgs['data']
    except BaseException as e:
        warning("error in get_essence_msg_list, error: {}".format(e))
        return []
def set_friend_add_request(flag, approve=True)->None:
    """处理加好友"""
    url = HTTP_URL+"/set_friend_add_request"
    params = {
        "flag": flag,
        "approve": approve
    }
    print(params)
    requests.get(url, params=params)