interview_transcriber.py

#!/usr/bin/env python3
"""
Audio Transcription Automation Specialist
Bulk transcription workflow with intelligent content filtering and summary generation
"""

import os
import sys
import json
import logging
from pathlib import Path
from typing import List, Dict, Tuple, Optional
import warnings
warnings.filterwarnings("ignore")

import whisper
import librosa
import numpy as np
import soundfile as sf
from textstat import flesch_reading_ease
import re

class InterviewTranscriber:
    """Specialized transcriber for interview content with smart filtering"""
    
    def __init__(self, input_folder: str, output_folder: str, model_size: str = "base", processing_mode: str = "normal"):
        self.input_folder = Path(input_folder)
        self.output_folder = Path(output_folder)
        self.model_size = model_size
        self.processing_mode = processing_mode  # "normal", "aggressive", "ml"
        
        # Create output directories
        self.output_folder.mkdir(parents=True, exist_ok=True)
        (self.output_folder / "transcripts").mkdir(exist_ok=True)
        (self.output_folder / "key_moments").mkdir(exist_ok=True)
        (self.output_folder / "cliff_notes").mkdir(exist_ok=True)
        
        # Initialize logging
        logging.basicConfig(
            level=logging.INFO,
            format='%(asctime)s - %(levelname)s - %(message)s',
            handlers=[
                logging.FileHandler(self.output_folder / 'transcription.log'),
                logging.StreamHandler()
            ]
        )
        self.logger = logging.getLogger(__name__)
        
        # Content relevance indicators
        self.relevant_keywords = [
            'important', 'key', 'main', 'crucial', 'essential', 'significant',
            'takeaway', 'insight', 'learned', 'discovered', 'found', 'realized',
            'question', 'answer', 'interview', 'discuss', 'talk about', 'explain',
            'because', 'since', 'therefore', 'however', 'although', 'despite'
        ]
        
        # Quote indicators
        self.quote_patterns = [
            r'"[^"]*"',  # Direct quotes
            r'\'[^\']*\'',  # Single quotes
            r'\b(said|told me|explained|mentioned|noted|added|pointed out)\b',
            r'\b(the main thing is|what i learned|the key is|important to note)\b'
        ]
        
        # Load Whisper model
        self.logger.info(f"Loading Whisper model: {model_size}")
        self.model = whisper.load_model(model_size)
        
    def enhance_audio(self, audio_file: Path) -> Tuple[np.ndarray, int]:
        """Enhance audio quality with noise reduction and normalization"""
        try:
            # Load audio
            y, sr = librosa.load(str(audio_file), sr=None)
            
            # Log audio information
            duration = len(y) / sr
            self.logger.info(f"Audio: {audio_file.name} - Duration: {duration:.2f}s, Sample Rate: {sr}Hz")
            
            # Remove silence from beginning and end
            y, _ = librosa.effects.trim(y, top_db=20)
            
            # Apply noise reduction using spectral subtraction
            stft = librosa.stft(y)
            magnitude = np.abs(stft)
            phase = np.angle(stft)
            
            # Estimate noise from first 0.5 seconds
            noise_frames = int(0.5 * sr / 512)  # Assuming hop length of 512
            noise_magnitude = np.mean(magnitude[:, :noise_frames], axis=1, keepdims=True)
            
            # Spectral subtraction
            alpha = 2.0  # Noise reduction factor
            enhanced_magnitude = magnitude - alpha * noise_magnitude
            enhanced_magnitude = np.maximum(enhanced_magnitude, magnitude * 0.1)
            
            # Reconstruct signal
            enhanced_stft = enhanced_magnitude * np.exp(1j * phase)
            y_enhanced = librosa.istft(enhanced_stft)
            
            # Normalize audio
            y_enhanced = librosa.util.normalize(y_enhanced)
            
            return y_enhanced, sr
            
        except Exception as e:
            self.logger.warning(f"Audio enhancement failed for {audio_file}: {e}")
            # Fallback to original audio
            return librosa.load(str(audio_file), sr=None)
    
    def detect_repetition(self, text: str, aggressive: bool = False) -> bool:
        """Detect if text contains excessive repetition"""
        words = text.lower().split()
        if len(words) < 2:
            return True  # Very short segments are often repetitive
        
        # AGGRESSIVE MODE: Much stricter filtering for ML datasets
        if aggressive:
            # If any word appears more than 2 times in a segment, it's repetitive
            word_counts = {}
            for word in words:
                word_counts[word] = word_counts.get(word, 0) + 1
            
            if any(count > 2 for count in word_counts.values()):
                return True
            
            # If the same word appears consecutively more than once
            for i in range(len(words) - 1):
                if words[i] == words[i + 1]:
                    return True
            
            # If the segment is mostly one word (80%+ same word)
            max_count = max(word_counts.values())
            if max_count > len(words) * 0.8:
                return True
        
        # NORMAL MODE: Standard filtering
        else:
            # Check for repeated words
            word_counts = {}
            for word in words:
                word_counts[word] = word_counts.get(word, 0) + 1
            
            # If any word appears more than 50% of the time, it's repetitive
            max_count = max(word_counts.values())
            if max_count > len(words) * 0.5:
                return True
        
        # Check for repeated short phrases (2-3 words)
        phrases = []
        for i in range(len(words) - 1):
            phrase = f"{words[i]} {words[i+1]}"
            phrases.append(phrase)
        
        for i in range(len(words) - 2):
            phrase = f"{words[i]} {words[i+1]} {words[i+2]}"
            phrases.append(phrase)
        
        phrase_counts = {}
        for phrase in phrases:
            phrase_counts[phrase] = phrase_counts.get(phrase, 0) + 1
        
        # If any phrase repeats, mark as repetitive
        if any(count > 1 for count in phrase_counts.values()):
            return True
        
        return False
    
    def filter_repetitive_segments(self, segments: List[Dict], aggressive: bool = False) -> List[Dict]:
        """Filter out repetitive segments from transcript"""
        filtered_segments = []
        recent_texts = []  # Track recent texts to catch temporal repetition
        
        for i, segment in enumerate(segments):
            text = segment['text'].strip()
            
            # Skip if segment is repetitive
            if self.detect_repetition(text, aggressive):
                self.logger.debug(f"Filtered repetitive segment: {text[:50]}...")
                continue
            
            # Skip very short segments that are likely filler
            if len(text.split()) < 2:
                continue
            
            # Skip segments with very low confidence
            if segment.get('avg_logprob', 0.0) < -2.0:
                continue
            
            # AGGRESSIVE MODE: Check for temporal repetition
            if aggressive:
                # Check if this text is very similar to recent texts
                for recent_text in recent_texts[-5:]:  # Check last 5 segments
                    if self.text_similarity(text, recent_text) > 0.8:
                        self.logger.debug(f"Filtered temporally repetitive segment: {text[:50]}...")
                        continue
                
                # Add to recent texts history
                recent_texts.append(text)
                if len(recent_texts) > 10:  # Keep only last 10
                    recent_texts.pop(0)
            
            filtered_segments.append(segment)
        
        return filtered_segments
    
    def text_similarity(self, text1: str, text2: str) -> float:
        """Calculate similarity between two texts"""
        words1 = set(text1.lower().split())
        words2 = set(text2.lower().split())
        
        if not words1 or not words2:
            return 0.0
        
        # Jaccard similarity
        intersection = len(words1.intersection(words2))
        union = len(words1.union(words2))
        
        return intersection / union if union > 0 else 0.0
    
    def calculate_content_relevance(self, text: str) -> float:
        """Calculate relevance score for content filtering"""
        text_lower = text.lower()
        score = 0.0
        
        # Keyword matching
        keyword_matches = sum(1 for keyword in self.relevant_keywords if keyword in text_lower)
        score += min(keyword_matches * 0.2, 0.6)
        
        # Quote pattern matching
        quote_matches = sum(1 for pattern in self.quote_patterns if re.search(pattern, text_lower))
        score += min(quote_matches * 0.3, 0.3)
        
        # Sentence structure analysis
        if '?' in text or '!' in text:
            score += 0.1
            
        # Length and complexity (prefer meaningful content)
        word_count = len(text.split())
        if 5 <= word_count <= 25:
            score += 0.1
        elif word_count > 25:
            score += 0.2
            
        return min(score, 1.0)
    
    def extract_key_moments(self, transcript: Dict) -> List[Dict]:
        """Extract key moments and quotes from transcript"""
        key_moments = []
        
        for segment in transcript.get('segments', []):
            text = segment['text'].strip()
            
            # Skip very short segments
            if len(text.split()) < 3:
                continue
                
            # Calculate relevance
            relevance_score = self.calculate_content_relevance(text)
            
            # Include relevant content (lowered threshold for better coverage)
            if relevance_score > 0.3:
                key_moments.append({
                    'timestamp': segment['start'],
                    'text': text,
                    'type': 'key_insight' if relevance_score > 0.7 else 'relevant_content',
                    'relevance_score': relevance_score,
                    'confidence': segment.get('avg_logprob', 0.0)
                })
        
        return key_moments
    
    def generate_cliff_notes(self, key_moments: List[Dict], max_points: int = 10) -> str:
        """Generate concise cliff notes from key moments"""
        # Sort by relevance and confidence
        sorted_moments = sorted(
            key_moments, 
            key=lambda x: (x['relevance_score'], x['confidence']), 
            reverse=True
        )
        
        # Select top moments
        top_moments = sorted_moments[:max_points]
        
        # Format cliff notes
        cliff_notes = []
        for i, moment in enumerate(top_moments, 1):
            timestamp = self.format_timestamp(moment['timestamp'])
            cliff_notes.append(f"{i}. **[{timestamp}]** {moment['text']}")
        
        return "\n\n".join(cliff_notes)
    
    def format_timestamp(self, seconds: float) -> str:
        """Format timestamp as MM:SS"""
        minutes = int(seconds // 60)
        secs = int(seconds % 60)
        return f"{minutes:02d}:{secs:02d}"
    
    def transcribe_file(self, audio_file: Path) -> Optional[Dict]:
        """Transcribe a single audio file with enhancement"""
        try:
            self.logger.info(f"Processing: {audio_file.name}")
            
            # Enhance audio
            enhanced_audio, sr = self.enhance_audio(audio_file)
            
            # Save enhanced audio temporarily
            temp_audio = self.output_folder / f"temp_{audio_file.stem}_enhanced.wav"
            sf.write(str(temp_audio), enhanced_audio, sr)
            
            # Transcribe with optimized settings
            result = self.model.transcribe(
                str(temp_audio),
                temperature=0.0,
                best_of=5,
                beam_size=5,
                fp16=False,
                verbose=False
            )
            
            # Clean up temp file
            temp_audio.unlink()
            
            # Apply repetition filtering based on processing mode
            original_segments = len(result.get('segments', []))
            aggressive_mode = self.processing_mode in ["aggressive", "ml"]
            result['segments'] = self.filter_repetitive_segments(result.get('segments', []), aggressive=aggressive_mode)
            filtered_segments = len(result['segments'])
            
            self.logger.info(f"Filtered {original_segments - filtered_segments} repetitive segments from {original_segments} total (mode: {self.processing_mode})")
            
            # Extract key moments
            key_moments = self.extract_key_moments(result)
            
            # Generate cliff notes
            cliff_notes = self.generate_cliff_notes(key_moments)
            
            return {
                'file': audio_file.name,
                'transcript': result,
                'key_moments': key_moments,
                'cliff_notes': cliff_notes,
                'stats': {
                    'duration': result.get('duration', 0),
                    'word_count': len(result.get('text', '').split()),
                    'key_moments_count': len(key_moments)
                }
            }
            
        except Exception as e:
            self.logger.error(f"Failed to transcribe {audio_file}: {e}")
            return None
    
    def save_results(self, results: Dict, audio_file: Path):
        """Save transcription results in organized format"""
        base_name = audio_file.stem
        
        # Save full transcript
        transcript_file = self.output_folder / "transcripts" / f"{base_name}_transcript.md"
        with open(transcript_file, 'w', encoding='utf-8') as f:
            f.write(f"# {audio_file.name} Transcript\n\n")
            f.write(f"**Duration:** {self.format_timestamp(results['stats']['duration'])}\n")
            f.write(f"**Word Count:** {results['stats']['word_count']}\n")
            f.write(f"**Key Moments:** {results['stats']['key_moments_count']}\n\n")
            
            f.write("## Full Transcript\n\n")
            for segment in results['transcript'].get('segments', []):
                timestamp = self.format_timestamp(segment['start'])
                f.write(f"[{timestamp}] {segment['text'].strip()}\n")
        
        # Save key moments
        if results['key_moments']:
            moments_file = self.output_folder / "key_moments" / f"{base_name}_key_moments.md"
            with open(moments_file, 'w', encoding='utf-8') as f:
                f.write(f"# {audio_file.name} - Key Moments & Quotes\n\n")
                
                for moment in results['key_moments']:
                    timestamp = self.format_timestamp(moment['timestamp'])
                    relevance_emoji = "⭐" if moment['relevance_score'] > 0.8 else "📌"
                    f.write(f"{relevance_emoji} **[{timestamp}]** {moment['text']}\n")
                    f.write(f"   *Relevance: {moment['relevance_score']:.2f}*\n\n")
        
        # Save cliff notes
        if results['cliff_notes']:
            cliff_notes_file = self.output_folder / "cliff_notes" / f"{base_name}_cliff_notes.md"
            with open(cliff_notes_file, 'w', encoding='utf-8') as f:
                f.write(f"# {audio_file.name} - Cliff Notes\n\n")
                f.write(results['cliff_notes'])
    
    def process_batch(self, file_pattern: str = "*.wav") -> Dict:
        """Process all matching files in input folder"""
        audio_files = list(self.input_folder.glob(f"**/{file_pattern}"))
        
        if not audio_files:
            self.logger.warning(f"No files found matching pattern: {file_pattern}")
            return {'processed': 0, 'failed': 0}
        
        self.logger.info(f"Found {len(audio_files)} files to process")
        
        results = {
            'processed': 0,
            'failed': 0,
            'files': []
        }
        
        for audio_file in audio_files:
            result = self.transcribe_file(audio_file)
            
            if result:
                self.save_results(result, audio_file)
                results['processed'] += 1
                results['files'].append({
                    'name': audio_file.name,
                    'stats': result['stats']
                })
                self.logger.info(f"✓ Completed: {audio_file.name}")
            else:
                results['failed'] += 1
                self.logger.error(f"✗ Failed: {audio_file.name}")
        
        # Generate summary report
        self.generate_summary_report(results)
        
        return results
    
    def generate_summary_report(self, results: Dict):
        """Generate overall processing summary"""
        report_file = self.output_folder / "processing_summary.md"
        
        with open(report_file, 'w', encoding='utf-8') as f:
            f.write("# Audio Transcription Summary\n\n")
            f.write(f"**Processed:** {results['processed']} files\n")
            f.write(f"**Failed:** {results['failed']} files\n\n")
            
            if results['files']:
                f.write("## File Statistics\n\n")
                f.write("| File | Duration | Words | Key Moments |\n")
                f.write("|------|----------|-------|-------------|\n")
                
                for file_info in results['files']:
                    duration = self.format_timestamp(file_info['stats']['duration'])
                    f.write(f"| {file_info['name']} | {duration} | {file_info['stats']['word_count']} | {file_info['stats']['key_moments_count']} |\n")
        
        self.logger.info(f"Summary report saved to: {report_file}")


def main():
    """Main execution function - configured for RodePro files"""
    # Configuration for RodePro audio files
    input_folder = r"e:\Video_Projects\Clients\Michelle Miller\Projects\2025_KWUA_Farm_Babe\audio\RodePro"
    output_folder = r"p:\local_tools\batch_transcribe\output\RodePro"
    
    # Initialize transcriber
    transcriber = InterviewTranscriber(
        input_folder=input_folder,
        output_folder=output_folder,
        model_size="base"  # Use "tiny" for faster processing, "medium" or "large" for better accuracy
    )
    
    # Process all WAV files
    results = transcriber.process_batch("*.wav")
    
    print(f"\nProcessing Complete!")
    print(f"✓ Processed: {results['processed']} files")
    print(f"✗ Failed: {results['failed']} files")
    print(f"📁 Output saved to: {output_folder}")


def main_dr10l():
    """Execution function for DR-10L files"""
    # Configuration for DR-10L audio files
    input_folder = r"e:\Video_Projects\Clients\Michelle Miller\Projects\2025_KWUA_Farm_Babe\audio\DR-10L"
    output_folder = r"p:\local_tools\batch_transcribe\output\DR-10L"
    
    # Initialize transcriber
    transcriber = InterviewTranscriber(
        input_folder=input_folder,
        output_folder=output_folder,
        model_size="base"
    )
    
    # Process all WAV files
    results = transcriber.process_batch("*.wav")
    
    print(f"\nDR-10L Processing Complete!")
    print(f"✓ Processed: {results['processed']} files")
    print(f"✗ Failed: {results['failed']} files")
    print(f"📁 Output saved to: {output_folder}")


if __name__ == "__main__":
    import sys
    if len(sys.argv) > 1 and sys.argv[1] == "dr10l":
        main_dr10l()
    else:
        main()