python -m venv venv
source venv/bin/activate
pip install openai-whisper pyannote.audio openai pydub
export OPENAI_API_KEY="sk-your-key"
export HUGGINGFACE_TOKEN="hf-your-token"

import logging
from typing import Any, Dict, List, Optional

logger = logging.getLogger(__name__)


class WhisperTranscriber:
    """Transcribe audio files using OpenAI Whisper with timestamp support."""

    def __init__(self, model_size: str = "base"):
        import whisper
        self.model = whisper.load_model(model_size)

    def transcribe(self, audio_path: str, language: Optional[str] = None) -> Dict[str, Any]:
        options = {}
        if language:
            options["language"] = language
        result = self.model.transcribe(audio_path, **options)
        return {
            "text": result["text"],
            "segments": [
                {
                    "start": seg["start"],
                    "end": seg["end"],
                    "text": seg["text"],
                }
                for seg in result["segments"]
            ],
            "language": result.get("language", "unknown"),
        }

    def transcribe_with_timestamps(self, audio_path: str) -> str:
        result = self.model.transcribe(audio_path, word_timestamps=True)
        output = []
        for seg in result["segments"]:
            start = self._format_time(seg["start"])
            end = self._format_time(seg["end"])
            output.append(f"[{start} -> {end}] {seg['text'].strip()}")
        return "\n".join(output)

    def _format_time(self, seconds: float) -> str:
        mins = int(seconds // 60)
        secs = int(seconds % 60)
        return f"{mins:02d}:{secs:02d}"

from typing import Any, Dict, List, Optional


class SpeakerDiarizer:
    """Identify speakers in audio using pyannote diarization."""

    def __init__(self, token: str):
        from pyannote.audio import Pipeline
        self.pipeline = Pipeline.from_pretrained(
            "pyannote/speaker-diarization-3.1",
            use_auth_token=token,
        )

    def diarize(self, audio_path: str, num_speakers: Optional[int] = None) -> List[Dict[str, Any]]:
        kwargs = {}
        if num_speakers:
            kwargs["num_speakers"] = num_speakers
        diarization = self.pipeline(audio_path, **kwargs)
        segments = []
        for turn, _, speaker in diarization.itertracks(yield_label=True):
            segments.append({
                "start": turn.start,
                "end": turn.end,
                "speaker": speaker,
            })
        return segments

    def merge_with_transcript(
        self,
        diarization: List[Dict[str, Any]],
        transcript: List[Dict[str, Any]],
    ) -> List[Dict[str, Any]]:
        merged = []
        for seg in transcript:
            speaker = self._find_speaker(diarization, seg["start"])
            merged.append({**seg, "speaker": speaker})
        return merged

    def _find_speaker(self, diarization: List[Dict[str, Any]], timestamp: float) -> str:
        for d in diarization:
            if d["start"] <= timestamp <= d["end"]:
                return d["speaker"]
        return "Unknown"

from openai import AsyncOpenAI
import json
from datetime import datetime
from typing import Any, Dict, List, Optional


class MeetingContentAnalyzer:
    """Analyze meeting transcripts for actions, decisions, and summaries."""

    def __init__(self, model: str = "gpt-4o"):
        self.client = AsyncOpenAI()
        self.model = model

    async def analyze(self, transcript: str, num_participants: int) -> Dict[str, Any]:
        response = await self.client.chat.completions.create(
            model=self.model,
            messages=[
                {
                    "role": "system",
                    "content": """Analyze this meeting transcript.
                    Return JSON:
                    {
                        "summary": "2-3 paragraph summary",
                        "key_decisions": [{"decision": "...", "context": "..."}],
                        "action_items": [{"action": "...", "assignee": "...", "deadline": "..."}],
                        "discussion_topics": ["topic1", "topic2"],
                        "follow_ups": ["items needing follow-up"],
                        "sentiment": "positive|neutral|negative",
                        "engagement_level": "high|medium|low"
                    }""",
                },
                {"role": "user", "content": f"Meeting transcript ({num_participants} participants):\n\n{transcript[:8000]}"},
            ],
            temperature=0.2,
        )
        try:
            return json.loads(response.choices[0].message.content)
        except json.JSONDecodeError:
            return {"summary": response.choices[0].message.content, "key_decisions": [], "action_items": []}


class MeetingNotesGenerator:
    """Generate formatted meeting notes from analysis results."""

    def generate_notes(
        self,
        analysis: Dict[str, Any],
        transcript_segments: List[Dict[str, Any]],
        metadata: Optional[Dict[str, Any]] = None,
    ) -> str:
        metadata = metadata or {}
        notes = f"# Meeting Notes\n\n"
        notes += f"**Date:** {metadata.get('date', datetime.now().strftime('%Y-%m-%d'))}\n"
        notes += f"**Duration:** {metadata.get('duration', 'N/A')}\n"
        notes += f"**Participants:** {metadata.get('participants', 'N/A')}\n\n"
        notes += "## Summary\n\n"
        notes += f"{analysis.get('summary', 'No summary available')}\n\n"
        notes += "## Key Decisions\n\n"
        for i, decision in enumerate(analysis.get("key_decisions", []), 1):
            notes += f"{i}. **{decision.get('decision', 'N/A')}**\n"
            notes += f"   - Context: {decision.get('context', 'N/A')}\n"
        notes += "\n## Action Items\n\n"
        notes += "| Action | Assignee | Deadline |\n"
        notes += "|--------|----------|----------|\n"
        for item in analysis.get("action_items", []):
            notes += f"| {item.get('action', 'N/A')} | {item.get('assignee', 'TBD')} | {item.get('deadline', 'TBD')} |\n"
        notes += "\n## Discussion Topics\n\n"
        for topic in analysis.get("discussion_topics", []):
            notes += f"- {topic}\n"
        notes += "\n## Follow-ups\n\n"
        for follow in analysis.get("follow_ups", []):
            notes += f"- {follow}\n"
        return notes

class MeetingNotesAgent:
    """Orchestrate the complete meeting notes pipeline."""

    def __init__(
        self,
        whisper_model: str = "base",
        hf_token: Optional[str] = None,
        llm_model: str = "gpt-4o",
    ):
        self.transcriber = WhisperTranscriber(whisper_model)
        self.diarizer = SpeakerDiarizer(hf_token) if hf_token else None
        self.analyzer = MeetingContentAnalyzer(llm_model)
        self.notes_gen = MeetingNotesGenerator()

    async def process_meeting(
        self,
        audio_path: str,
        num_speakers: Optional[int] = None,
        metadata: Optional[Dict[str, Any]] = None,
    ) -> Dict[str, Any]:
        transcript = self.transcriber.transcribe(audio_path)
        segments = transcript["segments"]
        if self.diarizer:
            diarization = self.diarizer.diarize(audio_path, num_speakers)
            segments = self.diarizer.merge_with_transcript(diarization, segments)
        full_text = " ".join(s["text"] for s in segments)
        analysis = await self.analyzer.analyze(full_text, num_speakers or 2)
        notes = self.notes_gen.generate_notes(analysis, segments, metadata)
        return {
            "transcript": transcript,
            "segments": segments,
            "analysis": analysis,
            "notes": notes,
            "duration_seconds": segments[-1]["end"] if segments else 0,
        }