从 PyPI 安装
pip install seeknetic 本地预处理、异步模型任务与托管视频工作流的 Python 和 JavaScript 示例。
代码、API 标识符和请求值保留原有技术语言。
Shot Detection 是独立的 Python 镜头分割工具,返回每个镜头的起止时间戳。搭配 Seeknetic SDK,构建按镜头处理视频的工作流。
使用 Shot Detection 分析视频,取得每个镜头的开始与结束时间。
将镜头时间范围传入 SDK 的本地预处理流程,再提交向量编码或标签任务。
在应用中保存返回的向量与元数据,并保留原文件、时间范围与检索结果的关联。
Shot Detection 负责镜头分割。模型编码与标注需另行安装 Seeknetic SDK、开通 API 访问,并按公开 API 价格计费。
pip install seeknetic 请先安装 ffmpeg,并确保 ffmpeg/ffprobe 已加入 PATH。
安装 SDK 后,先从最小文本 embedding 请求开始。
pip install seeknetic 首个示例使用同步 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(...)。
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) 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) 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) 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) 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) 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。
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) 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) 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) 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 的更高并发用法。
pip install seeknetic[aiohttp] 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 install seeknetic 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(...)。
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); 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); 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); 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); 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); 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。
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); 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); 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); 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);