大约 3 分钟约 1000 字
python
快速开始
可以使用pip来快速安装
pip install zyjj-open-sdk
访问 智游剪辑网页版 获取你的秘钥信息
这里以文章翻译
为例,使用方法如下
from zyjj_open_sdk import Client
# 初始化全局客户端(只需要初始化一次)
client = Client('sk-xxx')
# 直接调用文章翻译功能并获取返回结果
res = client.text.article_translate(
user="hello word",
language="中文"
).execute()
print(res.text)
文件上传
如果涉及到文件上传,需要使用FileObject
对象,该对象支持三种初始化方式
- 从本地路径初始化
- 从bytes初始化
- 从url初始化
这里以ncm转mp3
为例,我们可以这样使用
from zyjj_open_sdk import Client, FileObject
# 初始化全局客户端(只需要初始化一次)
client = Client('sk-xxx')
# 下面三种方法选一种即可
# 1.通过本地路径上传
file = FileObject.from_path('xxx.ncm')
# 2.通过bytes上传,注意需要带上文件名
file = FileObject.from_bytes('tmp.ncm', b'')
# 3.可以通过url上传
file = FileObject.from_url('https://xxx.com/xxx.ncm')
res = client.tool.ncm_to_mp3(ncm=file).execute()
print(res.mp3)
执行方式
目前sdk支持4种调用方式
部分任务不支持同步调用(所有任务均支持异步调用),请以文档说明为准
- 同步调用
- 异步等待模式
- 异步回调模式
- 异步查询模式
from zyjj_open_sdk import Client
import time
# 初始化全局客户端(只需要初始化一次)
client = Client('sk-xxx')
# 使用时会返回一个可执行对象,此时支持初始化了任务数据,不会立即执行
execute = client.text.article_translate("hello word", "中文")
# 1.我们可以直接同步执行获取执行结果,使用最简单
res = execute.execute()
print(res.text)
# 2.使用异步阻塞模式,异步阻塞等待任务完成,wait可以传入一个回调函数,用于进度监听
res = execute.execute_async().wait(lambda i: print(i))
print(res.text)
# 可以简化为下面这样的方式
res = execute.execute_async_wait(lambda i: print(i))
# 3.使用异步监听模型,不会阻塞流程,需要通过回调的方式来获取结果
execute.execute_async().listener(
on_progress=lambda i: print(i), # 任务进度回调
on_success=lambda data: print(data.text), # 任务执行成功回调
on_error=lambda e: print(e) # 任务执行失败回调
)
# 4.我们可以异步查询模式
task = execute.execute_async()
while True:
if task.status == 3: # 任务执行成功
print(task.output.text) # 打印任务结果
elif task.status == 4: # 任务执行失败
print(task.err) # 打印错误信息
else: # 其他情况为正在执行,这里可以打印执行进度
print(task.progress)
time.sleep(1) # 我们可以每秒轮询一次
实时结果
对于部分文字生成类的任务,我们可以通过详情回调获取实时生成结果,不需要等到全部生成完毕才显示,这样可以提高用户体验性。 目前详情回调只能通过异步任务触发,有两种方式
from zyjj_open_sdk import Client
# 初始化全局客户端(只需要初始化一次)
client = Client('sk-xxx')
# 使用时会返回一个可执行对象,此时支持初始化了任务数据,不会立即执行
execute = client.text.article_translate("hello word", "中文")
# 1. 直接通过wait的时候监听
execute.execute_async_wait(
detail_callback=lambda i: print(i), # 获取实时返回结果
)
# 2.直接通过监听器监听
execute.execute_async().listener(
on_detail=lambda i: print(i), # 获取实时返回结果
)
实际效果如下
{'text': ''}
{'text': '你好'}
{'text': '你好,'}
{'text': '你好,世界'}
{'text': '你好,世界'}
异步IO
为了让提高并发性,你还可以使用异步客户端,支持asyncio语法,使用方式和同步客户端基本一致
from zyjj_open_sdk import AsyncClient, FileObject
import asyncio
# 初始化异步客户端(只需要初始化一次)
client = AsyncClient('sk-xxx')
# 需要包装为一个异步函数
async def translate():
# 1.使用同步方式创建任务
res = await client.text.article_translate(
user="hello word",
language="中文"
).execute()
print(res.text)
# 2.使用异步方式创建任务
res = await client.text.article_translate(
user="hello word",
language="中文"
).execute_async_wait(progress_callback=lambda i: print(f"进度 {i}"))
print(res.text)
# 在事件循环中运行我们的任务
asyncio.run(translate())