"""Live transcription manager for meeting recordings.

Two VADChunker instances (one per stream — mic + system audio) push
audio chunks into this runner's queue as they become ready. A single
worker thread pulls from the queue and runs each chunk through the
shared Whisper transcriber. Segments get shifted by the chunk's
meeting-level offset and merged into a sorted timeline. A debounced
writer thread atomically rewrites `transcript.md` every ~3 seconds so
that:

  - The user sees text appear in near-realtime in the bank window.
  - A crash mid-meeting preserves everything up to the last successful
    chunk transcription (typical worst case: lose ~30-60 s of in-flight
    audio that hadn't reached the queue yet).

Why one worker (not two parallel):
  - MLX whisper-medium uses the same GPU compute pool, parallel calls
    would just contend and slow each other down.
  - Serial keeps memory predictable and avoids re-warming model state.
  - 6× realtime on M1 means a 60 s chunk transcribes in ~10 s. Even with
    both streams chunking simultaneously, queue depth stays manageable.

Why atomic transcript.md writes:
  - The bank UI re-reads transcript.md whenever it refreshes (1 Hz tick).
    A partially-written file would render as garbage briefly.
  - Atomic replace via tempfile + os.replace makes the swap a single
    inode pointer update — readers never see torn state.
"""

from __future__ import annotations

import os
import queue
import tempfile
import threading
import time
from datetime import datetime
from pathlib import Path
from typing import Any, Optional

import numpy as np


# Cadence at which we flush queued updates to transcript.md. Lower number
# = snappier UI, higher number = fewer disk writes for the same content.
# 3 s is the sweet spot — feels alive but doesn't thrash.
WRITE_INTERVAL_S = 3.0

# Hard wait limit for stop()'s drain phase. Average chunk = ~10 s of
# compute on M1; worst-case backlog (mic + sys both queued, ~3 chunks
# each, hard-cap chunks at 60 s = ~30 s compute each) caps at ~3 minutes.
# 5 minutes leaves slack for the long tail.
DEFAULT_FLUSH_TIMEOUT_S = 300.0


class LiveTranscribeRunner:
    """One instance per active meeting. Owns the transcription queue,
    worker thread, transcript writer, and the segments timeline.

    Lifetime:
      1. Construct (auto-starts worker + writer threads)
      2. submit_chunk(...) from each VADChunker callback
      3. flush_and_wait() before stopping (blocks until queue is empty)
      4. stop() to tear down threads
    """

    def __init__(self, meeting_dir: Path, transcriber: Any,
                 started_at: datetime, language: Optional[str] = None) -> None:
        self.meeting_dir = meeting_dir
        self.transcript_path = meeting_dir / "transcript.md"
        self.transcriber = transcriber
        self.started_at = started_at
        self.language = language  # informational only

        # Inbound chunks: (speaker, start_offset_s, audio_f32).
        self._queue: queue.Queue = queue.Queue()
        self._stop_event = threading.Event()
        self._inflight = 0  # chunks currently being transcribed (not in queue)
        self._inflight_lock = threading.Lock()

        # Master segments timeline — list of {start, end, speaker, text}
        # sorted by start. Mutated under lock; the writer thread reads a
        # snapshot.
        self._segments: list[dict[str, Any]] = []
        self._segments_lock = threading.Lock()

        # Writer wake event — set when there are new segments to flush
        # OR when stop_event arms a final write.
        self._dirty = threading.Event()
        # Chunk counter for meta.json updates; live-only stat.
        self._chunks_processed = 0

        self._worker = threading.Thread(
            target=self._worker_loop, name="meeting-live-worker", daemon=True,
        )
        self._writer = threading.Thread(
            target=self._writer_loop, name="meeting-live-writer", daemon=True,
        )
        self._worker.start()
        self._writer.start()
        print(f"[live_transcribe] runner started for {meeting_dir.name}",
              flush=True)

    # ── inbound: chunkers submit here ──────────────────────────────────

    def submit_chunk(self, speaker: str, start_offset_s: float,
                     audio_f32: np.ndarray) -> None:
        if self._stop_event.is_set():
            # Late chunks after stop are dropped — they'd contain stale
            # audio that doesn't belong to any user-relevant state.
            print(f"[live_transcribe] dropping {speaker} chunk after stop",
                  flush=True)
            return
        self._queue.put((speaker, start_offset_s, audio_f32))

    # ── worker thread: drains queue → transcribes → updates segments ──

    def _worker_loop(self) -> None:
        while True:
            try:
                item = self._queue.get(timeout=0.5)
            except queue.Empty:
                # Idle. Exit if we're shutting down and nothing's in
                # flight; otherwise loop and wait for more.
                if self._stop_event.is_set():
                    with self._inflight_lock:
                        if self._inflight == 0:
                            break
                continue

            speaker, start_offset_s, audio_f32 = item
            with self._inflight_lock:
                self._inflight += 1
            try:
                self._transcribe_one(speaker, start_offset_s, audio_f32)
            except Exception as e:
                print(f"[live_transcribe] worker exception: {e}", flush=True)
            finally:
                with self._inflight_lock:
                    self._inflight -= 1
                self._queue.task_done()

        # Final wake for the writer so it does its last pass before exit.
        self._dirty.set()
        print("[live_transcribe] worker exited", flush=True)

    def _transcribe_one(self, speaker: str, start_offset_s: float,
                        audio_f32: np.ndarray) -> None:
        if audio_f32.size == 0:
            return
        try:
            result = self.transcriber.transcribe(audio_f32, sample_rate=16000)
        except Exception as e:
            print(f"[live_transcribe] transcribe failed for "
                  f"{speaker} @ {start_offset_s:.1f}s: {e}", flush=True)
            return

        new_segs: list[dict[str, Any]] = []
        for seg in result.get("segments", []) or []:
            text = (seg.get("text") or "").strip()
            if not text:
                continue
            new_segs.append({
                "start": float(seg.get("start", 0.0)) + start_offset_s,
                "end": float(seg.get("end", 0.0)) + start_offset_s,
                "text": text,
                "speaker": speaker,
            })

        if not new_segs:
            self._chunks_processed += 1
            return

        with self._segments_lock:
            self._segments.extend(new_segs)
            self._segments.sort(key=lambda s: s["start"])
        self._chunks_processed += 1
        self._dirty.set()

    # ── writer thread: debounced atomic transcript.md updates ──────────

    def _writer_loop(self) -> None:
        last_write = 0.0
        while True:
            # Wait for either a dirty signal or the regular interval.
            signaled = self._dirty.wait(timeout=WRITE_INTERVAL_S)
            self._dirty.clear()

            # Respect minimum interval so a flurry of fast chunks doesn't
            # cause one rewrite per chunk.
            now = time.time()
            if signaled and (now - last_write) < WRITE_INTERVAL_S:
                # Sleep the remainder of the interval and then write.
                time.sleep(WRITE_INTERVAL_S - (now - last_write))

            try:
                self._write_transcript()
                last_write = time.time()
            except Exception as e:
                print(f"[live_transcribe] write failed: {e}", flush=True)

            # Exit if we're shutting down AND there are no pending writes
            # and the worker is done.
            if self._stop_event.is_set() and self._queue.empty():
                with self._inflight_lock:
                    if self._inflight == 0:
                        break
        # One final write to capture anything queued after our last pass.
        try:
            self._write_transcript()
        except Exception as e:
            print(f"[live_transcribe] final write failed: {e}", flush=True)
        print("[live_transcribe] writer exited", flush=True)

    def _write_transcript(self) -> None:
        with self._segments_lock:
            segs = list(self._segments)

        started_iso = self.started_at.isoformat(timespec="seconds")
        lines = [
            f"# Meeting — {started_iso}",
            "Captured: mic (You) + system audio (System)",
            "Status: live transcription in progress",
            "",
            "## Transcript",
            "",
        ]
        if not segs:
            lines.append("_(transcription will appear here as the meeting progresses)_")
        else:
            for seg in segs:
                lines.append(
                    f"[{_format_mmss(seg['start'])}] {seg['speaker']}: {seg['text']}"
                )

        content = "\n".join(lines) + "\n"
        # Atomic write: tmp file in same dir (same filesystem so rename
        # is atomic) → os.replace. Readers never see torn content.
        fd, tmp_name = tempfile.mkstemp(
            prefix=".transcript-",
            suffix=".md.tmp",
            dir=str(self.transcript_path.parent),
        )
        try:
            with os.fdopen(fd, "w", encoding="utf-8") as f:
                f.write(content)
            os.replace(tmp_name, self.transcript_path)
        except Exception:
            # Best effort cleanup on failure.
            try:
                os.unlink(tmp_name)
            except OSError:
                pass
            raise

    # ── shutdown ───────────────────────────────────────────────────────

    def flush_and_wait(self, timeout: float = DEFAULT_FLUSH_TIMEOUT_S) -> bool:
        """Block until every queued chunk has been transcribed. Called by
        MeetingRecording.stop() so the returned transcript.md is complete
        before the stop call returns. Returns True if queue fully drained,
        False on timeout (caller logs but proceeds anyway)."""
        deadline = time.time() + timeout
        while True:
            with self._inflight_lock:
                inflight = self._inflight
            if self._queue.empty() and inflight == 0:
                break
            if time.time() > deadline:
                print(f"[live_transcribe] flush_and_wait TIMEOUT after "
                      f"{timeout:.0f}s — queue={self._queue.qsize()} "
                      f"inflight={inflight}", flush=True)
                return False
            time.sleep(0.2)
        # Give the writer a moment to flush the last segments.
        self._dirty.set()
        time.sleep(0.5)
        return True

    def stop(self) -> None:
        """Signal worker + writer to exit. Must be called AFTER
        flush_and_wait() if you care about transcript completeness."""
        self._stop_event.set()
        self._dirty.set()
        self._worker.join(timeout=5.0)
        self._writer.join(timeout=5.0)
        # Mark final state in the transcript header.
        try:
            self._mark_transcript_final()
        except Exception as e:
            print(f"[live_transcribe] final header rewrite failed: {e}",
                  flush=True)

    def _mark_transcript_final(self) -> None:
        """Rewrite transcript.md with a 'complete' status header instead
        of 'live in progress'."""
        with self._segments_lock:
            segs = list(self._segments)
        started_iso = self.started_at.isoformat(timespec="seconds")
        lines = [
            f"# Meeting — {started_iso}",
            "Captured: mic (You) + system audio (System)",
            "",
            "## Transcript",
            "",
        ]
        for seg in segs:
            lines.append(
                f"[{_format_mmss(seg['start'])}] {seg['speaker']}: {seg['text']}"
            )
        content = "\n".join(lines) + "\n"
        fd, tmp_name = tempfile.mkstemp(
            prefix=".transcript-",
            suffix=".md.tmp",
            dir=str(self.transcript_path.parent),
        )
        try:
            with os.fdopen(fd, "w", encoding="utf-8") as f:
                f.write(content)
            os.replace(tmp_name, self.transcript_path)
        except Exception:
            try:
                os.unlink(tmp_name)
            except OSError:
                pass
            raise

    # ── stats for meta.json / UI ──────────────────────────────────────

    @property
    def chunks_processed(self) -> int:
        return self._chunks_processed

    @property
    def queue_depth(self) -> int:
        with self._inflight_lock:
            return self._queue.qsize() + self._inflight


def _format_mmss(seconds: float) -> str:
    s = int(round(seconds))
    return f"{s // 60:02d}:{s - (s // 60) * 60:02d}"
