以上代码，要实现需修改扩展库存“libraries”——“Spark.py”。文件所在路径：C:\Users\lenovo\AppData\Local\DFScratch\extensions\ac000108-spark-thirdex\python\libraries
import _thread as thread
import os
import base64
import hashlib
import hmac
import json
from urllib.parse import urlparse , urlencode
import ssl
from datetime import datetime
from time import mktime
from wsgiref.handlers import format_date_time
import websocket
import openpyxl
from concurrent.futures import ThreadPoolExecutor, as_completed
import requests
from PIL import Image
class Spark:
def __init__(self,appid,api_secret,api_key,Personality):
self.appid = appid
self.api_secret = api_secret
self.api_key = api_key
self.gpt_url = "wss://spark-api.xf-yun.com/v3.5/chat"
self.tti_url = "https://spark-api.cn-huabei-1.xf-yun.com/v2.1/tti"
self.domain = "generalv3.5"
self.Personality = Personality
self.msgs = ""
self.history = ""
self.flag = True
self.imageBase=""
# 收到websocket错误的处理
def on_error(self,ws, error):
#print("### error:", error)
pass
# 收到websocket关闭的处理
def on_close(self,ws):
#print("### closed ###")
pass
# 收到websocket连接建立的处理
def on_open(self,ws):
#print("### open ###")
thread.start_new_thread(self.run, (ws,))
def run(self,ws, *args):
#print("### run ###")
data = json.dumps(self.gen_params(appid=ws.appid, query=ws.query, domain=ws.domain))
ws.send(data)
# 收到websocket消息的处理
def on_message(self,ws, 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"]
self.msgs = self.msgs + content
#print(content,end='')
if status == 2:
self.flag = False
ws.close()
ws.keep_running = False
def gen_params(self,appid, query, domain):
text = []
text.extend(self.Personality)
text.extend(self.history)
text.append({"role": "user", "content": query})
data = {
"header": {
"app_id": appid,
"uid": "1234"
},
"parameter": {
"chat": {
"domain": domain,
"temperature": 0.5,
"max_tokens": 4096,
"auditing": "default"
}
},
"payload": {
"message": {
"text": text
}
}
}
return data
def ask(self,prompt,history):
self.flag = True
self.msgs = ""
self.history = history
try:
host = urlparse(self.gpt_url).netloc
path = urlparse(self.gpt_url).path
websocket.enableTrace(False)
now = datetime.now()
date = format_date_time(mktime(now.timetuple()))
signature_origin = "host: " + host + "\n"
signature_origin += "date: " + date + "\n"
signature_origin += "GET " + path + " HTTP/1.1"
signature_sha = hmac.new(self.api_secret.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.api_key}", 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": host
}
wsUrl = self.gpt_url + '?' + urlencode(v)
ws = websocket.WebSocketApp(wsUrl, on_message=self.on_message, on_open=self.on_open, on_error=self.on_error, on_close=self.on_close )
ws.appid = self.appid
ws.query = prompt
ws.domain = self.domain
ws.history = history
ws.run_forever(sslopt={"cert_reqs": ssl.CERT_NONE})
while(self.flag):
pass
return self.msgs
except:
pass
def tti(self,prompt):
stidx = self.tti_url.index("://")
host = self.tti_url[stidx + 3:]
schema = self.tti_url[:stidx + 3]
edidx = host.index("/")
if edidx <= 0:
raise AssembleHeaderException("invalid request url:" + self.tti_url)
path = host[edidx:]
host = host[:edidx]
now = datetime.now()
date = format_date_time(mktime(now.timetuple()))
signature_origin = "host: {}\ndate: {}\n{} {} HTTP/1.1".format(host, date, 'POST', path)
signature_sha = hmac.new(self.api_secret.encode('utf-8'), signature_origin.encode('utf-8'),
digestmod=hashlib.sha256).digest()
signature_sha = base64.b64encode(signature_sha).decode(encoding='utf-8')
authorization_origin = "api_key="%s", algorithm="%s", headers="%s", signature="%s"" % (
self.api_key, "hmac-sha256", "host date request-line", signature_sha)
authorization = base64.b64encode(authorization_origin.encode('utf-8')).decode(encoding='utf-8')
values = {
"host": host,
"date": date,
"authorization": authorization
}
url = self.tti_url + "?" + urlencode(values)
print(url)
body= {
"header": {
"app_id": self.appid,
"uid":"123456789"
},
"parameter": {
"chat": {
"domain": "general",
"width": 1280,
"height": 720
}
},
"payload": {
"message":{
"text":[
{
"role":"user",
"content":prompt
}
]
}
}
}
response = requests.post(url,json=body,headers={'content-type': "application/json"}).text
data = json.loads(response)
code = data['header']['code']
if code != 0:
print(f'请求错误: {code}, {data}')
else:
text = data["payload"]["choices"]["text"]
imageContent = text[0]
self.imageBase = imageContent["content"]
imgdata = base64.b64decode(self.imageBase)
with open("1.jpg",'wb') as f:
f.write(imgdata)
img = Image.open("1.jpg")
# 旋转方式二
img2 = img.rotate(90,expand=True) # 自定义旋转度数
img2 = img2.resize((240, 320)) # 改变图片尺寸
img2.save("1.png")
复制代码 Mind+ 连接行空板，在终端中使用命令， cd mindplus/.lib/thirdExtension/进入第三方扩展文件夹，使用 ls 命令查看是否存在 ac000108-spark-thirdex文件夹。如果存在，使用 rm -r ac000108-spark-thirdex删除。再次运行图形化程序， Mind+ 会将修改后的扩展，下载到行空板中。
置顶卡
变色卡
千斤顶
活跃会员
宣传大使
创客造
编辑选择奖
志“童”道合
编辑选择奖
编辑选择奖
