Comprehensive comparison for Speech-to-Text technology in AI applications

import assemblyai as aai
import os
from typing import Optional, Dict, List
from datetime import datetime
import logging

# Configure logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

class MeetingTranscriptionService:
    """
    Production-ready service for transcribing meeting recordings with speaker detection,
    sentiment analysis, and key topic extraction using AssemblyAI.
    """
    
    def __init__(self, api_key: Optional[str] = None):
        """
        Initialize the transcription service with API credentials.
        """
        self.api_key = api_key or os.environ.get('ASSEMBLYAI_API_KEY')
        if not self.api_key:
            raise ValueError("AssemblyAI API key is required")
        
        aai.settings.api_key = self.api_key
        self.transcriber = aai.Transcriber()
    
    def transcribe_meeting(
        self,
        audio_url: str,
        webhook_url: Optional[str] = None
    ) -> Dict:
        """
        Transcribe a meeting with advanced features including speaker diarization,
        sentiment analysis, and auto highlights.
        """
        try:
            logger.info(f"Starting transcription for audio: {audio_url}")
            
            # Configure transcription with production features
            config = aai.TranscriptionConfig(
                speaker_labels=True,
                speakers_expected=None,  # Auto-detect number of speakers
                sentiment_analysis=True,
                auto_highlights=True,
                entity_detection=True,
                iab_categories=True,
                language_detection=True,
                punctuate=True,
                format_text=True,
                webhook_url=webhook_url
            )
            
            # Submit transcription job
            transcript = self.transcriber.transcribe(
                audio_url,
                config=config
            )
            
            # Check for errors
            if transcript.status == aai.TranscriptStatus.error:
                logger.error(f"Transcription failed: {transcript.error}")
                raise Exception(f"Transcription error: {transcript.error}")
            
            logger.info(f"Transcription completed: {transcript.id}")
            
            # Process and structure the results
            result = self._process_transcript(transcript)
            
            return result
            
        except Exception as e:
            logger.error(f"Error during transcription: {str(e)}")
            raise
    
    def _process_transcript(self, transcript: aai.Transcript) -> Dict:
        """
        Process transcript results into structured format for API response.
        """
        # Extract speaker segments
        speaker_segments = []
        if transcript.utterances:
            for utterance in transcript.utterances:
                speaker_segments.append({
                    'speaker': utterance.speaker,
                    'text': utterance.text,
                    'start': utterance.start,
                    'end': utterance.end,
                    'confidence': utterance.confidence
                })
        
        # Extract sentiment analysis
        sentiments = []
        if transcript.sentiment_analysis:
            for sentiment in transcript.sentiment_analysis:
                sentiments.append({
                    'text': sentiment.text,
                    'sentiment': sentiment.sentiment,
                    'confidence': sentiment.confidence
                })
        
        # Extract key highlights
        highlights = []
        if transcript.auto_highlights:
            highlights = [
                {
                    'text': h.text,
                    'count': h.count,
                    'rank': h.rank
                }
                for h in transcript.auto_highlights.results
            ]
        
        # Extract detected entities
        entities = []
        if transcript.entities:
            for entity in transcript.entities:
                entities.append({
                    'text': entity.text,
                    'type': entity.entity_type,
                    'start': entity.start,
                    'end': entity.end
                })
        
        return {
            'transcript_id': transcript.id,
            'status': transcript.status.value,
            'text': transcript.text,
            'confidence': transcript.confidence,
            'audio_duration': transcript.audio_duration,
            'language_code': transcript.language_code,
            'speaker_segments': speaker_segments,
            'sentiments': sentiments,
            'highlights': highlights,
            'entities': entities,
            'created_at': datetime.now().isoformat()
        }

# Example usage in a Flask API endpoint
def create_meeting_transcript_endpoint():
    """
    Example Flask endpoint for meeting transcription.
    """
    from flask import Flask, request, jsonify
    
    app = Flask(__name__)
    service = MeetingTranscriptionService()
    
    @app.route('/api/v1/transcribe', methods=['POST'])
    def transcribe():
        try:
            data = request.get_json()
            audio_url = data.get('audio_url')
            webhook_url = data.get('webhook_url')
            
            if not audio_url:
                return jsonify({'error': 'audio_url is required'}), 400
            
            result = service.transcribe_meeting(audio_url, webhook_url)
            return jsonify(result), 200
            
        except Exception as e:
            logger.error(f"API error: {str(e)}")
            return jsonify({'error': str(e)}), 500
    
    return app

if __name__ == '__main__':
    # Example: Transcribe a sample meeting
    service = MeetingTranscriptionService()
    result = service.transcribe_meeting(
        'https://example.com/meeting-recording.mp3'
    )
    print(f"Transcription complete: {result['transcript_id']}")