Django+wechatpy+ChatGPT+Celery接入微信公众号、企业微信

发表于 更新于 分类于 Waline： 阅读次数：

Django+wechatpy+ChatGPT+Celery接入微信公众号、企业微信

安装模块

1
2
3
4
# 安装django
pip install django
# 安装WeChatpy
pip install wechatpy

新建项目

1
2
3
4
# 创建一个django项目
django-admin startproject wechat .
# 创建一个应用demo
python3 manage.py startapp demo

新增配置setting.py

1
TOKEN = 'xxxxx'

编写views.py

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
from wechatpy.utils import check_signature
from wechat import settings
from wechatpy.exceptions import InvalidSignatureException
from django.http import HttpResponse
from wechatpy import parse_message, create_reply
from wechatpy.replies import BaseReply
from wechatpy import WeChatClient
from wechatpy.oauth import WeChatOAuth
from django.shortcuts import redirect
from wechat.util import reply_text

# 连接微信公众号的方法
def serve(request):
    # GET 方式用于微信公众平台绑定验证
    if request.method == 'GET':
        signature = request.GET.get('signature', '')
        timestamp = request.GET.get('timestamp', '')
        nonce = request.GET.get('nonce', '')
        echo_str = request.GET.get('echostr', '')
        try:
            check_signature(settings.Token, signature, timestamp, nonce)
        except InvalidSignatureException:
            echo_str = '错误的请求'
        response = HttpResponse(echo_str)
    else:
        msg = parse_message(request.body)
        msg_dict = msg.__dict__['_data']
        print(type(msg))
        print(msg.id, msg.source, msg.create_time, msg.type, msg.target, msg.time, '====')
        if msg.type == 'text':
            reply = reply_text.do_reply(msg)
        elif msg.type == 'event':
            if msg_dict['Event'] == 'subscribe':
                pass
        else:
            pass
        response = HttpResponse(reply.render(), content_type="application/xml")
    return response


def getWxClient():
    return WeChatClient(settings.AppID, settings.AppSecret)


def getWxUserInfo(openid):
    wxClient = getWxClient()
    wxUserInfo = wxClient.user.get(openid)
    return wxUserInfo


def getWeChatOAuth(redirect_url):
    return WeChatOAuth(settings.AppID, settings.AppSecret, redirect_url)

# 定义授权装饰器
def oauth(method):
    def warpper(request):
        if request.session.get('user_info', None) is None:
            code = request.GET.get('code', None)
            wechat_oauth = getWeChatOAuth(request.get_raw_uri())
            url = wechat_oauth.authorize_url
            if code:
                try:
                    wechat_oauth.fetch_access_token(code)
                    user_info = wechat_oauth.get_user_info()
                except Exception as e:
                    print(str(e))
                    # 这里需要处理请求里包含的 code 无效的情况
                    # abort(403)
                else:
                    request.session['user_info'] = user_info
            else:
                return redirect(url)

        return method(request)
    return warpper

@oauth
def get_wx_user_info(request):
    user_info = request.session.get('user_info')
    return HttpResponse(str(user_info))

urls.py

1
2
3
4
5
6
from django.urls import path, include
from . import views
urlpatterns = [
    path('wechat/', views.serve),
    path('user/', views.get_wx_user_info)
]

ChatGPT

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
import openai
import os
# 境内服务器需要设置代理，否则无法访问
os.environ['http_proxy'] = 'http://172.17.0.1:1081'
os.environ['https_proxy'] = 'http://172.17.0.1:1081'

api_key = "sk-xxxxxxxxxxxxxxxxxxx"
openai.api_key = api_key

def askChatGPT(messages, system=None):
    if system is None:
        messages=[
            {"role": "user", "content": messages},
        ]
    else:
        messages=[
        {"role": "system","content": system},
        {"role": "user", "content": messages},
    ]
    MODEL = "gpt-3.5-turbo"
    response = openai.ChatCompletion.create(
        model=MODEL,
        messages = messages,
        temperature=0)
    return(response['choices'][0]['message']['content'])

测试运行

1
python manage.py runserver

实际运行

1
uwsgi --socket 127.0.0.1:9000 --chdir /var/www/html/wechat/demo  --module back.wsgi:application --processes 1 --enable-threads

Celery运行

1
/usr/local/python3/bin/celery -A wechat.celery.chatgpt worker --loglevel=info

参考

  1. django微信公众号开发入门

  2. 使用django快速搭建微信公众号后台服务

  3. Django+wechatpy接入微信公众平台以及授权登录

  4. wechatpy

  5. 编写你的第一个 Django 应用

  6. Linux 搭建Nginx+uWSGI+Django环境

  7. 手把手教你使用 Python 调用 ChatGPT-3.5-API

  8. 100个 ChatGPT 账号免费使用