coze_api/coze_bot_api_test.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 asyncio
import json
import re
import sys
import time
import uuid
from concurrent.futures import ThreadPoolExecutor
from xml.etree.ElementTree import fromstring

import requests
import uvicorn
from anyio import CapacityLimiter
from anyio.lowlevel import RunVar
from fastapi import FastAPI, Response, Request, Body, BackgroundTasks

from WXBizMsgCrypt3 import WXBizMsgCrypt
from commom import get_logger, request_id_context

timeout = 30

logger = get_logger()

# 加载配置文件
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()
executor = ThreadPoolExecutor(max_workers=200)


@app.on_event("startup")
def startup():
    print("start")
    RunVar("_default_thread_limiter").set(CapacityLimiter(210))


# 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": "object_string"
            }
        ]
    }
    res_create = requests.post(" https://api.coze.cn/v1/conversation/create", headers=req_head)
    conversation_id = res_create.json()["data"]["id"]
    res_chat = requests.post(f" https://api.coze.cn/v3/chat?conversation_id={conversation_id}", headers=req_head,
                             json=req_data)
    chat_id = res_chat.json()["data"]["id"]
    while True:
        # logger.info(f"{prompt=}")
        res_retrieve = requests.get(
            f" https://api.coze.cn/v3/chat/retrieve?chat_id={chat_id}&conversation_id={conversation_id}",
            headers=req_head)
        res_json = res_retrieve.json()
        # 首先判断网络状态是否为200
        if res_retrieve.status_code != 200:
            # logger.error(f"网络状态码失败，错误码：{res_retrieve.status_code}")
            coze_response = f"网络状态码失败，错误码：{res_retrieve.status_code}"
            return coze_response
        # 判断状态码是否为0
        if res_json["code"] != 0:
            # logger.error(f"API调用失败，错误码：{res_json['code']}")
            coze_response = f"API调用失败，错误码：{res_json['code']}"
            return coze_response
        # 打印并记录状态
        # logger.info(res_json["data"]["status"])
        status = res_json["data"]["status"]
        # 检查是否为错误状态
        error_statuses = {"failed", "requires_action", "canceled"}
        if status in error_statuses:
            error_message = res_json["data"]["last_error"]
            # logger.error(f"对话错误，状态：{status}，错误信息：{error_message}")
            coze_response = f"对话错误，状态：{status}，错误信息：{error_message}"
            return coze_response
        # 如果状态为completed，则获取消息
        if status == "completed":
            res_message = requests.get(
                f"https://api.coze.cn/v3/chat/message/list?chat_id={chat_id}&conversation_id={conversation_id}",
                headers=req_head)
            # 假设res_message是已经获取到的响应对象
            data = res_message.json()['data']

            # 使用列表推导式找到所有type为'answer'的记录，然后取最后一个
            last_answer_record = next((record for record in reversed(data) if record['type'] == 'answer'), None)
            coze_response = last_answer_record['content'].replace(" ", "")  # v3 删除图片url中的空格
            # coze_response = coze_response['data'][1]['content'].replace(" ", "")  # v3 删除图片url中的空格
            return coze_response
        time.sleep(1)


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},
        "image": {
            "media_id": "MEDIA_ID"
        },
        "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())
    logger.info(res.json())
    # return res.json()


# 问题传入字节服务器进行回答后发送给企业微信，行内服务器只进行接收然后发给字节，防止网络延迟
async def post_consumer_api(user_query, decrypt_data, request_id):
    data = {
        "user_query": user_query,
        "decrypt_data": decrypt_data,
        "request_id": request_id,
    }
    request_id_context.set(request_id)
    url = "https://101.126.81.2:18090/consumer"
    try:
        logger.info(f"post_consumer_api {user_query=}")

        def consumer(data):
            re = requests.post(url=url, json=data, verify=False)
            return re.text

        loop = asyncio.get_event_loop()
        result = await loop.run_in_executor(executor, consumer, data)
        logger.info(f"post_consumer_api {result=}")

        return result

        # t1 = threading.Thread(target=requests.post, kwargs={"url": url, "json": data, "verify": False})
        # t1.start()

        # response = requests.post(url, json=data, verify=False)  # 忽略SSL证书验证
        # response.raise_for_status()  # 检查响应状态码是否为200
        # logger.info(f"post_consumer_api 请求成功: {response.json()}")
    except requests.exceptions.RequestException as e:
        logger.error(f"post_consumer_api 请求失败: {e}")
        return "error"


# 创建一个字典来存储用户的bot_id状态
user_bot_id_mapping = {}

# 新增：存储用户是否已收到欢迎消息的状态
user_welcome_status = {}


@app.post("/consumer")
async def consumer(
        request_id: str = Body(...),
        user_query: str = Body(...),
        decrypt_data: dict = Body(...),
):
    # print(f"请求:{user_query}")
    # {
    #     "request_id":"",
    #     "user_query": "",
    #     "bot_id": "",
    #     "decrypt_data": {'ToUserName': 'ww5541cfeea51e3188', 'FromUserName': 'ZhuSanCheng', 'CreateTime': '1739241434', 'MsgType': 'text', 'Content': '产品介绍', 'MsgId': '7469985082764423235', 'AgentID': '1000002'}
    # }
    # body = body.decode()
    # body = json.loads(body)
    # request_id = body["request_id"]
    request_id_context.set(request_id)

    # user_query = body["user_query"] if body["user_query"] else "回答图片中的问题"
    # user_query = "回答图片中的问题"
    # decrypt_data = body["decrypt_data"]
    # bot_id = body["bot_id"]
    # decrypt_data = json.loads(decrypt_data)

    username = decrypt_data.get('FromUserName', '')
    agentid = decrypt_data.get('AgentID', '')
    msgtype = decrypt_data.get('MsgType', '')
    picurl = decrypt_data.get('PicUrl', '')
    event = decrypt_data.get('Event')
    msg_type = decrypt_data.get('MsgType')
    event_key = decrypt_data.get('EventKey')
    print(f"event_key: {event_key}")
    # 修改：仅在用户首次进入时发送欢迎消息
    if msg_type == 'event' and event == 'enter_agent':
        if not user_welcome_status.get(username, False):  # 检查是否首次进入
            welcome_message = "Hi，我是小微AI助手~你可以在屏幕底部“产品选择”菜单栏选择想咨询的产品，我会随时为你解答问题~"
            qiwei_post(username, welcome_message, agentid)
            user_welcome_status[username] = True  # 标记为已发送
        return Response(content="")

    # 根据用户发送的消息内容切换bot_id
    if msg_type == 'event' and event == 'click' and event_key == '#sendmsg#_0_0#7599826213209000':
        bot_id = "7456977536891846697"  # 当用户发送渝快振兴贷时使用的bot_id
        change_message = "您好，已切换为渝快振兴贷产品助手，请输入问题。"
        qiwei_post(username, change_message, agentid)
        user_bot_id_mapping[username] = bot_id  # 更新用户的bot_id状态
        return Response(content="")
    elif msg_type == 'event' and event == 'click' and event_key == '#sendmsg#_0_1#7599826213209001':
        bot_id = "7445101065005154313"  # 当用户发送房快贷时使用的bot_id
        change_message = "您好，已切换为房快贷产品助手，请输入问题。"
        qiwei_post(username, change_message, agentid)
        user_bot_id_mapping[username] = bot_id  # 更新用户的bot_id状态
        return Response(content="")
    else:
        # 如果用户之前已经选择了bot_id，则使用之前的bot_id
        bot_id = user_bot_id_mapping.get(username, "7456977536891846697")  # 默认bot_id

    qiwei_post(username, "我正在思考，请稍等...", agentid)
    logger.info("我正在思考，请稍等...")
    logger.info(f"consumer 请求:{user_query}")
    user_query = user_query if user_query else "回答图片中的问题"
    multimodal_content = [
        {"type": "text", "text": user_query},
        {"type": msgtype, "file_url": picurl},
        # {"type": "file", "file_id": "fileid2"},
        # {"type": "file", "file_url": "fileurl1"}
    ]
    user_query = json.dumps(multimodal_content, ensure_ascii=False)

    # 返回coze结果
    loop = asyncio.get_event_loop()
    future = loop.run_in_executor(executor, call_llm, user_query, bot_id, coze_access_token)

    try:
        coze_response = await asyncio.wait_for(future, timeout)
    except asyncio.exceptions.TimeoutError as err:
        return Response(content="timeout")

    # coze_response = call_llm(prompt=user_query, bot_id=bot_id, coze_access_token=coze_access_token)

    # answer = coze_response['messages'][1]['content']#v2
    # answer = coze_response['data'][1]['content'].replace(" ","") #v3 删除图片url中的空格

    ##处理图片链接
    image_counter = 1

    # 定义一个替换函数，用于在替换时添加序号
    def replace_with_counter(match):
        nonlocal image_counter
        alt_text = match.group(1) or f"示例图片{image_counter}"
        url = match.group(2)
        replacement = f'<a href="{url}">{alt_text}</a>'
        image_counter += 1
        return replacement

    # 将Markdown格式的图片链接转换为HTML格式的文字链接，并添加序号
    answer = re.sub(r'!\[(.*?)\]\((https?://[^)]+)\)', replace_with_counter, coze_response)

    # print(f"结果:{answer}")
    logger.info(f"结果:{answer}")
    qiwei_post(username, answer, agentid)
    return Response(content=answer)
    # 主动发结果给qiwei


    # choice_answer = "若需要切换产品助手请输入产品名称（渝快振兴贷、房快贷），无需切换请忽略"
    # qiwei_post(username, choice_answer, agentid)


@app.get("/ok")
def ok():
    return "ok"


@app.get("/bot")
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)
        logger.info(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()
    request_id = str(uuid.uuid4())
    request_id_context.set(request_id)
    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

    # print(body)
    # 获取用户发送的消息内容
    # decrypt_data = json.loads(body)
    user_query = decrypt_data.get('Content', '')
    # logger.info(f"start: {user_query}")

    # print(event_key)

    # 处理其他类型的消息
    logger.info(f"start: {user_query}")
    # background_tasks.add_task(post_consumer_api, user_query, decrypt_data, request_id, bot_id)

    result = await post_consumer_api(user_query, decrypt_data, request_id)
    # return Response(content="")
    return Response(content=result)

    # user_query = decrypt_data.get('Content', '')
    # logger.info(f"start: {user_query}")
    # background_tasks.add_task(post_consumer_api, user_query, decrypt_data, request_id)
    # data = {
    #     "user_query":user_query,
    #     "decrypt_data":decrypt_data
    # }
    # requests.post(
    #     f"https://101.126.81.2:18066/consumer",
    #     data=data)
    # return Response(content="")


if __name__ == "__main__":
    # coze_response = call_llm(prompt="房快贷是什么",bot_id=bot_id,coze_access_token = coze_access_token)
    # print(coze_response)
    try:
        port = sys.argv[1]
        int(port)
    except:
        port = 18090
    request_id_context.set("app start")
    logger.info(f'{port=}')
    print(f'{port=}')
    uvicorn.run("coze_bot_api_test:app", port=int(port), host='0.0.0.0', reload=False, ssl_keyfile="./key.pem",
                ssl_certfile="./cert.pem")