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-----------------------------
"""
import time

from django.db.models.fields import return_None
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 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 ={
        "bot_id": bot_id,
        "user_id": "123456789",
        "stream": False,
        "auto_save_history": True,
        "additional_messages": [
            {
                "role": "user",
                "content": prompt,
                "content_type": "text"
            }
        ]
    }
    res_create = requests.post(" https://api.coze.cn/v1/conversation/create", headers=req_head)
    coversition_id = res_create.json()["data"]["id"]
    res_chat = requests.post(f" https://api.coze.cn/v3/chat?conversation_id={coversition_id}", headers=req_head,json=req_data)
    chat_id = res_chat.json()["data"]["id"]
    while True:
        res_retrieve = requests.get(f" https://api.coze.cn/v3/chat/retrieve?chat_id={chat_id}&conversation_id={coversition_id}", headers=req_head)
        print(res_retrieve.json()["data"]["status"])
        status = res_retrieve.json()["data"]["status"]
        if status == "completed":
            res_message = requests.get(f" https://api.coze.cn/v3/chat/message/list?chat_id={chat_id}&conversation_id={coversition_id}", headers=req_head)
            # print(res_message.json())
            return res_message.json()
        time.sleep(1)
    # 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['data'][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")