SDK

SDK

Seeknetic 提供以下客戶端 SDK,協助你快速把平台能力整合到應用中。

SDK 示例

按工作流程整理的 Python 和 JavaScript SDK 核心示例。

Python SDK

透過 PyPI 安裝,快速完成伺服器端整合。

↗ 官方套件頁面

安裝

pip install seeknetic

驗證

透過環境變數 SEEKNETIC_API_KEY 傳入 API Key。

NOTE: 請先安裝 ffmpeg,並確保 ffmpeg/ffprobe 已加入 PATH。

快速開始

安裝 SDK 後,先從最小文字 embedding 請求開始。

從 PyPI 安裝
bash
pip install seeknetic
文字 embedding

首個示例使用同步 Python 客戶端,只有在需要更高並發時再切到 async。

python
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 步 · 預處理並上傳
python
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 步 · 提交非同步任務
python
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 步 · 輪詢結果
python
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 步 · 預處理並上傳
python
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 步 · 提交非同步任務
python
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 步 · 輪詢結果
python
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
python
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 步 · 輪詢結果
python
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
python
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 步 · 輪詢結果
python
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 擴展
bash
pip install seeknetic[aiohttp]
結合 DefaultAioHttpClient 使用 AsyncSeeknetic
python
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())