行空板之智能互动艺术展示大屏

        以上代码，要实现需修改扩展库存“libraries”——“Spark.py”。文件所在路径：C:\Users\lenovo\AppData\Local\DFScratch\extensions\ac000108-spark-thirdex\python\libraries

        修改“imageBase”，成为类变量。是从讯飞返回的图片base64编码字符串。下面是修改后的“Spark.py”文件内容。

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")
复制代码