AudioToTextNotes.py

"""Record microphone audio, transcribe it with Mega-ASR, and create notes."""

from __future__ import annotations

import argparse
import importlib
import json
import logging
import math
import os
import queue
import re
import sys
import tempfile
import threading
import time
from dataclasses import dataclass
from datetime import datetime
from pathlib import Path
from typing import Any, Sequence


LOGGER = logging.getLogger("ai_meeting_notes")
PROJECT_ROOT = Path(__file__).resolve().parent
SUPPORTED_PYTHON = ((3, 13),)
DEFAULT_SUMMARY_MODEL = "facebook/bart-large-cnn"
DEFAULT_SUMMARY_MODEL_REVISION = "37f520fa929c961707657b28798b30c003dd100b"
FORBIDDEN_MODEL_CONFIG_KEYS = frozenset(
    {"_attn_implementation_internal", "auto_map"}
)


@dataclass(frozen=True)
class AppConfig:
    sample_rate: int = 16_000
    chunk_seconds: float = 150.0
    max_backlog_seconds: float = 600.0
    summary_model: str = DEFAULT_SUMMARY_MODEL
    summary_model_revision: str = DEFAULT_SUMMARY_MODEL_REVISION
    language: str | None = "English"
    device: str = "auto"
    input_device: str | int | None = None
    output_dir: Path = Path(".")
    save_raw_transcript: bool = True
    final_summary: bool = False
    minimum_transcript_characters: int = 20
    summary_min_tokens: int = 30
    summary_max_tokens: int = 150
    asr_context_characters: int = 2_000

    @property
    def chunk_frames(self) -> int:
        return max(1, round(self.sample_rate * self.chunk_seconds))

    @property
    def callback_frames(self) -> int:
        return self.sample_rate

    @property
    def queue_blocks(self) -> int:
        seconds_per_block = self.callback_frames / self.sample_rate
        return max(1, math.ceil(self.max_backlog_seconds / seconds_per_block))


class AudioChunkAccumulator:
    """Accumulate audio blocks and return chunks with an exact frame count."""

    def __init__(self, numpy_module: Any) -> None:
        self._np = numpy_module
        self._blocks: list[Any] = []
        self.frame_count = 0

    def add(self, block: Any) -> None:
        self._blocks.append(block)
        self.frame_count += len(block)

    def pop(self, frame_count: int) -> Any:
        if frame_count <= 0 or frame_count > self.frame_count:
            raise ValueError("frame_count must be positive and available")

        remaining = frame_count
        pieces: list[Any] = []
        while remaining:
            block = self._blocks[0]
            take = min(remaining, len(block))
            pieces.append(block[:take])
            if take == len(block):
                self._blocks.pop(0)
            else:
                self._blocks[0] = block[take:]
            remaining -= take

        self.frame_count -= frame_count
        return self._np.concatenate(pieces, axis=0)

    def pop_all(self) -> Any | None:
        if not self.frame_count:
            return None
        return self.pop(self.frame_count)


def get_current_timestamp() -> str:
    return datetime.now().astimezone().isoformat(timespec="seconds")


def append_timestamped(path: Path, text: str) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    with path.open("a", encoding="utf-8", newline="\n") as output:
        output.write(f"{get_current_timestamp()} - {text.strip()}\n\n")


def generation_lengths(
    input_tokens: int, requested_minimum: int, requested_maximum: int
) -> tuple[int, int] | None:
    """Choose safe generation limits for both short and long inputs."""
    if input_tokens < 16:
        return None
    maximum = min(requested_maximum, max(8, input_tokens // 2))
    minimum = min(requested_minimum, maximum - 1, max(4, input_tokens // 8))
    return max(1, minimum), max(2, maximum)


def import_dependencies(names: Sequence[str]) -> dict[str, Any]:
    modules: dict[str, Any] = {}
    missing: list[str] = []
    for name in names:
        try:
            modules[name] = importlib.import_module(name)
        except ImportError:
            missing.append(name)
    if missing:
        raise RuntimeError(
            "Missing dependencies: "
            + ", ".join(missing)
            + ". Activate the project environment and run "
            "'python -m pip install -r requirements.txt'."
        )
    return modules


def validate_summary_model_snapshot(snapshot: Path) -> None:
    """Reject configuration fields that can cause remote code to be loaded."""

    primary_config = snapshot / "config.json"
    if not primary_config.is_file():
        raise RuntimeError(f"Summary model snapshot has no config.json: {snapshot}")
    if not any(snapshot.rglob("*.safetensors")):
        raise RuntimeError(
            f"Summary model snapshot has no SafeTensors weights: {snapshot}"
        )
    config_paths = list(snapshot.rglob("*config*.json"))

    def find_forbidden_keys(value: Any) -> set[str]:
        if isinstance(value, dict):
            found = set(FORBIDDEN_MODEL_CONFIG_KEYS.intersection(value))
            for nested in value.values():
                found.update(find_forbidden_keys(nested))
            return found
        if isinstance(value, list):
            found: set[str] = set()
            for nested in value:
                found.update(find_forbidden_keys(nested))
            return found
        return set()

    for config_path in config_paths:
        try:
            config = json.loads(config_path.read_text(encoding="utf-8"))
        except (OSError, json.JSONDecodeError) as error:
            raise RuntimeError(
                f"Summary model configuration is unreadable: {config_path}"
            ) from error
        forbidden = find_forbidden_keys(config)
        if forbidden:
            names = ", ".join(sorted(forbidden))
            raise RuntimeError(
                f"Summary model configuration contains unsafe field(s) {names}: "
                f"{config_path}"
            )


def download_summary_model(
    huggingface_hub_module: Any, model: str, revision: str
) -> Path:
    """Download an immutable model revision and validate it before loading."""

    if re.fullmatch(r"[0-9a-f]{40}", revision) is None:
        raise RuntimeError(
            "--summary-model-revision must be a full 40-character Hugging Face "
            "commit SHA"
        )
    snapshot = Path(
        huggingface_hub_module.snapshot_download(
            repo_id=model,
            revision=revision,
            allow_patterns=("*.json", "*.txt", "*.model", "*.safetensors"),
        )
    ).resolve()
    validate_summary_model_snapshot(snapshot)
    return snapshot


def create_summary_pipeline(
    transformers_module: Any, model_path: Path, device: Any
) -> Any:
    """Load a validated local snapshot without forwarding loader options."""

    return transformers_module.pipeline(
        "summarization",
        model=str(model_path),
        device=device,
        trust_remote_code=False,
    )


def resolve_device(torch_module: Any, requested: str) -> str:
    if requested != "auto":
        if requested == "cuda" and not torch_module.cuda.is_available():
            raise RuntimeError("CUDA was requested, but PyTorch cannot access a CUDA device.")
        return requested
    return "cuda" if torch_module.cuda.is_available() else "cpu"


def validate_mega_asr_installation(
    repository: Path, checkpoint: Path, routing_enabled: bool
) -> None:
    required_paths = [
        repository / "src" / "MegaASR" / "model" / "megaASR.py",
        checkpoint / "Qwen3-ASR-1.7B" / "config.json",
        checkpoint / "mega-asr-merged" / "adapter_config.json",
    ]
    if routing_enabled:
        required_paths.append(
            checkpoint
            / "audio_quality_router"
            / "best_acc_model.safetensors"
        )
    missing = [str(path) for path in required_paths if not path.exists()]
    if missing:
        formatted = "\n  - ".join(missing)
        raise RuntimeError(
            "Mega-ASR is not set up. Missing:\n  - "
            f"{formatted}\nRun 'python setup_mega_asr.py' first."
        )


def import_mega_asr(repository: Path) -> Any:
    source_dir = str((repository / "src").resolve())
    if source_dir not in sys.path:
        sys.path.insert(0, source_dir)
    try:
        module = importlib.import_module("MegaASR.model.megaASR")
    except ImportError as error:
        raise RuntimeError(
            "Mega-ASR could not be imported from "
            f"'{source_dir}'. Re-run 'python setup_mega_asr.py'."
        ) from error
    return module.MegaASR


@dataclass(frozen=True)
class TranscriptionResult:
    text: str
    used_mega_adapter: bool | None = None
    degraded_probability: float | None = None


class MegaASRTranscriber:
    """Adapt in-memory recorder chunks to Mega-ASR's routed file API."""

    def __init__(
        self,
        model: Any,
        soundfile_module: Any,
        sample_rate: int,
        language: str | None,
        routing_enabled: bool,
        router_sample_seconds: float,
    ) -> None:
        self._model = model
        self._soundfile = soundfile_module
        self._sample_rate = sample_rate
        self._language = language
        self._routing_enabled = routing_enabled
        self._router_sample_frames = max(
            1, round(router_sample_seconds * sample_rate)
        )

    @staticmethod
    def _normalize_text(value: Any) -> str:
        if isinstance(value, (list, tuple)):
            return " ".join(str(item).strip() for item in value if str(item).strip())
        return str(value or "").strip()

    def _write_temporary_wav(self, audio: Any) -> Path:
        descriptor, temporary_name = tempfile.mkstemp(
            prefix="ai-meeting-notes-", suffix=".wav"
        )
        os.close(descriptor)
        temporary_path = Path(temporary_name)
        try:
            self._soundfile.write(
                str(temporary_path),
                audio,
                self._sample_rate,
                subtype="PCM_16",
            )
        except Exception:
            temporary_path.unlink(missing_ok=True)
            raise
        return temporary_path

    def transcribe(self, audio_chunk: Any, context: str = "") -> TranscriptionResult:
        flattened = audio_chunk.flatten()
        full_audio_path = self._write_temporary_wav(flattened)
        route_audio_path: Path | None = None
        used_mega_adapter = True
        degraded_probability: float | None = None
        try:
            if self._routing_enabled:
                route_path = full_audio_path
                if len(flattened) > self._router_sample_frames:
                    start = (len(flattened) - self._router_sample_frames) // 2
                    route_audio = flattened[
                        start : start + self._router_sample_frames
                    ]
                    route_audio_path = self._write_temporary_wav(route_audio)
                    route_path = route_audio_path
                used_mega_adapter, degraded_probability = (
                    self._model.router.predict(route_path)
                )

            inference = (
                self._model.infer_with_lora
                if used_mega_adapter
                else self._model.infer_without_lora
            )
            result = inference(
                full_audio_path,
                language=self._language,
                context=context,
            )
        finally:
            full_audio_path.unlink(missing_ok=True)
            if route_audio_path is not None:
                route_audio_path.unlink(missing_ok=True)

        return TranscriptionResult(
            text=self._normalize_text(result),
            used_mega_adapter=used_mega_adapter,
            degraded_probability=degraded_probability,
        )


class TextSummarizer:
    def __init__(self, pipeline: Any, minimum_tokens: int, maximum_tokens: int) -> None:
        self._pipeline = pipeline
        self._tokenizer = pipeline.tokenizer
        reported_limit = getattr(self._tokenizer, "model_max_length", 1024)
        if not isinstance(reported_limit, int) or reported_limit > 100_000:
            reported_limit = 1024
        self._input_limit = max(64, reported_limit - 2)
        self._minimum_tokens = minimum_tokens
        self._maximum_tokens = maximum_tokens

    def _token_ids(self, text: str) -> list[int]:
        return self._tokenizer.encode(text, add_special_tokens=False)

    def _summarize_piece(self, text: str) -> str:
        input_tokens = len(self._token_ids(text))
        limits = generation_lengths(
            input_tokens, self._minimum_tokens, self._maximum_tokens
        )
        if limits is None:
            return text.strip()
        minimum, maximum = limits
        result = self._pipeline(
            text,
            min_new_tokens=minimum,
            max_new_tokens=maximum,
            do_sample=False,
            truncation=True,
        )
        return result[0]["summary_text"].strip()

    def summarize(self, text: str) -> str:
        """Summarize long text hierarchically without exceeding model context."""
        current = text.strip()
        for _ in range(4):
            token_ids = self._token_ids(current)
            if len(token_ids) <= self._input_limit:
                return self._summarize_piece(current)

            pieces = [
                self._tokenizer.decode(
                    token_ids[start : start + self._input_limit],
                    skip_special_tokens=True,
                )
                for start in range(0, len(token_ids), self._input_limit)
            ]
            reduced = "\n".join(self._summarize_piece(piece) for piece in pieces)
            if len(self._token_ids(reduced)) >= len(token_ids):
                return reduced
            current = reduced
        return self._summarize_piece(current)


class MeetingRecorder:
    def __init__(
        self,
        config: AppConfig,
        numpy_module: Any,
        transcriber: MegaASRTranscriber,
        summarizer: TextSummarizer,
    ) -> None:
        self.config = config
        self._np = numpy_module
        self._transcriber = transcriber
        self._summarizer = summarizer
        self._queue: queue.Queue[Any] = queue.Queue(maxsize=config.queue_blocks)
        self._stop = threading.Event()
        self._worker = threading.Thread(
            target=self._process_audio, name="meeting-notes-processor", daemon=False
        )
        self._counter_lock = threading.Lock()
        self._dropped_blocks = 0
        self._audio_statuses: list[str] = []
        self._session_transcripts: list[str] = []
        self._asr_context = ""
        self._fatal_error: BaseException | None = None

    @property
    def worker_alive(self) -> bool:
        return self._worker.is_alive()

    @property
    def fatal_error(self) -> BaseException | None:
        return self._fatal_error

    def start(self) -> None:
        self._worker.start()

    def audio_callback(self, indata: Any, frames: int, time_info: Any, status: Any) -> None:
        del frames, time_info
        if status:
            with self._counter_lock:
                self._audio_statuses.append(str(status))
        try:
            self._queue.put_nowait(indata.copy())
        except queue.Full:
            with self._counter_lock:
                self._dropped_blocks += 1

    def _report_audio_issues(self) -> None:
        with self._counter_lock:
            dropped, self._dropped_blocks = self._dropped_blocks, 0
            statuses, self._audio_statuses = self._audio_statuses, []
        if dropped:
            LOGGER.warning(
                "Audio processing fell behind; dropped %d second(s) of audio.", dropped
            )
        for status in sorted(set(statuses)):
            LOGGER.warning("Audio input reported: %s", status)

    def _transcribe(self, audio_chunk: Any, partial: bool = False) -> None:
        duration = len(audio_chunk) / self.config.sample_rate
        label = "final partial chunk" if partial else "chunk"
        LOGGER.info("Transcribing %s (%.1f seconds)...", label, duration)
        result = self._transcriber.transcribe(
            audio_chunk, context=self._asr_context
        )
        transcription = result.text
        if len(transcription) < self.config.minimum_transcript_characters:
            LOGGER.info("No substantial speech detected; skipping this chunk.")
            return

        if result.used_mega_adapter is not None:
            route = "Mega-ASR adapter" if result.used_mega_adapter else "base Qwen3-ASR"
            if result.degraded_probability is None:
                LOGGER.info("ASR route: %s", route)
            else:
                LOGGER.info(
                    "ASR route: %s (degraded probability %.3f)",
                    route,
                    result.degraded_probability,
                )

        if self.config.final_summary:
            self._session_transcripts.append(transcription)
        if self.config.asr_context_characters:
            updated_context = f"{self._asr_context}\n{transcription}".strip()
            self._asr_context = updated_context[
                -self.config.asr_context_characters :
            ]
        preview = transcription[:200] + ("..." if len(transcription) > 200 else "")
        LOGGER.info("Transcript: %s", preview)
        if self.config.save_raw_transcript:
            append_timestamped(self.config.output_dir / "raw_transcripts.txt", transcription)

        LOGGER.info("Creating notes...")
        summary = self._summarizer.summarize(transcription)
        append_timestamped(self.config.output_dir / "notes.txt", summary)
        LOGGER.info("Notes: %s", summary)

    def _process_audio(self) -> None:
        accumulator = AudioChunkAccumulator(self._np)
        try:
            while not self._stop.is_set() or not self._queue.empty():
                try:
                    block = self._queue.get(timeout=0.5)
                except queue.Empty:
                    self._report_audio_issues()
                    continue
                accumulator.add(block)
                self._queue.task_done()
                self._report_audio_issues()

                while accumulator.frame_count >= self.config.chunk_frames:
                    chunk = accumulator.pop(self.config.chunk_frames)
                    try:
                        self._transcribe(chunk)
                    except Exception:
                        LOGGER.exception(
                            "Failed to process an audio chunk; recording continues."
                        )

            partial = accumulator.pop_all()
            if partial is not None and len(partial) >= self.config.sample_rate:
                try:
                    self._transcribe(partial, partial=True)
                except Exception:
                    LOGGER.exception("Failed to process the final partial chunk.")
        except Exception as error:
            self._fatal_error = error
            LOGGER.exception("The audio processing worker stopped unexpectedly.")

    def stop(self) -> None:
        self._stop.set()
        self._worker.join()

    def save_final_summary(self) -> None:
        if not self.config.final_summary or not self._session_transcripts:
            return
        LOGGER.info("Creating the final meeting summary...")
        summary = self._summarizer.summarize("\n".join(self._session_transcripts))
        append_timestamped(self.config.output_dir / "meeting_summary.txt", summary)
        LOGGER.info("Final meeting summary: %s", summary)


def parse_input_device(value: str | None) -> str | int | None:
    if value is None:
        return None
    try:
        return int(value)
    except ValueError:
        return value


def build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(
        description="Record a meeting and create local AI transcripts and notes."
    )
    parser.add_argument("--chunk-seconds", type=float, default=150.0)
    parser.add_argument("--max-backlog-seconds", type=float, default=600.0)
    parser.add_argument(
        "--mega-asr-repo",
        type=Path,
        default=PROJECT_ROOT / "third_party" / "Mega-ASR",
        help="Path to the xzf-thu/Mega-ASR source checkout.",
    )
    parser.add_argument(
        "--mega-asr-checkpoint",
        type=Path,
        default=PROJECT_ROOT / "models" / "Mega-ASR",
        help="Path to the downloaded zhifeixie/Mega-ASR checkpoint.",
    )
    parser.add_argument(
        "--routing",
        action=argparse.BooleanOptionalAction,
        default=True,
        help="Route clean audio to base Qwen3-ASR and degraded audio to Mega-ASR.",
    )
    parser.add_argument("--router-threshold", type=float, default=0.5)
    parser.add_argument("--router-sample-seconds", type=float, default=30.0)
    parser.add_argument("--max-asr-tokens", type=int, default=1_024)
    parser.add_argument(
        "--keep-delta-on-gpu",
        action=argparse.BooleanOptionalAction,
        default=True,
        help="Keep Mega-ASR adapter deltas on the GPU for faster route changes.",
    )
    parser.add_argument("--asr-context-characters", type=int, default=2_000)
    parser.add_argument("--summary-model", default=DEFAULT_SUMMARY_MODEL)
    parser.add_argument(
        "--summary-model-revision",
        default=DEFAULT_SUMMARY_MODEL_REVISION,
        help="Full Hugging Face commit SHA for the summarization model.",
    )
    parser.add_argument(
        "--language",
        default="English",
        help="Qwen3-ASR language name, or 'auto' to detect it.",
    )
    parser.add_argument("--device", choices=("auto", "cpu", "cuda"), default="auto")
    parser.add_argument("--input-device", help="Microphone name or numeric device ID.")
    parser.add_argument("--list-devices", action="store_true")
    parser.add_argument("--output-dir", type=Path, default=Path("."))
    parser.add_argument("--no-raw-transcript", action="store_true")
    parser.add_argument("--final-summary", action="store_true")
    parser.add_argument("--minimum-transcript-characters", type=int, default=20)
    parser.add_argument("--summary-min-tokens", type=int, default=30)
    parser.add_argument("--summary-max-tokens", type=int, default=150)
    parser.add_argument("--verbose", action="store_true")
    return parser


def validate_args(parser: argparse.ArgumentParser, args: argparse.Namespace) -> None:
    positive_values = {
        "--chunk-seconds": args.chunk_seconds,
        "--max-backlog-seconds": args.max_backlog_seconds,
        "--minimum-transcript-characters": args.minimum_transcript_characters,
        "--max-asr-tokens": args.max_asr_tokens,
        "--router-sample-seconds": args.router_sample_seconds,
        "--summary-min-tokens": args.summary_min_tokens,
        "--summary-max-tokens": args.summary_max_tokens,
    }
    for option, value in positive_values.items():
        if value <= 0:
            parser.error(f"{option} must be greater than zero")
    if args.summary_min_tokens >= args.summary_max_tokens:
        parser.error("--summary-min-tokens must be less than --summary-max-tokens")
    if not 0.0 <= args.router_threshold <= 1.0:
        parser.error("--router-threshold must be between zero and one")
    if args.asr_context_characters < 0:
        parser.error("--asr-context-characters cannot be negative")


def run(args: argparse.Namespace) -> int:
    if args.list_devices:
        sounddevice = import_dependencies(("sounddevice",))["sounddevice"]
        print(sounddevice.query_devices())
        return 0

    if sys.version_info[:2] not in SUPPORTED_PYTHON:
        supported = ", ".join(".".join(map(str, version)) for version in SUPPORTED_PYTHON)
        raise RuntimeError(
            f"Python {supported} is supported; this interpreter is "
            f"{sys.version_info.major}.{sys.version_info.minor}."
        )

    repository = args.mega_asr_repo.expanduser().resolve()
    checkpoint = args.mega_asr_checkpoint.expanduser().resolve()
    validate_mega_asr_installation(repository, checkpoint, args.routing)
    dependencies = import_dependencies(
        (
            "numpy",
            "huggingface_hub",
            "qwen_asr",
            "safetensors",
            "scipy",
            "sounddevice",
            "soundfile",
            "torch",
            "torchaudio",
            "transformers",
        )
    )
    torch_module = dependencies["torch"]
    device = resolve_device(torch_module, args.device)
    LOGGER.info("Using %s for model inference.", device.upper())

    LOGGER.info("Loading Mega-ASR from '%s'...", checkpoint)
    MegaASR = import_mega_asr(repository)
    mega_asr_model = MegaASR(
        model_path=checkpoint / "Qwen3-ASR-1.7B",
        lora_dir=checkpoint / "mega-asr-merged",
        router_checkpoint=(
            checkpoint
            / "audio_quality_router"
            / "best_acc_model.safetensors"
        ),
        routing_enabled=args.routing,
        quality_threshold=args.router_threshold,
        device_map="cuda:0" if device == "cuda" else "cpu",
        quality_device=device,
        max_inference_batch_size=1,
        max_new_tokens=args.max_asr_tokens,
        keep_delta_on_gpu=args.keep_delta_on_gpu and device == "cuda",
        backend="transformers",
    )
    LOGGER.info(
        "Downloading summarization model '%s' at revision %s...",
        args.summary_model,
        args.summary_model_revision,
    )
    summary_model_path = download_summary_model(
        dependencies["huggingface_hub"],
        args.summary_model,
        args.summary_model_revision,
    )
    LOGGER.info("Loading validated summarization model from '%s'...", summary_model_path)
    pipeline_device: Any = 0 if device == "cuda" else -1
    summary_pipeline = create_summary_pipeline(
        dependencies["transformers"], summary_model_path, pipeline_device
    )
    summarizer = TextSummarizer(
        summary_pipeline, args.summary_min_tokens, args.summary_max_tokens
    )

    config = AppConfig(
        chunk_seconds=args.chunk_seconds,
        max_backlog_seconds=args.max_backlog_seconds,
        summary_model=args.summary_model,
        summary_model_revision=args.summary_model_revision,
        language=None if args.language.lower() == "auto" else args.language,
        device=device,
        input_device=parse_input_device(args.input_device),
        output_dir=args.output_dir.expanduser().resolve(),
        save_raw_transcript=not args.no_raw_transcript,
        final_summary=args.final_summary,
        minimum_transcript_characters=args.minimum_transcript_characters,
        summary_min_tokens=args.summary_min_tokens,
        summary_max_tokens=args.summary_max_tokens,
        asr_context_characters=args.asr_context_characters,
    )
    config.output_dir.mkdir(parents=True, exist_ok=True)

    transcriber = MegaASRTranscriber(
        mega_asr_model,
        dependencies["soundfile"],
        config.sample_rate,
        config.language,
        args.routing,
        args.router_sample_seconds,
    )
    recorder = MeetingRecorder(
        config, dependencies["numpy"], transcriber, summarizer
    )
    recorder.start()
    LOGGER.info("Recording. Press Ctrl+C to stop.")
    try:
        with dependencies["sounddevice"].InputStream(
            samplerate=config.sample_rate,
            blocksize=config.callback_frames,
            channels=1,
            dtype="float32",
            device=config.input_device,
            callback=recorder.audio_callback,
        ):
            while recorder.worker_alive:
                time.sleep(0.5)
    except KeyboardInterrupt:
        LOGGER.info("Stopping recording...")
    finally:
        recorder.stop()

    if recorder.fatal_error is not None:
        raise RuntimeError(
            "The audio processing worker stopped unexpectedly."
        ) from recorder.fatal_error
    recorder.save_final_summary()
    LOGGER.info("Finished. Output directory: %s", config.output_dir)
    return 0


def main(argv: Sequence[str] | None = None) -> int:
    parser = build_parser()
    args = parser.parse_args(argv)
    validate_args(parser, args)
    logging.basicConfig(
        level=logging.DEBUG if args.verbose else logging.INFO,
        format="%(asctime)s %(levelname)s %(message)s",
    )
    try:
        return run(args)
    except RuntimeError as error:
        LOGGER.error("%s", error)
        return 2
    except Exception:
        LOGGER.exception("AiMeetingNotes stopped because of an unexpected error.")
        return 1


if __name__ == "__main__":
    raise SystemExit(main())