Los vídeos de YouTube contienen grandes cantidades de conocimientos encerrados en audio. Al recopilar transcripciones a través de la API de Scavio e indexarlas en BigQuery, puedes crear una base de conocimientos con capacidad de búsqueda en miles de videos. La función SEARCH() de BigQuery permite la búsqueda de texto completo sobre el texto de la transcripción, lo que le permite encontrar temas, citas o instrucciones específicas mencionadas en cualquier video indexado. Este tutorial recopila transcripciones, las carga en BigQuery y ejecuta consultas de texto completo.
Requisitos previos
- Python 3.9+ instalado
- solicitudes y bibliotecas de google-cloud-bigquery instaladas
- Una clave API de Scavio de scavio.dev
- Un proyecto de Google Cloud con BigQuery habilitado
Guia paso a paso
Paso 1: Recopile transcripciones de YouTube a través de API
Busque en YouTube videos sobre un tema, extraiga ID de video y luego obtenga transcripciones de cada uno. Almacene los datos de transcripción sin procesar para cargar BigQuery.
Python
import os, requests, json, time, re
SCAVIO_KEY = os.environ['SCAVIO_API_KEY']
H = {'x-api-key': SCAVIO_KEY, 'Content-Type': 'application/json'}
URL = 'https://api.scavio.dev/api/v1/search'
def search_youtube_videos(topic: str, num: int = 10) -> list:
resp = requests.post(URL, headers=H,
json={'query': f'{topic} site:youtube.com', 'country_code': 'us', 'num_results': num})
results = resp.json().get('organic_results', [])
videos = []
for r in results:
vid_match = re.search(r'v=([^&]+)', r.get('link', ''))
if vid_match:
videos.append({'video_id': vid_match.group(1), 'title': r['title'], 'url': r['link']})
return videos
def get_transcript(video_id: str) -> list:
resp = requests.post(URL, headers=H,
json={'platform': 'youtube', 'action': 'transcript', 'video_id': video_id})
return resp.json().get('transcript', [])
videos = search_youtube_videos('python FastAPI tutorial')
print(f'Found {len(videos)} YouTube videos')
for v in videos[:3]:
print(f' {v["video_id"]}: {v["title"][:50]}')Paso 2: Preparar datos de transcripción para BigQuery
Recopile transcripciones, compántelas en texto completo y estructure los datos en formato JSONL para cargar BigQuery.
Python
def collect_transcripts(videos: list) -> list:
records = []
for v in videos:
segments = get_transcript(v['video_id'])
if not segments:
continue
full_text = ' '.join(seg.get('text', '') for seg in segments)
records.append({
'video_id': v['video_id'],
'title': v['title'],
'url': v['url'],
'transcript_text': full_text,
'segment_count': len(segments),
'word_count': len(full_text.split()),
'duration_seconds': sum(seg.get('duration', 0) for seg in segments),
})
time.sleep(0.3)
return records
def save_for_bigquery(records: list, output_file: str = 'transcripts.jsonl'):
with open(output_file, 'w') as f:
for r in records:
f.write(json.dumps(r) + '\n')
print(f'Saved {len(records)} transcripts to {output_file}')
total_words = sum(r['word_count'] for r in records)
print(f'Total words: {total_words:,}')
transcripts = collect_transcripts(videos[:5])
save_for_bigquery(transcripts)Paso 3: Cargue en BigQuery y ejecute consultas de BÚSQUEDA
Cree una tabla de BigQuery, cargue los datos de la transcripción y use la función SEARCH() para realizar búsquedas de texto completo en todas las transcripciones indexadas.
Python
from google.cloud import bigquery
def setup_bigquery(project_id: str, dataset: str = 'youtube_data'):
client = bigquery.Client(project=project_id)
# Create dataset if needed
dataset_ref = bigquery.Dataset(f'{project_id}.{dataset}')
try:
client.create_dataset(dataset_ref)
except Exception:
pass # Already exists
# Create table with search index
schema = [
bigquery.SchemaField('video_id', 'STRING'),
bigquery.SchemaField('title', 'STRING'),
bigquery.SchemaField('url', 'STRING'),
bigquery.SchemaField('transcript_text', 'STRING'),
bigquery.SchemaField('segment_count', 'INTEGER'),
bigquery.SchemaField('word_count', 'INTEGER'),
bigquery.SchemaField('duration_seconds', 'FLOAT'),
]
table_ref = f'{project_id}.{dataset}.transcripts'
table = bigquery.Table(table_ref, schema=schema)
client.create_table(table, exists_ok=True)
# Load data
job_config = bigquery.LoadJobConfig(
source_format=bigquery.SourceFormat.NEWLINE_DELIMITED_JSON)
with open('transcripts.jsonl', 'rb') as f:
job = client.load_table_from_file(f, table_ref, job_config=job_config)
job.result()
print(f'Loaded data into {table_ref}')
return client
def search_transcripts(client, project_id: str, query: str):
sql = f"""
SELECT video_id, title, url,
SUBSTR(transcript_text, 1, 200) AS excerpt
FROM `{project_id}.youtube_data.transcripts`
WHERE SEARCH(transcript_text, @query)
LIMIT 10
"""
job_config = bigquery.QueryJobConfig(
query_parameters=[bigquery.ScalarQueryParameter('query', 'STRING', query)])
results = client.query(sql, job_config=job_config)
print(f'Search results for: "{query}"')
for row in results:
print(f' [{row.video_id}] {row.title[:50]}')
print(f' {row.excerpt[:100]}...')
# Example usage (requires GCP credentials)
# client = setup_bigquery('my-project-id')
# search_transcripts(client, 'my-project-id', 'FastAPI async endpoints')
print('BigQuery transcript search ready')Ejemplo en Python
Python
import os, requests, json, time, re
SCAVIO_KEY = os.environ['SCAVIO_API_KEY']
H = {'x-api-key': SCAVIO_KEY, 'Content-Type': 'application/json'}
def collect_transcripts(topic, num=5):
# Search YouTube
resp = requests.post('https://api.scavio.dev/api/v1/search', headers=H,
json={'query': f'{topic} site:youtube.com', 'country_code': 'us', 'num_results': num})
videos = []
for r in resp.json().get('organic_results', []):
m = re.search(r'v=([^&]+)', r.get('link', ''))
if m: videos.append({'id': m.group(1), 'title': r['title']})
# Get transcripts
for v in videos:
resp = requests.post('https://api.scavio.dev/api/v1/search', headers=H,
json={'platform': 'youtube', 'action': 'transcript', 'video_id': v['id']})
segs = resp.json().get('transcript', [])
v['text'] = ' '.join(s.get('text', '') for s in segs)
v['words'] = len(v['text'].split())
print(f'{v["id"]}: {v["words"]} words | {v["title"][:40]}')
time.sleep(0.3)
print(f'Cost: ${len(videos) * 0.010:.3f}')
collect_transcripts('Python FastAPI tutorial')Ejemplo en JavaScript
JavaScript
const SCAVIO_KEY = process.env.SCAVIO_API_KEY;
async function collectTranscripts(topic) {
const resp = await fetch('https://api.scavio.dev/api/v1/search', {
method: 'POST',
headers: { 'x-api-key': SCAVIO_KEY, 'Content-Type': 'application/json' },
body: JSON.stringify({ query: `${topic} site:youtube.com`, country_code: 'us', num_results: 5 })
});
const videos = (await resp.json()).organic_results || [];
for (const v of videos.filter(r => r.link.includes('youtube.com/watch'))) {
const vidId = v.link.match(/v=([^&]+)/)?.[1];
if (!vidId) continue;
const t = await fetch('https://api.scavio.dev/api/v1/search', {
method: 'POST',
headers: { 'x-api-key': SCAVIO_KEY, 'Content-Type': 'application/json' },
body: JSON.stringify({ platform: 'youtube', action: 'transcript', video_id: vidId })
}).then(r => r.json());
const text = (t.transcript || []).map(s => s.text).join(' ');
console.log(`${vidId}: ${text.split(' ').length} words | ${v.title.slice(0, 40)}`);
}
}
collectTranscripts('Python FastAPI tutorial');Salida esperada
JSON
Found 8 YouTube videos
dQw4w9WgXcQ: Python FastAPI Tutorial - Build a REST API in 30
abc123def4: FastAPI vs Django: Complete Comparison 2026
xyz789ghi0: Advanced FastAPI Patterns for Production
Saved 5 transcripts to transcripts.jsonl
Total words: 42,350
Search results for: "FastAPI async endpoints"
[dQw4w9WgXcQ] Python FastAPI Tutorial - Build a REST API in 30
FastAPI uses async def by default for all route handlers...
[abc123def4] FastAPI vs Django: Complete Comparison 2026
When it comes to async endpoints FastAPI has a clear advant...
BigQuery transcript search ready