Transcribe multi-lingual videos

async def transcribe_audio(audio_path: str, use_diarization: bool = True) -> tuple:
    """Transcribe audio file using GPT-4o with diarization."""
    if use_diarization:
        model = Model("gpt-4o-transcribe-diarize")

        with open(audio_path, "rb") as audio_file:
            result = await model.speech_to_text(
                audio_file=audio_file,
                language=None,  # Auto-detect language
                model="gpt-4o-transcribe-diarize",
                response_format="diarized_json",
                chunking_strategy="auto",
            )

        transcript = format_diarized_transcript(result)
        return transcript, result

async def transcribe_audio_chunked(
    video_path: str,
    temp_dir: str,
    use_diarization: bool = True,
    progress_callback=None,
) -> str:
    """Transcribe a video by extracting and processing audio in chunks."""
    total_duration = await get_video_duration(video_path)

    MAX_CHUNK_SIZE_MB = 5
    ESTIMATED_MB_PER_MIN = 1.0
    MAX_CHUNK_DURATION = 300  # 5 minutes

    chunk_duration = min(
        (MAX_CHUNK_SIZE_MB / ESTIMATED_MB_PER_MIN) * 60,
        MAX_CHUNK_DURATION
    )

    transcripts = []
    current_time = 0
    chunk_num = 0

    while current_time < total_duration:
        chunk_num += 1
        chunk_audio = Path(temp_dir) / f"chunk_{chunk_num}.mp3"

        # Extract audio chunk
        success = await extract_audio_chunk(
            video_path, str(chunk_audio), current_time, chunk_duration
        )

        if success:
            # Transcribe chunk with diarization
            chunk_transcript, _ = await transcribe_audio(
                str(chunk_audio), use_diarization=use_diarization
            )

            # Prefix speaker labels with chunk number for later matching
            if use_diarization:
                chunk_transcript = prefix_speakers_with_chunk(
                    chunk_transcript, chunk_num
                )

            transcripts.append(chunk_transcript.strip())

        current_time += chunk_duration

    return "\n\n".join(transcripts)

SPEAKER_MATCHER_INSTRUCTIONS = """
You are an expert at analyzing transcripts to identify and match speakers.

You will receive a transcript transcribed in multiple chunks. Each chunk has
its own speaker labels (e.g., "Speaker 1A", "Speaker 1B" for chunk 1).

Analyze the transcript and return ONLY a JSON mapping of speaker labels.

CLUES TO LOOK FOR:
- Self-introductions: "I'm Maria", "My name is Dr. Lopez"
- Being addressed by name: "Thank you, Maria"
- Role indicators: "As the host...", "In my research..."
- Speaking patterns: Who asks questions (host) vs who answers (guest)

OUTPUT FORMAT (JSON only):
{
  "mapping": {
    "Speaker 1A": "Maria (Host)",
    "Speaker 1B": "Dr. Lopez (Guest)",
    "Speaker 2A": "Maria (Host)"
  },
  "confidence": "high",
  "notes": "Maria identified as host from introduction."
}
"""

async def match_speakers_across_chunks(transcript: str) -> str:
    """Unify speaker labels across transcript chunks using LLM analysis."""
    model = Model("claude-sonnet-4-5")
    messages = [
        {"role": "system", "content": SPEAKER_MATCHER_INSTRUCTIONS},
        {
            "role": "user",
            "content": f"Analyze this transcript and return a JSON speaker mapping:\n\n{transcript}",
        },
    ]

    response = await model.complete_chat(messages, stream=False)
    result = response.choices[0].message.content.strip()

    # Parse the JSON mapping
    mapping_data = extract_json_from_response(result)
    mapping = mapping_data.get("mapping", {})

    # Apply mapping to transcript
    return apply_speaker_mapping(transcript, mapping)

TRANSLATOR_INSTRUCTIONS = """
You are a professional translator. You will receive transcribed text that may be
in any language. The transcription includes speaker labels.

Your job is to:
1. Identify the source language
2. Translate the text accurately to English
3. PRESERVE all existing speaker labels exactly as they appear

Output format:
---
Source Language: [detected language]

Translation:
[speaker label]: [translated text]
---

CRITICAL RULES:
- PRESERVE existing speaker labels EXACTLY as they appear in the input
- DO NOT modify, rename, or add any speaker labels
- Maintain natural, fluent English while preserving the original meaning
"""

async def initialize(node: Node) -> bool:
    """Initialize the translator agent."""
    global _agent

    _agent = await Agent.start(
        node=node,
        name="translator",
        instructions=TRANSLATOR_INSTRUCTIONS,
        model=Model("claude-sonnet-4-5"),
    )

    return True

def split_transcript_into_chunks(
    text: str, max_chars: int = 15000
) -> list[str]:
    """Split transcript into chunks at speaker boundaries."""
    if len(text) <= max_chars:
        return [text]

    chunks = []
    current_chunk = []
    current_length = 0

    # Split on double newlines (speaker boundaries)
    segments = text.split("\n\n")

    for segment in segments:
        segment_length = len(segment) + 2

        if current_length + segment_length > max_chars and current_chunk:
            chunks.append("\n\n".join(current_chunk))
            current_chunk = []
            current_length = 0

        current_chunk.append(segment)
        current_length += segment_length

    if current_chunk:
        chunks.append("\n\n".join(current_chunk))

    return chunks

@app.post("/webhook/box")
async def box_webhook(request: Request, background_tasks: BackgroundTasks):
    """Handle Box webhook notifications for FILE.UPLOADED events."""
    payload = await request.json()

    # Validate payload and determine action
    result = box.validate_webhook_payload(payload, VIDEO_EXTENSIONS)

    if result["action"] == "challenge":
        return JSONResponse({"challenge": result["challenge"]})

    if result["action"] == "ignore":
        return JSONResponse(result["response"])

    # Process the video in the background
    file_id = result["file_id"]
    filename = result["filename"]

    background_tasks.add_task(
        box.process_video,
        file_id,
        filename,
        process_video,
        generate_result_markdown,
    )

    return JSONResponse(
        {"status": "processing", "file_id": file_id, "filename": filename}
    )

async def handle_websocket_upload(scope, receive, send, process_video_func):
    """Handle chunked video uploads via raw ASGI WebSocket."""
    await send({"type": "websocket.accept"})

    while True:
        message = await receive()

        if message["type"] == "websocket.receive":
            if "text" in message:
                data = json.loads(message["text"])

                if data.get("type") == "start":
                    # Initialize upload
                    filename = data.get("filename")
                    temp_dir = tempfile.mkdtemp()
                    video_path = Path(temp_dir) / filename
                    video_file = open(video_path, "wb")
                    await send_json({"type": "ready"})

                elif data.get("type") == "end":
                    # Process the video
                    video_file.close()
                    result = await process_video_func(
                        str(video_path), filename, progress_callback
                    )
                    await send_json({"type": "result", **result})

            elif "bytes" in message:
                # Write chunk to file
                video_file.write(message["bytes"])
                await send_json({"type": "chunk_ack"})

@dataclass
class ProcessingJob:
    """Represent a video processing job in the queue."""
    id: str
    filename: str
    status: str  # "queued", "processing", "complete", "error"
    progress: int
    queued_at: datetime

processing_lock = asyncio.Lock()
current_job: Optional[ProcessingJob] = None
job_queue: deque[ProcessingJob] = deque()

while job.status == "queued":
    await asyncio.sleep(2)

    async with processing_lock:
        position = get_job_position(job_id)
        if position:
            await send_json({
                "type": "queue_update",
                "position": position,
                "message": f"You are #{position} in queue.",
            })