coze_api/coze_bot_api.py

#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
-----------------File Info-----------------------
Name: web.py
Description: web api support
Author: GentleCP
Email: me@gentlecp.com
Create Date: 2021/6/19
-----------------End-----------------------------
"""
from fastapi import FastAPI, Response, Request, BackgroundTasks
from WXBizMsgCrypt3 import WXBizMsgCrypt
from xml.etree.ElementTree import fromstring
import uvicorn
import requests
import json

# 加载配置文件
with open('config.json', 'r') as f:
    config = json.load(f)

# 从配置文件中提取参数
token = config['token']
aeskey = config['aeskey']
corpid = config['corpid']
corpsecret = config['corpsecret']
coze_access_token = config['coze_access_token']
bot_id = config['bot_id']
port = config['port']


# token = "EcSp"#企业微信应用api信息
# aeskey = "OTZoY8N67kOnGosEpS3jw4Rsjea0Gu6D7X4IWxoYKtY"#企业微信应用api信息
# corpid = "ww5541cfeea51e3188"#企业id
# corpsecret = "SbyG25s1LsMsW0nAMiaNprrQIHYrWKQP4f2mNLLDnwE"##api成功后的secret
# coze_access_token = "pat_HNBYQOWE5h4r1tzXi8S2PuY4ddoVRH3DpTbE3NsYBjtcWHTYw5ffrVmKPh26hSLW"#豆包access_token
# bot_id="7397619068440182793"#豆包机器人id
# port = 18090#服务器端口

wxcpt = WXBizMsgCrypt(token, aeskey, corpid)


app = FastAPI()


def call_llm(prompt: str, bot_id: str,coze_access_token:str):
    req_head = {
        "Authorization":f"Bearer {coze_access_token}",
        "Content-Type": "application/json",
    }
    req_data = {
        "conversation_id": "123",
        "bot_id": bot_id,
        "user": "test",
        "query": prompt,
        "stream": False
    }
    res = requests.post("https://api.coze.cn/open_api/v2/chat", headers=req_head, json=req_data)
    res.raise_for_status()  # 检查响应状态码是否为200
    return res.json()

def qiwei_get():
    res = requests.get(f"https://qyapi.weixin.qq.com/cgi-bin/gettoken?corpid={corpid}&corpsecret={corpsecret}")
    qw_access_token = res.json()["access_token"]
    return qw_access_token

def qiwei_post(username: str, answer: str,agentid:str):
    req_data = {
            "touser": username,
            "toparty": "",
            "totag": "",
            "msgtype": "text",
            "agentid": agentid,
            "text": {"content": answer},
            "safe": 0,
            "enable_id_trans": 0,
            "enable_duplicate_check": 0,
            "duplicate_check_interval": 1800
    }
    res = requests.post(f"https://qyapi.weixin.qq.com/cgi-bin/message/send?access_token={qiwei_get()}", json=req_data)
    print(res.json())
    #return res.json()





def consumer(user_query,decrypt_data):
    print(f"请求:{user_query}")
    username = decrypt_data.get('FromUserName', '')
    agentid = decrypt_data.get('AgentID', '')
    # 返回coze结果
    coze_response = call_llm(prompt=user_query,bot_id=bot_id,coze_access_token = coze_access_token)
    answer = coze_response['messages'][1]['content']
    print(f"结果:{answer}")
    # 主动发结果给qiwei
    qiwei_post(username, answer, agentid)

@app.get("/bot")
async def verify(msg_signature: str, timestamp: str, nonce: str, echostr: str):
    ret, sEchoStr = wxcpt.VerifyURL(msg_signature, timestamp, nonce, echostr)
    if ret == 0:
        return Response(content=sEchoStr.decode('utf-8'))
    else:
        print(sEchoStr)

@app.post("/bot")
async def recv(msg_signature: str, timestamp: str, nonce: str, request: Request, background_tasks: BackgroundTasks):
    #start_time = time.time()
    body = await request.body()
    ret, sMsg = wxcpt.DecryptMsg(body.decode('utf-8'), msg_signature, timestamp, nonce)
    decrypt_data = {}
    for node in list(fromstring(sMsg.decode('utf-8'))):
        decrypt_data[node.tag] = node.text
    user_query = decrypt_data.get('Content', '')
    background_tasks.add_task(consumer, user_query, decrypt_data)
    return Response(content="")




if __name__ == "__main__":
    uvicorn.run("coze_bot_api:app", port=port, host='0.0.0.0', reload=False,ssl_keyfile="./key.pem", ssl_certfile="./cert.pem")