init

coze_bot_api.py

@@ -0,0 +1,124 @@
+#!/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")