🚀 Auto-deploy backend from GitHub (23fbade)

main.py

@@ -80,6 +80,7 @@ from services.user_provisioning_service import (
 from routes.rag_routes import router as rag_router
 from routes.admin_model_routes import router as admin_model_router
 from routes.diagnostic import router as diagnostic_router
+from routes.video_routes import router as video_router
 from rag.curriculum_rag import (
     build_analysis_curriculum_context,
     build_lesson_prompt,
@@ -363,6 +364,7 @@ ROLE_POLICIES: Dict[str, Set[str]] = {
     "/api/admin/model-config/profile": ADMIN_ONLY,
     "/api/admin/model-config/override": ADMIN_ONLY,
     "/api/admin/model-config/reset": ADMIN_ONLY,
+    "/api/lessons/videos/search": ALL_APP_ROLES,
 }
 
 if not os.getenv("DEEPSEEK_API_KEY"):
@@ -1013,6 +1015,7 @@ app.add_middleware(AuthMiddleware)
 app.include_router(rag_router)
 app.include_router(admin_model_router)
 app.include_router(diagnostic_router)
+app.include_router(video_router)
 
 
 # ─── Global Exception Handler ─────────────────────────────────

requirements.txt

@@ -22,3 +22,4 @@ redis[hiredis]>=5.0.0
 PyYAML>=6.0.0
 mypy>=1.20.0
 pytest>=9.0.0
+google-api-python-client>=2.0.0

routes/rag_routes.py

@@ -173,17 +173,31 @@ async def rag_health():
         }
 
 
-def _fetch_youtube_video(lesson_title: str, subject: str, competency: str, quarter: int) -> dict:
+def _fetch_youtube_videos(
+    lesson_title: str,
+    subject: str,
+    competency: str,
+    quarter: int,
+    lesson_id: Optional[str] = None,
+) -> List[Dict]:
+    """Fetch up to 3 relevant YouTube videos for a lesson."""
     try:
-        from backend.services.youtube_service import get_video_for_lesson
+        from services.youtube_service import get_video_search_results
     except ImportError:
-        return {}
+        return []
     try:
-        video = get_video_for_lesson(lesson_title, subject, competency, quarter)
-        return video or {}
+        result = get_video_search_results(
+            topic=lesson_title,
+            subject=subject,
+            lesson_context=competency,
+            grade_level=f"Grade {quarter + 10}",
+            lesson_id=lesson_id,
+            max_results=3,
+        )
+        return result.get("videos", [])
     except Exception as e:
-        logger.warning("YouTube search failed: %s", e)
-        return {}
+        logger.warning("YouTube video search failed: %s", e)
+        return []
 
 
 def _ensure_7_sections(lesson_data: dict, lesson_title: str) -> dict:
@@ -332,23 +346,28 @@ async def rag_lesson(request: Request, payload: RagLessonRequest):
             },
         )
 
-    # ── Step 5: Enrich with video ────────────────────────────────────────────
+    # ── Step 5: Enrich with videos ───────────────────────────────────────────
     if parsed_lesson.get("sections"):
         video_section = next((s for s in parsed_lesson["sections"] if s.get("type") == "video"), None)
         if video_section:
             try:
-                video_data = _fetch_youtube_video(
+                videos = _fetch_youtube_videos(
                     payload.lessonTitle or payload.topic,
                     payload.subject,
                     payload.learningCompetency or "",
                     payload.quarter,
+                    lesson_id=payload.lessonId,
                 )
-                if video_data:
-                    video_section["videoId"] = video_data.get("videoId", "")
-                    video_section["videoTitle"] = video_data.get("videoTitle", "")
-                    video_section["videoChannel"] = video_data.get("videoChannel", "")
-                    video_section["embedUrl"] = video_data.get("embedUrl", "")
-                    video_section["thumbnailUrl"] = video_data.get("thumbnailUrl", "")
+                if videos:
+                    # Primary video for backwards compatibility
+                    primary = videos[0]
+                    video_section["videoId"] = primary.get("videoId", "")
+                    video_section["videoTitle"] = primary.get("title", "")
+                    video_section["videoChannel"] = primary.get("channelTitle", "")
+                    video_section["embedUrl"] = f"https://www.youtube.com/embed/{primary.get('videoId', '')}"
+                    video_section["thumbnailUrl"] = primary.get("thumbnailUrl", "")
+                    # New: full videos array for Smart Video Integration
+                    video_section["videos"] = videos
             except Exception as exc:
                 logger.warning("YouTube enrichment skipped: %s", exc)
 

routes/video_routes.py

@@ -0,0 +1,102 @@
+"""
+Video lesson search routes for MathPulse AI.
+POST /api/lessons/videos/search — smart YouTube video search with RAG enrichment.
+"""
+
+from __future__ import annotations
+
+import logging
+from typing import List, Optional
+
+from fastapi import APIRouter, HTTPException, Request
+from pydantic import BaseModel, Field
+
+from services.youtube_service import (
+    get_video_search_results,
+    YOUTUBE_API_KEY,
+)
+
+logger = logging.getLogger("mathpulse.videos")
+router = APIRouter(prefix="/api/lessons/videos", tags=["videos"])
+
+
+class VideoSearchRequest(BaseModel):
+    topic: str = Field(..., min_length=1, max_length=200)
+    grade_level: str = Field(default="Grade 11", max_length=50)
+    subject: str = Field(default="General Mathematics", max_length=100)
+    lesson_context: str = Field(default="", max_length=1000)
+    lesson_id: Optional[str] = Field(default=None, max_length=100)
+
+
+class VideoResult(BaseModel):
+    videoId: str
+    title: str
+    channelTitle: str
+    thumbnailUrl: str
+    durationSeconds: int
+
+
+class VideoSearchResponse(BaseModel):
+    videos: List[VideoResult]
+    cached: bool = False
+
+
+@router.post("/search", response_model=VideoSearchResponse)
+async def search_videos(request: Request, payload: VideoSearchRequest):
+    """
+    Search for relevant educational YouTube videos for a lesson topic.
+
+    - Checks Firestore video_cache first (7-day TTL)
+    - Enriches the search query with RAG curriculum keywords
+    - Filters for educational channels, medium/long duration, HD quality
+    - Returns up to 3 video results
+    """
+    # Graceful degradation: if YouTube API key is not configured, return 503
+    # so the frontend can hide the video section silently
+    if not YOUTUBE_API_KEY:
+        logger.warning("YouTube API key not configured")
+        raise HTTPException(
+            status_code=503,
+            detail={
+                "error": "youtube_api_not_configured",
+                "message": "YouTube API key is not configured on the server.",
+            },
+        )
+
+    try:
+        result = get_video_search_results(
+            topic=payload.topic,
+            subject=payload.subject,
+            lesson_context=payload.lesson_context,
+            grade_level=payload.grade_level,
+            lesson_id=payload.lesson_id,
+            max_results=3,
+        )
+
+        videos = [
+            VideoResult(
+                videoId=v["videoId"],
+                title=v["title"],
+                channelTitle=v["channelTitle"],
+                thumbnailUrl=v["thumbnailUrl"],
+                durationSeconds=v["durationSeconds"],
+            )
+            for v in result.get("videos", [])
+        ]
+
+        return VideoSearchResponse(
+            videos=videos,
+            cached=result.get("cached", False),
+        )
+
+    except HTTPException:
+        raise
+    except Exception as exc:
+        logger.error("Video search endpoint error: %s", exc)
+        raise HTTPException(
+            status_code=500,
+            detail={
+                "error": "video_search_failed",
+                "message": f"Failed to search videos: {exc}",
+            },
+        )

services/youtube_service.py

@@ -1,170 +1,384 @@
 """
-YouTube search service for lesson video embeddings.
-Uses YouTube Data API v3 to find relevant educational videos.
+Smart YouTube Video Search Service for MathPulse AI.
+Uses YouTube Data API v3 (googleapiclient.discovery) to find relevant
+educational math videos, enriched with RAG curriculum context.
+Results are cached in Firestore video_cache/{lessonId} with 7-day TTL.
 """
 
 from __future__ import annotations
 
-import os
+import hashlib
 import logging
-from typing import Optional
+import os
+import re
+from datetime import datetime, timezone
+from typing import Dict, List, Optional
 
 logger = logging.getLogger("mathpulse.youtube")
 
-YOUTUBE_API_KEY = os.getenv("YOUTUBE_API_KEY", "")
+YOUTUBE_API_KEY = os.getenv("YOUTUBE_API_KEY", "").strip()
+
+# Known educational channel keywords and exact names for post-filtering
+_EDUCATIONAL_CHANNEL_KEYWORDS = [
+    "khan", "math", "academy", "education", "teacher", "professor",
+    "tutorial", "lesson", "school", "university", "college", "deped",
+    "philippines", "filipino", "pinoy", "stem", "learning", "study",
+    "organic chemistry tutor", "patrickjmt", "3blue1brown", "numberphile",
+    "math antics", "bright side", "crashcourse", "ted-ed", "ted ed",
+    "nancy pi", "professor leonard", "mit", "stanford", "harvard",
+]
+
+_EDUCATIONAL_CHANNEL_EXACT = {
+    "khan academy", "patrickjmt", "3blue1brown", "numberphile",
+    "math antics", "the organic chemistry tutor", "professor leonard",
+    "nancy pi", "ted-ed", "crashcourse", "bright side",
+    "mit opencourseware", "stanford", "harvard",
+}
+
+# Minimum duration in seconds to filter out shorts (3 minutes)
+_MIN_DURATION_SECONDS = 180
+# Maximum duration in seconds to avoid extremely long videos (60 minutes)
+_MAX_DURATION_SECONDS = 3600
+# Cache TTL in seconds (7 days)
+_CACHE_TTL_SECONDS = 7 * 24 * 60 * 60
+
+
+def _build_youtube_client():
+    """Lazy-init googleapiclient YouTube client. Returns None if no API key."""
+    if not YOUTUBE_API_KEY:
+        return None
+    try:
+        from googleapiclient.discovery import build
+        return build("youtube", "v3", developerKey=YOUTUBE_API_KEY, cache_discovery=False)
+    except Exception as exc:
+        logger.warning("Failed to build YouTube client: %s", exc)
+        return None
 
 
 def _parse_iso8601_duration(duration: str) -> int:
     """Parse ISO 8601 duration string like 'PT5M30S' to seconds."""
-    import re
     if not duration:
         return 0
-    hours_match = re.search(r'(\d+)H', duration)
-    minutes_match = re.search(r'(\d+)M', duration)
-    seconds_match = re.search(r'(\d+)S', duration)
+    hours_match = re.search(r"(\d+)H", duration)
+    minutes_match = re.search(r"(\d+)M", duration)
+    seconds_match = re.search(r"(\d+)S", duration)
     hours = int(hours_match.group(1)) if hours_match else 0
     minutes = int(minutes_match.group(1)) if minutes_match else 0
     seconds = int(seconds_match.group(1)) if seconds_match else 0
     return hours * 3600 + minutes * 60 + seconds
 
 
-def search_youtube_video(
-    query: str,
-    max_results: int = 5,
-    min_duration_seconds: int = 180,
-    language: str = "en",
-) -> Optional[dict]:
-    """
-    Search YouTube Data API v3 for relevant educational videos.
+def _is_educational_channel(channel_title: str) -> bool:
+    """Check if a channel appears to be educational."""
+    lowered = channel_title.lower().strip()
+    if lowered in _EDUCATIONAL_CHANNEL_EXACT:
+        return True
+    return any(kw in lowered for kw in _EDUCATIONAL_CHANNEL_KEYWORDS)
+
 
-    Args:
-        query: Search query combining lesson title, subject, and competency
-        max_results: Maximum number of results to return
-        min_duration_seconds: Minimum video duration (filter out shorts)
-        language: Preferred video language
+def _score_video_result(item: dict, query: str) -> float:
+    """Score a video result for relevance. Higher is better."""
+    score = 0.0
+    title = (item.get("title") or "").lower()
+    description = (item.get("description") or "").lower()
+    channel = (item.get("channelTitle") or "").lower()
+    query_lower = query.lower()
 
-    Returns:
-        Best video match with videoId, title, channel, embedUrl, thumbnailUrl
+    # Title contains key math/education terms
+    math_terms = ["tutorial", "lesson", "explain", "math", "mathematics",
+                  "solution", "problem", "example", "learn", "how to"]
+    for term in math_terms:
+        if term in title:
+            score += 2.0
+
+    # Query terms appear in title
+    for word in query_lower.split():
+        if len(word) > 2 and word in title:
+            score += 1.5
+
+    # Educational channel bonus
+    if _is_educational_channel(channel):
+        score += 5.0
+
+    # Description relevance
+    for word in query_lower.split():
+        if len(word) > 2 and word in description:
+            score += 0.5
+
+    # Duration sweet spot: 5-20 minutes
+    duration = item.get("durationSeconds", 0)
+    if 300 <= duration <= 1200:
+        score += 2.0
+    elif duration >= _MIN_DURATION_SECONDS:
+        score += 1.0
+
+    return score
+
+
+def _enrich_query_with_rag(topic: str, subject: str, lesson_context: str = "") -> str:
     """
-    if not YOUTUBE_API_KEY:
-        logger.warning("YOUTUBE_API_KEY not set. Video search disabled.")
-        return None
+    Query the RAG vectorstore to extract curriculum keywords and enrich
+    the YouTube search query for higher relevance.
+    """
+    enriched = topic
+    if subject:
+        enriched = f"{enriched} {subject}"
+    if lesson_context:
+        enriched = f"{enriched} {lesson_context}"
 
-    import urllib.parse
-    import json
+    try:
+        from rag.curriculum_rag import retrieve_curriculum_context
+        chunks = retrieve_curriculum_context(
+            query=topic,
+            subject=subject if subject else None,
+            top_k=5,
+        )
+        if chunks:
+            # Extract key terms from top chunk contents
+            keywords: List[str] = []
+            for chunk in chunks[:3]:
+                content = str(chunk.get("content", "")).strip()
+                # Extract meaningful words (skip math symbols, numbers, stop words)
+                if content:
+                    # Clean content: remove special chars, keep only alphabetic words
+                    cleaned = re.sub(r'[^\w\s]', ' ', content)
+                    words = [w for w in cleaned.split() if len(w) > 3 and w.isalpha()]
+                    # Take up to 5 key words per chunk
+                    keywords.extend(words[:5])
+            if keywords:
+                # Deduplicate and limit
+                unique_keywords = list(dict.fromkeys(keywords))[:8]
+                keyword_str = " ".join(unique_keywords)
+                enriched = f"{enriched} {keyword_str}"
+    except Exception as exc:
+        logger.debug("RAG enrichment skipped: %s", exc)
+
+    # Append standard DepEd/Philippines math context
+    enriched = f"{enriched} DepEd Philippines mathematics tutorial"
+    return enriched[:300]
+
+
+def _get_cache_key(topic: str, subject: str, grade_level: str) -> str:
+    """Generate a deterministic Firestore document ID for caching."""
+    raw = f"{subject}|{topic}|{grade_level}"
+    return hashlib.sha256(raw.encode("utf-8")).hexdigest()[:32]
+
+
+def get_cached_videos(lesson_id: str) -> Optional[List[Dict]]:
+    """Check Firestore video_cache/{lessonId} for cached results (TTL 7 days)."""
+    try:
+        import firebase_admin
+        from firebase_admin import firestore
+        if not firebase_admin._apps:
+            return None
 
-    search_query = f"{query} DepEd Philippines Grade 11 Grade 12 mathematics"
-    encoded_query = urllib.parse.quote(search_query)
+        db = firestore.client()
+        doc_ref = db.collection("video_cache").document(lesson_id)
+        doc = doc_ref.get()
+        if not doc.exists:
+            return None
 
-    search_url = (
-        f"https://www.googleapis.com/youtube/v3/search"
-        f"?part=snippet&type=video&q={encoded_query}"
-        f"&maxResults={max_results}&relevanceLanguage={language}"
-        f"&key={YOUTUBE_API_KEY}"
-    )
+        data = doc.to_dict()
+        if not data:
+            return None
 
+        cached_at = data.get("cachedAt")
+        if cached_at:
+            # Firestore timestamps have a .timestamp() method or are datetime objects
+            if hasattr(cached_at, "timestamp"):
+                cached_epoch = cached_at.timestamp()
+            elif isinstance(cached_at, datetime):
+                cached_epoch = cached_at.timestamp()
+            else:
+                cached_epoch = float(cached_at)
+            now_epoch = datetime.now(timezone.utc).timestamp()
+            if (now_epoch - cached_epoch) > _CACHE_TTL_SECONDS:
+                logger.info("Video cache expired for lesson %s", lesson_id)
+                return None
+
+        videos = data.get("videos")
+        if isinstance(videos, list) and len(videos) > 0:
+            logger.info("Video cache hit for lesson %s (%d videos)", lesson_id, len(videos))
+            return videos
+    except Exception as exc:
+        logger.debug("Could not read video cache: %s", exc)
+    return None
+
+
+def cache_videos(lesson_id: str, videos: List[Dict], topic: str) -> None:
+    """Store search results in Firestore video_cache/{lessonId}."""
     try:
-        import urllib.request
-        with urllib.request.urlopen(search_url, timeout=10) as response:
-            data = json.loads(response.read().decode())
+        import firebase_admin
+        from firebase_admin import firestore
+        if not firebase_admin._apps:
+            return
+
+        db = firestore.client()
+        db.collection("video_cache").document(lesson_id).set({
+            "videos": videos,
+            "cachedAt": firestore.SERVER_TIMESTAMP,
+            "topic": topic,
+        })
+        logger.info("Cached %d videos for lesson %s", len(videos), lesson_id)
+    except Exception as exc:
+        logger.warning("Could not cache videos in Firestore: %s", exc)
+
+
+def search_youtube_videos(
+    topic: str,
+    subject: str = "",
+    lesson_context: str = "",
+    grade_level: str = "",
+    max_results: int = 3,
+    language: str = "en",
+) -> List[Dict]:
+    """
+    Search YouTube Data API v3 for relevant educational math videos.
+
+    Returns up to `max_results` videos after applying filters:
+    - Educational channels (post-filter by channel name)
+    - Medium/long duration (>= 3 minutes, <= 60 minutes)
+    - HD quality preferred (videoDefinition = high)
+    - English or Filipino language
+
+    Each result contains: videoId, title, channelTitle, thumbnailUrl, durationSeconds.
+    """
+    client = _build_youtube_client()
+    if client is None:
+        logger.warning("YOUTUBE_API_KEY not set. Video search disabled.")
+        return []
+
+    # Step 1: Enrich query with RAG curriculum context
+    enriched_query = _enrich_query_with_rag(topic, subject, lesson_context)
+    logger.info("YouTube search query (enriched): %s", enriched_query)
 
-        video_results = []
-        for item in data.get("items", []):
+    try:
+        # Step 2: Search for videos
+        search_response = client.search().list(
+            part="snippet",
+            q=enriched_query,
+            type="video",
+            maxResults=15,  # Fetch more to allow post-filtering
+            relevanceLanguage=language,
+            videoDefinition="high",
+            videoDuration="medium",  # 4-20 minutes
+            safeSearch="strict",
+            order="relevance",
+        ).execute()
+
+        items = search_response.get("items", [])
+        if not items:
+            logger.info("No YouTube results for query: %s", enriched_query)
+            return []
+
+        # Step 3: Get video details (duration, etc.)
+        video_ids = [item["id"]["videoId"] for item in items if item.get("id", {}).get("videoId")]
+        if not video_ids:
+            return []
+
+        details_response = client.videos().list(
+            part="contentDetails,statistics,snippet",
+            id=",".join(video_ids),
+        ).execute()
+
+        details_map = {}
+        for detail in details_response.get("items", []):
+            vid = detail.get("id")
+            if vid:
+                details_map[vid] = detail
+
+        # Step 4: Build results with filtering
+        results = []
+        for item in items:
             video_id = item.get("id", {}).get("videoId", "")
             if not video_id:
                 continue
 
-            title = item.get("snippet", {}).get("title", "")
-            channel = item.get("snippet", {}).get("channelTitle", "")
-            description = item.get("snippet", {}).get("description", "")
-
-            video_details_url = (
-                f"https://www.googleapis.com/youtube/v3/videos"
-                f"?part=contentDetails,statistics&id={video_id}&key={YOUTUBE_API_KEY}"
-            )
-
-            try:
-                with urllib.request.urlopen(video_details_url, timeout=10) as vd_response:
-                    vd_data = json.loads(vd_response.read().decode())
-                    vd_item = vd_data.get("items", [{}])[0]
-                    content_details = vd_item.get("contentDetails", {})
-                    duration = content_details.get("duration", "")
+            detail = details_map.get(video_id, {})
+            snippet = detail.get("snippet", item.get("snippet", {}))
+            content_details = detail.get("contentDetails", {})
 
-                    duration_secs = _parse_iso8601_duration(duration)
-            except Exception:
-                duration_secs = 600
+            title = snippet.get("title", "")
+            channel_title = snippet.get("channelTitle", "")
+            description = snippet.get("description", "")
+            duration = content_details.get("duration", "")
+            duration_secs = _parse_iso8601_duration(duration)
 
-            if duration_secs < min_duration_seconds:
+            # Filter: duration
+            if duration_secs < _MIN_DURATION_SECONDS or duration_secs > _MAX_DURATION_SECONDS:
                 continue
 
-            embed_url = f"https://www.youtube.com/embed/{video_id}"
-            thumbnail_url = f"https://img.youtube.com/vi/{video_id}/mqdefault.jpg"
+            # Filter: educational channels
+            if not _is_educational_channel(channel_title):
+                # Still allow if title strongly suggests math tutorial
+                lowered_title = title.lower()
+                if not any(term in lowered_title for term in ["tutorial", "lesson", "math", "explain"]):
+                    continue
 
-            video_results.append({
+            thumbnail_url = f"https://img.youtube.com/vi/{video_id}/mqdefault.jpg"
+            # Prefer hqdefault if available
+            thumbs = snippet.get("thumbnails", {})
+            if "high" in thumbs:
+                thumbnail_url = thumbs["high"]["url"]
+            elif "medium" in thumbs:
+                thumbnail_url = thumbs["medium"]["url"]
+
+            results.append({
                 "videoId": video_id,
-                "videoTitle": title,
-                "videoChannel": channel,
-                "embedUrl": embed_url,
+                "title": title,
+                "channelTitle": channel_title,
                 "thumbnailUrl": thumbnail_url,
                 "durationSeconds": duration_secs,
-                "description": description[:200],
+                "description": description[:200] if description else "",
             })
 
-        if not video_results:
-            return None
+        # Step 5: Score and sort, return top N
+        for r in results:
+            r["_score"] = _score_video_result(r, enriched_query)
 
-        for vr in video_results:
-            if any(term in vr["videoTitle"].lower() or term in vr["description"].lower()
-                   for term in ["tutorial", "lesson", "explain", "math", "solution"]):
-                return vr
+        results.sort(key=lambda x: x["_score"], reverse=True)
+        for r in results:
+            r.pop("_score", None)
 
-        return video_results[0] if video_results else None
+        top_results = results[:max_results]
+        logger.info("YouTube search returned %d results (top %d)", len(results), len(top_results))
+        return top_results
 
-    except Exception as e:
-        logger.error("YouTube search failed: %s", e)
-        return None
+    except Exception as exc:
+        logger.error("YouTube search failed: %s", exc)
+        return []
 
 
-def get_video_for_lesson(
-    lesson_title: str,
-    subject: str,
-    competency: str = "",
-    quarter: int = 1,
-) -> Optional[dict]:
-    """Get the best YouTube video for a lesson."""
-    query = " ".join(filter(None, [lesson_title, subject, competency]))[:200]
-    return search_youtube_video(query)
-
+def get_video_search_results(
+    topic: str,
+    subject: str = "",
+    lesson_context: str = "",
+    grade_level: str = "",
+    lesson_id: Optional[str] = None,
+    max_results: int = 3,
+) -> Dict:
+    """
+    High-level wrapper: check cache first, then search YouTube, then cache results.
 
-def store_video_in_firestore(lesson_id: str, video_data: dict):
-    """Persist chosen video to Firestore for caching."""
-    try:
-        import firebase_admin
-        from firebase_admin import firestore
-        if not firebase_admin._apps:
-            return
-        db = firestore.client()
-        doc_ref = db.collection("curriculumDocuments").document(lesson_id)
-        doc_ref.collection("videoEmbed").document("primary").set({
-            **video_data,
-            "storedAt": firestore.SERVER_TIMESTAMP,
-        })
-    except Exception as e:
-        logger.warning("Could not store video in Firestore: %s", e)
+    Returns {"videos": [...], "cached": bool}.
+    """
+    cache_key = lesson_id or _get_cache_key(topic, subject, grade_level)
+
+    # Check cache first
+    cached = get_cached_videos(cache_key)
+    if cached is not None:
+        return {"videos": cached, "cached": True}
+
+    # Search YouTube
+    videos = search_youtube_videos(
+        topic=topic,
+        subject=subject,
+        lesson_context=lesson_context,
+        grade_level=grade_level,
+        max_results=max_results,
+    )
 
+    if videos:
+        cache_videos(cache_key, videos, topic)
 
-def get_cached_video(lesson_id: str) -> Optional[dict]:
-    """Retrieve cached video from Firestore."""
-    try:
-        import firebase_admin
-        from firebase_admin import firestore
-        if not firebase_admin._apps:
-            return None
-        db = firestore.client()
-        doc = db.collection("curriculumDocuments").document(lesson_id)
-        video_doc = doc.collection("videoEmbed").document("primary").get()
-        if video_doc.exists:
-            return video_doc.to_dict()
-    except Exception:
-        pass
-    return None
+    return {"videos": videos, "cached": False}

tests/test_video_routes.py

@@ -0,0 +1,209 @@
+"""
+Tests for the video search endpoint and YouTube service.
+"""
+
+from __future__ import annotations
+
+import os
+import sys
+from unittest.mock import MagicMock, patch
+
+import pytest
+from fastapi.testclient import TestClient
+
+# Add backend directory to path
+sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))
+
+# Mock Firebase auth BEFORE importing the app
+from main import app as _app_import
+import main as main_module
+
+if getattr(main_module, "firebase_auth", None) is None:
+    main_module.firebase_auth = MagicMock()
+main_module.firebase_auth.verify_id_token = MagicMock(
+    return_value={
+        "uid": "test-student-uid",
+        "email": "student@example.com",
+        "role": "student",
+    }
+)
+
+client = TestClient(_app_import, headers={"Authorization": "Bearer test-auth-token"})
+
+
+# ─── Fixtures ───────────────────────────────────────────────────
+
+@pytest.fixture
+def mock_youtube_api_key(monkeypatch):
+    monkeypatch.setenv("YOUTUBE_API_KEY", "test_youtube_api_key")
+
+
+@pytest.fixture
+def no_youtube_api_key(monkeypatch):
+    monkeypatch.setenv("YOUTUBE_API_KEY", "")
+
+
+# ─── YouTube Service Tests ──────────────────────────────────────
+
+def test_parse_iso8601_duration():
+    from services.youtube_service import _parse_iso8601_duration
+    assert _parse_iso8601_duration("PT5M30S") == 330
+    assert _parse_iso8601_duration("PT1H2M3S") == 3723
+    assert _parse_iso8601_duration("PT0S") == 0
+    assert _parse_iso8601_duration("") == 0
+
+
+def test_is_educational_channel():
+    from services.youtube_service import _is_educational_channel
+    assert _is_educational_channel("Khan Academy") is True
+    assert _is_educational_channel("Math Antics") is True
+    assert _is_educational_channel("3Blue1Brown") is True
+    assert _is_educational_channel("Gaming Channel") is False
+    assert _is_educational_channel("Random Vlogs") is False
+
+
+def test_enrich_query_with_rag_fallback(monkeypatch):
+    """When RAG is unavailable, enrichment falls back to topic + subject."""
+    from services.youtube_service import _enrich_query_with_rag
+    # Mock RAG to simulate unavailability — patch where it's used, not where it's imported
+    with patch("rag.curriculum_rag.retrieve_curriculum_context", side_effect=Exception("RAG unavailable")):
+        result = _enrich_query_with_rag("quadratic equations", "General Mathematics")
+    assert "quadratic equations" in result
+    assert "General Mathematics" in result
+    assert "DepEd Philippines mathematics" in result
+
+
+def test_get_cache_key():
+    from services.youtube_service import _get_cache_key
+    key1 = _get_cache_key("quadratic equations", "General Mathematics", "Grade 11")
+    key2 = _get_cache_key("quadratic equations", "General Mathematics", "Grade 11")
+    key3 = _get_cache_key("linear equations", "General Mathematics", "Grade 11")
+    assert key1 == key2
+    assert key1 != key3
+    assert len(key1) == 32
+
+
+def test_cache_and_retrieve(mock_youtube_api_key, monkeypatch):
+    from services.youtube_service import cache_videos, get_cached_videos
+
+    lesson_id = "test-lesson-123"
+    videos = [
+        {"videoId": "abc123", "title": "Test Video", "channelTitle": "Test Channel",
+         "thumbnailUrl": "http://example.com/thumb.jpg", "durationSeconds": 300}
+    ]
+
+    # Mock Firebase at the module level where it's imported inside functions
+    mock_doc = MagicMock()
+    mock_doc.get.return_value.exists = False
+    mock_db = MagicMock()
+    mock_db.collection.return_value.document.return_value = mock_doc
+
+    with patch("firebase_admin.firestore.client", return_value=mock_db):
+        with patch("firebase_admin._apps", {"default": MagicMock()}):
+            # Store should call set
+            cache_videos(lesson_id, videos, "quadratic equations")
+            mock_doc.set.assert_called_once()
+
+            # Retrieve should return None since we mock doc.exists = False
+            result = get_cached_videos(lesson_id)
+            assert result is None
+
+
+def test_search_youtube_videos_no_api_key(no_youtube_api_key):
+    from services.youtube_service import search_youtube_videos
+    result = search_youtube_videos("quadratic equations")
+    assert result == []
+
+
+# ─── Route Tests ────────────────────────────────────────────────
+
+def test_video_search_endpoint_no_api_key(no_youtube_api_key):
+    """Should return 503 when YouTube API key is not configured."""
+    response = client.post("/api/lessons/videos/search", json={
+        "topic": "quadratic equations",
+        "subject": "General Mathematics",
+        "grade_level": "Grade 11",
+    })
+    assert response.status_code == 503
+    data = response.json()
+    assert data["detail"]["error"] == "youtube_api_not_configured"
+
+
+def test_video_search_endpoint_success(mock_youtube_api_key):
+    """Should return video results when search succeeds."""
+    mock_videos = [
+        {"videoId": "vid1", "title": "Video 1", "channelTitle": "Channel 1",
+         "thumbnailUrl": "http://example.com/1.jpg", "durationSeconds": 300},
+        {"videoId": "vid2", "title": "Video 2", "channelTitle": "Channel 2",
+         "thumbnailUrl": "http://example.com/2.jpg", "durationSeconds": 450},
+    ]
+
+    with patch("routes.video_routes.YOUTUBE_API_KEY", "test_key"):
+        with patch("routes.video_routes.get_video_search_results") as mock_search:
+            mock_search.return_value = {"videos": mock_videos, "cached": False}
+            response = client.post("/api/lessons/videos/search", json={
+                "topic": "quadratic equations",
+                "subject": "General Mathematics",
+                "grade_level": "Grade 11",
+                "lesson_id": "lesson-123",
+            })
+
+    assert response.status_code == 200
+    data = response.json()
+    assert len(data["videos"]) == 2
+    assert data["cached"] is False
+    assert data["videos"][0]["videoId"] == "vid1"
+
+
+def test_video_search_endpoint_empty_results(mock_youtube_api_key):
+    """Should return empty list when no videos found."""
+    with patch("routes.video_routes.YOUTUBE_API_KEY", "test_key"):
+        with patch("routes.video_routes.get_video_search_results") as mock_search:
+            mock_search.return_value = {"videos": [], "cached": False}
+            response = client.post("/api/lessons/videos/search", json={
+                "topic": "very obscure topic xyz123",
+                "subject": "General Mathematics",
+            })
+
+    assert response.status_code == 200
+    data = response.json()
+    assert data["videos"] == []
+    assert data["cached"] is False
+
+
+def test_video_search_endpoint_cached(mock_youtube_api_key):
+    """Should return cached results."""
+    mock_videos = [
+        {"videoId": "vid1", "title": "Cached Video", "channelTitle": "Channel 1",
+         "thumbnailUrl": "http://example.com/1.jpg", "durationSeconds": 300},
+    ]
+
+    with patch("routes.video_routes.YOUTUBE_API_KEY", "test_key"):
+        with patch("routes.video_routes.get_video_search_results") as mock_search:
+            mock_search.return_value = {"videos": mock_videos, "cached": True}
+            response = client.post("/api/lessons/videos/search", json={
+                "topic": "linear equations",
+                "lesson_id": "lesson-456",
+            })
+
+    assert response.status_code == 200
+    data = response.json()
+    assert data["cached"] is True
+    assert len(data["videos"]) == 1
+
+
+def test_video_search_endpoint_validation_error(mock_youtube_api_key):
+    """Should return 422 when topic is missing or too long."""
+    with patch("routes.video_routes.YOUTUBE_API_KEY", "test_key"):
+        response = client.post("/api/lessons/videos/search", json={
+            "topic": "",
+            "subject": "General Mathematics",
+        })
+    assert response.status_code == 422
+
+    with patch("routes.video_routes.YOUTUBE_API_KEY", "test_key"):
+        response = client.post("/api/lessons/videos/search", json={
+            "topic": "x" * 201,
+            "subject": "General Mathematics",
+        })
+    assert response.status_code == 422