Django+ WeChat official account development project

         Recently, I did something because 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 startapp userInfo register app


Configure access routes

# Modify template_message/template_message/

from django.urls import path, include

urlpatterns = [
    # path('admin/',,
    path("template/", include('userInfo.urls'))

Create 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 files for WeChat official account backend service docking

Before docking, install wechatpy and djangorest framework

pip install wechatpy[cryptography]
pip install djangorestframework



reference resources:

Documents used by wechatpy:

The official account of WeChat:

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")
            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)
            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

    check_signature(token, signature, timestamp, nonce)
except InvalidSignatureException:
    # Handle exceptions or ignore

Parsing xml pushed by wechat server

from wechatpy import parse_message
msg = parse_message(xml)

Public attribute

idMessage id, 64 bit integer.
sourceThe source user of the message, that is, the user sending the message.
targetThe target user of the message.
create_timeMessage sending time, UNIX timestamp
typeType of message

After successfully launching WeChat official account, access_ is configured. Token interface

Configure routing in userInfo/

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

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

import redis
import requests
import configparser

r = redis.Redis(host='localhost', port=6379, db=1, decode_responses=True)  # Create redis object
config = configparser.ConfigParser(), 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/ file

from celery import Celery

celery_app = Celery('wechat_template')

# Import profile

# Automatically register cellery tasks


broker_url = "redis://"
result_backend = "redis://"

Create cellery_ Tasks / sends package, create new

# -*-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 = '{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')
        send_url = model_url.format(access_token=access_token)
            for _ in range(3):
                response =, json=temp_dict)
                res_dict = json.loads(response.text)
                if res_dict.get('errcode') == 0:
                    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:
        except Exception as e:
  "template_id") + "Sending error, the error message is:" + str(e))
            return str(e)

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 status

Start 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

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': '',
            '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.

Keywords: Python Django wechat

Added by mattcairns on Wed, 15 Sep 2021 20:45:48 +0300