audio_processor.py

import os
from typing import Dict, Optional
from whisper import load_model  # Import directly from whisper package
import librosa
import soundfile as sf
from datetime import datetime
from rich.console import Console
from rich.progress import Progress
from content_generator import ContentGenerator
import config

class AudioProcessor:
    def __init__(self):
        self.console = Console()
        try:
            # Use tiny model instead of base for faster processing
            self.model = load_model("tiny")
            self.console.print("[green]Successfully loaded Whisper model (tiny)[/green]")
        except Exception as e:
            self.console.print(f"[red]Error loading Whisper model:[/red] {str(e)}")
            raise
        self.content_generator = ContentGenerator()
        
    def process_audio(
        self,
        audio_path: str,
        language: str = config.DEFAULT_LANGUAGE,
        content_type: str = "news",  # "news" or "blog"
        generate_content: bool = True
    ) -> Dict:
        """
        Process audio file: transcribe and optionally generate content.
        
        Args:
            audio_path (str): Path to the audio file
            language (str): Language code for transcription and content generation
            content_type (str): Type of content to generate ("news" or "blog")
            generate_content (bool): Whether to generate content from transcript
            
        Returns:
            Dict: Contains transcript and optionally generated content
        """
        self.console.print(f"[yellow]Processing audio file:[/yellow] {audio_path}")
        
        try:
            # Transcribe audio with highly optimized settings
            self.console.print("[yellow]Transcribing audio...[/yellow]")
            result = self.model.transcribe(
                audio_path,
                language=language if language != "tr" else "turkish",
                fp16=False,
                beam_size=1,  # Minimum beam size for fastest processing
                best_of=1,    # Single candidate for fastest processing
                condition_on_previous_text=False,
                compression_ratio_threshold=2.4,
                logprob_threshold=-1.0,
                no_speech_threshold=0.6,
                initial_prompt="Bu bir haber metnidir."  # Add context for better transcription
            )
            
            transcript = result["text"]
            
            # Generate content if requested
            generated_content = None
            if generate_content:
                self.console.print(f"[yellow]Generating {content_type} content from transcript...[/yellow]")
                if content_type == "news":
                    generated_content = self._generate_news_from_transcript(transcript, language)
                else:
                    generated_content = self._generate_blog_from_transcript(transcript, language)
            
            output = {
                "transcript": transcript,
                "language": language,
                "date": datetime.now().strftime("%Y-%m-%d"),
                "audio_file": os.path.basename(audio_path),
                "content_type": content_type
            }
            
            if generated_content:
                output["generated_content"] = generated_content
                
            return output
            
        except Exception as e:
            self.console.print(f"[red]Error processing audio:[/red] {str(e)}")
            raise
    
    def _generate_news_from_transcript(
        self,
        transcript: str,
        language: str
    ) -> Optional[Dict]:
        """Generate a news article from the transcript."""
        try:
            news_content = self.content_generator.generate_content(
                topic=transcript,
                keywords=["news", "professional", "factual"],
                language=language
            )
            
            # Validate the generated content
            if news_content and "title" in news_content:
                if len(news_content["content"].split('\n')) < 3:  # Minimum 3 paragraphs
                    return None
                    
            return news_content
            
        except Exception as e:
            self.console.print(f"[red]Error generating news article:[/red] {str(e)}")
            return None

    def _generate_blog_from_transcript(
        self,
        transcript: str,
        language: str
    ) -> Optional[Dict]:
        """Generate a blog post from the transcript."""
        try:
            blog_content = self.content_generator.generate_content(
                topic=transcript,
                keywords=["blog", "engaging", "informative"],
                language=language
            )
            return blog_content
        except Exception as e:
            self.console.print(f"[red]Error generating blog post:[/red] {str(e)}")
            return None
    
    def save_results(
        self,
        results: Dict,
        output_dir: str = "data/transcripts"
    ) -> None:
        """
        Save transcription and generated content results.
        
        Args:
            results (Dict): Processing results including transcript and content
            output_dir (str): Directory to save the output files
        """
        os.makedirs(output_dir, exist_ok=True)
        
        # Create base filename from audio file
        base_name = os.path.splitext(results["audio_file"])[0]
        date_prefix = results["date"]
        
        # Save transcript
        transcript_file = os.path.join(
            output_dir,
            f"{date_prefix}-{base_name}-transcript.txt"
        )
        with open(transcript_file, "w", encoding="utf-8") as f:
            f.write(results["transcript"])
        
        # Save generated content if available and valid
        if "generated_content" in results and results["generated_content"]:
            content_type = results["content_type"]
            content_file = os.path.join(
                output_dir,
                f"{date_prefix}-{base_name}-{content_type}.md"
            )
            
            try:
                with open(content_file, "w", encoding="utf-8") as f:
                    if content_type == "news":
                        # Add metadata and format for news articles
                        f.write(f"# {results['generated_content']['title']}\n\n")
                        
                        # Extract subtitle if it exists (first non-empty line after title)
                        content_lines = results['generated_content']['content'].split('\n')
                        first_line = next((line for line in content_lines if line.strip()), '')
                        if first_line and not first_line.startswith('*') and not first_line.startswith('#'):
                            f.write(f"*{first_line}*\n\n")
                            content = '\n'.join(content_lines[content_lines.index(first_line) + 1:])
                        else:
                            content = results['generated_content']['content']
                        
                        # Add metadata
                        f.write(f"**Tarih:** {date_prefix}\n\n")
                        f.write("---\n\n")  # Separator line
                        
                        # Write main content with proper formatting
                        f.write(content)
                    else:
                        # Blog format
                        f.write(f"# {results['generated_content']['title']}\n\n")
                        f.write(f"*Yazar: Mete*\n")
                        f.write(f"*Tarih: {date_prefix}*\n\n")
                        f.write(results['generated_content']['content'])
                
                self.console.print(f"[green]{results['content_type'].title()} content saved to:[/green] {content_file}")
            except Exception as e:
                self.console.print(f"[red]Error saving content:[/red] {str(e)}")
        else:
            if results.get("content_type") == "news":
                self.console.print("[yellow]Warning:[/yellow] Could not generate news article from this audio content.")
            else:
                self.console.print("[yellow]Warning:[/yellow] Could not generate blog post from this audio content.")
        
        self.console.print(f"[green]Transcript saved to:[/green] {transcript_file}")