Python Tkinter GUI界面调用梦AI文生图片API接口实战指南

背景

目前字节跳动公司提供了即梦AI的接口免费试用,但是并发量只有1,不过足够我们使用了。我这里想做个使用python+tkinter实现的GUI可视化界面客户端,这样就不用每次都登录官方网站去进行文生图片,当然文生视频,或者图生视频也是如此的逻辑。

实现思路

1. 编写一个窗口

2. 窗口放置一个文本输入框,一个按钮和一个消息对话框就可以了

最后界面如下:

3. 给按钮绑定事件函数,当用户输入文字提示词后,点击按钮则生成对应的图片

4. 获取生成的图片链接地址,然后通过打开图片将图片写入对话框中 

Tkinter的python实现代码 

import time
import tkinter
from tkinter import scrolledtext, END, RAISED
from io import BytesIO
import requests
from tttt.ai_chat import AI_chat
from PIL import Image, ImageTk
from tttt.play_voice import play_audio
from tttt.record_voice import record_audio
from tttt.streaming_asr_demo import test_one
from tttt.voice_generate import txt_to_voice
from tttt.即梦AI文生图片 import txt_to_image

# 创建根窗口
root = tkinter.Tk()
# 设置属性
root.title('即梦AI文生图片python客户端')
# 设置窗口大小
root.geometry("650x560")

# 文本历史记录
txt_history = scrolledtext.ScrolledText(root, width=80, height=30)
txt_history.tag_config('green', foreground='#008B00')
txt_history.tag_config('red', foreground='#FF0000')
txt_history.grid(column=0, row=0, columnspan=3, padx=10, pady=10)

# 标签
lbl = tkinter.Label(root, text="请输入文字:")
lbl.grid(column=0, row=1, padx=5, pady=5)

# 文本输入框
txt_input = tkinter.Entry(root, width=50)
txt_input.grid(column=1, row=1, padx=5, pady=5)

# 按钮样式
button_style = {
    "font": ("Arial", 10, "bold"),  # 字体样式
    "bg": "#4CAF50",  # 背景颜色
    "fg": "white",  # 文字颜色
    "activebackground": "#45a049",  # 按钮按下时的背景颜色
    "activeforeground": "white",  # 按钮按下时的文字颜色
    "relief": RAISED,  # 按钮边框样式
    "bd": 2,  # 按钮边框宽度
    "padx": 5,  # 按钮内部水平方向的填充
    "pady": 5  # 按钮内部垂直方向的填充
}



# 从网页下载图片并加载
def download_and_display_image(url):
    try:
        response = requests.get(url)
        response.raise_for_status()  # 确保请求成功
        image_data = BytesIO(response.content)
        image = Image.open(image_data)
        photo = ImageTk.PhotoImage(image)
        # 在历史消息对话框中插入图片
        txt_history.image = photo  # 保持对PhotoImage的引用,防止被垃圾回收
        txt_history.insert(END, "\n")
        txt_history.image_create(END, image=photo)
        txt_history.insert(END, "\n")
    except requests.RequestException as e:
        txt_history.insert(END, f"无法下载图片: {e}\n")


# 发送文生图片
def txt_generate_image():
    # 获取输入框文本内容
    user_message = txt_input.get()
    # 将用户输入的内容插入对话框中
    txt_history.insert(END, '即梦AI:' + time.strftime("%Y-%m-%d %H:%M:%S", time.localtime()) + '\n',
                       'purple')  # 将输入的内容添加到文本历史记录中
    txt_history.insert(END, "正在请求处理中..." + "\n")
    try:
        # 文生图片接口
        image_path = txt_to_image(user_message)
        if image_path:
            # 清空输入框
            txt_input.delete(0, END)
            # 清空所有历史记录
            txt_history.delete("1.0", END)
            # 插入最新消息
            txt_history.insert(END, "成功生成图片如下" + "\n")
        # 显示图片到历史对话框中
        download_and_display_image(image_path)
    except Exception as e:
        txt_history.insert(END, "后台处理失败,原因如下:" + "\n")
        txt_history.insert(END, str(e) + "\n")


btn = tkinter.Button(root, text="文生图片", command=txt_generate_image, **button_style)
btn.grid(column=1, row=3, padx=5, pady=5)

# 运行主窗口
root.mainloop()

火山引擎的即梦AI接口及改造

即梦AI文生图片接口文档

即梦AI-图像生成–即梦AI-火山引擎

http接口示例:

import json
import sys
import os
import base64
import datetime
import hashlib
import hmac
import requests


method = 'POST'
host = 'visual.volcengineapi.com'
region = 'cn-north-1'
endpoint = 'https://visual.volcengineapi.com'
service = 'cv'

def sign(key, msg):
    return hmac.new(key, msg.encode('utf-8'), hashlib.sha256).digest()

def getSignatureKey(key, dateStamp, regionName, serviceName):
    kDate = sign(key.encode('utf-8'), dateStamp)
    kRegion = sign(kDate, regionName)
    kService = sign(kRegion, serviceName)
    kSigning = sign(kService, 'request')
    return kSigning

def formatQuery(parameters):
    request_parameters_init = ''
    for key in sorted(parameters):
        request_parameters_init += key + '=' + parameters[key] + '&'
    request_parameters = request_parameters_init[:-1]
    return request_parameters

def signV4Request(access_key, secret_key, service, req_query, req_body):
    if access_key is None or secret_key is None:
        print('No access key is available.')
        sys.exit()

    t = datetime.datetime.utcnow()
    current_date = t.strftime('%Y%m%dT%H%M%SZ')
    # current_date = '20210818T095729Z'
    datestamp = t.strftime('%Y%m%d')  # Date w/o time, used in credential scope
    canonical_uri = '/'
    canonical_querystring = req_query
    signed_headers = 'content-type;host;x-content-sha256;x-date'
    payload_hash = hashlib.sha256(req_body.encode('utf-8')).hexdigest()
    content_type = 'application/json'
    canonical_headers = 'content-type:' + content_type + '\n' + 'host:' + host + \
        '\n' + 'x-content-sha256:' + payload_hash + \
        '\n' + 'x-date:' + current_date + '\n'
    canonical_request = method + '\n' + canonical_uri + '\n' + canonical_querystring + \
        '\n' + canonical_headers + '\n' + signed_headers + '\n' + payload_hash
    # print(canonical_request)
    algorithm = 'HMAC-SHA256'
    credential_scope = datestamp + '/' + region + '/' + service + '/' + 'request'
    string_to_sign = algorithm + '\n' + current_date + '\n' + credential_scope + '\n' + hashlib.sha256(
        canonical_request.encode('utf-8')).hexdigest()
    # print(string_to_sign)
    signing_key = getSignatureKey(secret_key, datestamp, region, service)
    # print(signing_key)
    signature = hmac.new(signing_key, (string_to_sign).encode(
        'utf-8'), hashlib.sha256).hexdigest()
    # print(signature)

    authorization_header = algorithm + ' ' + 'Credential=' + access_key + '/' + \
        credential_scope + ', ' + 'SignedHeaders=' + \
        signed_headers + ', ' + 'Signature=' + signature
    # print(authorization_header)
    headers = {'X-Date': current_date,
               'Authorization': authorization_header,
               'X-Content-Sha256': payload_hash,
               'Content-Type': content_type
               }
    # print(headers)

    # ************* SEND THE REQUEST *************
    request_url = endpoint + '?' + canonical_querystring

    print('\nBEGIN REQUEST++++++++++++++++++++++++++++++++++++')
    print('Request URL = ' + request_url)
    try:
        r = requests.post(request_url, headers=headers, data=req_body)
    except Exception as err:
        print(f'error occurred: {err}')
        raise
    else:
        print('\nRESPONSE++++++++++++++++++++++++++++++++++++')
        print(f'Response code: {r.status_code}\n')
        # 使用 replace 方法将 \u0026 替换为 &
        resp_str = r.text.replace("\\u0026", "&")
        print(f'Response body: {resp_str}\n')


if __name__ == "__main__":
    # 请求凭证,从访问控制申请
    access_key = 'AK*****'
    secret_key = '*****=='

    # 请求Query,按照接口文档中填入即可
    query_params = {
        'Action': 'CVProcess',
        'Version': '2022-08-31',
    }
    formatted_query = formatQuery(query_params)

    # 请求Body,按照接口文档中填入即可
    body_params = {
        "req_key": "******",
        # ......
    }
    formatted_body = json.dumps(body_params)
    
    signV4Request(access_key, secret_key, service,
                  formatted_query, formatted_body)

这里要求我们开通服务,并创建对应的AK和SK

账号登录-火山引擎欢迎登录火山引擎,火山引擎是字节跳动旗下的云服务平台。

账号登录-火山引擎

这里我已经开通了,自行选择开通试用就行,不需要点击正式调用,不然收费

 

接口改造

原来的接口只能查看是否调用成功,我们想获取图片地址,并将图片显示在对话框中,因此需要获取返回值,请求体里面加了"return_url":True,就是用来获取图片地址的。同时将认证的AK和SK也填充到自定义函数中,这样接口地址可以正常使用认证并返回我们要的结果。

import json
import sys
import os
import base64
import datetime
import hashlib
import hmac
import requests

method = 'POST'
host = 'visual.volcengineapi.com'
region = 'cn-north-1'
endpoint = 'https://visual.volcengineapi.com'
service = 'cv'


def sign(key, msg):
    return hmac.new(key, msg.encode('utf-8'), hashlib.sha256).digest()


def getSignatureKey(key, dateStamp, regionName, serviceName):
    kDate = sign(key.encode('utf-8'), dateStamp)
    kRegion = sign(kDate, regionName)
    kService = sign(kRegion, serviceName)
    kSigning = sign(kService, 'request')
    return kSigning


def formatQuery(parameters):
    request_parameters_init = ''
    for key in sorted(parameters):
        request_parameters_init += key + '=' + parameters[key] + '&'
    request_parameters = request_parameters_init[:-1]
    return request_parameters


def signV4Request(access_key, secret_key, service, req_query, req_body):
    if access_key is None or secret_key is None:
        print('No access key is available.')
        sys.exit()

    t = datetime.datetime.utcnow()
    current_date = t.strftime('%Y%m%dT%H%M%SZ')
    # current_date = '20210818T095729Z'
    datestamp = t.strftime('%Y%m%d')  # Date w/o time, used in credential scope
    canonical_uri = '/'
    canonical_querystring = req_query
    signed_headers = 'content-type;host;x-content-sha256;x-date'
    payload_hash = hashlib.sha256(req_body.encode('utf-8')).hexdigest()
    content_type = 'application/json'
    canonical_headers = 'content-type:' + content_type + '\n' + 'host:' + host + \
                        '\n' + 'x-content-sha256:' + payload_hash + \
                        '\n' + 'x-date:' + current_date + '\n'
    canonical_request = method + '\n' + canonical_uri + '\n' + canonical_querystring + \
                        '\n' + canonical_headers + '\n' + signed_headers + '\n' + payload_hash
    # print(canonical_request)
    algorithm = 'HMAC-SHA256'
    credential_scope = datestamp + '/' + region + '/' + service + '/' + 'request'
    string_to_sign = algorithm + '\n' + current_date + '\n' + credential_scope + '\n' + hashlib.sha256(
        canonical_request.encode('utf-8')).hexdigest()
    # print(string_to_sign)
    signing_key = getSignatureKey(secret_key, datestamp, region, service)
    # print(signing_key)
    signature = hmac.new(signing_key, (string_to_sign).encode(
        'utf-8'), hashlib.sha256).hexdigest()
    # print(signature)

    authorization_header = algorithm + ' ' + 'Credential=' + access_key + '/' + \
                           credential_scope + ', ' + 'SignedHeaders=' + \
                           signed_headers + ', ' + 'Signature=' + signature
    # print(authorization_header)
    headers = {'X-Date': current_date,
               'Authorization': authorization_header,
               'X-Content-Sha256': payload_hash,
               'Content-Type': content_type
               }
    # print(headers)

    # ************* SEND THE REQUEST *************
    request_url = endpoint + '?' + canonical_querystring

    print('\nBEGIN REQUEST++++++++++++++++++++++++++++++++++++')
    # print('Request URL = ' + request_url)
    try:
        r = requests.post(request_url, headers=headers, data=req_body)
    except Exception as err:
        print(f'error occurred: {err}')
        raise
    else:
        # print('\nRESPONSE++++++++++++++++++++++++++++++++++++')
        print(f'Response code: {r.status_code}\n')
        # 注释调原来的返回处理,我们不需要文本类型的,我们需要json格式的
        # 使用 replace 方法将 \u0026 替换为 &
        # resp_str = r.text.replace("\\u0026", "&")
        # print(f'Response body: {resp_str}\n')
        # 获取json类型的返回对象,可进行数据处理,获取返回的image_url
        print(r.json()['data']['image_urls'][0])
        return r.json()['data']['image_urls'][0]


def txt_to_image(prompt_text):
    # 请求凭证,从访问控制申请
    access_key = ''
    secret_key = ''

    # 请求Query,按照接口文档中填入即可
    query_params = {
        'Action': 'CVProcess',
        'Version': '2022-08-31',
    }
    formatted_query = formatQuery(query_params)

    # 请求Body,按照接口文档中填入即可
    body_params = {
        "req_key": "jimeng_high_aes_general_v21_L",
        "prompt": prompt_text,
        "return_url": True
    }
    formatted_body = json.dumps(body_params)

    image_url = signV4Request(access_key, secret_key, service,
                  formatted_query, formatted_body)
    return image_url


if __name__ == "__main__":
    txt_to_image('一张海报,上面文字写着:"新年快乐"')

请求的body体两个参数必填,因此在body_params中必须放置,其中req_key取固定值,prompt就是我们的输入框中的文本提示词。

测试结果:

到这里就大功告成了!可以愉快的玩耍了 

作者:xingdiango

物联沃分享整理
物联沃-IOTWORD物联网 » Python Tkinter GUI界面调用梦AI文生图片API接口实战指南

发表回复