Update app.py

app.py

@@ -4,57 +4,42 @@ import gradio as gr
 from googleapiclient.discovery import build
 from googleapiclient.errors import HttpError
 from youtube_transcript_api import YouTubeTranscriptApi, TranscriptsDisabled, NoTranscriptFound
-import time # Added for potential delays and UI updates
+import time
+import traceback
 
 # --- Constants ---
 YOUTUBE_API_SERVICE_NAME = "youtube"
 YOUTUBE_API_VERSION = "v3"
-DEFAULT_KEYWORDS = (
-    '3d, texturing, rigging, vfx, cgi'
-)
+DEFAULT_KEYWORDS = "3d, texturing, rigging, vfx, cgi"
+API_KEY = "AIzaSyCcxSkhTp6aowcyowuBkHIFTSrl_HJ79J0"  # Replace with your actual YouTube Data API key
 
-# --- YouTube API Helper Functions (modified slightly for Gradio context) ---
+# --- YouTube API Helper Functions ---
 
-# Cache the service object to potentially reuse if API key doesn't change
-# NOTE: In a real multi-user scenario, this simple cache isn't safe.
-# For Gradio's typical single-process use, it might offer minor speedup
-# if the same key is used repeatedly, but building it each time is safer.
-# We will build it each time within the main Gradio function for simplicity & correctness.
-
-def get_youtube_service(api_key):
-    """Initializes and returns the YouTube API service."""
-    if not api_key:
+def get_youtube_service():
+    """Initializes and returns the YouTube API service with hardcoded API key."""
+    if not API_KEY:
         raise ValueError("API Key is missing.")
     try:
-        # Disable cache discovery for reliability in different environments
-        # Might slightly slow down the first request but avoids potential issues.
         service = build(YOUTUBE_API_SERVICE_NAME, YOUTUBE_API_VERSION,
-                        developerKey=api_key, cache_discovery=False)
+                       developerKey=API_KEY, cache_discovery=False)
         return service
     except HttpError as e:
         if e.resp.status == 400:
-             raise ValueError(f"Invalid API Key or API not enabled. Error: {e.content}")
+            raise ValueError(f"Invalid API Key or API not enabled. Error: {e.content}")
         elif e.resp.status == 403:
-             raise ValueError(f"API Key valid, but Quota Exceeded or Forbidden. Error: {e.content}")
+            raise ValueError(f"API Key valid, but Quota Exceeded or Forbidden. Error: {e.content}")
         else:
-             raise ConnectionError(f"Could not connect to YouTube API: {e}")
+            raise ConnectionError(f"Could not connect to YouTube API: {e}")
     except Exception as e:
         raise ConnectionError(f"Error building YouTube service: {e}")
 
-
 def get_channel_id(service, channel_identifier):
     """Gets the channel ID using the channel handle or ID."""
     if not channel_identifier:
         raise ValueError("Channel Handle or ID is missing.")
-
-    # If it looks like a channel ID already
     if channel_identifier.startswith("UC") and len(channel_identifier) == 24:
-        # Optional: Verify it's a valid channel ID (API call)
-        # For speed, we'll trust the user if it looks like an ID
         print(f"Assuming '{channel_identifier}' is a Channel ID.")
         return channel_identifier
-
-    # If it looks like a handle
     if channel_identifier.startswith('@'):
         handle = channel_identifier
         print(f"Attempting to find Channel ID for handle: {handle}")
@@ -65,9 +50,8 @@ def get_channel_id(service, channel_identifier):
                 type="channel",
                 maxResults=1
             ).execute()
-
             if not search_response.get("items"):
-                 raise ValueError(f"Could not find channel for handle '{handle}'. Check the handle.")
+                raise ValueError(f"Could not find channel for handle '{handle}'. Check the handle.")
             channel_id = search_response["items"][0]["id"]["channelId"]
             print(f"Found Channel ID: {channel_id}")
             return channel_id
@@ -78,7 +62,6 @@ def get_channel_id(service, channel_identifier):
     else:
         raise ValueError("Invalid Channel Identifier. Use '@handle' or 'UC...' Channel ID.")
 
-
 def get_channel_uploads_playlist_id(service, channel_id):
     """Gets the uploads playlist ID for a given channel ID."""
     try:
@@ -86,26 +69,22 @@ def get_channel_uploads_playlist_id(service, channel_id):
             id=channel_id,
             part="contentDetails"
         ).execute()
-
         if not channels_response.get("items"):
             raise ValueError(f"Could not find channel details for ID '{channel_id}'.")
-
-        playlist_id = channels_response["items"][0].get("contentDetails", {}).get("relatedPlaylists", {}).get("uploads")
+        playlist_id = channels_response["items"][0]["contentDetails"]["relatedPlaylists"]["uploads"]
         if not playlist_id:
-            raise ValueError(f"Could not find the 'uploads' playlist ID for channel {channel_id}. The channel might have no public videos or API returned unexpected data.")
+            raise ValueError(f"Could not find uploads playlist for channel {channel_id}.")
         return playlist_id
     except HttpError as e:
         raise ConnectionError(f"API error getting uploads playlist for {channel_id}: {e.content}")
     except Exception as e:
         raise Exception(f"Error getting uploads playlist for {channel_id}: {e}")
 
-
 def get_all_video_ids(service, playlist_id):
-    """Gets all video IDs from a specific playlist. Yields status updates."""
+    """Gets all video IDs from a playlist, yielding status updates."""
     video_ids = []
     next_page_token = None
     fetched_count = 0
-
     yield f"Fetching video IDs from playlist: {playlist_id}..."
     while True:
         try:
@@ -115,147 +94,99 @@ def get_all_video_ids(service, playlist_id):
                 maxResults=50,
                 pageToken=next_page_token
             ).execute()
-
             current_page_ids = [item["contentDetails"]["videoId"]
-                                for item in playlist_items_response.get("items", [])
-                                if item.get("contentDetails", {}).get("videoId")] # Ensure videoId exists
+                              for item in playlist_items_response.get("items", [])]
             video_ids.extend(current_page_ids)
             fetched_count += len(current_page_ids)
-
             next_page_token = playlist_items_response.get("nextPageToken")
-
             yield f"Fetched {fetched_count} video IDs so far..."
-
             if not next_page_token:
                 break
-            # time.sleep(0.05) # Small delay between page fetches
-
         except HttpError as e:
-             # Check for quota error specifically
             if e.resp.status == 403:
-                yield f"API Quota Error while fetching video list page. Stopping early. Results might be incomplete. Error: {e.content}"
-                break # Stop fetching, return what we have
+                yield f"API Quota Error: {e.content}. Stopping early."
+                break
             else:
-                yield f"API Error fetching video list page: {e.content}. Stopping early."
-                break # Stop trying on other errors
+                yield f"API Error fetching video list: {e.content}. Stopping early."
+                break
         except Exception as e:
-            yield f"Error fetching video list page: {e}. Stopping early."
+            yield f"Error fetching video list: {e}. Stopping early."
             break
-
     yield f"Finished fetching. Total unique video IDs found: {len(video_ids)}"
     return video_ids
 
-
 def process_video(service, video_id, keywords_set):
-    """Fetches video details, transcript, and searches for keywords. Returns dict or None."""
+    """Fetches video details and transcript, searches for keywords."""
     video_url = f"https://www.youtube.com/watch?v={video_id}"
     found_data = {
         "video_id": video_id,
         "video_url": video_url,
-        "title": f"Video ID: {video_id}", # Default title
+        "title": f"Video ID: {video_id}",
         "transcript_mentions": set(),
         "description_mentions": set(),
         "description_links": []
     }
     has_mention = False
-    status_updates = [] # Collect status messages for this video
+    status_updates = []
 
-    # 1. Get Video Details (Title and Description)
     try:
         video_response = service.videos().list(
             id=video_id,
             part="snippet"
         ).execute()
-
         if video_response.get("items"):
-            snippet = video_response["items"][0].get("snippet", {})
+            snippet = video_response["items"][0]["snippet"]
             found_data["title"] = snippet.get("title", f"Video ID: {video_id}")
             description = snippet.get("description", "").lower()
-
-            # Search description for keywords
             for keyword in keywords_set:
-                # Use word boundaries for more precise matching (optional, but reduces false positives)
-                # pattern = r'\b' + re.escape(keyword) + r'\b'
-                # if re.search(pattern, description):
-                if keyword in description: # Simpler check
+                if keyword in description:
                     found_data["description_mentions"].add(keyword)
                     has_mention = True
-
-            # Extract links from description
             found_data["description_links"] = re.findall(r'https?://\S+', snippet.get("description", ""))
         else:
-             status_updates.append(f"  - Could not retrieve details for video {video_id}")
-
+            status_updates.append(f"  - Could not retrieve details for video {video_id}")
     except HttpError as e:
         status_updates.append(f"  - API error getting details for {video_id}: {e.resp.status}")
     except Exception as e:
         status_updates.append(f"  - Error getting details for {video_id}: {e}")
 
-    # 2. Get Transcript
     transcript_text = ""
     try:
         transcript_list = YouTubeTranscriptApi.list_transcripts(video_id)
-        try:
-            transcript = transcript_list.find_transcript(['en', 'en-US', 'en-GB']) # Prefers manual, falls back to generated
-        except NoTranscriptFound:
-            # Try common non-English potentially auto-translatable? Might be too broad.
-            # Let's stick to English for now. User can add keywords in other languages if needed.
-            status_updates.append(f"  - No English transcript found for {video_id}.")
-            transcript = None
-
+        transcript = transcript_list.find_transcript(['en', 'en-US', 'en-GB'])
         if transcript:
             full_transcript = transcript.fetch()
-            # Limit transcript length for processing? Maybe not needed for keyword search.
             transcript_text = " ".join([segment['text'] for segment in full_transcript]).lower()
-
-            # Search transcript for keywords
             for keyword in keywords_set:
-                 # pattern = r'\b' + re.escape(keyword) + r'\b'
-                 # if re.search(pattern, transcript_text):
-                 if keyword in transcript_text: # Simpler check
+                if keyword in transcript_text:
                     found_data["transcript_mentions"].add(keyword)
                     has_mention = True
-
-    except TranscriptsDisabled:
-        status_updates.append(f"  - Transcripts disabled for {video_id}")
-    except NoTranscriptFound:
-         # Handled above but catch just in case logic changes
-         status_updates.append(f"  - No transcript found (checked again) for {video_id}")
+    except (TranscriptsDisabled, NoTranscriptFound):
+        status_updates.append(f"  - No transcript available for {video_id}")
     except Exception as e:
-        # Catch potential network errors or library issues
         status_updates.append(f"  - Error fetching transcript for {video_id}: {type(e).__name__}")
 
-    if has_mention:
-        # Convert sets back to lists for consistency if needed later, but sets are fine for display logic
-        return found_data, status_updates
-    else:
-        return None, status_updates # Return None for data if no mentions
-
+    return found_data if has_mention else None, status_updates
 
 # --- Gradio Main Function ---
 
-def scan_channel_videos(api_key, channel_identifier, keywords_str, progress=gr.Progress(track_ τότε=True)):
-    """Main function called by Gradio interface."""
+def scan_channel_videos(channel_identifier, keywords_str, progress=gr.Progress(track_tqdm=True)):
+    """Main function called by Gradio interface with hardcoded API key."""
     start_time = time.time()
     status_log = []
     results = []
 
     def log_status(message):
-        print(message) # Log to console as well
+        print(message)
         status_log.append(message)
-        # Yield intermediate updates to the status textbox
-        yield "\n".join(status_log), gr.Markdown("### Processing...") # Keep results empty/processing
+        yield "\n".join(status_log), gr.Markdown("### Processing...")
 
     try:
         yield from log_status("1. Initializing YouTube Service...")
-        if not api_key:
-            raise ValueError("API Key is required.")
-        service = get_youtube_service(api_key)
+        service = get_youtube_service()
         yield from log_status("   Service Initialized.")
 
         yield from log_status(f"2. Finding Channel ID for '{channel_identifier}'...")
-        if not channel_identifier:
-             raise ValueError("Channel Handle or ID is required.")
         channel_id = get_channel_id(service, channel_identifier)
         yield from log_status(f"   Found Channel ID: {channel_id}")
 
@@ -264,62 +195,33 @@ def scan_channel_videos(api_key, channel_identifier, keywords_str, progress=gr.P
         yield from log_status(f"   Found Uploads Playlist ID: {uploads_playlist_id}")
 
         yield from log_status("4. Fetching all Video IDs...")
-        # Use a generator to yield status updates from within get_all_video_ids
         video_ids_generator = get_all_video_ids(service, uploads_playlist_id)
         video_ids = []
-        while True:
-            try:
-                status_or_list = next(video_ids_generator)
-                if isinstance(status_or_list, str):
-                     yield from log_status(f"   {status_or_list}")
-                else: # We expect the final list at the end
-                    video_ids = status_or_list
-                    # The last status is already yielded by the function
-                    # yield from log_status(f"   Finished fetching. Total unique video IDs found: {len(video_ids)}")
-                    break
-            except StopIteration:
-                 # Generator finished, means it yielded the list last
-                 # This case might happen if the generator structure changes, handle defensively
-                 if not video_ids: # If we never got the list (e.g., error occurred)
-                      yield from log_status("   Warning: Video ID fetching may have stopped unexpectedly.")
-                 break # Exit loop
-            except Exception as e:
-                 yield from log_status(f"   Error during video ID fetching: {e}")
-                 break # Stop if error occurs
-
+        for status_or_list in video_ids_generator:
+            if isinstance(status_or_list, str):
+                yield from log_status(f"   {status_or_list}")
+            else:
+                video_ids = status_or_list
+                break
 
         if not video_ids:
-            yield from log_status("   No video IDs found or fetching failed. Cannot proceed.")
-            # Final yield before returning
-            yield "\n".join(status_log), gr.Markdown("### Scan Stopped\nNo videos found or API error during fetch.")
+            yield from log_status("   No video IDs found or fetching failed.")
+            yield "\n".join(status_log), gr.Markdown("### Scan Stopped\nNo videos found.")
             return
 
-        # Prepare keywords
         keywords_list = [k.strip().lower() for k in keywords_str.split(',') if k.strip()]
         if not keywords_list:
             raise ValueError("Please provide at least one keyword.")
-        keywords_set = set(keywords_list) # Use set for faster lookups
+        keywords_set = set(keywords_list)
         yield from log_status(f"5. Scanning {len(video_ids)} videos for keywords: {', '.join(keywords_list)}")
 
         total_videos = len(video_ids)
-        processed_count = 0
-        # Process videos with progress tracking
         for video_id in progress.tqdm(video_ids, desc="Scanning Videos"):
-            processed_count += 1
             video_result, video_statuses = process_video(service, video_id, keywords_set)
-            # Log minor statuses only if needed for debugging, keep main log cleaner
-            # for status in video_statuses:
-            #     yield from log_status(status) # This can make the log very long
-
             if video_result:
                 results.append(video_result)
                 yield from log_status(f"   Found mention in: {video_result['title']} ({video_id})")
 
-            # Add a small delay to be nice to APIs, especially transcript API
-            # time.sleep(0.1) # Adjust as needed
-
-
-        # 6. Format Results
         yield from log_status("\n6. Scan Complete. Formatting results...")
         final_md = f"## Scan Results for {channel_identifier}\n\n"
         final_md += f"Searched for keywords: `{', '.join(keywords_list)}`\n"
@@ -331,75 +233,56 @@ def scan_channel_videos(api_key, channel_identifier, keywords_str, progress=gr.P
             final_md += "\n**No mentions found for the specified keywords.**"
         else:
             for res in results:
-                final_md += f"\n### [{res['title']}]({res['video_url']})\n" # Clickable title
-                final_md += f"*Video URL: <{res['video_url']}>*\n\n" # Explicit URL
+                final_md += f"\n### [{res['title']}]({res['video_url']})\n"
+                final_md += f"*Video URL: <{res['video_url']}>*\n\n"
                 if res['transcript_mentions']:
-                     mentions = ", ".join(sorted(list(res['transcript_mentions'])))
-                     final_md += f"**Transcript Mentions:** `{mentions}`\n"
+                    mentions = ", ".join(sorted(res['transcript_mentions']))
+                    final_md += f"**Transcript Mentions:** `{mentions}`\n"
                 if res['description_mentions']:
-                     mentions = ", ".join(sorted(list(res['description_mentions'])))
-                     final_md += f"**Description Mentions:** `{mentions}`\n"
+                    mentions = ", ".join(sorted(res['description_mentions']))
+                    final_md += f"**Description Mentions:** `{mentions}`\n"
                 if res['description_links']:
-                     final_md += f"\n**Links in Description:**\n"
-                     # Try to highlight links potentially related to the software
-                     possible_keywords = set()
-                     for k in keywords_set:
-                         parts = k.split()
-                         possible_keywords.add(parts[0]) # e.g., add 'blender' from 'blender foundation'
-
-                     for link in res['description_links']:
-                         is_potentially_related = any(pk in link.lower() for pk in possible_keywords if pk)
-                         marker = " (*)" if is_potentially_related else ""
-                         final_md += f"- <{link}>{marker}\n" # Use angle brackets for auto-linking in Markdown
-                final_md += "\n---\n" # Separator between videos
+                    final_md += f"\n**Links in Description:**\n"
+                    for link in res['description_links']:
+                        final_md += f"- <{link}>\n"
+                final_md += "\n---\n"
 
-        yield "\n".join(status_log), gr.Markdown(final_md) # Final update
+        yield "\n".join(status_log), gr.Markdown(final_md)
 
     except ValueError as ve:
         yield from log_status(f"Configuration Error: {ve}")
         yield "\n".join(status_log), gr.Markdown(f"### Error\n**Input Error:** {ve}")
     except ConnectionError as ce:
-         yield from log_status(f"API Connection Error: {ce}")
-         yield "\n".join(status_log), gr.Markdown(f"### Error\n**API Connection Error:** {ce}\nCheck your API key, internet connection, and API quota.")
+        yield from log_status(f"API Connection Error: {ce}")
+        yield "\n".join(status_log), gr.Markdown(f"### Error\n**API Connection Error:** {ce}")
     except HttpError as he:
         yield from log_status(f"API HTTP Error: {he.resp.status} - {he.content}")
-        yield "\n".join(status_log), gr.Markdown(f"### Error\n**API HTTP Error:** Status {he.resp.status}\n{he.content}\nCheck your API key and quota.")
+        yield "\n".join(status_log), gr.Markdown(f"### Error\n**API HTTP Error:** Status {he.resp.status}\n{he.content}")
     except Exception as e:
-        import traceback
-        traceback.print_exc() # Print full traceback to console for debugging
+        traceback.print_exc()
         yield from log_status(f"An unexpected error occurred: {e}")
-        yield "\n".join(status_log), gr.Markdown(f"### Error\n**An unexpected error occurred:** {e}")
-
+        yield "\n".join(status_log), gr.Markdown(f"### Error\n**Unexpected Error:** {e}")
 
 # --- Gradio Interface Definition ---
 
 with gr.Blocks(theme=gr.themes.Soft()) as app:
     gr.Markdown("# YouTube Channel 3D Software Scanner")
-    gr.Markdown("Find mentions of 3D software in video transcripts and descriptions for a specific YouTube channel.")
+    gr.Markdown("Find mentions of 3D software in video transcripts and descriptions.")
 
     with gr.Row():
         with gr.Column(scale=1):
             gr.Markdown("## Settings")
-            api_key_input = gr.Textbox(
-                label="YouTube Data API Key",
-                placeholder="Enter your Google Cloud API Key here",
-                type="password",
-                info="Keep this key secure. Get one from Google Cloud Console."
-            )
             channel_input = gr.Textbox(
                 label="Channel Handle or ID",
-                placeholder="e.g., @theAIsearch or UCxxxxxxxxxxxxxx",
-                info="Use the channel's @handle or its full Channel ID."
+                placeholder="e.g., @theAIsearch or UCxxxxxxxxxxxxxx"
             )
             keywords_input = gr.Textbox(
                 label="Keywords to Search (comma-separated)",
-                value=DEFAULT_KEYWORDS,
-                info="List software or terms like 'Blender', '3D Modeling', 'Unreal Engine'."
+                value=DEFAULT_KEYWORDS
             )
             scan_button = gr.Button("Scan Channel", variant="primary")
             clear_button = gr.Button("Clear All")
 
-
         with gr.Column(scale=2):
             gr.Markdown("## Status & Logs")
             status_output = gr.Textbox(
@@ -407,38 +290,29 @@ with gr.Blocks(theme=gr.themes.Soft()) as app:
                 lines=10,
                 max_lines=20,
                 interactive=False,
-                autoscroll=True,
-                placeholder="Scan status updates will appear here..."
+                autoscroll=True
             )
             gr.Markdown("## Results")
-            results_output = gr.Markdown(
-                 value="Results will be displayed here after scanning.",
-                 # Allow HTML for links? Markdown should handle <url>
-            )
+            results_output = gr.Markdown(value="Results will be displayed here after scanning.")
 
-    # Button Click Actions
     scan_button.click(
         fn=scan_channel_videos,
-        inputs=[api_key_input, channel_input, keywords_input],
-        outputs=[status_output, results_output],
-        # api_name="scan_channel" # Optional: for API usage if needed
+        inputs=[channel_input, keywords_input],
+        outputs=[status_output, results_output]
     )
 
     def clear_outputs():
-        return "", "Results cleared.", "", "", DEFAULT_KEYWORDS # Clear status, results, api_key, channel, reset keywords
+        return "", "Results cleared.", "", DEFAULT_KEYWORDS
 
     clear_button.click(
         fn=clear_outputs,
         inputs=[],
-        outputs=[status_output, results_output, api_key_input, channel_input, keywords_input]
+        outputs=[status_output, results_output, channel_input, keywords_input]
     )
 
     gr.Markdown("---")
-    gr.Markdown("**(Note:** Scans can take time depending on the number of videos and API quotas. Ensure your API key has the YouTube Data API v3 enabled and sufficient quota.*)")
-
+    gr.Markdown("**Note:** Scans may take time depending on video count and API quotas.")
 
 # --- Run the App ---
 if __name__ == "__main__":
-    # share=True creates a public link (use with caution, especially with API keys)
-    # Set debug=True for more detailed logs in console during development
     app.launch(debug=False)