# Google Gemini Video Intelligence API

> Source: <https://gist.github.com/gitgithan/1515a2e093463b9fdb4163e6039778ee>
> Published: 2026-09-03 18:59:08+00:00

|  | import json | 
|  | import os | 
|  | from typing import Any | 
|  | from google import genai | 
|  | # 1. Configuration & Input | 
|  | # video_id = "88I6IidylGc" | 
|  | video_id = "2lm1DFGFxPs" | 
|  | YOUTUBE_URL = f"https://www.youtube.com/watch?v={video_id}" | 
|  | processing_mode = "agentic" | 
|  | # Use the script's directory as the base for all output files | 
|  | _SCRIPT_DIR = os.path.dirname(os.path.abspath(__file__)) | 
|  | OUTPUT_FILE = os.path.join(_SCRIPT_DIR, f"transcript_concepts_{video_id}_{processing_mode}.json") | 
|  | THOUGHTS_OUTPUT_FILE = os.path.join(_SCRIPT_DIR, f"transcript_concepts_thoughts_{video_id}_{processing_mode}.json") | 
|  | STEPS_OUTPUT_FILE = os.path.join(_SCRIPT_DIR, f"transcript_concepts_steps_{video_id}_{processing_mode}.json") | 
|  | # 2. Schema Definition (Standard JSON Schema with lowercase types) | 
|  | _YOUTUBE_RESPONSE_SCHEMA: dict[str, Any] = { | 
|  | "type": "object", | 
|  | "properties": { | 
|  | "videoTitle": {"type": "string"}, | 
|  | "contentType": {"type": "string", "enum": ["educational", "entertainment", "mixed"]}, | 
|  | "concepts": { | 
|  | "type": "array", | 
|  | "items": { | 
|  | "type": "object", | 
|  | "properties": { | 
|  | "label": {"type": "string"}, | 
|  | "timestamps": { | 
|  | "type": "array", | 
|  | "items": { | 
|  | "type": "object", | 
|  | "properties": { | 
|  | "seconds": {"type": "string"}, | 
|  | "description": {"type": "string"}, | 
|  | "quote": {"type": "string"}, | 
|  | "visual": {"type": "boolean"}, | 
|  | "visualDescription": {"type": "string"}, | 
|  | }, | 
|  | "required": ["seconds", "description", "quote"], | 
|  | }, | 
|  | }, | 
|  | }, | 
|  | "required": ["label", "timestamps"], | 
|  | }, | 
|  | }, | 
|  | "edges": { | 
|  | "type": "array", | 
|  | "items": { | 
|  | "type": "object", | 
|  | "properties": { | 
|  | "fromLabel": {"type": "string"}, | 
|  | "toLabel": {"type": "string"}, | 
|  | "relation": {"type": "string", "enum": ["detail", "sibling", "abstraction", "related"]}, | 
|  | "rationale": {"type": "string"}, | 
|  | }, | 
|  | "required": ["fromLabel", "toLabel", "relation"], | 
|  | }, | 
|  | }, | 
|  | "checkpoints": { | 
|  | "type": "array", | 
|  | "items": { | 
|  | "type": "object", | 
|  | "properties": { | 
|  | "conceptLabel": {"type": "string"}, | 
|  | "question": {"type": "string"}, | 
|  | "answer": {"type": "string"}, | 
|  | }, | 
|  | "required": ["conceptLabel", "question", "answer"], | 
|  | }, | 
|  | }, | 
|  | }, | 
|  | "required": ["videoTitle", "contentType", "concepts"], | 
|  | } | 
|  | def serialize_item(item: Any) -> Any: | 
|  | if hasattr(item, "model_dump"): | 
|  | return item.model_dump() | 
|  | elif hasattr(item, "__dict__"): | 
|  | return { | 
|  | k: str(v) if not isinstance(v, (dict, list, int, float, bool, type(None))) else v | 
|  | for k, v in item.__dict__.items() | 
|  | } | 
|  | return str(item) | 
|  | def extract_thoughts(interaction: Any) -> list[str]: | 
|  | thoughts = [] | 
|  | # May 2026 Steps Schema parsing | 
|  | for step in getattr(interaction, "steps", []) or []: | 
|  | step_type = getattr(step, "type", None) or (step.get("type") if isinstance(step, dict) else None) | 
|  | if step_type == "thought": | 
|  | summary = getattr(step, "summary", []) or (step.get("summary") if isinstance(step, dict) else []) | 
|  | for item in summary: | 
|  | text = getattr(item, "text", None) or (item.get("text") if isinstance(item, dict) else None) | 
|  | if text: | 
|  | thoughts.append(text) | 
|  | # Legacy fallback for older API versions | 
|  | if not thoughts: | 
|  | for candidate in getattr(interaction, "candidates", []) or []: | 
|  | for part in getattr(getattr(candidate, "content", None), "parts", []): | 
|  | if getattr(part, "thought", False): | 
|  | thoughts.append(part.text) | 
|  | return thoughts | 
|  | def get_interaction_text(interaction: Any) -> str: | 
|  | if hasattr(interaction, "outputs") and interaction.outputs: | 
|  | for output in reversed(interaction.outputs): | 
|  | if hasattr(output, "text") and output.text: | 
|  | return output.text | 
|  | if hasattr(output, "content") and getattr(output.content, "parts", None): | 
|  | for part in output.content.parts: | 
|  | if getattr(part, "text", None) and not getattr(part, "thought", False): | 
|  | return part.text | 
|  | if hasattr(interaction, "steps") and interaction.steps: | 
|  | for step in reversed(interaction.steps): | 
|  | model_res = getattr(step, "model_response", None) or getattr(step, "response", None) | 
|  | if model_res and hasattr(model_res, "candidates"): | 
|  | for candidate in model_res.candidates or []: | 
|  | for part in getattr(getattr(candidate, "content", None), "parts", []): | 
|  | if getattr(part, "text", None) and not getattr(part, "thought", False): | 
|  | return part.text | 
|  | return getattr(interaction, "output_text", "{}") | 
|  | def analyze_youtube_video(url: str, output_path: str, thoughts_path: str, steps_path: str): | 
|  | print("Sending request to Gemini via Interactions API...") | 
|  | client = genai.Client() | 
|  | prompt = ( | 
|  | f"Analyze the YouTube video ({url}) and identify its key structural units. " | 
|  | "For each unit you must log the EXACT timecode in 'MM:SS' format as it first appears in the video's underlying audio stream. " | 
|  | "Try to find at least 3 timestamps per concept label" | 
|  | "Return in the seconds field in the response schema, every timestamp where a unit is first discussed in HH:MM:SS format or MM:SS if video is less than an hour." | 
|  | ) | 
|  | interaction = client.interactions.create( | 
|  | model="gemini-3.5-flash-lite", | 
|  | input=[ | 
|  | { | 
|  | "type": "text", | 
|  | "text": prompt, | 
|  | }, | 
|  | { | 
|  | "type": "video", | 
|  | "uri": url, | 
|  | "processing": processing_mode | 
|  | } | 
|  | ], | 
|  | response_format={ | 
|  | "type": "text", | 
|  | "mime_type": "application/json", | 
|  | "schema": _YOUTUBE_RESPONSE_SCHEMA, | 
|  | }, | 
|  | generation_config={ | 
|  | "temperature": 0.1, | 
|  | "thinking_summaries": "auto", | 
|  | }, | 
|  | ) | 
|  | # 1. Save thinking process | 
|  | thoughts = extract_thoughts(interaction) | 
|  | with open(thoughts_path, "w", encoding="utf-8") as f: | 
|  | json.dump({"interaction_id": getattr(interaction, "id", None), "thoughts": thoughts}, f, indent=2) | 
|  | # 2. Save steps | 
|  | raw_steps = getattr(interaction, "steps", []) or [] | 
|  | serialized_steps = [serialize_item(step) for step in raw_steps] | 
|  | with open(steps_path, "w", encoding="utf-8") as f: | 
|  | json.dump({"interaction_id": getattr(interaction, "id", None), "steps": serialized_steps}, f, indent=2) | 
|  | # 3. Extract and parse structured text | 
|  | raw_text = get_interaction_text(interaction) | 
|  | data = json.loads(raw_text) | 
|  | with open(output_path, "w", encoding="utf-8") as f: | 
|  | json.dump(data, f, indent=2, ensure_ascii=False) | 
|  | print(f"Successfully saved structured output to '{output_path}'!") | 
|  | return data | 
|  | result = analyze_youtube_video(YOUTUBE_URL, OUTPUT_FILE, THOUGHTS_OUTPUT_FILE, STEPS_OUTPUT_FILE) | 
|  | print(json.dumps(result, indent=2)) |
