import gradio as gr import asyncio from asyncio import Semaphore # Added for concurrency control from pathlib import Path import os import tempfile # Added for temporary chunk files import traceback # Import traceback for better error logging import re import pandas as pd 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 # 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): self.current_prompts = self._load_default_prompts() self.client: genai.Client | None = None def _load_default_prompts(self) -> Dict[str, str]: # (Implementation identical to previous version) prompts = {} 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: # (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: 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 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: 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: if not video_id: raise ValueError("Invalid Video ID") try: 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_TRANSCRIPT_FETCH: Failed for ID '{video_id}'. Reason: {e}" # --- TranscriptProcessor Class (Refactored for Concurrency Control) --- class TranscriptProcessor: def __init__(self): self.generator = ContentGenerator() # (Helper _get_youtube_transcript remains unchanged) def _get_youtube_transcript(self, url: str) -> str: # ... (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: 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) try: 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}") # --- 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: # (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 (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 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="Your GCP account needs to have billing enabled to use the 2.5 pro model." ) input_audio = gr.File( label="Upload Audio File", file_count="single", file_types=["audio", ".mp3", ".wav", ".ogg", ".flac", ".m4a", ".aac"] ) submit_btn = gr.Button("Generate with Gemini", 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(label="", value="...", interactive=False, lines=3, max_lines=10) gr.Markdown(f"### {display_desc}") output_desc = gr.Textbox(label="", value="...", interactive=False, lines=3, max_lines=10) gr.Markdown(f"### {display_previews}") output_previews = gr.Textbox(label="", value="...", interactive=False, lines=3, max_lines=10) gr.Markdown(f"### {display_clips}") output_clips = gr.Textbox(label="", value="...", interactive=False, lines=3, max_lines=10) gr.Markdown(f"### {display_timestamps}") output_timestamps = gr.Textbox(label="", value="...", interactive=False, lines=3, max_lines=10) 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: 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: 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=[google_api_key_input, input_audio], outputs=outputs_list ) with gr.Tab("Customize Prompts"): # (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]) 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=reset_prompts_ui, inputs=None, outputs=[status_prompt_tab] + prompt_inputs ) return app # --- Main Execution Block (Unchanged) --- if __name__ == "__main__": 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.")