Spaces:
Runtime error
Runtime error
| import logging | |
| from typing import Dict, Any, List | |
| from dataclasses import asdict | |
| logger = logging.getLogger(__name__) | |
| class PodcastTool: | |
| """ | |
| MCP Tool for podcast generation from documents | |
| """ | |
| def __init__(self, podcast_generator): | |
| """ | |
| Initialize Podcast Tool | |
| Args: | |
| podcast_generator: PodcastGeneratorService instance | |
| """ | |
| self.podcast_generator = podcast_generator | |
| async def generate_podcast( | |
| self, | |
| document_ids: List[str], | |
| style: str = "conversational", | |
| duration_minutes: int = 10, | |
| host1_voice: str = "Rachel", | |
| host2_voice: str = "Adam" | |
| ) -> Dict[str, Any]: | |
| """ | |
| MCP Tool: Generate podcast from documents | |
| Args: | |
| document_ids: List of document IDs to generate podcast from | |
| style: Podcast style (conversational, educational, technical, casual) | |
| duration_minutes: Target duration in minutes | |
| host1_voice: Voice name for first host | |
| host2_voice: Voice name for second host | |
| Returns: | |
| Dictionary with podcast ID, audio URL, transcript, and metadata | |
| """ | |
| try: | |
| if not document_ids or len(document_ids) == 0: | |
| return { | |
| "success": False, | |
| "error": "No documents provided. Please select at least one document." | |
| } | |
| logger.info(f"Generating podcast from {len(document_ids)} documents") | |
| # Generate podcast using service | |
| result = await self.podcast_generator.generate_podcast( | |
| document_ids=document_ids, | |
| style=style, | |
| duration_minutes=duration_minutes, | |
| host1_voice=host1_voice, | |
| host2_voice=host2_voice | |
| ) | |
| if result.success: | |
| return { | |
| "success": True, | |
| "podcast_id": result.podcast_id, | |
| "audio_file": result.audio_file_path, | |
| "audio_url": f"/data/podcasts/{result.podcast_id}.mp3", | |
| "transcript": result.transcript, | |
| "metadata": asdict(result.metadata) if result.metadata else {}, | |
| "generation_time": result.generation_time, | |
| "message": f"Podcast generated successfully! Duration: {result.metadata.duration_seconds/60:.1f} minutes" | |
| } | |
| else: | |
| return { | |
| "success": False, | |
| "error": result.error or "Unknown error during podcast generation" | |
| } | |
| except Exception as e: | |
| logger.error(f"Podcast generation failed: {str(e)}") | |
| return { | |
| "success": False, | |
| "error": str(e) | |
| } | |
| async def generate_transcript( | |
| self, | |
| document_ids: List[str], | |
| style: str = "conversational", | |
| duration_minutes: int = 10 | |
| ) -> Dict[str, Any]: | |
| """ | |
| MCP Tool: Generate podcast transcript ONLY (no audio) | |
| Args: | |
| document_ids: List of document IDs | |
| style: Podcast style | |
| duration_minutes: Target duration | |
| Returns: | |
| Dictionary with transcript and analysis | |
| """ | |
| try: | |
| if not document_ids: | |
| return {"success": False, "error": "No documents provided"} | |
| logger.info(f"Generating transcript for {len(document_ids)} docs") | |
| # 1. Analyze | |
| analysis = await self.podcast_generator.analyze_documents(document_ids) | |
| # 2. Generate Script | |
| script = await self.podcast_generator.generate_script( | |
| analysis, style, duration_minutes | |
| ) | |
| return { | |
| "success": True, | |
| "transcript": script.to_text(), | |
| "word_count": script.word_count, | |
| "estimated_duration": script.total_duration_estimate, | |
| "key_insights": analysis.key_insights, | |
| "topics": analysis.topics | |
| } | |
| except Exception as e: | |
| logger.error(f"Transcript generation failed: {str(e)}") | |
| return {"success": False, "error": str(e)} | |
| def get_podcast_audio(self, podcast_id: str) -> Dict[str, Any]: | |
| """ | |
| MCP Tool: Get audio file path for a podcast | |
| Args: | |
| podcast_id: Podcast ID | |
| Returns: | |
| Dictionary with audio file path | |
| """ | |
| try: | |
| podcast = self.podcast_generator.get_podcast(podcast_id) | |
| if not podcast: | |
| return {"success": False, "error": "Podcast not found"} | |
| # Construct absolute path (assuming local running) | |
| # In a real remote setup, this might return a URL | |
| audio_path = f"/data/podcasts/{podcast_id}.mp3" | |
| return { | |
| "success": True, | |
| "podcast_id": podcast_id, | |
| "audio_path": audio_path, | |
| "exists": True | |
| } | |
| except Exception as e: | |
| logger.error(f"Failed to get audio path: {str(e)}") | |
| return {"success": False, "error": str(e)} | |
| def list_podcasts(self, limit: int = 10) -> Dict[str, Any]: | |
| """ | |
| List previously generated podcasts | |
| Args: | |
| limit: Maximum number of podcasts to return | |
| Returns: | |
| Dictionary with list of podcast metadata | |
| """ | |
| try: | |
| podcasts = self.podcast_generator.list_podcasts(limit=limit) | |
| return { | |
| "success": True, | |
| "podcasts": [asdict(p) for p in podcasts], | |
| "total": len(podcasts) | |
| } | |
| except Exception as e: | |
| logger.error(f"Failed to list podcasts: {str(e)}") | |
| return { | |
| "success": False, | |
| "error": str(e), | |
| "podcasts": [] | |
| } | |
| def get_podcast(self, podcast_id: str) -> Dict[str, Any]: | |
| """ | |
| Get specific podcast by ID | |
| Args: | |
| podcast_id: Podcast identifier | |
| Returns: | |
| Dictionary with podcast metadata | |
| """ | |
| try: | |
| podcast = self.podcast_generator.get_podcast(podcast_id) | |
| if podcast: | |
| return { | |
| "success": True, | |
| "podcast": asdict(podcast) | |
| } | |
| else: | |
| return { | |
| "success": False, | |
| "error": "Podcast not found" | |
| } | |
| except Exception as e: | |
| logger.error(f"Failed to get podcast: {str(e)}") | |
| return { | |
| "success": False, | |
| "error": str(e) | |
| } | |