小白Python脚本：使用Python对接微信公众号测试号推送消息

　　使用Python对接微信公众平台接口测试账号推送消息，比如将收集的数据或者一些消息及时推送给自己或者团队成员，本小白曾长时间运行类似脚本，将监测的异常情况及时发送给自己，提高了异常发现的及时性、提高工作效率。

　　这是一个对于初学者可以尝试的一个小小的测试项目，具有一定的较易入手性与可玩性，比较容易上手。

　　本文以使用微信公众平台接口测试帐号为例，直接使用微信公众号的用法是一样的，在生产环境中有一定的适用性，但是具体的实现逻辑需要根据实际需要进行编制。

　　‍

一、 申请微信公众平台接口测试帐号

　　链接： https://mp.weixin.qq.com/debug/cgi-bin/sandbox?t=sandbox/login

　　点击“登录”，在新打开的页面使用自己的微信进行扫一扫。

​

　　在手机微信端确认登录后，电脑上的网页将会自动进入测试号管理界面，如下：

　　​

​

　　appID与appsecret可记录一下，后续会用到。

　　​

　　关于接口配置信息部分，可在服务器端配置完成后，再进行配置。

二、 服务器端配置

　　对于服务器端，需有要公网可访问的服务器。由于一般的家用宽带的80、443等入端口被封禁等情况，难以直接使用家用宽带中的电脑作为服务器，当然可以通过使用一些公网服务器端口转发本地端口的方式；也可以申请云服务器，把服务端脚本部署在服务器上，也是比较方便的方式。像当前的阿里云、百度云、或者腾讯云的一些类似轻量服务器，对于新用户，一年几十元即可购得，成本较低，后续可使用的地方也比较多。亚马逊云的EC2对于一些配置的云服务器可以免费使用一年，但是需要校验绑定信用卡。

　　我的服务器端环境：操作系统是采用的是Linux ，Ubuntu kylin 20.04发行版；Python为Python3.8。

　　如果没有安装pip3的话，可以使用sudo apt install python3-pip安装。

　　需要安装flask及sqlite，安装指令如下：

sudo pip3 install flask
sudo apt-get install sqlite3

　　‍

　　以下是服务器端的python脚本wechatOfficialAccountsServer.py文件内容：

#!/usr/bin/env python3
# -*- coding: UTF-8 -*-
import os
import hashlib
import datetime
import time
import urllib3
import json
import random
import string
from json import *
import requests
from flask import Flask, g, request, render_template ,redirect, url_for, escape, session, flash

app = Flask(__name__)
app.secret_key = 'A0Zr98j/3yX R~XHH!jmN]LWXXSXAE89'

import sqlite3
DATABASE = 'OfficialAccounts.db'

def connect_db():
    return sqlite3.connect(DATABASE)

@app.before_request
def before_request():
    g.db = connect_db()

@app.teardown_request
def teardown_request(exception):
    if hasattr(g, 'db'):
        g.db.close()




def fetchaccesstoken_APPID(APPID):
    timenow = datetime.datetime.now()
    c=g.db.cursor()
    sql = "select  APPID,APPSECRET,accesstoken,updatetime  from OfficialAccounts where APPID=\"%s\";"  %APPID
    c*ex.e**cute(sql)
    row= c.fetchone()
    appId=row[0]
    secret=row[1]
    accesstoken=row[2]
    #time=row[3]
    time=datetime.datetime.strptime(row[3],"%Y-%m-%d %H:%M:%S")
    if (accesstoken is None) or ((timenow-time).total_seconds()>6000) :
        url='https://api.weixin.qq.com/cgi-bin/token?grant_type=client_credential&appid=' + appId + '&secret=' + secret
        r = requests.get(url)
        tokeninfo = json.loads(r.text)
        accesstokenstr=tokeninfo['access_token']
        sql_set_accesstoken="update OfficialAccounts set accesstoken=\"" + accesstokenstr +"\",updatetime=dateTime('now','localtime') where appid=\"%s\";"  %APPID
        c*ex.e**cute(sql_set_accesstoken)
        g.db.commit()
        accesstoken=accesstokenstr
    return accesstoken


@app.route('/wechatMsgReceiver', methods=['GET', 'POST'])
def wemsgre():
    if request.method == 'GET':
        #verify wechat connect configuration.
        id=request.args.get("id")
        if id is not None:
            signature=request.args.get("signature")
            timestamp=request.args.get("timestamp")
            nonce=request.args.get("nonce")
            echostr=request.args.get("echostr")
            c=g.db.cursor()
            sql="select Token from OfficialAccounts where id=%d;"  %(int(id))
            c*ex.e**cute(sql)
            row= c.fetchone()
            token= row[0]
            tmplist = [ token, timestamp, nonce ]
            tmplist.sort()
            tmpstr = ''.join( tmplist )
            hashstr = hashlib.sha1( tmpstr.encode() ).hexdigest()
            if hashstr == signature:
                return echostr
            else:
                 return 'Error' + echostr
        else:
            return "fault"
    elif request.method == 'POST':
        return 'SUCCESS'



#send text message.
@app.route('/sendtxtmsg', methods=['GET', 'POST'])
def sendtxtmsg():
    if request.method == 'POST':
        #print(request.data)
        receivedData=json.loads(request.data.decode())
        #print(receivedData["appID"])
        token=fetchaccesstoken_APPID(receivedData["appID"])
        data=json.dumps(receivedData["messageContent"],ensure_ascii=False).encode("utf-8")
        url = 'https://api.weixin.qq.com/cgi-bin/message/custom/send?access_token='+token
        r = requests.post(url,data)
        response = json.loads(r.text)
        return response



#send text message.
@app.route('/', methods=['GET', 'POST'])
def index():
    return "Just a test!"




if __name__ == "__main__":
    app.run(host='0.0.0.0')

　　以上脚本的app.secret_key = 'A0Zr98j/3yX R~XHH!jmN]LWXXSXAE89' 建议自己再修改一下，当然测试使用就无所谓了。

　　上传wechatOfficialAccountsServer.py脚本到某一路径，以下操作在同路径中进行。

　　创建数据库、建立表、插入数据。

　　对于insert into OfficialAccounts (Id, Name, APPID, APPSECRET, Token) values ("<Id>", "<Name>", "<APPID>", "<APPSECRET>", "<Token>"); 这条指令中的"<Id>", "<Name>", "<APPID>", "<APPSECRET>", "<Token>"这几个参数，根据配置进行修改，这里的<Id>，<Name>是服务端系统中自己随便定义的参数，比如Id 为 1， Name为testAccounts, 而"<APPID>", "<APPSECRET>"是来自于测试公众号中的参数，见上一步“申请微信公众平台接口测试帐号”中的参数，<Token>是自定义的，与“申请微信公众平台接口测试帐号"中设置一致即可。

　　sqlite指令示例：

sqlite3   OfficialAccounts.db
  
CREATE TABLE 'OfficialAccounts' ( 
 'Id' integer,  
 'Name' char,  
 'APPID' char, 
 'APPSECRET' char,  
 'Token' char,  
 'accesstoken' char,  
 'EncodingAESKey' char, 
  'updatetime' TIMESTAMP DEFAULT current_timestamp);

#exapmle : insert into OfficialAccounts (Id, Name, APPID, APPSECRET, Token) values ("<Id>", "<Name>", "<APPID>", "<APPSECRET>", "<Token>");
insert into OfficialAccounts (Id, Name, APPID, APPSECRET, Token) values (2, "testAccount", "wx9276144e23c95xxx", "a320ddd3ae51c363b294086f031d0xxx", "testtoken");
.exit

　　‍

　　运行脚本 wechatOfficialAccountsServer.py。以下为在开发测试环境中直接使用flask运行，如果长期使用或者在生产环境中可以配置使用apache、ngnix等。在脚本所在目录中执行：

export FLASK_APP=wechatOfficialAccountsServer.py
flask run --host 0.0.0.0 -p 8080

　　​

​

　　端口根据实际使用定义，如80,8080等。如果遇到缺少某些库，自行安装即可。

　　‍

三、 微信公众平台接口测试帐号的接口配置确认：

　　返回第1步微信公众平台接口测试帐号配置页面，在如下”接口配置信息“部分进行配置。

　　​

　　在“接口配置信息”部分，需要填写URL与Token，结合自己的服务器端资源情况进行填写，如我的填写为：

　　URL： http://<xx.xx.xx.xx>:8080/wechatMsgReceiver?id=2

　　Token： testtoken

　　以上示例中的<xx.xx.xx.xx>代替实际IP，请按实际配置更改。

　　在服务器端正常运行后，再进行此处的接口配置修改。一切正常的话，将会看到"配置成功"的提示。

　　‍

四、推送文本消息测试：

　　​

　　在测试号管理页面可以看到"测试号二维码"，使用微信客户端扫描关注后，此页面可以看到用户，测试号最多可以100个用户关注。如下，所示：

　　记录下用户微信号那一长串字符串，后面将会使用。

　　本处使用的脚本sendMsgClient.py，在任意可以与服务器端正常通信的电脑上使用都可以，也可以直接放在服务器端。

　　脚本中的Server、appID，根据实际配置进行修改，receiveUserList为接收者微信号列表，多个接收者使用逗号分隔。（接收者微信号标识是在微信公众号管理页面里看到的用户的那一长串用户标识。）

　　sendMsgClient.py脚本内容如下：

#!/usr/bin/env python3
import json
import requests
import datetime

###修改此部分配置信息
Server="http://xx.xx.xx.xx:8080"  #根据实际配置修改服务端信息
appID="xxxxxxxx"  #修改为实际使用的appID
receiveUserList=["yyyyyyyyyyyyyyyyyyyy"]  #接收消息用户列表


def send_wechat_msg(content):
    for touser in receiveUserList:
        data={
                "appID":appID,
                "messageContent":
                    {
                    "touser":touser,
                    "msgtype":"text",
                    "text":{
                    "content": content}
                    }
                }
        data = json.dumps(data,ensure_ascii=False).encode("utf-8")
        url = Server + "/sendtxtmsg"
        r = requests.post(url,data)
        result = json.loads(r.text)
        print(result)


def main():
    message="This is just a test message.\n这是一条测试消息。"
    send_wechat_msg(message)

if __name__ == "__main__":
    main()

　　message="This is just a test message.\n这是一条测试消息。" 这部分是定义的发送的测试消息内容，发送内容可根据需要修改。

　　执行发送消息脚本，运行如下：

　　​

　​

　　手机微信微收到消息，示例如下：

　　​

​

　　‍

　　这里是使用微信公众号的客服消息接口进行的推送消息，会有一定的下发额度及有效期限制，如果由于用户长时间与测试号未交互造成不在客户消息下发规则内，可以使用手机微信端先给测试号发送一下消息，再进行推送消息测试。

　　发送客服消息的额度与有效期如下：

　　​

​

　　如果格式比较固定，也可以使用”模板消息（业务通知）“进行消息发送，在用户长时间没有与公众号交互的情况下也可以推送模板消息。相应的脚本、配置等，需要进行新的适配，相关可参考微信公众号体验接口权限表接口链接。

　　​

　　‍

　　‍

　　以上是关于使用python对接微信公众号测试号推送消息的记录，也是学习途中一点小小的快乐，如果对于探索途中的初学者有所帮助，本人将深感幸甚。