app.py

#该应用创建工具共包含三个区域，顶部工具栏，左侧代码区，右侧交互效果区，其中右侧交互效果是通过左侧代码生成的，存在对照关系。
#顶部工具栏：运行、保存、新开浏览器打开、实时预览开关，针对运行和在浏览器打开选项进行重要说明：
#[运行]：交互效果并非实时更新，代码变更后，需点击运行按钮获得最新交互效果。
#[在浏览器打开]：新建页面查看交互效果。
#以下为应用创建工具的示例代码

import gradio as gr
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


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):
    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='')
        with open('output.txt', 'a', encoding='utf-8') as f:
            f.write(content)
        if status == 2:
            ws.close()


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


def score(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})


def greet(appid, api_secret, api_key, topic, answer):
    with open('output.txt', 'w') as f:
        pass
    if appid and api_secret and api_key:
        appid = appid
        api_secret = api_secret
        api_key = api_key
    else:
        appid="39951d99"
        api_secret="ZjVhNmFiMjg2YWExODlmNjhjMDliOWMz"
        api_key="1CopZeypHcpbuerDSNGYT8ryNJE3izi3Uu"
    question = f"作为专业的雅思考试作文指导老师，需要你根据雅思考试第二篇作文的题目要求：{topic}为学生的答案：{answer}分别从Task Response、Coherence and Cohesion、Lexical Resource、Grammatical Range and Accuracy四项评分角度从0、0.5、1、1.5、2、2.5、3、3.5、4、4.5、5、5.5、6中选一个数值为考生的答案给出各个角度的分数结果，并结合学生答案的具体句子或词汇给出具体的评分依据和修改意见，最后结合以上四项评分结果，严格从0、0.5、1、1.5、2、2.5、3、3.5、4、4.5、5、5.5、6以上这13个数值中选择一个数值作为该考生答案的综合得分，并给出综合的评分依据和修改意见。你要严格按照以下模板格式进行回复：Task Response项分数结果:Task Response项角度润色意见:Coherence and Cohesion项分数结果:Coherence and Cohesion项角度润色意见:Lexical Resource项分数结果:Lexical Resource项角度润色意见:Grammatical Range and Accuracy项分数结果:Grammatical Range and Accuracy项角度润色意见:综合得分:综合润色意见:"
    print(question)
    score(appid=appid,
         api_secret=api_secret,  
         api_key=api_key,  
         gpt_url="ws://spark-api.xf-yun.com/v2.1/chat",
         question=question)
    with open('output.txt', 'r') as f:
        opinions = f.read()
    return opinions

with gr.Blocks() as demo:
    gr.Markdown(
    """
    # 欢迎使用雅思大作文评分助手!
    若您没有讯飞星火大模型API接口权限，则将使用作者的API，额度有限，请您尽量使用自己的API接口信息。
    """)
    # 设置输入组件
    with gr.Row() as row:
        appid = gr.Textbox(label="请输入您的appid")
        api_secret = gr.Textbox(label="请输入您的api_secret")
        api_key = gr.Textbox(label="请输入您的api_key")
    with gr.Row() as row:
        with gr.Column():
            gr.Markdown("## 作文题目及答案输入")
            topic = gr.Textbox(label="请输入作文题目", lines=5)
            answer = gr.Textbox(label="请输入您的答案",lines=25)
        with gr.Column():
            # 设置输出组件
            greet_btn = gr.Button("提交作文题目及答案")  # 设置按钮
            output = gr.TextArea(label="评分及润色意见", lines=33)
    # 设置按钮点击事件
    greet_btn.click(fn=greet, inputs=[appid, api_secret, api_key, topic, answer], outputs=output)

demo.launch()