mirror of
https://github.com/nubenetes/awesome-kubernetes.git
synced 2026-07-12 18:00:37 +00:00
112 lines
4.7 KiB
Python
112 lines
4.7 KiB
Python
import yaml
|
|
import os
|
|
import re
|
|
import asyncio
|
|
import httpx
|
|
from src.logger import log_event
|
|
from src.gemini_utils import call_gemini_with_retry, fetch_youtube_metadata
|
|
from src.inventory_manager import load_inventory, save_inventory
|
|
|
|
INVENTORY_PATH = "data/inventory.yaml"
|
|
|
|
async def enrich_video_entry(url: str, entry: dict):
|
|
log_event(f"[*] Enriching video: {url}")
|
|
meta = await fetch_youtube_metadata(url)
|
|
|
|
# Strategy A: Use metadata from direct fetch
|
|
# Strategy B: Fallback to AI Grounding if fetch was blocked or generic
|
|
use_grounding = False
|
|
is_generic = False
|
|
if meta:
|
|
title = meta.get('raw_title', '').lower()
|
|
if title in ['youtube', 'before you continue', 'n/a', '']:
|
|
is_generic = True
|
|
|
|
if not meta or is_generic:
|
|
log_event(f" [!] Missing or generic metadata for {url}. Forcing AI Grounding (Pro Model)...")
|
|
use_grounding = True
|
|
context = f"YouTube URL: {url}\nNote: The local scraper was blocked and returned generic platform metadata. YOU MUST SEARCH FOR THE REAL TITLE."
|
|
else:
|
|
context = f"Local Context (from YouTube extraction):\nTitle: {meta['raw_title']}\nDescription: {meta['raw_description']}"
|
|
|
|
prompt = f"""
|
|
You are a Senior Cloud Architect. Your goal is to provide a high-fidelity summary of a technical YouTube video.
|
|
|
|
INPUT CONTEXT:
|
|
{context}
|
|
|
|
CRITICAL INSTRUCTIONS:
|
|
1. If the 'Local Context' above is missing, generic, or mentions 'YouTube' without a specific title, you MUST use your internal GOOGLE SEARCH GROUNDING to find the actual title and technical description of this video URL: {url}.
|
|
2. Identify the ACTUAL technical content based on the verified video metadata.
|
|
3. Generate a high-density architectural summary (2-3 sentences) explaining its specific value for a 2026 Cloud Native context.
|
|
4. DO NOT describe generic YouTube platform infrastructure unless the video is specifically about it.
|
|
5. Crossover AI Agents Detection: Pay special attention to AI Agents & MCP, and how they integrate with Cloud Native (e.g. SRE with AI Agents, Kubernetes with AI Agents, IaC with Terraform and AI, DevOps with AI, MLOps with AI). If the video covers these topics, reflect it in the technology field.
|
|
6. Select the primary technology (e.g., Kubernetes + AI Agents, IaC + Terraform + AI Agents, DevOps + AI Agents, SRE + AI Agents, MLOps + AI Agents, or standard core tech like Kubernetes, Istio) and a category from: [Fundamentals and Documentaries, Architecture and Cloud Strategy, Networking and Service Mesh, Infrastructure as Code, Observability and Monitoring, AI and Future Operations, Security and Compliance].
|
|
|
|
Return ONLY a JSON object:
|
|
{{
|
|
"title": "Actual Verified Title",
|
|
"summary": "Specific and faithful architectural summary...",
|
|
"technology": "...",
|
|
"category": "..."
|
|
}}
|
|
"""
|
|
|
|
try:
|
|
# Use grounding only if direct fetch failed
|
|
data = await call_gemini_with_retry(prompt, prefer_flash=True, use_grounding=use_grounding)
|
|
|
|
entry["title"] = data.get("title", meta['raw_title'] if meta else "YouTube Video")
|
|
entry["ai_summary"] = data["summary"]
|
|
entry["technology"] = data["technology"]
|
|
entry["category"] = data["category"]
|
|
entry["is_enriched"] = True
|
|
log_event(f" [OK] {url} enriched successfully.")
|
|
except Exception as e:
|
|
log_event(f" [ERROR] Failed to parse AI response for {url}: {e}")
|
|
|
|
return entry
|
|
|
|
async def main():
|
|
if not os.path.exists(INVENTORY_PATH):
|
|
return
|
|
|
|
force_enrich = os.getenv("FORCE_ENRICH", "false").lower() == "true"
|
|
|
|
inventory = load_inventory()
|
|
|
|
video_urls = [u for u, e in inventory.items() if e.get("is_featured_video")]
|
|
|
|
tasks = []
|
|
skipped = 0
|
|
for url in video_urls:
|
|
if inventory[url].get("is_enriched") and not force_enrich:
|
|
skipped += 1
|
|
continue
|
|
tasks.append(enrich_video_entry(url, inventory[url]))
|
|
|
|
if skipped > 0:
|
|
log_event(f"[*] Skipped {skipped} already enriched videos. Use FORCE_ENRICH=true to re-process.")
|
|
|
|
if not tasks:
|
|
log_event("[*] No videos need enrichment.")
|
|
return
|
|
|
|
# Process in small batches to respect rate limits
|
|
batch_size = 5
|
|
for i in range(0, len(tasks), batch_size):
|
|
batch = tasks[i:i+batch_size]
|
|
await asyncio.gather(*batch)
|
|
|
|
# Incremental Persistence: Save after each batch
|
|
save_inventory(inventory)
|
|
log_event(f" [💾] Saved progress: {min(i + batch_size, len(tasks))}/{len(tasks)} videos.")
|
|
|
|
if i + batch_size < len(tasks):
|
|
await asyncio.sleep(2) # Safety delay
|
|
|
|
log_event("✅ Video Hub Enrichment Complete.")
|
|
|
|
if __name__ == "__main__":
|
|
asyncio.run(main())
|