fix request metadata issue

src/pages/_home_page.py

@@ -1,8 +1,6 @@
 from __future__ import annotations
 
 from typing import Any, Dict, List, Tuple
-
-import requests
 import streamlit as st
 
 from utils.helpers import (
@@ -11,6 +9,7 @@ from utils.helpers import (
     fetch_job_status,
     fetch_quiz,
     fetch_summary,
+    fetch_youtube_metadata_v3,
     get_image_base64,
     get_remote_image_url,
     list_cached_videos,
@@ -135,12 +134,12 @@ def _render_api_status() -> None:
 def _fetch_youtube_metadata(video_url: str) -> Dict[str, Any] | None:
     if not video_url:
         return None
-    oembed_url = f"https://www.youtube.com/oembed?url={video_url}&format=json"
     try:
-        response = requests.get(oembed_url, timeout=10)
-        if response.status_code == 200:
-            return response.json()
-    except requests.RequestException:
+        metadata = fetch_youtube_metadata_v3(video_url)
+        if metadata:
+            return metadata
+    except RuntimeError as exc:
+        st.warning(str(exc))
         return None
     return None
 

src/utils/helpers.py

@@ -26,6 +26,7 @@ _API_KEY_ENV_KEY = "CHAPTLY_API_KEY"
 _HMAC_SECRET_ENV_KEY = "CHAPTLY_API_SECRET"
 _S3_BUCKET_ENV_KEY = "CHAPTLY_S3_BUCKET"
 _AWS_REGION_ENV_KEY = "AWS_REGION"
+_YOUTUBE_API_KEY_ENV = "YOUTUBE_V3_DATA_API_KEY"
 _SIG_HEADER = "X-Signature"
 _TS_HEADER = "X-Timestamp"
 
@@ -33,6 +34,7 @@ DEFAULT_S3_BUCKET = "chaptly-rag"
 DEFAULT_AWS_REGION = "ap-southeast-1"
 IMAGE_REQUEST_TIMEOUT = 10
 _IMG_DIR = Path(__file__).resolve().parents[2] / "img"
+YOUTUBE_API_URL = "https://www.googleapis.com/youtube/v3/videos"
 
 
 @dataclass(frozen=True)
@@ -315,3 +317,46 @@ def list_cached_videos() -> list[Dict[str, Any]]:
 				metadata = _load_video_metadata(video_id)
 				items.append(metadata)
 	return sorted(items, key=lambda item: item.get("title", ""))
+
+
+def extract_youtube_video_id(url: str) -> Optional[str]:
+	"""Best-effort extraction of the YouTube video ID from a URL."""
+	if not url:
+		return None
+	url = url.strip()
+	if "v=" in url:
+		return url.split("v=")[1].split("&")[0]
+	if "youtu.be" in url:
+		return url.rsplit("/", 1)[-1].split("?")[0]
+	return None
+
+
+def fetch_youtube_metadata_v3(video_url: str) -> Optional[Dict[str, Any]]:
+	"""Fetch video metadata using the YouTube Data API v3."""
+	video_id = extract_youtube_video_id(video_url)
+	if not video_id:
+		return None
+	api_key = _get_secret_or_env(_YOUTUBE_API_KEY_ENV)
+	if not api_key:
+		raise RuntimeError("YOUTUBE_V3_DATA_API_KEY is not configured in this environment.")
+	params = {
+		"part": "snippet",
+		"id": video_id,
+		"key": api_key,
+	}
+	try:
+		response = requests.get(YOUTUBE_API_URL, params=params, timeout=DEFAULT_TIMEOUT)
+		response.raise_for_status()
+		payload = response.json()
+	except requests.RequestException as exc:
+		raise RuntimeError(f"Unable to contact YouTube Data API: {exc}") from exc
+	items = payload.get("items") or []
+	if not items:
+		return None
+	snippet = items[0].get("snippet", {})
+	return {
+		"title": snippet.get("title"),
+		"author_name": snippet.get("channelTitle"),
+		"thumbnail_url": (snippet.get("thumbnails", {}).get("high", {}) or snippet.get("thumbnails", {}).get("default", {})).get("url"),
+		"video_id": video_id,
+	}