Update app.py

app.py

@@ -1,187 +1,122 @@
-from fastapi import FastAPI, Query, HTTPException
-import httpx
-from bs4 import BeautifulSoup
-import json
+import os
 import re
-from typing import List, Dict, Any
+from fastapi import FastAPI, Body, HTTPException, Query
+from googleapiclient.discovery import build
+from dotenv import load_dotenv
+from typing import List, Dict, Any, Optional
+
+# Load environment variables from .env file
+load_dotenv()
 
 app = FastAPI()
 
-async def _extract_videos_from_html(html_content: str) -> List[Dict[str, Any]]:
+# YouTube Data API Key
+GOOGLE_API_KEY = os.getenv("GOOGLE_API_KEY")
+
+def find_relevant_youtube_shorts(movie_title: str, order: str = "relevance", max_results: int = 10) -> List[Dict[str, Any]]:
     """
-    Helper function to parse YouTube HTML content and extract video data.
+    Finds relevant YouTube Shorts for a given movie title using YouTube Data API v3.
+
+    Args:
+        movie_title (str): The title of the movie or TV series.
+        order (str): The order to sort the search results. Can be "relevance", "viewCount", or "rating".
+        max_results (int): The maximum number of shorts to return.
+
+    Returns:
+        list: A list of dictionaries, each representing a found YouTube Short
+              with its title, link, video_id, channel, and published_at.
     """
-    soup = BeautifulSoup(html_content, 'html.parser')
-    
-    # Set headers to mimic a browser to reduce blocking
-    headers = {
-        "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/121.0.0.0 Safari/537.36",
-        "Accept-Language": "en-US,en;q=0.9",
-    }
-    
-    script_tags = soup.find_all('script')
-    yt_initial_data = None
-    for script in script_tags:
-        if script.string and 'var ytInitialData = ' in script.string:
-            json_str = script.string.split('var ytInitialData = ', 1)[1]
-            
-            # Robust JSON extraction:
-            # YouTube's JSON can sometimes have a trailing semicolon followed by other JS.
-            # We need to ensure we parse only the JSON object.
-            # Find the balance of curly braces to identify the end of the JSON.
-            balance = 0
-            json_end_index = -1
-            # Simple heuristic to find balanced brace for the main JSON object
-            # This might need more sophistication for edge cases
-            for i, char in enumerate(json_str):
-                if char == '{':
-                    balance += 1
-                elif char == '}':
-                    balance -= 1
-                
-                if balance == 0 and char == '}':
-                    json_end_index = i
-                    break
+    if not GOOGLE_API_KEY:
+        return []
+
+    youtube = build("youtube", "v3", developerKey=GOOGLE_API_KEY)
+
+    # Refined search query: More focused on movie title and direct shorts indicators.
+    # We'll rely more on post-filtering for excluding long-form.
+    # The query now explicitly requests "#shorts" or "YouTube Shorts" alongside the movie title.
+    search_query = f'"{movie_title}" #shorts OR "{movie_title}" "YouTube Shorts"'
+
+    try:
+        request = youtube.search().list(
+            q=search_query,
+            part="snippet",
+            type="video",
+            videoDuration="short",  # Filters videos under 4 minutes
+            order=order,
+            maxResults=max_results * 5, # Fetch more to allow for aggressive filtering
+        )
+        response = request.execute()
+
+        potential_shorts = []
+        for item in response.get('items', []):
+            if item['id']['kind'] == 'youtube#video':
+                video_id = item['id']['videoId']
+                snippet = item['snippet']
+
+                potential_shorts.append({
+                    "title": snippet['title'],
+                    "link": f"https://www.youtube.com/watch?v={video_id}",
+                    "video_id": video_id,
+                    "channel_title": snippet['channelTitle'],
+                    "published_at": snippet['publishedAt'],
+                    "description": snippet['description'],
+                    "thumbnails": snippet['thumbnails']['default']['url'],
+                })
+
+        # Stronger post-API filtering
+        final_shorts = []
+        # Indicators that strongly suggest long-form content
+        long_form_indicators = [
+            "full movie", "full episode", "season", "documentary", 
+            "trailer review", "official trailer", "compilation", "movie review",
+            "best scenes", "explained", "summary", "recap", "analysis", "episodes",
+            "movie in 10 minutes", "full story", "watch full"
+        ]
+        # Strong indicators that it is a short video
+        shorts_indicators = ["#shorts", "youtube shorts", "short video", "short clip", "vertical video"]
+
+        movie_title_lower = movie_title.lower()
+
+        for short in potential_shorts:
+            title_lower = short['title'].lower()
+            description_lower = short['description'].lower()
             
-            if json_end_index != -1:
-                json_str = json_str[:json_end_index + 1]
-            else: # Fallback if balancing failed, try splitting by semicolon
-                 json_str = json_str.split(');', 1)[0]
+            # Rule 1: Must contain a strong shorts indicator
+            has_strong_shorts_indicator = any(indicator in title_lower or indicator in description_lower for indicator in shorts_indicators)
+
+            # Rule 2: Ensure movie title (or part of it) is present in title/description for strong relevance
+            # Using split for multi-word titles to check for partial matches too
+            is_relevant_to_movie = movie_title_lower in title_lower or movie_title_lower in description_lower or \
+                                   any(word in title_lower or word in description_lower for word in movie_title_lower.split())
 
 
-            try:
-                yt_initial_data = json.loads(json_str)
+            # Rule 3: Must NOT contain strong long-form indicators
+            is_not_long_form = not any(indicator in title_lower or indicator in description_lower for indicator in long_form_indicators)
+            
+            if has_strong_shorts_indicator and is_relevant_to_movie and is_not_long_form:
+                final_shorts.append(short)
+            
+            if len(final_shorts) >= max_results:
                 break
-            except json.JSONDecodeError as e:
-                print(f"DEBUG: JSON Decode Error while parsing ytInitialData: {e}")
-                # print(f"DEBUG: Attempted JSON string (start): {json_str[:500]}...") # Too verbose for production
-                # print(f"DEBUG: Attempted JSON string (end): ...{json_str[-500:]}") # Too verbose for production
-                yt_initial_data = None # Reset to try next script tag if any
 
-    if not yt_initial_data:
-        print("DEBUG: ytInitialData not found or could not be parsed from any script tag.")
-        return [] 
+        return final_shorts
 
-    # Debugging: Print a snippet of the parsed ytInitialData structure (use sparingly)
-    # print(f"DEBUG: First 1000 chars of ytInitialData (for structure inspection): {json.dumps(yt_initial_data, indent=2)[:1000]}...")
+    except Exception as e:
+        print(f"An error occurred during YouTube Data API search for '{movie_title}': {e}")
+        return []
 
-    raw_videos_data = []
-    try:
-        # Common path for video renderers in search results
-        # This path is subject to YouTube UI changes. We try multiple known paths.
-        contents = yt_initial_data.get('contents', {})
-        
-        # Path 1: twoColumnSearchResultsRenderer -> primaryContents -> sectionListRenderer
-        sections = contents.get('twoColumnSearchResultsRenderer', {})\
-                           .get('primaryContents', {})\
-                           .get('sectionListRenderer', {})\
-                           .get('contents', [])
-
-        # Path 2: Fallback for different structures (e.g., singleColumnBrowseResultsRenderer for hashtag pages)
-        if not sections:
-             # This path is more common for hashtag pages or channels that are directly linked
-             sections = contents.get('singleColumnBrowseResultsRenderer', {})\
-                                .get('tabs', [{}])[0]\
-                                .get('tabRenderer', {})\
-                                .get('content', {})\
-                                .get('sectionListRenderer', {})\
-                                .get('contents', [])
-             if not sections:
-                 # Another common path for some layouts
-                 sections = contents.get('twoColumnBrowseResultsRenderer', {})\
-                                    .get('tabs', [{}])[0]\
-                                    .get('tabRenderer', {})\
-                                    .get('content', {})\
-                                    .get('sectionListRenderer', {})\
-                                    .get('contents', [])
-
-
-        for section in sections:
-            item_section_renderer = section.get('itemSectionRenderer', {})
-            for item in item_section_renderer.get('contents', []):
-                if 'videoRenderer' in item:
-                    video_data = item['videoRenderer']
-                    
-                    extracted_data = {
-                        "id": video_data.get('videoId'),
-                        "title": video_data.get('title', {}).get('runs', [{}])[0].get('text'),
-                        "description": video_data.get('descriptionSnippet', {}).get('runs', [{}])[0].get('text'),
-                        "duration_text": video_data.get('lengthText', {}).get('simpleText'),
-                        "view_count": video_data.get('viewCountText', {}).get('simpleText'),
-                        "published_time": video_data.get('publishedTimeText', {}).get('simpleText'),
-                        "channel_name": video_data.get('ownerText', {}).get('runs', [{}])[0].get('text'),
-                        "channel_id": video_data.get('ownerText', {}).get('runs', [{}])[0].get('navigationEndpoint', {}).get('browseEndpoint', {}).get('browseId'),
-                        "thumbnails": [thumb.get('url') for thumb in video_data.get('thumbnail', {}).get('thumbnails', [])],
-                        "url": f"https://www.youtube.com/watch?v={video_data.get('videoId')}" if video_data.get('videoId') else None,
-                        "badges": [badge.get('metadataBadgeRenderer', {}).get('label') for badge in video_data.get('badges', []) if 'metadataBadgeRenderer' in badge],
-                        "is_short_url": "shorts/" in (f"https://www.youtube.com/watch?v={video_data.get('videoId')}" if video_data.get('videoId') else "") # Indication if URL is a shorts URL
-                    }
-                    raw_videos_data.append(extracted_data)
-
-    except KeyError as ke:
-        print(f"DEBUG: YouTube data structure changed (KeyError) during extraction: {ke}")
-        return raw_videos_data
-    except IndexError as ie:
-        print(f"DEBUG: YouTube data structure changed (IndexError) during extraction: {ie}")
-        return raw_videos_data
-    except Exception as data_extract_e:
-        print(f"DEBUG: An unexpected error occurred during data extraction: {data_extract_e}")
-        return raw_videos_data
-
-    return raw_videos_data
-
-
-@app.get("/scrape_raw_shorts")
-async def scrape_raw_youtube_shorts(
-    query: str = Query(..., description="The search query for YouTube Shorts")
-) -> List[Dict[str, Any]]:
-    if not query:
-        raise HTTPException(status_code=400, detail="'query' parameter is required.")
-
-    all_shorts_combined = []
-    seen_video_ids = set()
-    
-    # Common headers to mimic a browser
-    headers = {
-        "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/121.0.0.0 Safari/537.36",
-        "Accept-Language": "en-US,en;q=0.9",
-    }
-
-    async with httpx.AsyncClient(headers=headers) as client:
-        # --- Method 1: Original Search Query --- #
-        print(f"\n--- Processing results from Search Query: {query} ---") # Separator for Method 1
-        search_url_1 = f"https://www.youtube.com/results?search_query={query}&sp=EgIQCQ%253D%253D"
-        try:
-            response_1 = await client.get(search_url_1)
-            response_1.raise_for_status()
-            shorts_from_search = await _extract_videos_from_html(response_1.text)
-            print(f"DEBUG: Found {len(shorts_from_search)} potential shorts from Search Query.") # Debugging count
-            for short in shorts_from_search:
-                if short and short.get('id') and short['id'] not in seen_video_ids:
-                    all_shorts_combined.append(short)
-                    seen_video_ids.add(short['id'])
-        except (httpx.HTTPStatusError, httpx.RequestError, Exception) as e:
-            print(f"ERROR: Error scraping from search_query ({search_url_1}): {e}")
-
-        # --- Method 2: Hashtag Search --- #
-        print(f"\n--- Processing results from Hashtag Search: {query} ---") # Separator for Method 2
-        query_hashtag = query.replace(" ", "_").replace("'", "").replace("(", "").replace(")", "").replace(".", "").lower() # Clean query for hashtag
-        # Ensure hashtag query is not empty after cleaning
-        if query_hashtag:
-            search_url_2 = f"https://www.youtube.com/hashtag/{query_hashtag}/shorts"
-            try:
-                response_2 = await client.get(search_url_2)
-                response_2.raise_for_status()
-                shorts_from_hashtag = await _extract_videos_from_html(response_2.text)
-                print(f"DEBUG: Found {len(shorts_from_hashtag)} potential shorts from Hashtag Search.") # Debugging count
-                for short in shorts_from_hashtag:
-                    if short and short.get('id') and short['id'] not in seen_video_ids:
-                        all_shorts_combined.append(short)
-                        seen_video_ids.add(short['id'])
-            except (httpx.HTTPStatusError, httpx.RequestError, Exception) as e:
-                print(f"ERROR: Error scraping from hashtag query ({search_url_2}): {e}")
-        else:
-            print(f"DEBUG: Hashtag query '{query_hashtag}' is empty after cleaning, skipping hashtag search.")
-
-    print(f"\n--- Total unique shorts found: {len(all_shorts_combined)} ---") # Final count
-    return all_shorts_combined
+@app.get("/get_shorts_for_movie")
+async def get_shorts_for_movie(
+    movie_title: str = Query(..., description="Title of the movie or TV series"),
+    search_order: Optional[str] = Query("relevance", description="Order by relevance, viewCount, or rating"),
+    max_shorts: Optional[int] = Query(10, description="Maximum number of shorts to return")
+):
+    if not movie_title:
+        raise HTTPException(status_code=400, detail="'movie_title' query parameter is required.")
+
+    if not GOOGLE_API_KEY:
+        raise HTTPException(status_code=500, detail="Google API Key is not configured.")
+
+    relevant_shorts = find_relevant_youtube_shorts(movie_title, order=search_order, max_results=max_shorts)
+
+    return {"movie_title": movie_title, "related_shorts": relevant_shorts}