# URL
video = coll.upload(url="https://example.com/video.mp4")
# YouTube
video = coll.upload(url="https://www.youtube.com/watch?v=VIDEO_ID")
# Local file
video = coll.upload(file_path="/path/to/video.mp4")
转录 + 字幕
# force=True skips the error if the video is already indexed
video.index_spoken_words(force=True)
text = video.get_transcript_text()
stream_url = video.add_subtitle()
在视频内搜索
from videodb.exceptions import InvalidRequestError
video.index_spoken_words(force=True)
# search() raises InvalidRequestError when no results are found.# Always wrap in try/except and treat "No results found" as empty.try:
results = video.search("product demo")
shots = results.get_shots()
stream_url = results.compile()
except InvalidRequestError as e:
if"No results found"instr(e):
shots = []
else:
raise
场景搜索
import re
from videodb import SearchType, IndexType, SceneExtractionType
from videodb.exceptions import InvalidRequestError
# index_scenes() has no force parameter — it raises an error if a scene# index already exists. Extract the existing index ID from the error.try:
scene_index_id = video.index_scenes(
extraction_type=SceneExtractionType.shot_based,
prompt="Describe the visual content in this scene.",
)
except Exception as e:
match = re.search(r"id\s+([a-f0-9]+)", str(e))
ifmatch:
scene_index_id = match.group(1)
else:
raise# Use score_threshold to filter low-relevance noise (recommended: 0.3+)try:
results = video.search(
query="person writing on a whiteboard",
search_type=SearchType.semantic,
index_type=IndexType.scene,
scene_index_id=scene_index_id,
score_threshold=0.3,
)
shots = results.get_shots()
stream_url = results.compile()
except InvalidRequestError as e:
if"No results found"instr(e):
shots = []
else:
raise
import json
import os
import time
from pathlib import Path
events_dir = Path(os.environ.get("VIDEODB_EVENTS_DIR", Path.home() / ".local" / "state" / "videodb"))
events_file = events_dir / "videodb_events.jsonl"
events = []
if events_file.exists():
with events_file.open(encoding="utf-8") as handle:
for line in handle:
try:
events.append(json.loads(line))
except json.JSONDecodeError:
continue
transcripts = [e["data"]["text"] for e in events if e.get("channel") == "transcript"]
cutoff = time.time() - 300
recent_visual = [
e for e in events
if e.get("channel") == "visual_index"and e["unix_ts"] > cutoff
]