xx-xhai / XhApi.py
friklogff
Add application file
9cb42b3
# -*- coding = utf-8 -*-
"""
# @Time : 2023/7/20 12:37
# @Author : CSDN:FriKlogff
# @File : XhApi.py
# @Software: PyCharm
# @Function: 星火大模型API
"""
import os
os.system("""python -m pip install -i https://mirrors.aliyun.com/pypi/simple/ --upgrade pip setuptools
pip install -i https://mirrors.aliyun.com/pypi/simple/ websocket
pip install -i https://mirrors.aliyun.com/pypi/simple/ websocket-client
pip install -i https://mirrors.aliyun.com/pypi/simple/ gradio
pip install -i https://mirrors.aliyun.com/pypi/simple/ sxtwl
""")
import _thread as thread # 导入线程模块
import base64 # 导入base64编码模块
import datetime # 导入datetime模块
import hashlib # 导入hashlib模块
import hmac # 导入hmac模块
import json # 导入json模块
from urllib.parse import urlparse # 从urllib.parse导入urlparse用于url解析
import ssl # 导入ssl模块
from datetime import datetime # 从datetime导入datetime类
from time import mktime # 从time导入mktime用于生成时间戳
from urllib.parse import urlencode # 从urllib.parse导入urlencode用于编码请求参数
from wsgiref.handlers import format_date_time # 从wsgiref.handlers导入format_date_time用于格式化时间
import websocket # 导入websocket模块
response_content = ""
# 请求参数类
class Ws_Param:
# 初始化
def __init__(self, APPID, APIKey, APISecret, gpt_url):
self.APPID = APPID # 应用ID
self.APIKey = APIKey # API Key
self.APISecret = APISecret # API Secret
self.host = urlparse(gpt_url).netloc # 从url解析出host
self.path = urlparse(gpt_url).path # 从url解析出path
self.gpt_url = gpt_url # 完整的url
# 生成签名和url的方法
def create_url(self):
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"
# 生成签名
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')
# 生成授权header
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')
# 生成url参数字典
v = {
"authorization": authorization,
"date": date,
"host": self.host
}
# 构造最终url
url = self.gpt_url + '?' + urlencode(v)
return url
# 收到websocket错误的处理
def on_error(ws, error):
print("### error:", error)
# 收到websocket关闭的处理
def on_close(ws):
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='')
global response_content
response_content += 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
ws.run_forever(sslopt={"cert_reqs": ssl.CERT_NONE})
return response_content