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

本地视频工作流

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

Node.js SDK

通过 npm 安装，支持 JavaScript/TypeScript 工作流。

↗ 官方包页面

安装

npm install seeknetic

鉴权

通过环境变量 SEEKNETIC_API_KEY 传入 API Key。

NOTE: 请先安装 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);

本地视频工作流

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

Python SDK

Node.js SDK

SDK 示例

Python SDK

快速开始

从 PyPI 安装

文本 embedding

本地视频工作流

Embedding · 第 1 步 · 预处理并上传

Embedding · 第 2 步 · 提交异步任务

Embedding · 第 3 步 · 轮询结果

Tagging · 第 1 步 · 预处理并上传

Tagging · 第 2 步 · 提交异步任务

Tagging · 第 3 步 · 轮询结果

远程或已上传视频工作流

Embedding · 第 1 步 · 提交 video_url

Embedding · 第 2 步 · 轮询结果

Tagging · 第 1 步 · 提交 video_url

Tagging · 第 2 步 · 轮询结果

异步客户端

安装 aiohttp 扩展

结合 DefaultAioHttpClient 使用 AsyncSeeknetic

Node.js SDK

快速开始

从 npm 安装

文本 embedding

本地视频工作流

Embedding · 第 1 步 · 预处理并上传

Embedding · 第 2 步 · 提交异步任务

Embedding · 第 3 步 · 轮询结果

Tagging · 第 1 步 · 预处理并上传

Tagging · 第 2 步 · 提交异步任务

Tagging · 第 3 步 · 轮询结果

远程或已上传视频工作流

Embedding · 第 1 步 · 提交 video_url

Embedding · 第 2 步 · 轮询结果

Tagging · 第 1 步 · 提交 video_url

Tagging · 第 2 步 · 轮询结果