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())

JavaScript SDK

官方包页面
安装
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);