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.")