0805try

SparkApi_jayce.py

@@ -0,0 +1,164 @@
+import _thread as thread
+import base64
+import datetime
+import hashlib
+import hmac
+import json
+from urllib.parse import urlparse
+import ssl
+from datetime import datetime
+from time import mktime
+from urllib.parse import urlencode
+from wsgiref.handlers import format_date_time
+
+import websocket
+
+result = ""
+
+class Ws_Param(object):
+    # 初始化
+    def __init__(self, APPID, APIKey, APISecret, gpt_url):
+        self.APPID = APPID
+        self.APIKey = APIKey
+        self.APISecret = APISecret
+        self.host = urlparse(gpt_url).netloc
+        self.path = urlparse(gpt_url).path
+        self.gpt_url = gpt_url
+
+    # 生成url
+    def create_url(self):
+        # 生成RFC1123格式的时间戳
+        now = datetime.now()
+        date = format_date_time(mktime(now.timetuple()))
+
+        # 拼接字符串
+        signature_origin = "host: " + self.host + "\n"
+        signature_origin += "date: " + date + "\n"
+        signature_origin += "GET " + self.path + " HTTP/1.1"
+
+        # 进行hmac-sha256进行加密
+        signature_sha = hmac.new(self.APISecret.encode('utf-8'), signature_origin.encode('utf-8'),
+                                 digestmod=hashlib.sha256).digest()
+
+        signature_sha_base64 = base64.b64encode(signature_sha).decode(encoding='utf-8')
+
+        authorization_origin = f'api_key="{self.APIKey}", algorithm="hmac-sha256", headers="host date request-line", signature="{signature_sha_base64}"'
+
+        authorization = base64.b64encode(authorization_origin.encode('utf-8')).decode(encoding='utf-8')
+
+        # 将请求的鉴权参数组合为字典
+        v = {
+            "authorization": authorization,
+            "date": date,
+            "host": self.host
+        }
+        # 拼接鉴权参数，生成url
+        url = self.gpt_url + '?' + urlencode(v)
+        # 此处打印出建立连接时候的url,参考本demo的时候可取消上方打印的注释，比对相同参数时生成的url与自己代码生成的url是否一致
+        return url
+
+
+# 收到websocket错误的处理
+def on_error(ws, error):
+    print("### error:", error)
+
+
+# 收到websocket关闭的处理
+def on_close(ws,  *args):
+    print("### closed ###")
+
+
+# 收到websocket连接建立的处理
+def on_open(ws):
+    thread.start_new_thread(run, (ws,))
+
+
+def run(ws, *args):
+    data = json.dumps(gen_params(appid=ws.appid, question=ws.question))
+    ws.send(data)
+
+
+# 收到websocket消息的处理
+def on_message(ws, message):
+    # print(message)
+    global result
+    data = json.loads(message)
+    code = data['header']['code']
+    if code != 0:
+        print(f'请求错误: {code}, {data}')
+        ws.close()
+    else:
+        choices = data["payload"]["choices"]
+        status = choices["status"]
+        content = choices["text"][0]["content"]
+
+        # by zxd
+        #print(content, end='')
+        result += content
+        #received_data.append(content)
+
+        if status == 2:
+            ws.close()
+
+
+def gen_params(appid, question):
+    """
+    通过appid和用户的提问来生成请参数
+    """
+    data = {
+        "header": {
+            "app_id": appid,
+            "uid": "1234"
+        },
+        "parameter": {
+            "chat": {
+                "domain": "general",
+                "random_threshold": 0.5,
+                "max_tokens": 2048,
+                "auditing": "default"
+            }
+        },
+        "payload": {
+            "message": {
+                "text": [
+                    {"role": "user", "content": question}
+                ]
+            }
+        }
+    }
+    return data
+
+
+def main(appid, api_key, api_secret, gpt_url, question):
+    wsParam = Ws_Param(appid, api_key, api_secret, gpt_url)
+    websocket.enableTrace(False)
+    wsUrl = wsParam.create_url()
+    ws = websocket.WebSocketApp(wsUrl, on_message=on_message, on_error=on_error, on_close=on_close, on_open=on_open)
+    ws.appid = appid
+    ws.question = question
+
+    # by zxd
+    result = ws.run_forever(sslopt={"cert_reqs": ssl.CERT_NONE})
+    return result
+
+
+def create(question):
+    global result
+    result = ""
+    main(appid="d2ff57e0",
+        api_secret="YjlmNDdkYjFmMGMzYjc5MmJiODFjN2Fi",
+        api_key="07963fbc530a42f4ad223517decfd5fe",
+        gpt_url="ws://spark-api.xf-yun.com/v1.1/chat",
+        question=question)
+    return result
+
+
+
+if __name__ == "__main__":
+    # 测试时候在此处正确填写相关信息即可运行
+    #print(create("你是谁"))
+    A = create("你是谁")
+    print(type(A))
+    print(A)
+
+

app.py

@@ -4,7 +4,7 @@ import openai
 import time
 import gradio as gr
 from threading import Thread #线程  用于定时器
-import testlinkspark
+import SparkApi_jayce
 
 #from assets.char_poses_base64 import ( #角色动作
  #   CHAR_IDLE_HTML, CHAR_THINKING_HTML, CHAR_TALKING_HTML)
@@ -79,17 +79,17 @@ def get_response(history, audio_input):
         question = 'hello'
     
     #answer = conv_model.run(question)
-    answer = testlinkspark.main(appid="d2ff57e0",
+    answer = SparkApi_jayce.create(appid="d2ff57e0",
             api_secret="YjlmNDdkYjFmMGMzYjc5MmJiODFjN2Fi",
             api_key="07963fbc530a42f4ad223517decfd5fe",
             gpt_url="ws://spark-api.xf-yun.com/v1.1/chat",
             question= question)
-    answer1 = str(answer)
-    LOGGER.info("\ndocument_response: %s", answer1)
-    print('\ndocument_response:', answer1)
+    #answer1 = str(answer)
+    LOGGER.info("\ndocument_response: %s", answer)
+    print('\ndocument_response:', answer)
 
     for trigger in GENERAL_RSPONSE_TRIGGERS:
-        if trigger in str(answer1):    
+        if trigger in answer:    
             MESSAGES.append({"role": "user", "content": question})
     #chat = openai.ChatCompletion.create(
     #            model="gpt-3.5-turbo", 
@@ -99,13 +99,12 @@ def get_response(history, audio_input):
     #            stop="\n"
     #            )
     #answer = chat.choices[0].message.content
-            MESSAGES.append({"role": "assistant", "content": answer1})
-            LOGGER.info("general_response: %s", answer1)
-            print('\ngeneral_response:', answer1)
-
-    AUDIO_HTML = text_to_speech_gen(answer1)
-    history[-1][1] = answer1
+            MESSAGES.append({"role": "assistant", "content": answer})
+            LOGGER.info("general_response: %s", answer)
+            print('\ngeneral_response:', answer)
 
+    AUDIO_HTML = text_to_speech_gen(answer)
+    history[-1][1] = answer
     return history, AUDIO_HTML
 
 # buzz_usr_proc = Thread(target=idle_timer)