Python SDK
透過 PyPI 安裝,快速完成伺服器端整合。
安裝
pip install seeknetic 驗證
透過環境變數 SEEKNETIC_API_KEY 傳入 API Key。
快速開始
安裝 SDK 後,先從最小文字 embedding 請求開始。
從 PyPI 安裝
pip install seeknetic 文字 embedding
首個示例使用同步 Python 客戶端,只有在需要更高並發時再切到 async。
import os
from seeknetic import Seeknetic
client = Seeknetic(
api_key=os.environ["SEEKNETIC_API_KEY"],
)
response = client.video_embedding.encode_text(
text="a person walking on the beach"
)
print(len(response.embedding)) 本機影片工作流程
推薦這是處理本機檔案的推薦方式。使用 preprocess_and_upload(...) 前,請先安裝 ffmpeg。
原始影片會保留在本機,這個工作流程只上傳預處理後的 tensors。service="embedding" 只應配合 video_embedding.encode_video.submit_async(...);service="tagging" 只應配合 video_tagging.submit_async_job(...)。
Embedding · 第 1 步 · 預處理並上傳
import os
from seeknetic import Seeknetic
client = Seeknetic(api_key=os.environ["SEEKNETIC_API_KEY"])
upload = client.preprocess_and_upload(
video_path="sample.mp4",
service="embedding",
request_id="emb-task-0001",
start_ms=0,
end_ms=10_000,
)
print(upload.request_id)
print(upload.r2_keys) Embedding · 第 2 步 · 提交非同步任務
import os
from seeknetic import Seeknetic
client = Seeknetic(api_key=os.environ["SEEKNETIC_API_KEY"])
upload = client.preprocess_and_upload(
video_path="sample.mp4",
service="embedding",
request_id="emb-task-0001",
)
job = client.video_embedding.encode_video.submit_async(
tensor={
"video_input_key": upload.r2_keys["video_input"],
"audio_input_key": upload.r2_keys["audio_input"],
},
request_id=upload.request_id,
)
print(job.request_id, job.status) Embedding · 第 3 步 · 輪詢結果
import os
from seeknetic import Seeknetic
client = Seeknetic(api_key=os.environ["SEEKNETIC_API_KEY"])
status = client.async_status.retrieve("emb-task-0001")
print(status.status)
print(status.embedding) Tagging · 第 1 步 · 預處理並上傳
import os
from seeknetic import Seeknetic
client = Seeknetic(api_key=os.environ["SEEKNETIC_API_KEY"])
upload = client.preprocess_and_upload(
video_path="sample.mp4",
service="tagging",
request_id="tag-task-0001",
start_ms=0,
end_ms=10_000,
)
print(upload.request_id)
print(upload.r2_keys) Tagging · 第 2 步 · 提交非同步任務
import os
from seeknetic import Seeknetic
client = Seeknetic(api_key=os.environ["SEEKNETIC_API_KEY"])
upload = client.preprocess_and_upload(
video_path="sample.mp4",
service="tagging",
request_id="tag-task-0001",
)
job = client.video_tagging.submit_async_job(
tensor={
"video_input_key": upload.r2_keys["video_input"],
"video_kwargs_key": upload.r2_keys["video_kwargs"],
},
request_id=upload.request_id,
)
print(job.request_id, job.status) Tagging · 第 3 步 · 輪詢結果
import os
from seeknetic import Seeknetic
client = Seeknetic(api_key=os.environ["SEEKNETIC_API_KEY"])
status = client.async_status.retrieve("tag-task-0001")
print(status.status)
print(status.tags) 遠端或已上傳影片工作流程
如果不需要本機預處理,可以直接提交 video_url。
Embedding · 第 1 步 · 提交 video_url
import os
from seeknetic import Seeknetic
client = Seeknetic(api_key=os.environ["SEEKNETIC_API_KEY"])
job = client.video_embedding.encode_video.submit_async(
video_url="https://example.com/video.mp4",
)
print(job.request_id, job.status) Embedding · 第 2 步 · 輪詢結果
import os
from seeknetic import Seeknetic
client = Seeknetic(api_key=os.environ["SEEKNETIC_API_KEY"])
status = client.async_status.retrieve("emb-task-0001")
print(status.status)
print(status.embedding) Tagging · 第 1 步 · 提交 video_url
import os
from seeknetic import Seeknetic
client = Seeknetic(api_key=os.environ["SEEKNETIC_API_KEY"])
job = client.video_tagging.submit_async_job(
video_url="https://example.com/video.mp4",
)
print(job.request_id, job.status) Tagging · 第 2 步 · 輪詢結果
import os
from seeknetic import Seeknetic
client = Seeknetic(api_key=os.environ["SEEKNETIC_API_KEY"])
status = client.async_status.retrieve("tag-task-0001")
print(status.status)
print(status.tags) 非同步客戶端
README 亦提供基於 aiohttp 的更高並發用法。
安裝 aiohttp 擴展
pip install seeknetic[aiohttp] 結合 DefaultAioHttpClient 使用 AsyncSeeknetic
import os
import asyncio
from seeknetic import DefaultAioHttpClient
from seeknetic import AsyncSeeknetic
async def main() -> None:
async with AsyncSeeknetic(
api_key=os.environ.get("SEEKNETIC_API_KEY"),
http_client=DefaultAioHttpClient(),
) as client:
response = await client.video_embedding.encode_text(
text="a person walking on the beach"
)
print(len(response.embedding))
asyncio.run(main())