下载或打开 PDF
ChatGPT接入微信 from flask import Flask, request, make_response import requests import hashlib import xml.etree.cElementTree as ET import time # from WXBizMsgCrypt import WXBizMsgCrypt import base64 import redis app = Flask(__name__) # 见下图 headers={'Authorization': 'Bearer **-********************'} appid = '***********' token = '*******************' encodingAESKey = '************************' r = redis.Redis() @app.get('/completions') def sig(): args = request.args print(args) signature = args['signature'] echostr = args['echostr'] timestamp = args['timestamp'] nonce = args['nonce'] arr = [token, timestamp, nonce] arr.sort() s = ''.join(arr) sha1 = hashlib.sha1(s.encode('utf-8')).hexdigest() if sha1 == signature: return echostr, 200 return 'error', 500 @app.post('/completions') def completions(): t0 = time.time() xmlStr = request.stream.read() print(f'xmlStr = {xmlStr}') xmlData = None try: xmlData = ET.fromstring(xmlStr) except Exception: return '', 200 msg_type = xmlData.find('MsgType').text if msg_type == 'text': ToUserName = xmlData.find('ToUserName').text FromUserName = xmlData.find('FromUserName').text Content = xmlData.find('Content').text json = { "model": "text-davinci-003", "prompt": Content.strip(), "max_tokens": 4000, "temperature": 1.0 } key = str(FromUserName + ':' + Content.strip()) print(f'question:{key}') reply = ''' <xml> <ToUserName><![CDATA[%s]]></ToUserName> <FromUserName><![CDATA[%s]]></FromUserName> <CreateTime>%s</CreateTime> <MsgType><![CDATA[text]]></MsgType> <Content><![CDATA[%s]]></Content> </xml> ''' answer = '没听清，再说一遍' cache = r.get(key) if cache: answer = cache.decode('utf-8') r.delete(key) elif cache is None: r.set(key, '', ex=60) text = '' try: res = requests.post('https://api.openai.com/v1/completions', headers=headers, json=json).json() text = res['choices'][0]['text'] print(f'res = {res}') except Exception as e: print(f'error:{e}') r.set(key, None) return '', 555 t1 = time.time() if t1 - t0 > 4.5: r.set(key, text, ex=3600) return '', 555 else: answer = res['choices'][0]['text'] elif cache.decode('utf-8') == '': print('sleep') time.sleep(4.5) answer = f'思考中...\n过会儿再对我说一遍[{Content.strip()}]' else: return 'success' response = make_response(reply % (FromUserName, ToUserName, str(int(time.time())), answer.strip())) response.content_type = 'application/xml' return response else: return 'success' if __name__ == '__main__': app.run(port = 9999) appid：必填 token：必填 encodingAESKey：选填（当需要加密时必填）
...