Python SDK

from vidnavigator import VidNavigatorClient

# Reads the VIDNAVIGATOR_API_KEY environment variable
client = VidNavigatorClient()

# Or pass the key explicitly
client = VidNavigatorClient(api_key="YOUR_API_KEY")

client = VidNavigatorClient(
    api_key="YOUR_API_KEY",       # or set VIDNAVIGATOR_API_KEY
    base_url="https://...",       # override the API base URL
    timeout=60,                   # per HTTP request, in seconds (default: 30)
    session=my_requests_session,  # bring your own requests.Session
)

with VidNavigatorClient() as client:
    resp = client.get_transcript(video_url="https://www.youtube.com/watch?v=dQw4w9WgXcQ")
    print(resp.data.transcript)

from vidnavigator import VidNavigatorClient

client = VidNavigatorClient()

# 1. Captions of a YouTube video
resp = client.get_transcript(video_url="https://www.youtube.com/watch?v=dQw4w9WgXcQ", transcript_text=True)
print(resp.data.transcript)

# 2. Speech-to-text for a video without captions (any length)
resp = client.transcribe_video(video_url="https://www.instagram.com/reel/C86ZvEaqRmo/", transcript_text=True)
print(resp.data.transcript)

# 3. Structured data from a video
resp = client.extract_video_data(
    video_url="https://www.youtube.com/watch?v=dQw4w9WgXcQ",
    schema={"mood": {"type": "String", "description": "Overall mood in one word"}},
)
print(resp.data)  # {"mood": "upbeat"}

resp = client.get_transcript(video_url="https://www.youtube.com/watch?v=dQw4w9WgXcQ")

print(resp.data.video_info.title)
# "Rick Astley - Never Gonna Give You Up (Official Music Video)"

for segment in resp.data.transcript:
    print(f"[{segment.start:.1f}s] {segment.text}")
# [0.0s] We're no strangers to love
# [3.2s] You know the rules and so do I

resp = client.get_transcript(video_url="https://www.youtube.com/watch?v=dQw4w9WgXcQ", transcript_text=True)
print(resp.data.transcript)
# "We're no strangers to love. You know the rules and so do I..."

resp = client.transcribe_video(video_url="https://www.instagram.com/reel/C86ZvEaqRmo/")

print(resp.data.video_info.title)
for segment in resp.data.transcript:
    print(f"[{segment.start:.1f}s] {segment.text}")

resp = client.transcribe_video(
    video_url="https://www.instagram.com/p/ABC123/",
    all_videos=True,
)

print(resp.data.carousel_info.video_count)
for video in resp.data.videos:
    print(f"Video {video.index} ({video.status}): {video.transcript}")

from vidnavigator import BadRequestError, NotFoundError

def best_transcript(url):
    try:
        return client.get_transcript(video_url=url, transcript_text=True).data.transcript
    except (NotFoundError, BadRequestError):  # no captions, or a platform without captions
        return client.transcribe_video(video_url=url, transcript_text=True).data.transcript

resp = client.extract_video_data(
    video_url="https://www.youtube.com/watch?v=dQw4w9WgXcQ",
    schema={
        "mood": {"type": "String", "description": "Overall mood of the video in one word"},
        "main_topics": {
            "type": "Array",
            "description": "List of main topics discussed",
            "items": {"type": "String", "description": "A topic"},
        },
        "has_spoken_lyrics": {
            "type": "Boolean",
            "description": "Whether the video contains spoken or sung lyrics",
        },
    },
    what_to_extract="Focus on the tone and content of the lyrics.",
)

print(resp.data)
# {"mood": "upbeat", "main_topics": ["love", "commitment"], "has_spoken_lyrics": True}

print(resp.video_info.title, resp.video_info.duration)
# "Rick Astley - Never Gonna Give You Up (Official Music Video)" 212.0

schema = {
    "sentiment": {
        "type": "Enum",
        "description": "Overall sentiment of the video",
        "enum": ["positive", "negative", "neutral"],
    },
    "speaker": {
        "type": "Object",
        "description": "The main speaker or singer",
        "properties": {
            "name": {"type": "String", "description": "Their name"},
            "role": {"type": "String", "description": "Their role in the video"},
        },
    },
    "key_takeaway": {"type": "String", "description": "The single most important takeaway"},
}

resp = client.extract_file_data(
    file_id="your_file_id",
    schema_file="./meeting-schema.yaml",  # or schema={...}
    what_to_extract="Extract action items and deadlines.",
)
print(resp.data)

resp = client.extract_video_data(
    video_url="...",
    schema={"summary": {"type": "String", "description": "One-line summary"}},
    include_usage=True,
)
tokens = resp.usage.analysis_tokens
print(f"LLM tokens: {tokens.total_tokens} (prompt {tokens.prompt_tokens}, completion {tokens.completion_tokens})")

resp = client.get_tweet_statement(tweet_id="1585841080431321088")

print(resp.data.final_statement)      # the core claim, in the author's voice
print(resp.data.detailed_analysis)    # claim, context, evidence and stance
print(resp.data.claim_type, resp.data.intent, resp.data.tone)
print(resp.data.topics, resp.data.entities)
print(resp.data.tweet_media_summary)  # summary of attached video/images, if any

job = client.submit_transcribe_video(video_url="https://www.instagram.com/reel/C86ZvEaqRmo/")
print(job.task_id)   # "550e8400-e29b-41d4-a716-446655440000" -- save it
print(job.status())  # "processing"

# ... do other work ...

resp = job.result()  # waits if needed, then returns the same TranscriptResponse as transcribe_video

from vidnavigator import JobTimeoutError

try:
    resp = client.transcribe_video(video_url=url, timeout=600)
except JobTimeoutError as exc:
    print("Still running:", exc.task_id)
    resp = exc.job.result()  # keep waiting on the same job

job = client.resume_job("transcribe", task_id)
if job.done():
    resp = job.result()

from vidnavigator import BadRequestError, PaymentRequiredError, VidNavigatorError

try:
    resp = client.transcribe_video(video_url=url)
except PaymentRequiredError:
    print("Out of transcription credit")
except BadRequestError as exc:
    print(exc.error_code, exc)  # e.g. "unsupported_platform"
except VidNavigatorError as exc:
    print("Job failed:", exc)

urls = [
    "https://www.tiktok.com/@a/video/1",
    "https://www.tiktok.com/@b/video/2",
    "https://www.tiktok.com/@c/video/3",
]

jobs = [client.submit_transcribe_video(video_url=url, transcript_text=True) for url in urls]
results = [job.result() for job in jobs]

for url, resp in zip(urls, results):
    print(url, resp.data.transcript[:80])

from vidnavigator import VidNavigatorError

for url, job in zip(urls, jobs):
    try:
        print(url, job.result().data.transcript[:80])
    except VidNavigatorError as exc:
        print(url, "failed:", exc.error_code, exc)

import time

pending = list(jobs)
while pending:
    for job in list(pending):
        if job.done():            # one free poll
            handle(job.result())  # already finished: returns immediately
            pending.remove(job)
    time.sleep(3)

import time
from vidnavigator import TooManyActiveJobsError, VidNavigatorError

MAX_RUNNING = 5
queue = list(urls)
running = []  # (url, job) pairs
results, errors = {}, {}

while queue or running:
    # Fill free slots
    while queue and len(running) < MAX_RUNNING:
        try:
            running.append((queue[0], client.submit_transcribe_video(video_url=queue[0])))
            queue.pop(0)
        except TooManyActiveJobsError:
            break  # the account is at its limit; retry after some jobs finish

    # Collect finished jobs
    for url, job in list(running):
        if job.done():
            try:
                results[url] = job.result()
            except VidNavigatorError as exc:
                errors[url] = exc
            running.remove((url, job))

    time.sleep(3)

import json

job = client.submit_extract_video_data(video_url=url, schema=schema)
with open("jobs.json", "w") as f:
    json.dump({"job_type": job.job_type, "task_id": job.task_id}, f)

# ... after a restart ...
with open("jobs.json") as f:
    saved = json.load(f)
resp = client.resume_job(saved["job_type"], saved["task_id"]).result()

profile = client.scrape_tiktok_profile(profile_url="https://www.tiktok.com/@tiktok", max_posts=20, limit=20)

jobs = [
    client.submit_transcribe_video(video_url=video.url, transcript_text=True)
    for video in profile.data.videos or []
    if video.url
]
for job in jobs:
    print(job.result().data.transcript[:80])

import asyncio
from vidnavigator import AsyncVidNavigatorClient

async def main():
    async with AsyncVidNavigatorClient() as client:  # reads VIDNAVIGATOR_API_KEY
        resp = await client.transcribe_video(video_url="https://www.instagram.com/reel/C86ZvEaqRmo/")
        print(resp.data.transcript)

asyncio.run(main())

async with AsyncVidNavigatorClient() as client:
    results = await asyncio.gather(
        *(client.transcribe_video(video_url=url, transcript_text=True) for url in urls),
        return_exceptions=True,  # one failed job doesn't cancel the others
    )

for url, resp in zip(urls, results):
    if isinstance(resp, Exception):
        print(url, "failed:", resp)
    else:
        print(url, resp.data.transcript[:80])

limit = asyncio.Semaphore(5)

async def transcribe(url):
    async with limit:
        return await client.transcribe_video(video_url=url)

results = await asyncio.gather(*(transcribe(url) for url in urls))

job = await client.submit_extract_video_data(video_url=url, schema=schema)
print(job.task_id)

if not await job.done():
    ...  # do other work
resp = await job.result()
print(resp.data, resp.video_info.title)

from contextlib import asynccontextmanager
from fastapi import FastAPI
from vidnavigator import AsyncVidNavigatorClient

@asynccontextmanager
async def lifespan(app: FastAPI):
    app.state.vidnavigator = AsyncVidNavigatorClient()
    yield
    await app.state.vidnavigator.aclose()

app = FastAPI(lifespan=lifespan)

@app.post("/jobs")
async def start_transcription(video_url: str):
    job = await app.state.vidnavigator.submit_transcribe_video(video_url=video_url)
    return {"task_id": job.task_id}

@app.get("/jobs/{task_id}")
async def job_status(task_id: str):
    job = app.state.vidnavigator.resume_job("transcribe", task_id)
    resp = await job.refresh()
    return {"status": resp.data.task_status}

import httpx

client = AsyncVidNavigatorClient(
    api_key="YOUR_API_KEY",   # or set VIDNAVIGATOR_API_KEY
    base_url="https://...",   # override the API base URL
    timeout=60,               # per HTTP request, in seconds (default: 30)
    http_client=httpx.AsyncClient(limits=httpx.Limits(max_connections=20)),  # optional; you close it yourself
)

job = client.submit_extract_video_data(
    video_url="https://www.tiktok.com/@user/video/1234567890",
    schema={"summary": {"type": "String", "description": "One-line summary"}},
    webhook_url="https://example.com/hooks/vidnavigator",
)
print(job.webhook_url)  # which webhook the job will call, or None

from flask import Flask, request
from vidnavigator import VidNavigatorClient, WebhookSignatureError, construct_webhook_event

app = Flask(__name__)
client = VidNavigatorClient()
WEBHOOK_SECRET = "your_signing_secret"

@app.post("/hooks/vidnavigator")
def vidnavigator_webhook():
    try:
        event = construct_webhook_event(
            request.get_data(),
            request.headers.get("X-VidNavigator-Signature"),
            WEBHOOK_SECRET,
        )
    except WebhookSignatureError:
        return "invalid signature", 400

    delivery_id = request.headers.get("X-VidNavigator-Delivery")  # stable across retries
    if already_processed(delivery_id):
        return "", 200

    if event.type == "transcribe.completed":
        # The job handle gives the typed result, even when the event omits it
        resp = client.resume_job("transcribe", event.data.task_id).result()
        save_transcript(event.data.task_id, resp.data.transcript)
    elif event.type.endswith(".failed"):
        print(event.data.task_id, event.data.error.error, event.data.error.message)

    return "", 200

result = client.scrape_tiktok_profile(
    profile_url="https://www.tiktok.com/@tiktok",
    max_posts=100,
    after_datetime="2024-01-01",
    limit=50,
)
task_id = result.data.task_id

for video in result.data.videos or []:
    print(video.title, video.published_at, video.likes, video.url)

pagination = result.data.pagination

while pagination and pagination.has_next:
    page = client.get_tiktok_profile_scrape(task_id, cursor=pagination.next_cursor, limit=50)
    for video in page.data.videos or []:
        print(video.title, video.url)
    pagination = page.data.pagination

import json
from urllib.request import urlopen

if result.data.download_url:
    with urlopen(result.data.download_url) as response:
        full_profile = json.load(response)

    for video in full_profile.get("videos", []):
        print(video.get("title"), video.get("url"))  # plain dicts, not typed models

result = client.search_tiktok(
    query="ai tools",
    max_results=100,
    sort_by="most_liked",
    published_within="this_month",
    min_views=1000,
    limit=50,
)

for item in result.data.results or []:
    print(item.description, item.published_at, item.stats.views if item.stats else None, item.url)

print(result.data.stats.sort_by, result.data.stats.published_within)  # what the search actually ran with

results = client.search_youtube(
    query="how to train a neural network from scratch",
    start_year=2023,
    max_results=5,
)

print(f"Found {results.data.total_found} results")
print(results.data.explanation)

for r in results.data.results:
    print(f"{r.title} (score: {r.relevance_score})")
    print(f"  {r.transcript_summary}")

results = client.search_files(
    query="customer feedback about pricing",
    namespace_ids=["ns_client_calls"],  # optional: scope to specific namespaces
)

for r in results.data.results:
    print(f"{r.name} (score: {r.relevance_score})")
    print(f"  Timestamps: {r.timestamps}")
    print(f"  {r.relevant_text}")

result = client.upload_file("./meeting.mp4", wait_for_completion=True)
file_id = result["file_id"]
print(result["file_status"])  # "completed"

resp = client.get_file(file_id)  # pass transcript_text=True for a single string
info = resp.data.file_info

print(f"{info.name} ({info.duration}s, {info.status})")
print(f"Namespaces: {[ns.name for ns in info.namespaces or []]}")

for segment in resp.data.transcript or []:
    print(f"[{segment.start:.1f}s] {segment.text}")

resp = client.get_files(limit=20, status="completed")
for f in resp.data.files:
    print(f"{f.name} -- {f.status}")

resp = client.get_files(namespace_id="ns_client_calls")  # filter by namespace
resp = client.get_files(limit=20, offset=20)             # paginate
print(resp.data.has_more)

client.get_file_url(file_id)            # signed URL to view or download the file
client.retry_file_processing(file_id)   # retry after a failure
client.cancel_file_upload(file_id)      # cancel in-progress processing
client.delete_file(file_id)             # permanently delete

ns = client.create_namespace(name="Client Calls")
print(ns.data.id)  # "ns_abc123"

for ns in client.get_namespaces().data:
    print(f"{ns.id}: {ns.name}")

client.update_namespace("ns_abc123", name="Sales Calls")

# Upload a file straight into a namespace
client.upload_file("./call.mp3", namespace_ids=["ns_abc123"])

# Reassign a file's namespaces
resp = client.update_file_namespaces(file_id, namespace_ids=["ns_abc123", "ns_def456"])
print(resp.data.namespaces)

# Delete (files are unlinked, not deleted)
client.delete_namespace("ns_abc123")

resp = client.search_youtube(query="react best practices", include_usage=True)

usage = resp.usage
if usage:
    print(f"Credits this call: {usage.total_credits}")
    for charge in usage.charges or []:
        print(f"  {charge.service_type}: {charge.quantity} unit(s), {charge.credits} credits")

    tokens = usage.analysis_tokens  # analyze / extract / YouTube search / tweet analysis
    if tokens:
        print(f"  LLM tokens: {tokens.total_tokens}")

usage = client.get_usage().data

print(f"Credits: {usage.credits.monthly_remaining} / {usage.credits.monthly_total}")
print(f"Storage: {usage.storage.used_formatted} / {usage.storage.limit_formatted}")

if usage.usage:
    print(f"Standard requests: {usage.usage.standard_request.used}")
    print(f"Transcription hours: {usage.usage.transcription_hour.used}")

from vidnavigator import BadRequestError, NotFoundError, RateLimitExceeded, VidNavigatorError

try:
    client.get_file("nonexistent-id")
except NotFoundError:
    print("File not found")
except RateLimitExceeded:
    print("Rate limit hit -- slow down")
except VidNavigatorError as exc:
    print("Something else went wrong:", exc)

try:
    client.extract_video_data(video_url=url, schema={"x": {"type": "Nope", "description": "?"}})
except BadRequestError as exc:
    print(exc.status_code)  # 400
    print(exc.error_code)   # "invalid_schema" -- machine-readable, safe to branch on
    print(exc.docs_url)     # link to the relevant docs, when provided
    print(exc.payload)      # the full JSON error body