Update app.py

app.py

@@ -1,266 +1,662 @@
 import gradio as gr
 import asyncio
+from asyncio import Semaphore # Added for concurrency control
 from pathlib import Path
-from google import genai
-from google.genai import types
 import os
-from dataclasses import dataclass
-from typing import Dict
-from youtube_transcript_api import YouTubeTranscriptApi
+import tempfile # Added for temporary chunk files
+import traceback # Import traceback for better error logging
 import re
 import pandas as pd
-import assemblyai as aai
+from dataclasses import dataclass
+from typing import Dict, AsyncGenerator, Tuple, Any, List
+
+# Use standard import convention for genai
+# Assuming genai is installed and configured elsewhere
+from google import genai
+from youtube_transcript_api import YouTubeTranscriptApi
 
-# Move relevant classes and functions into app.py
+# Import pydub for audio manipulation
+from pydub import AudioSegment
+from pydub.exceptions import CouldntDecodeError
+
+
+# --- Constants ---
+PROMPT_KEYS = ["titles_and_thumbnails", "description", "previews", "clips", "timestamps"]
+PROMPT_DISPLAY_NAMES = {
+    "titles_and_thumbnails": "Titles and Thumbnails",
+    "description": "Twitter Description",
+    "previews": "Preview Clips",
+    "clips": "Twitter Clips",
+    "timestamps": "Timestamps"
+}
+# --- MODIFIED: Increased chunk size to 30 minutes ---
+AUDIO_CHUNK_DURATION_MS = 30 * 60 * 1000 # Process audio in 30-minute chunks
+# --- ADDED: Concurrency Limits ---
+MAX_CONCURRENT_TRANSCRIPTIONS = 3 # Limit simultaneous transcription API calls
+MAX_CONCURRENT_GENERATIONS = 4    # Limit simultaneous content generation API calls
+
+# --- Core Classes (ContentRequest, ContentGenerator) ---
+# (ContentRequest and ContentGenerator remain unchanged)
 @dataclass
 class ContentRequest:
     prompt_key: str
 
 class ContentGenerator:
-    def __init__(self,api_key):
+    def __init__(self):
         self.current_prompts = self._load_default_prompts()
-        self.client = genai.Client(api_key=api_key)
+        self.client: genai.Client | None = None
 
     def _load_default_prompts(self) -> Dict[str, str]:
-        """Load default prompts and examples from files and CSVs."""
-
-        # Load CSV examples
-        try:
-            timestamps_df = pd.read_csv("data/Timestamps.csv")
-            titles_df = pd.read_csv("data/Titles & Thumbnails.csv")
-            descriptions_df = pd.read_csv("data/Viral Episode Descriptions.csv")
-            clips_df = pd.read_csv("data/Viral Twitter Clips.csv")
-
-            # Format timestamp examples
-            timestamp_examples = "\n\n".join(timestamps_df['Timestamps'].dropna().tolist())
-
-            # Format title examples
-            title_examples = "\n".join([
-                f'Title: "{row.Titles}"\nThumbnail: "{row.Thumbnail}"'
-                for _, row in titles_df.iterrows()
-            ])
-
-            # Format description examples
-            description_examples = "\n".join([
-                f'Tweet: "{row["Tweet Text"]}"'
-                for _, row in descriptions_df.iterrows()
-            ])
-
-            # Format clip examples
-            clip_examples = "\n\n".join([
-                f'Tweet Text: "{row["Tweet Text"]}"\nClip Transcript: "{row["Clip Transcript"]}"'
-                for _, row in clips_df.iterrows() if pd.notna(row["Tweet Text"])
-            ])
-
-        except Exception as e:
-            print(f"Warning: Error loading CSV examples: {e}")
-            timestamp_examples = ""
-            title_examples = ""
-            description_examples = ""
-            clip_examples = ""
-
-        # Load base prompts and inject examples
+        # (Implementation identical to previous version)
         prompts = {}
-        for key in ["previews", "clips", "description", "timestamps", "titles_and_thumbnails"]:
-            prompt = Path(f"prompts/{key}.txt").read_text()
-
-            # Inject relevant examples
-            if key == "timestamps":
-                prompt = prompt.replace("{timestamps_examples}", timestamp_examples)
-            elif key == "titles_and_thumbnails":
-                prompt = prompt.replace("{title_examples}", title_examples)
-            elif key == "description":
-                prompt = prompt.replace("{description_examples}", description_examples)
-            elif key == "clips":
-                prompt = prompt.replace("{clip_examples}", clip_examples)
-
-            prompts[key] = prompt
-
+        timestamp_examples, title_examples, description_examples, clip_examples = "", "", "", ""
+        try:
+            data_dir = Path("data")
+            if data_dir.is_dir():
+                try: timestamps_df = pd.read_csv(data_dir / "Timestamps.csv"); timestamp_examples = "\n\n".join(timestamps_df['Timestamps'].dropna().tolist())
+                except Exception as e: print(f"Warning: Loading Timestamps.csv failed: {e}")
+                try: titles_df = pd.read_csv(data_dir / "Titles & Thumbnails.csv"); title_examples = "\n".join([f'Title: "{r.Titles}"\nThumbnail: "{r.Thumbnail}"' for _, r in titles_df.iterrows() if pd.notna(r.Titles) and pd.notna(r.Thumbnail)])
+                except Exception as e: print(f"Warning: Loading Titles & Thumbnails.csv failed: {e}")
+                try: descriptions_df = pd.read_csv(data_dir / "Viral Episode Descriptions.csv"); description_examples = "\n".join([f'Tweet: "{r["Tweet Text"]}"' for _, r in descriptions_df.iterrows() if pd.notna(r["Tweet Text"])])
+                except Exception as e: print(f"Warning: Loading Viral Episode Descriptions.csv failed: {e}")
+                try: clips_df = pd.read_csv(data_dir / "Viral Twitter Clips.csv"); clip_examples = "\n\n".join([f'Tweet Text: "{r["Tweet Text"]}"\nClip Transcript: "{r["Clip Transcript"]}"' for _, r in clips_df.iterrows() if pd.notna(r["Tweet Text"]) and pd.notna(r["Clip Transcript"])])
+                except Exception as e: print(f"Warning: Loading Viral Twitter Clips.csv failed: {e}")
+            else: print("Warning: 'data' directory not found.")
+        except Exception as e: print(f"Warning: Error accessing 'data' directory: {e}")
+
+        prompts_dir = Path("prompts")
+        if not prompts_dir.is_dir():
+                print("Error: 'prompts' directory not found.")
+                return {key: f"ERROR: Prompt directory missing." for key in PROMPT_KEYS}
+        for key in PROMPT_KEYS:
+            try:
+                prompt = (prompts_dir / f"{key}.txt").read_text(encoding='utf-8')
+                if key == "timestamps": prompt = prompt.replace("{timestamps_examples}", timestamp_examples)
+                elif key == "titles_and_thumbnails": prompt = prompt.replace("{title_examples}", title_examples)
+                elif key == "description": prompt = prompt.replace("{description_examples}", description_examples)
+                elif key == "clips": prompt = prompt.replace("{clip_examples}", clip_examples)
+                prompts[key] = prompt
+            except Exception as e:
+                print(f"Warning: Loading prompt prompts/{key}.txt failed: {e}")
+                prompts[key] = f"Generate {key} based on the transcript. Do not use markdown formatting." # Fallback
+        for key in PROMPT_KEYS: prompts.setdefault(key, f"Generate {key} based on the transcript. Do not use markdown formatting.")
         return prompts
 
     async def generate_content(self, request: ContentRequest, transcript: str) -> str:
-        """Generate content using Gemini asynchronously."""
+        # (Implementation identical to previous version)
+        if not self.client: return "ERROR_CONFIGURATION: Gemini Client not initialized."
+        if not transcript: return "ERROR_INTERNAL: Empty transcript provided for content generation."
         try:
-            print(f"\nFull prompt for {request.prompt_key}:")
-            print("=== SYSTEM PROMPT ===")
-            print(self.current_prompts[request.prompt_key])
-            print("=== END SYSTEM PROMPT ===\n")
-
-            response = self.client.models.generate_content(
-                model="gemini-2.5-pro-exp-03-25",
-                config=types.GenerateContentConfig(system_instruction=self.current_prompts[request.prompt_key]),
-                contents=transcript
+            system_prompt = self.current_prompts.get(request.prompt_key)
+            if not system_prompt: return f"ERROR_INTERNAL: System prompt for '{request.prompt_key}' missing."
+            contents_for_api = [system_prompt, transcript]
+            # --- IMPORTANT: Model kept as gemini-1.5-flash ---
+            model_name = "gemini-2.5-pro-preview-03-25"
+            response = await asyncio.to_thread(
+                self.client.models.generate_content, model=model_name, contents=contents_for_api
             )
-
-            if response and hasattr(response, 'candidates'):
-                return response.text
-            else:
-                return f"Error: Unexpected response structure for {request.prompt_key}"
-
+            if not response: return f"ERROR_API: No response received for {request.prompt_key}."
+            try:
+                    if response.text:
+                            try:
+                                    if hasattr(response, 'prompt_feedback') and response.prompt_feedback.block_reason:
+                                            reason = response.prompt_feedback.block_reason.name; return f"ERROR_BLOCKED: Blocked by API. Reason: {reason}"
+                            except AttributeError: pass
+                            return str(response.text.strip())
+                    else:
+                            if response.candidates and response.candidates[0].content and response.candidates[0].content.parts:
+                                    full_text = "".join(part.text for part in response.candidates[0].content.parts if hasattr(part, 'text')).strip()
+                                    if full_text:
+                                        print(f"Warning: Used fallback text extraction via candidates for {request.prompt_key}")
+                                        return str(full_text)
+                            return f"ERROR_NO_TEXT: Could not extract text from response for {request.prompt_key}."
+            except (ValueError, AttributeError) as e:
+                    print(f"Error accessing response text/feedback for {request.prompt_key} (potentially blocked): {e}")
+                    reason = "Unknown"
+                    try:
+                        if hasattr(response, 'prompt_feedback') and response.prompt_feedback.block_reason: reason = response.prompt_feedback.block_reason.name
+                    except AttributeError: pass
+                    return f"ERROR_BLOCKED: Content generation failed (possibly blocked). Reason: {reason}"
         except Exception as e:
-            return f"Error generating content: {str(e)}"
-
-def extract_video_id(url: str) -> str:
-    """Extract video ID from various YouTube URL formats."""
-    match = re.search(
-        r"(?:youtube\.com\/watch\?v=|youtu\.be\/|youtube\.com\/embed\/|youtube\.com\/v\/)([A-Za-z0-9_-]+)",
-        url
-    )
-    return match.group(1) if match else None
+            print(f"Error generating content for {request.prompt_key}: {traceback.format_exc()}")
+            error_str = str(e).lower()
+            # Add specific check for rate limit errors if the API provides clear indicators
+            if "rate limit exceeded" in error_str or "quota exceeded" in error_str or "429" in error_str:
+                 return f"ERROR_RATE_LIMIT: API limit likely exceeded. Details: {str(e)}"
+            elif "permission denied" in error_str or "api key not valid" in error_str: return f"ERROR_PERMISSION_DENIED: API Error (Permission Denied?). Check Key. Details: {str(e)}"
+            # elif "quota" in error_str: return f"ERROR_QUOTA: API Quota Error. Details: {str(e)}" # Covered by rate limit check above
+            elif "model" in error_str and "not found" in error_str: return f"ERROR_MODEL_NOT_FOUND: Model name likely incorrect. Details: {str(e)}"
+            else: return f"ERROR_API_GENERAL: API Error during generation. Details: {str(e)}"
+
+    def update_prompts(self, *values):
+        # (Implementation identical to previous version)
+        updated_keys = []
+        for key, value in zip(PROMPT_KEYS, values):
+            if isinstance(value, str): self.current_prompts[key] = value; updated_keys.append(key)
+        return f"Prompts updated: {', '.join(updated_keys)}" if updated_keys else "No prompts updated."
+
+# (extract_video_id and get_transcript remain unchanged)
+def extract_video_id(url: str) -> str | None:
+    patterns = [r"(?:v=|\/)([0-9A-Za-z_-]{11}).*", r"youtu\.be\/([0-9A-Za-z_-]{11})"]
+    for pattern in patterns:
+        match = re.search(pattern, url);
+        if match: return match.group(1)
+    return None
 
 def get_transcript(video_id: str) -> str:
-    """Get transcript from YouTube video ID."""
+    if not video_id: raise ValueError("Invalid Video ID")
     try:
-        transcript = YouTubeTranscriptApi.list_transcripts(video_id).find_transcript(["en"])
-        return " ".join(entry["text"] for entry in transcript.fetch())
+        t_list = YouTubeTranscriptApi.list_transcripts(video_id)
+        transcript = t_list.find_transcript(['en', 'en-US'])
+        fetched = transcript.fetch()
+        if not fetched: raise ValueError("Fetched transcript empty")
+        return " ".join(entry.get("text", "") for entry in fetched).strip()
     except Exception as e:
-        return f"Error fetching transcript: {str(e)}"
+        return f"ERROR_TRANSCRIPT_FETCH: Failed for ID '{video_id}'. Reason: {e}"
 
+# --- TranscriptProcessor Class (Refactored for Concurrency Control) ---
 class TranscriptProcessor:
     def __init__(self):
-        self.generator = ContentGenerator(api_key=os.getenv("GOOGLE_API_KEY"))
-
+        self.generator = ContentGenerator()
 
+    # (Helper _get_youtube_transcript remains unchanged)
     def _get_youtube_transcript(self, url: str) -> str:
-        """Get transcript from YouTube URL."""
+       # ... (identical implementation)
+        print(f"Extracting Video ID from: {url}")
+        video_id = extract_video_id(url)
+        if not video_id: raise ValueError(f"Invalid YouTube URL/ID: {url}")
+        print(f"Video ID: {video_id}. Fetching transcript...")
         try:
-            if video_id := extract_video_id(url):
-                return get_transcript(video_id)
-            raise Exception("Invalid YouTube URL")
-        except Exception as e:
-            raise Exception(f"Error fetching YouTube transcript: {str(e)}")
+                transcript = get_transcript(video_id)
+                if transcript.startswith("ERROR_TRANSCRIPT_FETCH"): raise Exception(transcript)
+                if not transcript: raise ValueError(f"Empty transcript for ID: {video_id}")
+                print(f"Transcript fetched (length: {len(transcript)}).")
+                return transcript
+        except Exception as e: print(f"Error fetching YouTube transcript: {e}"); raise Exception(f"Failed to get YouTube transcript: {str(e)}")
+
+
+    # --- MODIFIED: Added Semaphore argument ---
+    async def _transcribe_chunk(self, client: genai.Client, chunk_path: Path, chunk_index: int, total_chunks: int, semaphore: Semaphore) -> str:
+        """Transcribes a single audio chunk using Gemini API, respecting the semaphore."""
+        # Acquire semaphore before proceeding
+        async with semaphore:
+            print(f"Semaphore acquired for chunk {chunk_index + 1}/{total_chunks}. Processing...")
+            gemini_audio_file_ref = None
+            try:
+                print(f"Uploading chunk {chunk_index + 1}/{total_chunks}: {chunk_path.name}")
+                gemini_audio_file_ref = await asyncio.to_thread(client.files.upload, file=chunk_path)
+                print(f"Chunk {chunk_index + 1} uploaded. File Ref: {gemini_audio_file_ref.name}")
+
+                prompt_for_transcription = "Transcribe the following audio file accurately."
+                contents = [prompt_for_transcription, gemini_audio_file_ref]
+                # --- IMPORTANT: Model kept as gemini-1.5-flash ---
+                model_name = "gemini-2.5-pro-preview-03-25"
+
+                print(f"Requesting transcription for chunk {chunk_index + 1}...")
+                # Make the API call *within* the semaphore lock
+                transcription_response = await asyncio.to_thread(
+                    client.models.generate_content, model=model_name, contents=contents
+                )
+                print(f"Transcription response received for chunk {chunk_index + 1}.")
+
+                # Extract transcript text (identical logic)
+                transcript_piece = ""
+                try:
+                    if transcription_response.text:
+                        transcript_piece = transcription_response.text.strip()
+                    elif transcription_response.candidates and transcription_response.candidates[0].content and transcription_response.candidates[0].content.parts:
+                        transcript_piece = "".join(part.text for part in transcription_response.candidates[0].content.parts if hasattr(part, 'text')).strip()
+
+                    if not transcript_piece and hasattr(transcription_response, 'prompt_feedback') and transcription_response.prompt_feedback.block_reason:
+                         reason = transcription_response.prompt_feedback.block_reason.name
+                         print(f"Warning: Transcription blocked for chunk {chunk_index + 1}. Reason: {reason}")
+                         return f"[CHUNK_ERROR: Blocked - {reason}]"
+
+                    print(f"Chunk {chunk_index + 1} transcript length: {len(transcript_piece)}")
+                    return str(transcript_piece)
+
+                except (ValueError, AttributeError, Exception) as extraction_err:
+                    print(f"Error extracting transcript for chunk {chunk_index + 1}: {extraction_err}. Response: {transcription_response}")
+                    return f"[CHUNK_ERROR: Extraction Failed - {str(extraction_err)}]"
+
+            except Exception as e:
+                print(f"Error processing chunk {chunk_index + 1} (within semaphore): {traceback.format_exc()}")
+                error_str = str(e).lower()
+                # Add specific check for rate limit errors
+                if "rate limit exceeded" in error_str or "quota exceeded" in error_str or "429" in error_str:
+                    return f"[CHUNK_ERROR: API Rate Limit Exceeded - {str(e)}]"
+                elif "permission denied" in error_str or "api key not valid" in error_str:
+                     return f"[CHUNK_ERROR: API Permission Denied - {str(e)}]"
+                elif "file size" in error_str:
+                     return f"[CHUNK_ERROR: File Size Limit Exceeded - {str(e)}]"
+                else:
+                     return f"[CHUNK_ERROR: General API/Processing Error - {str(e)}]"
+            finally:
+                # Cleanup happens *before* semaphore is released automatically by 'async with'
+                if gemini_audio_file_ref:
+                    # Run cleanup in background to avoid blocking semaphore release if deletion is slow
+                    asyncio.create_task(self.delete_uploaded_file(client, gemini_audio_file_ref.name, f"chunk {chunk_index + 1} cleanup"))
+                if chunk_path.exists():
+                    try:
+                        os.remove(chunk_path)
+                    except OSError as e:
+                        print(f"Warning: Could not delete local temp chunk file {chunk_path}: {e}")
+                print(f"Semaphore released for chunk {chunk_index + 1}/{total_chunks}.")
+        # Semaphore is automatically released when exiting 'async with' block
+
+
+    async def process_transcript(self, client: genai.Client, audio_file: Any) -> AsyncGenerator[Tuple[str, Any], None]:
+        """
+        Processes audio with larger chunks and controlled concurrency using Semaphores.
+        """
+        if AudioSegment is None:
+            yield "error", "Audio processing library (pydub) not loaded. Cannot proceed."
+            return
+        if not client:
+            yield "error", "Gemini Client object was not provided."
+            return
+        self.generator.client = client
+        if not audio_file:
+            yield "error", "No audio file provided."
+            return
+
+        audio_path_str = getattr(audio_file, 'name', None)
+        if not audio_path_str:
+            yield "error", "Invalid audio file object."
+            return
+        original_audio_path = Path(audio_path_str)
+        if not original_audio_path.exists():
+            yield "error", f"Audio file not found: {original_audio_path}"
+            return
+
+        # --- ADDED: Initialize Semaphores ---
+        transcription_semaphore = Semaphore(MAX_CONCURRENT_TRANSCRIPTIONS)
+        generation_semaphore = Semaphore(MAX_CONCURRENT_GENERATIONS)
 
-    async def process_transcript(self, audio_file):
-        """Process input and generate all content."""
-        audio_path = audio_file.name
         try:
-            aai.settings.api_key = os.getenv("ASSEMBLYAI_API_KEY")
-            config = aai.TranscriptionConfig(speaker_labels=True, language_code="en")
-            transcript_iter = aai.Transcriber().transcribe(str(audio_path), config=config)
-            transcript = transcript_iter.text
-
-            # Process each type sequentially
-            sections = {}
-            for key in ["titles_and_thumbnails", "description", "previews", "clips", "timestamps"]:
-                result = await self.generator.generate_content(ContentRequest(key), transcript)
-                sections[key] = result
-
-            # Combine into markdown with H2 headers
-            markdown = f"""
-## Titles and Thumbnails
-{sections['titles_and_thumbnails']}
-## Twitter Description
-{sections['description']}
-## Preview Clips
-{sections['previews']}
-## Twitter Clips
-{sections['clips']}
-## Timestamps
-{sections['timestamps']}
-"""
-            return markdown
+            yield "status", f"Loading audio file: {original_audio_path.name}..."
+            print(f"Loading audio file with pydub: {original_audio_path}")
+            try:
+                 file_format = original_audio_path.suffix.lower().replace('.', '')
+                 audio = AudioSegment.from_file(original_audio_path, format=file_format if file_format else None)
+            except CouldntDecodeError as decode_error:
+                 print(f"pydub decode error: {decode_error}. Make sure ffmpeg is installed.")
+                 yield "error", f"Failed to load/decode audio file: {original_audio_path.name}. Ensure valid format and ffmpeg."
+                 return
+            except Exception as load_err:
+                 print(f"Error loading audio with pydub: {traceback.format_exc()}")
+                 yield "error", f"Error loading audio file {original_audio_path.name}: {load_err}"
+                 return
+
+            duration_ms = len(audio)
+            # --- MODIFIED: Chunk duration increased ---
+            total_chunks = (duration_ms + AUDIO_CHUNK_DURATION_MS - 1) // AUDIO_CHUNK_DURATION_MS
+            print(f"Audio loaded. Duration: {duration_ms / 1000:.2f}s. Splitting into {total_chunks} x {AUDIO_CHUNK_DURATION_MS / 60000:.1f}min chunks.")
+            yield "status", f"Audio loaded ({duration_ms / 1000:.2f}s). Transcribing in {total_chunks} chunks (max {MAX_CONCURRENT_TRANSCRIPTIONS} concurrent)..."
+
+            transcript_pieces = [""] * total_chunks # Pre-allocate list to store pieces in order
+            transcription_tasks = []
+
+            # --- MODIFIED: Create tasks with semaphore ---
+            for i in range(total_chunks):
+                start_ms = i * AUDIO_CHUNK_DURATION_MS
+                end_ms = min((i + 1) * AUDIO_CHUNK_DURATION_MS, duration_ms)
+                chunk = audio[start_ms:end_ms]
+
+                with tempfile.NamedTemporaryFile(suffix=".wav", delete=False) as temp_chunk_file:
+                    chunk_path = Path(temp_chunk_file.name)
+                try:
+                    chunk.export(chunk_path, format="wav")
+                except Exception as export_err:
+                     print(f"Error exporting chunk {i+1}: {traceback.format_exc()}")
+                     yield "error", f"Failed to create temporary audio chunk file: {export_err}"
+                     if chunk_path.exists(): os.remove(chunk_path)
+                     return
+
+                # Pass semaphore to the chunk transcription function
+                task = asyncio.create_task(self._transcribe_chunk(client, chunk_path, i, total_chunks, transcription_semaphore))
+                # Store task along with its index to place result correctly
+                transcription_tasks.append((i, task))
+
+            # Process transcription results as they complete, maintaining order
+            processed_chunks = 0
+            # Wait for all tasks using gather, but process results as they come in via callbacks or checking task states?
+            # Using asyncio.gather might be simpler here if we need all results before proceeding. Let's try gather.
+            # results = await asyncio.gather(*(task for _, task in transcription_tasks), return_exceptions=True)
+
+            # Alternative: Process as completed, but store in correct order
+            temp_results = {}
+            for index, task in transcription_tasks:
+                 try:
+                     result = await task
+                     temp_results[index] = result
+                     processed_chunks += 1
+                     yield "status", f"Transcribed chunk {processed_chunks}/{total_chunks}..."
+                     # Check for critical chunk errors immediately if needed
+                     if isinstance(result, str) and ("[CHUNK_ERROR: API Rate Limit Exceeded" in result or \
+                                                      "[CHUNK_ERROR: API Permission Denied" in result or \
+                                                      "[CHUNK_ERROR: API Quota Exceeded" in result):
+                         print(f"Critical API error in chunk {index + 1}, stopping transcription. Error: {result}")
+                         yield "error", f"Transcription stopped. Critical API error in chunk {index + 1}: {result.split('-', 1)[-1].strip()}"
+                         # Cancel remaining tasks (important!)
+                         for j, other_task in transcription_tasks:
+                            if not other_task.done():
+                                other_task.cancel()
+                         return # Stop processing
+                 except asyncio.CancelledError:
+                     print(f"Transcription task for chunk {index + 1} was cancelled.")
+                     temp_results[index] = "[CHUNK_ERROR: Cancelled]"
+                     # If one task is cancelled due to an error in another, we might stop everything
+                     if processed_chunks < total_chunks: # Avoid double error message if already stopped
+                         yield "error", "Transcription process was cancelled."
+                         return
+                 except Exception as e:
+                     print(f"Error waiting for transcription task {index + 1}: {traceback.format_exc()}")
+                     temp_results[index] = f"[CHUNK_ERROR: Task Processing Failed - {str(e)}]"
+                     processed_chunks += 1 # Count as processed even though it failed
+
+            # Reconstruct the transcript in the correct order
+            transcript_pieces = [temp_results.get(i, "[CHUNK_ERROR: Missing Result]") for i in range(total_chunks)]
+            full_transcript = " ".join(transcript_pieces).strip()
+
+            # Improved check for transcription failure
+            if not full_transcript or full_transcript.isspace() or all(s.startswith("[CHUNK_ERROR") for s in transcript_pieces if s):
+                 error_summary = " ".join(p for p in transcript_pieces if p.startswith("[CHUNK_ERROR"))
+                 print(f"Transcription failed or resulted in only errors. Summary: {error_summary}")
+                 yield "error", f"Failed to transcribe audio or all chunks failed. Errors: {error_summary[:200]}"
+                 return
+
+            print(f"Full transcript concatenated (length: {len(full_transcript)}).")
+            yield "status", "Transcription complete. Generating content..."
+
+            # --- Generate other content using the FULL transcript with Semaphore ---
+            generation_tasks = []
+            for key in PROMPT_KEYS:
+                # Pass generation semaphore to the item generation function
+                task = asyncio.create_task(self._generate_single_item(key, full_transcript, generation_semaphore))
+                generation_tasks.append(task)
+
+            generated_items = 0
+            total_items = len(PROMPT_KEYS)
+            # Process generation results as they complete
+            for future in asyncio.as_completed(generation_tasks):
+                try:
+                    key, result = await future # Result from _generate_single_item
+                    yield "progress", (key, result)
+                    generated_items += 1
+                    # More granular status for generation
+                    yield "status", f"Generating content ({key} done, {generated_items}/{total_items} total)..."
+                except asyncio.CancelledError:
+                     # Should not happen unless transcription failed and cancelled tasks
+                     print("Content generation task was cancelled.")
+                     yield "error", "Content generation cancelled."
+                     return
+                except Exception as e:
+                    print(f"Error processing completed generation task: {traceback.format_exc()}")
+                    yield "status", f"Error during content generation phase: {str(e)}"
+                    # Optionally yield an error for the specific item?
+                    # key_if_possible = "unknown_key" # How to get key here? Task doesn't easily pass it back on exception
+                    # yield "progress", (key_if_possible, f"ERROR_GENERATION: {str(e)}")
+
+            yield "status", "All content generation tasks complete."
+
+        except FileNotFoundError as e:
+            yield "error", f"File Error: {str(e)}"
+            return
+        except Exception as e: # Catch-all for transcription setup phase
+            print(f"Error during transcription setup/chunking phase: {traceback.format_exc()}")
+            yield "error", f"System Error during transcription setup: {str(e)}"
+            return
+
+
+    async def delete_uploaded_file(self, client: genai.Client, file_name: str, context: str):
+        # (Implementation identical - called in background now)
+        if not client or not file_name:
+            # print(f"Skipping deletion: Invalid client or file name ({context}).") # Reduce noise
+            return
+        try:
+            # print(f"Attempting background cleanup: {file_name} ({context})")
+            await asyncio.to_thread(client.files.delete, name=file_name)
+            print(f"Successfully cleaned up Gemini file: {file_name} ({context})")
+        except Exception as cleanup_err:
+            if "not found" in str(cleanup_err).lower() or "404" in str(cleanup_err):
+                 pass # Ignore file not found during cleanup
+                # print(f"Info: File {file_name} likely already deleted ({context}).")
+            else:
+                print(f"Warning: Failed Gemini file cleanup for {file_name} ({context}): {cleanup_err}")
 
-        except Exception as e:
-            return f"Error processing input: {str(e)}"
+
+    # --- MODIFIED: Added Semaphore argument ---
+    async def _generate_single_item(self, key: str, transcript: str, semaphore: Semaphore) -> Tuple[str, str]:
+        """Helper to generate one piece of content, respecting the semaphore."""
+        # Acquire semaphore before calling the API
+        async with semaphore:
+            print(f"Semaphore acquired for generating: {key}. Calling API...")
+            result = await self.generator.generate_content(ContentRequest(key), transcript)
+            print(f"Finished generation task for: {key}. Semaphore released.")
+            # Semaphore is released automatically by 'async with'
+            return key, result
 
     def update_prompts(self, *values) -> str:
-        """Update the current session's prompts."""
-        self.generator.current_prompts.update(zip(
-            ["previews", "clips", "description", "timestamps", "titles_and_thumbnails"],
-            values
-        ))
-        return "Prompts updated for this session!"
+        # (Implementation identical to previous version)
+        return self.generator.update_prompts(*values)
 
 
+# --- Gradio Interface Creation (UI remains unchanged from previous version) ---
 
 def create_interface():
-    """Create the Gradio interface."""
+    """Create the Gradio interface (UI definition identical to last version)."""
     processor = TranscriptProcessor()
 
+    key_titles = "titles_and_thumbnails"
+    key_desc = "description"
+    key_previews = "previews"
+    key_clips = "clips"
+    key_timestamps = "timestamps"
+    display_titles = PROMPT_DISPLAY_NAMES[key_titles]
+    display_desc = PROMPT_DISPLAY_NAMES[key_desc]
+    display_previews = PROMPT_DISPLAY_NAMES[key_previews]
+    display_clips = PROMPT_DISPLAY_NAMES[key_clips]
+    display_timestamps = PROMPT_DISPLAY_NAMES[key_timestamps]
+
     with gr.Blocks(title="Gemini Podcast Content Generator") as app:
         gr.Markdown(
             """
             # Gemini Podcast Content Generator
-            Generate preview clips, timestamps, descriptions and more from an audio file using Gemini.
-            Simply upload an audio file to get started and Gemini handles the rest.
+            Generate social media content from podcast audio using Gemini.
+            Enter your Google API key below and upload an audio file.
+            Audio will be processed in larger (~30min) chunks with controlled concurrency.
             """
-        )
+        ) # Updated description slightly
 
         with gr.Tab("Generate Content"):
+            google_api_key_input = gr.Textbox(
+                label="Google API Key", type="password",
+                placeholder="Enter your Google API Key here",
+                info="Ensure the API key is valid and has necessary permissions."
+            )
             input_audio = gr.File(
-                label="Upload Audio File",
-                file_count="single",
-                file_types=["audio"]
+                label="Upload Audio File", file_count="single",
+                file_types=["audio", ".mp3", ".wav", ".ogg", ".flac", ".m4a", ".aac"]
             )
-            submit_btn = gr.Button("Generate Content with Gemini")
+            submit_btn = gr.Button("Generate Content", variant="huggingface")
+
+            gr.Markdown("### Processing Status")
+            output_status = gr.Textbox(label="Current Status", value="Idle.", interactive=False, lines=1, max_lines=5)
+
+            gr.Markdown(f"### {display_titles}")
+            output_titles = gr.Textbox(value="...", interactive=False, lines=3, max_lines=10) # No label
+
+            gr.Markdown(f"### {display_desc}")
+            output_desc = gr.Textbox(value="...", interactive=False, lines=3, max_lines=10) # No label
+
+            gr.Markdown(f"### {display_previews}")
+            output_previews = gr.Textbox(value="...", interactive=False, lines=3, max_lines=10) # No label
 
-            output = gr.Markdown()  # Single markdown output
+            gr.Markdown(f"### {display_clips}")
+            output_clips = gr.Textbox(value="...", interactive=False, lines=3, max_lines=10) # No label
 
-            async def process_wrapper(text):
-                print("Process wrapper started")
-                print(f"Input text: {text[:100]}...")
+            gr.Markdown(f"### {display_timestamps}")
+            output_timestamps = gr.Textbox(value="...", interactive=False, lines=3, max_lines=10) # No label
 
+            outputs_list = [
+                output_status,
+                output_titles, output_desc, output_previews,
+                output_clips, output_timestamps
+            ]
+            results_component_map = {
+                key_titles: output_titles, key_desc: output_desc, key_previews: output_previews,
+                key_clips: output_clips, key_timestamps: output_timestamps
+            }
+
+            # --- process_wrapper (UI Update Logic - largely unchanged) ---
+            async def process_wrapper(google_key, audio_file_obj, progress=gr.Progress(track_tqdm=True)):
+                print("Started Processing...")
+                initial_updates = {
+                    output_status: gr.update(value="Initiating..."),
+                    output_titles: gr.update(value="⏳ Pending..."),
+                    output_desc: gr.update(value="⏳ Pending..."),
+                    output_previews: gr.update(value="⏳ Pending..."),
+                    output_clips: gr.update(value="⏳ Pending..."),
+                    output_timestamps: gr.update(value="⏳ Pending..."),
+                }
+                yield initial_updates
+
+                if not google_key:
+                    yield {output_status: gr.update(value="🛑 Error: Missing Google API Key.")}
+                    return
+                if not audio_file_obj:
+                    yield {output_status: gr.update(value="🛑 Error: No audio file uploaded.")}
+                    return
+
+                masked_key = f"{'*'*(len(google_key)-4)}{google_key[-4:]}" if len(google_key) > 4 else "****"
+                print(f"Using Google Key: {masked_key}")
+                print(f"Audio file: Name='{getattr(audio_file_obj, 'name', 'N/A')}'")
+                client: genai.Client | None = None
                 try:
-                    result = await processor.process_transcript(text)
-                    print("Process completed, got results")
-                    return result
+                    yield {output_status: gr.update(value="⏳ Initializing Gemini Client...")}
+                    client = await asyncio.to_thread(genai.Client, api_key=google_key)
+                    print("Gemini Client initialized successfully.")
+                    yield {output_status: gr.update(value="✅ Client Initialized.")}
                 except Exception as e:
-                    print(f"Error in process_wrapper: {str(e)}")
-                    return f"# Error\n\n{str(e)}"
+                    error_msg = f"🛑 Error: Failed Client Initialization: {e}"
+                    print(f"Client Init Error: {traceback.format_exc()}")
+                    yield {output_status: gr.update(value=error_msg)}
+                    return
+
+                updates_to_yield = {}
+                try:
+                    # Call the refactored processor
+                    async for update_type, data in processor.process_transcript(client, audio_file_obj):
+                        updates_to_yield = {}
+                        if update_type == "status":
+                            updates_to_yield[output_status] = gr.update(value=f"⏳ {data}")
+                        elif update_type == "progress":
+                            key, result = data
+                            component_to_update = results_component_map.get(key)
+                            if component_to_update:
+                                ui_result = ""
+                                if isinstance(result, str) and result.startswith("ERROR_"):
+                                    # Handle specific rate limit error display
+                                    if result.startswith("ERROR_RATE_LIMIT"):
+                                         ui_result = f"❌ Error (Rate Limit):\n{result.split(':', 1)[-1].strip()}"
+                                    else:
+                                        try:
+                                            error_type, error_detail = result.split(':', 1)
+                                            error_type_display = error_type.replace('ERROR_', '').replace('_', ' ').title()
+                                            ui_result = f"❌ Error ({error_type_display}):\n{error_detail.strip()}"
+                                        except ValueError:
+                                            ui_result = f"❌ Error:\n{result}"
+                                else:
+                                    ui_result = str(result)
+                                updates_to_yield[component_to_update] = gr.update(value=ui_result)
+                            else:
+                                print(f"Warning: No UI component mapped for result key '{key}'")
+                        elif update_type == "error":
+                            error_message = f"🛑 Processing Error: {data}"
+                            updates_to_yield[output_status] = gr.update(value=error_message)
+                            yield updates_to_yield
+                            return
+
+                        if updates_to_yield:
+                           yield updates_to_yield
+
+                    final_success_update = {output_status: gr.update(value="✅ Processing Complete.")}
+                    final_success_update.update(updates_to_yield) # Include any final progress updates
+                    yield final_success_update
+                    print("Process wrapper finished successfully.")
+
+                except Exception as e:
+                    print(f"Error in process_wrapper async loop: {traceback.format_exc()}")
+                    error_msg = f"🛑 Unexpected wrapper error: {e}"
+                    yield {output_status: gr.update(value=error_msg)}
 
             submit_btn.click(
                 fn=process_wrapper,
-                inputs=input_audio,
-                outputs=output,
-                queue=True
+                inputs=[google_api_key_input, input_audio],
+                outputs=outputs_list
             )
 
         with gr.Tab("Customize Prompts"):
-            gr.Markdown(
-                """
-                ## Customize Generation Prompts
-                Here you can experiment with different prompts during your session.
-                Changes will remain active until you reload the page.
-                Tip: Copy your preferred prompts somewhere safe if you want to reuse them later!
-                """
-            )
+            # (Customize Prompts tab UI remains unchanged)
+            gr.Markdown("## Customize Generation Prompts")
+            prompt_inputs = []
+            default_prompts = processor.generator.current_prompts
+            for key in PROMPT_KEYS:
+                display_name = PROMPT_DISPLAY_NAMES.get(key, key.replace('_', ' ').title())
+                default_value = default_prompts.get(key, "")
+                prompt_inputs.append(gr.Textbox(label=f"{display_name} Prompt", lines=10, value=default_value or ""))
+
+            status_prompt_tab = gr.Textbox(label="Status", interactive=False)
+            update_btn = gr.Button("Update Session Prompts")
+            update_btn.click(fn=processor.update_prompts, inputs=prompt_inputs, outputs=[status_prompt_tab])
 
-            prompt_inputs = [
-                gr.Textbox(
-                    label=f"{key.replace('_', ' ').title()} Prompt",
-                    lines=10,
-                    value=processor.generator.current_prompts[key]
-                )
-                for key in [
-                    "previews",
-                    "clips",
-                    "description",
-                    "timestamps",
-                    "titles_and_thumbnails"
-                ]
-            ]
-            status = gr.Textbox(label="Status", interactive=False)
-
-            # Update prompts when they change
-            for prompt in prompt_inputs:
-                prompt.change(
-                    fn=processor.update_prompts,
-                    inputs=prompt_inputs,
-                    outputs=[status]
-                )
-
-            # Reset button
             reset_btn = gr.Button("Reset to Default Prompts")
+            def reset_prompts_ui():
+                try:
+                    defaults = processor.generator._load_default_prompts()
+                    if any(isinstance(v, str) and v.startswith("ERROR:") for v in defaults.values()): raise ValueError("Failed to load one or more default prompts.")
+                    processor.generator.current_prompts = defaults
+                    updates = {status_prompt_tab: gr.update(value="Prompts reset to defaults!")}
+                    for i, key in enumerate(PROMPT_KEYS):
+                        updates[prompt_inputs[i]] = gr.update(value=defaults.get(key, ""))
+                    return updates
+                except Exception as e:
+                    print(f"Error during prompt reset: {e}")
+                    return {status_prompt_tab: gr.update(value=f"Error resetting prompts: {str(e)}")}
+
             reset_btn.click(
-                fn=lambda: (
-                    processor.update_prompts(*processor.generator.current_prompts.values()),
-                    *processor.generator.current_prompts.values(),
-                ),
-                outputs=[status] + prompt_inputs,
+                 fn=reset_prompts_ui,
+                 inputs=None,
+                 outputs=[status_prompt_tab] + prompt_inputs
             )
 
     return app
 
+# --- Main Execution Block (Unchanged) ---
 if __name__ == "__main__":
-    create_interface().launch()
-    app.launch()
+    if AudioSegment is None:
+        print("\nFATAL ERROR: pydub is required but could not be imported.")
+        print("Please install it ('pip install pydub') and ensure ffmpeg is available.")
+        print("Application cannot start correctly.")
+        exit(1)
+
+    Path("prompts").mkdir(exist_ok=True)
+    Path("data").mkdir(exist_ok=True)
+    _prompt_dir = Path("prompts")
+    for key in PROMPT_KEYS:
+        prompt_file = _prompt_dir / f"{key}.txt"
+        if not prompt_file.exists():
+            # Ensure default prompts advise against markdown
+            default_content = f"This is the default placeholder prompt for {PROMPT_DISPLAY_NAMES[key]}. Process the transcript provided. Important: Generate the response as plain text only. Do not use any Markdown formatting (no '#', '*', '_', list formatting, bolding, etc.)."
+            if key == "titles_and_thumbnails": default_content += "\n\nExamples:\n{title_examples}"
+            elif key == "description": default_content += "\n\nExamples:\n{description_examples}"
+            elif key == "clips": default_content += "\n\nExamples:\n{clip_examples}"
+            elif key == "timestamps": default_content += "\n\nExamples:\n{timestamps_examples}"
+            prompt_file.write_text(default_content, encoding='utf-8')
+            print(f"Created dummy prompt file: {prompt_file}")
+
+    print("Starting Gradio application...")
+    app = create_interface()
+    app.launch()
+    print("Gradio application stopped.")