chatGPT_Web/SparkApi.py

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
import threading
from threading import Condition


class Ws_Param(object):
    # 初始化
    def __init__(self, APPID, APIKey, APISecret, gpt_url, condition):
        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
        self.resp = ''
        self.condition = condition


    # 生成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_message(self, ws, message):
        # print(message)
        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"]
            print(content, end='')
            self.resp += content
            if status == 2:
                ws.close()

    def on_close(self, ws, *args):
        print("### closed ###")
        # self.condition.notify()


# 收到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)
    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"]
        print(content, end='')
        if status == 2:
            ws.close()


def gen_params(appid, question):
    """
    通过appid和用户的提问来生成请参数
    """
    data = {
        "header": {
            "app_id": appid,
            "uid": "1234"
        },
        "parameter": {
            "chat": {
                "domain": "general",
                "temperature": 0.01,
                "max_tokens": 2048,
                "auditing": "default"
            }
        },
        "payload": {
            "message": {
                "text": [
                    # {"role": "user", "content": question},
                    {"role": "user", "content": question}


                ]
            }
        }
    }
    return data


def main(appid, api_key, api_secret, gpt_url, question):
    condition = threading.Condition()
    condition.acquire()
    wsParam = Ws_Param(appid, api_key, api_secret, gpt_url, condition)
    websocket.enableTrace(False)
    wsUrl = wsParam.create_url()
    print(wsUrl)
    ws = websocket.WebSocketApp(wsUrl, on_message=wsParam.on_message, on_error=on_error, on_close=wsParam.on_close, on_open=on_open)
    ws.appid = appid
    ws.question = question
    ws.run_forever(sslopt={"cert_reqs": ssl.CERT_NONE})
    # condition.wait()
    print("回答：" + wsParam.resp)
    return wsParam.resp

if __name__ == "__main__":
    # 测试时候在此处正确填写相关信息即可运行
    main(appid="3d9282da",
         api_secret="ZTRiNDQwMTRlOTlmZDQwMDUwYTdjMDM0",
         api_key="7c217b3a313f4b66fcc14a8e97f85103",
         gpt_url="ws://spark-api.xf-yun.com/v1.1/chat",
         # question='你是谁')
         # question='将给定语段转换为含义最相似的意图 。意图是如下列表中的一项。如果没有对应意图或者语段缺失则返回0.列表为{1、你喜欢的状态是什么。2、公司高管有谁。3、打印机怎么使用。4、员工的福利有哪些。5、公司的价值观是什么。6、工作时间是几点。7、公司的愿景是什么。8、公司的使命是什么。9、公司有哪些产品。10、员工守则是什么。11、介绍一下你们公司。12、洗手间怎么走。13、开门。14、关门。15、取消你的静音。16、调小你的音量。17、调大你的音量。18、把你的声音关掉。19、介绍你自己。20、介绍我。21、预约个访客。22、预约会议。23、大会议室的地点。24.小会议室的地点。25、打招呼。26.停止。27、找人。28、在不在。}。输出为JSON格式，键为number。不要输出任何说明和解释。语段为“[用户]：' \
         question='假设你是公司前台，你在与面前的人对话，语段代表对方的问话，首先生成一个应对，然后从列表中选出一个选项作为回答的内容，如果没有对应意图或者语段缺失则返回0。列表为{1、喜欢的状态。2、公司高管名单。3、打印机使用指南。4、员工的福利待遇。5、公司的价值观。6、工作时间。7、公司的愿景。8、公司的使命。9、公司产品。10、员工守则。11、洗手间位置。12、开门。13、关门。14、取消静音。15、调小音量。16、调大音量。17、把声音关掉。18、自我介绍。19、对方介绍。20、预约访客。21、预约会议。22、王总的位置。23、介绍一下你们公司。24、怎么联系财务。}。输出为JSON格式，键只有number，代表应对的编号。不要输出任何说明和解释。语段为：产品介绍')