SDK

你的管線。 我們的模型。

本地預處理、異步模型任務與托管影片工作流程的 Python 和 JavaScript 示例。

程式碼、API 標識符和請求值保留原有技術語言。

開源工具 / Shot Detection

從鏡頭分割開始。

Shot Detection 是獨立的 Python 鏡頭分割工具,傳回每個鏡頭的起止時間戳。配合 Seeknetic SDK,建立按鏡頭處理影片的工作流程。

  1. 01

    分割鏡頭

    使用 Shot Detection 分析影片,取得每個鏡頭的開始與結束時間。

  2. 02

    編碼或標註

    將鏡頭時間範圍傳入 SDK 的本機預處理流程,再提交向量編碼或標籤任務。

  3. 03

    組織檢索

    在應用程式中儲存傳回的向量與元資料,並保留原始檔案、時間範圍與檢索結果的關聯。

Shot Detection 負責鏡頭分割。模型編碼與標註需另行安裝 Seeknetic SDK、開通 API 存取,並按公開 API 價格計費。

安裝
pip install seeknetic

請先安裝 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())
安裝
npm install seeknetic

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

快速開始

安裝 SDK 後,先從 README 中的最小文字 embedding 示例開始。

從 npm 安裝

bash
npm install seeknetic

文字 embedding

javascript
import Seeknetic from 'seeknetic';

const client = new Seeknetic({
  apiKey: process.env['SEEKNETIC_API_KEY'],
});

const response = await client.videoEmbedding.encodeText({
  text: 'a person walking on the beach',
});

console.log(response.embedding.length);

本機影片工作流程

這是處理本機檔案的推薦方式。使用 preprocessAndUpload(...) 前,請先安裝 ffmpeg。

原始影片會保留在本機,這個工作流程只上傳預處理後的 tensors。service: "embedding" 只應配合 videoEmbedding.encodeVideo.submitAsync(...);service: "tagging" 只應配合 videoTagging.submitAsyncJob(...)。

Embedding · 第 1 步 · 預處理並上傳

typescript
import Seeknetic from 'seeknetic';

const client = new Seeknetic({
  apiKey: process.env['SEEKNETIC_API_KEY'],
});

const upload = await client.preprocessAndUpload({
  videoPath: 'sample.mp4',
  service: 'embedding',
  requestId: 'emb-task-0001',
  startMs: 0,
  endMs: 10_000,
});

console.log(upload.request_id);
console.log(upload.r2_keys);

Embedding · 第 2 步 · 提交非同步任務

typescript
import Seeknetic from 'seeknetic';

const client = new Seeknetic({
  apiKey: process.env['SEEKNETIC_API_KEY'],
});

const upload = await client.preprocessAndUpload({
  videoPath: 'sample.mp4',
  service: 'embedding',
  requestId: 'emb-task-0001',
});

const job = await client.videoEmbedding.encodeVideo.submitAsync({
  tensor: {
    video_input_key: upload.r2_keys['video_input'],
    audio_input_key: upload.r2_keys['audio_input'],
  },
  request_id: upload.request_id,
});

console.log(job.request_id, job.status);

Embedding · 第 3 步 · 輪詢結果

typescript
import Seeknetic from 'seeknetic';

const client = new Seeknetic({
  apiKey: process.env['SEEKNETIC_API_KEY'],
});

const status = await client.asyncStatus.retrieve('emb-task-0001');

console.log(status.status);
console.log(status.embedding);

Tagging · 第 1 步 · 預處理並上傳

typescript
import Seeknetic from 'seeknetic';

const client = new Seeknetic({
  apiKey: process.env['SEEKNETIC_API_KEY'],
});

const upload = await client.preprocessAndUpload({
  videoPath: 'sample.mp4',
  service: 'tagging',
  requestId: 'tag-task-0001',
  startMs: 0,
  endMs: 10_000,
});

console.log(upload.request_id);
console.log(upload.r2_keys);

Tagging · 第 2 步 · 提交非同步任務

typescript
import Seeknetic from 'seeknetic';

const client = new Seeknetic({
  apiKey: process.env['SEEKNETIC_API_KEY'],
});

const upload = await client.preprocessAndUpload({
  videoPath: 'sample.mp4',
  service: 'tagging',
  requestId: 'tag-task-0001',
});

const job = await client.videoTagging.submitAsyncJob({
  tensor: {
    video_input_key: upload.r2_keys['video_input'],
    video_kwargs_key: upload.r2_keys['video_kwargs'],
  },
  request_id: upload.request_id,
});

console.log(job.request_id, job.status);

Tagging · 第 3 步 · 輪詢結果

typescript
import Seeknetic from 'seeknetic';

const client = new Seeknetic({
  apiKey: process.env['SEEKNETIC_API_KEY'],
});

const status = await client.asyncStatus.retrieve('tag-task-0001');

console.log(status.status);
console.log(status.tags);

遠端或已上傳影片工作流程

如果不需要本機預處理,可以直接提交 video_url。

Embedding · 第 1 步 · 提交 video_url

typescript
import Seeknetic from 'seeknetic';

const client = new Seeknetic({
  apiKey: process.env['SEEKNETIC_API_KEY'],
});

const job = await client.videoEmbedding.encodeVideo.submitAsync({
  video_url: 'https://example.com/video.mp4',
});

console.log(job.request_id, job.status);

Embedding · 第 2 步 · 輪詢結果

typescript
import Seeknetic from 'seeknetic';

const client = new Seeknetic({
  apiKey: process.env['SEEKNETIC_API_KEY'],
});

const status = await client.asyncStatus.retrieve('emb-task-0001');

console.log(status.status);
console.log(status.embedding);

Tagging · 第 1 步 · 提交 video_url

typescript
import Seeknetic from 'seeknetic';

const client = new Seeknetic({
  apiKey: process.env['SEEKNETIC_API_KEY'],
});

const job = await client.videoTagging.submitAsyncJob({
  video_url: 'https://example.com/video.mp4',
});

console.log(job.request_id, job.status);

Tagging · 第 2 步 · 輪詢結果

typescript
import Seeknetic from 'seeknetic';

const client = new Seeknetic({
  apiKey: process.env['SEEKNETIC_API_KEY'],
});

const status = await client.asyncStatus.retrieve('tag-task-0001');

console.log(status.status);
console.log(status.tags);