可爱猫+python3+Flask+aiohttp简单搭建微信机器人

之前一直在研究QQ机器人，最近发现一个微信机器人框架，在办公中来增加工作效率。

一、需要环境：

可爱猫（http://www.keaimao.com/）
python3.8+ flask、aiohttp
微信 for windows 3.4.0.38

1.启动可爱猫、登录微信

2.在可爱猫应用中找到HTTP个人对接版，双击打开

其中
消息回调地址，相当于消息接收地址
调用本地api地址，相当于消息发送地址

用python整个run.py，顺便从隔壁投一个flask使用async的方法

二、核心代码（run.py）

import flask
import logging
from flask import Flask, jsonify, has_request_context, copy_current_request_context, request
from functools import wraps
from concurrent.futures import Future, ThreadPoolExecutor
import time
from getmsg import getmsg
import asyncio

defrun_async(func):@wraps(func)def_wrapper(*args,**kwargs):
        call_result = Future()def_run():
            loop = asyncio.new_event_loop()try:
                result = loop.run_until_complete(func(*args,**kwargs))except Exception as error:
                call_result.set_exception(error)else:
                call_result.set_result(result)finally:
                loop.close()
 
        loop_executor = ThreadPoolExecutor(max_workers=1)if has_request_context():
            _run = copy_current_request_context(_run)
        loop_future = loop_executor.submit(_run)
        loop_future.result()return call_result.result()return _wrapper
 

app = flask.Flask(__name__)@app.route('/', methods=['POST'])@run_asyncasyncdefmain():
    msg,from_name,final_from_name,time_str,from_wxid,final_from_wxid,msg_type = getmsg()#这一块后面可以写插件主要内容#...return'ok'if __name__ =='__main__':
    app.config['JSON_AS_ASCII']=False
    log = logging.getLogger('werkzeug')
    log.disabled =True
    app.run(host='0.0.0.0', port=8074, debug=True)

三、可爱猫提供的接收与发送方法

1.接收消息（getmsg.py）

import flask,time
defgetmsg():#取msg
    allmsg = flask.request.get_data()
    msg_type =flask.request.form.get('type')if msg_type =='200':
        msg_type_name ='群聊消息'elif msg_type =='100':
        msg_type_name ='私聊消息'
    msg = flask.request.form.get('msg')
    from_name = flask.request.form.get('from_name').encode('utf-8').decode('utf-8')
    final_from_name = flask.request.form.get('final_from_name').encode('utf-8').decode('utf-8')
    time_str = flask.request.form.get('time')
    time_str = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(int(time_str)))
    from_wxid = flask.request.form.get('from_wxid')
    final_from_wxid = flask.request.form.get('final_from_wxid')print(f'\033[1;36m[INFO] [{time_str}]\033[0m [{msg_type_name}(\033[1;34m{from_name}\033[0m)]\033[1;35m {final_from_name} \033[0m: \033[1;32m{msg}\033[0m ')return msg,from_name,final_from_name,time_str,from_wxid,final_from_wxid,msg_type

2.发送消息（httpapi.py）

这里按照可爱猫提供的方法写了三种，基本上已经够用了其中wxid_xxxxx为机器人的wxid

import json
import time
import aiohttp

url ='http://127.0.0.1:8073/send'asyncdefsend_msg_private(msg, to_wxid):
    data ={"type":"100","msg": msg,"to_wxid": to_wxid,"robot_wxid":"wxid_xxxxx"}
    data = json.dumps(data)asyncwith aiohttp.ClientSession()as session:asyncwith session.post(url, data=data)as resp:print(await resp.text())asyncdefsend_msg_group(msg, to_wxid):
    data ={"type":"100","msg": msg,"to_wxid": to_wxid,"robot_wxid":"wxid_xxxxx"}
    data = json.dumps(data)asyncwith aiohttp.ClientSession()as session:asyncwith session.post(url, data=data)as resp:print(await resp.text())asyncdefsend_image_msg(picpath, to_wxid):
    data ={"type":"106","msg": picpath,"to_wxid": to_wxid,"robot_wxid":"wxid_xxxxx"}
    data = json.dumps(data)asyncwith aiohttp.ClientSession()as session:asyncwith session.post(url, data=data)as resp:print(await resp.text())

四、插件例子（百度文心的AI对对联duilian.py）

插件我放在plugin文件夹下，这个随意

import wenxin_api 
from wenxin_api.tasks.couplet import Couplet
defduilian(str1):
    wenxin_api.ak =""#自己申请
    wenxin_api.sk =""#自己申请
    input_dict ={"text":f"上联：{str1}\n下联：","seq_len":512,"topp":0.9,"penalty_score":1.0,"min_dec_len":2,"is_unidirectional":0,"task_prompt":"couplet"}
    rst = Couplet.create(**input_dict)#返回resultreturn rst['result']

五、在run.py中添加

...from plugin.duilian import duilian

#上面main()中getmsg后面asyncdefmain()if'#上联'in msg and msg_type =='200':
        args = msg.split('上联')[-1]if args =='':pass#这里可以自行添加判断这里图方便我就没写else:#print(args)
            result ='下联：'+duilian(args)#发送微信消息await send_msg_group(result,from_wxid)

运行效果