Recently, I did something because web.py did not expand and reuse WeChat official account, and developed WeChat official account using Django. Use celery to push template messages to users' wechat, which is ultimately convenient for reuse and function increase in the future.
01 preparation
-
Python 3 environment
-
WeChat official account
-
Available domain names
-
Mysql database
-
redis database
02 Official account configuration
The official account in the WeChat public address requires two values of AppID and AppSecret for developers to be saved.
After obtaining the value of AppSecret, the IP white list will appear.
The administrator sets the IP white list and calls access_token interface plays a protective role.
The administrator sets the server configuration.
There is no need to click the submit button at this time, because we have not developed a web listening service for response.
03 development
Create virtual environment
python3 -m venv venv # Activate virtual environment source ./venv/bin/activate
Create app app
python manager.py startapp userInfo
settings.py register app
INSTALLED_APPS = [
...,
'userInfo',
]Configure access routes
# Modify template_message/template_message/urls.py
from django.urls import path, include
urlpatterns = [
# path('admin/', admin.site.urls),
path("template/", include('userInfo.urls'))
]
Create urls.py file in userInfo application and configure view function.
from django.conf.urls import url
from . import views
urlpatterns = [
url("message/$", views.Message.as_view(), name='message'),
]
Edit views.py files for WeChat official account backend service docking
Before docking, install wechatpy and djangorest framework
pip install wechatpy[cryptography] pip install djangorestframework
Modify settings.py
INSTALLED_APPS = [
...,
'rest-framework',
]
reference resources:
Documents used by wechatpy:
http://docs.wechatpy.org/zh_CN/stable/index.html
The official account of WeChat: https://developers.weixin.qq.com/doc/offiaccount/Getting_Started/Overview.html
3.1 user information acquisition
from rest_framework.views import APIView
from wechatpy.utils import check_signature
from wechatpy import parse_message, create_reply
from wechatpy.exceptions import InvalidSignatureException
class Message(APIView):
def get(self, request):
signature = request.GET.get('signature', '')
timestamp = request.GET.get('timestamp', '')
nonce = request.GET.get('nonce', '')
echostr = request.GET.get('echostr', '')
token = wx_config.get("token")
try:
check_signature(token, signature, timestamp, nonce)
except InvalidSignatureException:
echostr = 'Bad request'
response = HttpResponse(echostr)
return response
def post(self, request):
msg = parse_message(request.body)
wel_msg = "Welcome to WeChat official account: programmer 9527"
openid = msg.source # Get user openid
if msg.type == 'text':
reply = create_reply(content, msg)
elif msg.type == 'image':
reply = create_reply(content, msg)
elif msg.type == 'voice':
reply = create_reply(content, msg)
else:
reply = create_reply(content, msg)
if hasattr(msg, 'event') and msg.event == "subscribe":
print("User attention", openid)
elif hasattr(msg, 'event') and msg.event == 'unsubscribe':
print("Cancel attention", openid)
response = HttpResponse(reply.render(), content_type="application/xml")
return response
According to the example code of WeChat official account, the message pushed by WeChat server is verified and returned.
from wechatpy.utils import check_signature
from wechatpy.exceptions import InvalidSignatureException
try:
check_signature(token, signature, timestamp, nonce)
except InvalidSignatureException:
# Handle exceptions or ignoreParsing xml pushed by wechat server
from wechatpy import parse_message msg = parse_message(xml)
Public attribute
| name | value |
|---|---|
| id | Message id, 64 bit integer. |
| source | The source user of the message, that is, the user sending the message. |
| target | The target user of the message. |
| create_time | Message sending time, UNIX timestamp |
| type | Type of message |
After successfully launching WeChat official account, access_ is configured. Token interface
Configure routing in userInfo/urls.py
from django.conf.urls import url
from . import views
urlpatterns = [
url("message/$", views.Message.as_view(), name='message'),
url("access/token/$", views.AccessToken.as_view(), name='token'),
]
pip install redis
Create the config.ini configuration file in the root directory of the project
[wechat] token = official account token appid = official account AppID appsecret = official account AppSecret token_exp = access_toke Expiration time token_url = access_token Access address for
3.2 obtaining access_token
view.py
import redis
import requests
import configparser
r = redis.Redis(host='localhost', port=6379, db=1, decode_responses=True) # Create redis object
config = configparser.ConfigParser()
config.read(settings.WECHAT, encoding="utf-8")
wx_config = config.items("wechat")
wx_config = dict(map(lambda x: [x[0], x[1]], wx_config))
wx_config.update({"token_exp": int(wx_config.get("token_exp"))})
class AccessToken(APIView):
def get(self, request):
access_token = r.get("access_token") # Obtain access from redis_ TOKEN
ifnot access_token:
appid = wx_config.get("appid")
appsecret = wx_config.get("appsecret")
token_api = wx_config.get("token_api")
exp = wx_config.get("token_exp")
api = token_api.format(appid=appid, secret=appsecret)
response = requests.get(api, headers=settings.HEADER).json()
access_token = response.get("access_token")
r.setex('access_token', exp, access_token)
return JsonResponse({"code": 1, "token": access_token})
Using redis to cache access_token, avoid calling the official account every time. See the interface authority for details.
Call the interface to verify whether access can be obtained_ token
3.3 template messages
to configure
Configuring message templates to be sent in official account number
After the template ID is configured, the next step is to use celery for task execution.
pip install celery
Create cellery_ Tasks package file to configure relevant celery information
celery_tasks/main.py file
from celery import Celery
celery_app = Celery('wechat_template')
# Import profile
celery_app.config_from_object('celery_tasks.config')
# Automatically register cellery tasks
celery_app.autodiscover_tasks(['celery_tasks.sends'])celery_tasks/config.py
broker_url = "redis://127.0.0.1/9" result_backend = "redis://127.0.0.1/10"
Create cellery_ Tasks / sends package, create new tasks.py
# -*-coding=utf-8 -*-
import json
import time
import requests
from celery_tasks.logs import Logger
from celery_tasks.main import celery_app
token_url = "Your own access_token"
# Template message api
model_url = 'https://api.weixin.qq.com/cgi-bin/message/template/send?access_token={access_token}'
def request_model(temp_dict):
"""Send template message network request"""
if temp_dict:
token_str = requests.get(token_url).json()
access_token = token_str.get('token')
model_url
send_url = model_url.format(access_token=access_token)
try:
for _ in range(3):
response = requests.post(send_url, json=temp_dict)
res_dict = json.loads(response.text)
if res_dict.get('errcode') == 0:
returnTrue
else:
msg = 'Send template id:' + temp_dict.get("template_id") + "Sending error, error message is" + str(res_dict) + '\n'
with open('msg.log', 'a+', encoding="utf-8") as f:
f.write(msg)
time.sleep(1)
except Exception as e:
log.logger.info(temp_dict.get("template_id") + "Sending error, the error message is:" + str(e))
return str(e)
@celery_app.task(name='send_server_sign')
def send_server_sign(openid=None, temp_id=None, **kwargs):
"""
Signing reminder
"""
temp_dict = {
"touser": openid,
"template_id": temp_id,
"data": {
"first": {
"value": 'Dear customers, you have successfully purchased our products!',
"color": "#FF0000"
},
"keyword1": {
"value": kwargs.get("name"),
"color": "#FF0000"
},
"keyword2": {
"value": "{}to{}".format(kwargs.get("start_time"), kwargs.get("end_time")),
"color": "#000000"
},
"remark": {
"value": "If you have any questions or inquiries, please contact our customer service personnel.",
"color": "#000000"
}
}
}
status = request_model(temp_dict)
return statusStart the celery command
celery multi start worker -A celery_tasks.main -l info --pool=gevent --concurrency=60 --logfile=celery.log
celery stop
celery multi stop worker -A celery_tasks.main -l info
views.py
from django.conf import settings
from rest_framework.views import APIView
from django.http import JsonResponse
from celery_tasks.sends.tasks import send_server_msg, send_server_sign
config = settings.EXPORT_CONFIG
class Complete(APIView):
def get(self, request):
items = {
'url': 'http://www.baidu.com',
'name': 'Sundries',
'title': 'The fund's income advantage is highlighted to attract public offering intensive layout',
'about': 'Test data: 11 provinces increased year-on-year, but the increase was lower than that of the whole country'
}
openid = "asdasPh0AfSasdasddsadwf"
status = send_server_msg.delay(openid=openid, temp_id=config.get("modelID").get("complete_msg"), **items)
return JsonResponse({"code": 1, 'msg': str(status)})
class SignMsg(APIView):
def get(self, request):
name = request.GET.get('name') # Service name
start_time = request.GET.get('start_time') # Service signing time
end_time = request.GET.get('end_time') # Service expiration time
openid = request.GET.get('openid') # User openid
items = {"name": name, "start_time": start_time, "end_time": end_time}
model_server_sign = config.get("modelID").get("sign_msg")
status = send_server_sign.delay(openid=openid, temp_id=model_server_sign, **items)
return JsonResponse({"code": 1, 'msg': str(status), 'model_server_sign': model_server_sign})results of enforcement
The above is all content developed by django and WeChat official account. We hope to help you. If you have any problems, you can contact the author.
Interested parties can be concerned about the author's WeChat official account: programmer 9527.
