"""Keep Whisper's transcript and add Oruk's timed emotion estimates.

Python 3.11 and ffmpeg. Install the pinned Whisper and current Oruk SDK versions
from https://oruk.ai/whisper-emotion-detection before running this script.
Live: python whisper-emotions.py recording.wav --output combined.json
Saved example: add --oruk-response response.json --manifest sample-manifest.json

Whisper runs locally. Live mode sends the same original audio to Oruk's
metered emotions endpoint. Saved mode makes no Oruk request and verifies the
audio/response pair against the supplied manifest. Model weights may download
on the first Whisper run. Code: MIT; samples have separate attribution.
"""

from __future__ import annotations

import argparse
import hashlib
import io
import json
import math
import os
import sys
import tempfile
from bisect import bisect_left, bisect_right
from datetime import datetime, timezone
from pathlib import Path
from urllib.parse import urlsplit


def finite_time(value: object, field: str) -> float:
    if isinstance(value, bool) or not isinstance(value, (int, float)):
        raise TypeError(f"{field} must be a number")
    if not math.isfinite(value) or value < 0:
        raise ValueError(f"{field} must be finite and nonnegative")
    return float(value)


def validate_emotions(value: object) -> None:
    if not isinstance(value, list):
        raise TypeError("emotions must be an array")
    for item in value:
        if not isinstance(item, dict) or not isinstance(item.get("label"), str):
            raise TypeError("each emotion needs a label and score")
        if finite_time(item.get("score"), "emotion.score") > 1:
            raise ValueError("emotion.score must be between 0 and 1")


def timed_segments(value: object, provider: str) -> list[dict]:
    if not isinstance(value, list) or len(value) > 20_000:
        raise ValueError(
            f"{provider}.segments must be an array of at most 20,000 items"
        )
    for segment in value:
        if not isinstance(segment, dict):
            raise TypeError(f"{provider} segment must be an object")
        start = finite_time(segment.get("start"), f"{provider}.start")
        end = finite_time(segment.get("end"), f"{provider}.end")
        if end < start:
            raise ValueError(f"{provider} segment ends before it starts")
        if provider == "Oruk":
            validate_emotions(segment.get("emotions"))
        elif not isinstance(segment.get("text"), str):
            raise ValueError("Whisper segment is missing text")
    return value


def validate_oruk(result: object) -> dict:
    if not isinstance(result, dict) or result.get("object") != "speech.result":
        raise ValueError("expected an Oruk speech.result response")
    if result.get("task") not in {"emotions", "analysis", "affect"}:
        raise ValueError("Oruk response must include emotion analysis")
    if not isinstance(result.get("model"), str):
        raise TypeError("Oruk response is missing its model")
    finite_time(result.get("duration"), "Oruk duration")
    validate_emotions(result.get("emotions"))
    timed_segments(result.get("segments"), "Oruk")
    return result


def combine(whisper_result: object, oruk_result: object) -> dict:
    """Associate overlapping intervals without changing either provider's output."""
    if not isinstance(whisper_result, dict) or not isinstance(
        whisper_result.get("text"), str
    ):
        raise TypeError("Whisper response is missing its transcript")
    transcription = timed_segments(whisper_result.get("segments"), "Whisper")
    emotion = validate_oruk(oruk_result)
    ordered = sorted(enumerate(emotion["segments"]), key=lambda pair: pair[1]["start"])
    starts = [segment["start"] for _, segment in ordered]
    latest_end = []
    for _, segment in ordered:
        latest_end.append(max(latest_end[-1] if latest_end else 0, segment["end"]))
    associations = []
    link_count = 0
    for index, words in enumerate(transcription):
        matches = []
        first = bisect_right(latest_end, words["start"])
        last = bisect_left(starts, words["end"])
        for emotion_index, segment in ordered[first:last]:
            overlap = min(words["end"], segment["end"]) - max(
                words["start"], segment["start"]
            )
            if overlap > 0:
                link_count += 1
                if link_count > 100_000:
                    raise ValueError(
                        "too many overlapping intervals for this example (100,000 links)"
                    )
                matches.append(
                    {
                        "oruk_segment_index": emotion_index,
                        "overlap_seconds": round(overlap, 6),
                    }
                )
        matches.sort(key=lambda match: match["oruk_segment_index"])
        associations.append({"whisper_segment_index": index, "oruk_intervals": matches})
    return {
        "whisper": whisper_result,
        "oruk": emotion,
        "alignment": {
            "method": "positive interval overlap; no averaging or word-level emotion attribution",
            "segments": associations,
        },
    }


def read_json(path: Path) -> tuple[object, str]:
    if path.stat().st_size > 32 * 1024 * 1024:
        raise ValueError("JSON file exceeds this example's 32 MiB limit")
    data = path.read_bytes()
    return json.loads(data), hashlib.sha256(data).hexdigest()


def saved_response(
    audio_hash: str, response_path: Path, manifest_path: Path
) -> tuple[dict, str]:
    manifest, _ = read_json(manifest_path)
    result, response_hash = read_json(response_path)
    if not isinstance(manifest, dict) or manifest.get("schema_version") != 1:
        raise ValueError("unsupported sample manifest")
    audio_entry, response_entry = manifest.get("audio"), manifest.get("oruk_response")
    if not isinstance(audio_entry, dict) or not isinstance(response_entry, dict):
        raise TypeError("sample manifest needs audio and oruk_response records")
    if (
        audio_entry.get("sha256") != audio_hash
        or response_entry.get("sha256") != response_hash
    ):
        raise ValueError("audio/response hashes do not match the sample manifest")
    return validate_oruk(result), response_hash


def api_base_url() -> str:
    base = os.environ.get("ORUK_API_BASE_URL", "https://speech-api.oruk.ai").rstrip("/")
    parsed = urlsplit(base)
    local = parsed.hostname in {"127.0.0.1", "localhost", "::1"}
    if (
        parsed.scheme != "https" and not (parsed.scheme == "http" and local)
    ) or not parsed.hostname:
        raise ValueError(
            "ORUK_API_BASE_URL requires HTTPS; HTTP is allowed only for localhost fixtures"
        )
    if (
        parsed.username
        or parsed.password
        or parsed.query
        or parsed.fragment
        or parsed.path
    ):
        raise ValueError(
            "ORUK_API_BASE_URL must be an origin without credentials, a path, or query"
        )
    return base


def main() -> int:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("audio", type=Path)
    parser.add_argument(
        "--output",
        required=True,
        type=Path,
        help="new JSON file; existing files are preserved",
    )
    parser.add_argument(
        "--whisper-model",
        default="tiny.en",
        help="local Whisper model, default tiny.en",
    )
    parser.add_argument("--threads", type=int, default=4, help="CPU threads, 1–16")
    parser.add_argument(
        "--oruk-response",
        type=Path,
        help="use a saved Oruk response instead of an API request",
    )
    parser.add_argument(
        "--manifest",
        type=Path,
        help="hash-bound recording/response manifest for saved mode",
    )
    args = parser.parse_args()
    if bool(args.oruk_response) != bool(args.manifest):
        parser.error("--oruk-response and --manifest must be supplied together")
    if not 1 <= args.threads <= 16:
        parser.error("--threads must be from 1 to 16")
    if args.output.exists() or not args.output.parent.is_dir():
        parser.error("--output must be a new file in an existing directory")
    key = os.environ.get("ORUK_API_KEY")
    if not args.oruk_response and not key:
        parser.error(
            "set ORUK_API_KEY for live mode, or provide a saved response and manifest"
        )

    try:
        if (
            not args.audio.is_file()
            or not 0 < args.audio.stat().st_size <= 30 * 1024 * 1024
        ):
            raise ValueError("audio must be a nonempty local file up to 30 MiB")
        with args.audio.open("rb") as audio_file:
            original = audio_file.read(30 * 1024 * 1024 + 1)
        if not 0 < len(original) <= 30 * 1024 * 1024:
            raise ValueError("audio must be a nonempty local file up to 30 MiB")
        audio_hash = hashlib.sha256(original).hexdigest()
        cached = (
            saved_response(audio_hash, args.oruk_response, args.manifest)
            if args.oruk_response
            else None
        )
        base_url = api_base_url() if not cached else None

        import torch
        import whisper

        torch.set_num_threads(args.threads)
        if args.whisper_model not in whisper.available_models():
            raise ValueError("choose a model name from whisper.available_models()")
        # Both providers receive the same immutable original bytes. Whisper's
        # decoder uses its own 16 kHz mono representation for local inference.
        with tempfile.TemporaryDirectory(prefix="oruk-whisper-") as temporary:
            snapshot = Path(temporary) / ("recording" + args.audio.suffix)
            snapshot.write_bytes(original)
            waveform = whisper.load_audio(str(snapshot))
        duration = len(waveform) / whisper.audio.SAMPLE_RATE
        if not 0 < duration <= 3600:
            raise ValueError(
                "decoded audio must be nonempty and no longer than 60 minutes"
            )
        model = whisper.load_model(args.whisper_model, device="cpu")
        words = model.transcribe(
            waveform, language="en", fp16=False, temperature=0, verbose=None
        )
        if cached:
            emotion, response_hash = cached
            source = {"kind": "saved_response", "sha256": response_hash}
        else:
            from oruk import Oruk

            if not key or not base_url:
                raise ValueError("live mode requires an API key and API origin")
            audio = io.BytesIO(original)
            audio.name = args.audio.name
            # A failed request is surfaced. This example never automatically
            # replays audio after a response timeout or server failure.
            with Oruk(api_key=key, base_url=base_url, max_retries=0) as client:
                emotion = client.emotions(audio, model="oruk-resonance")
            source = {"kind": "live_request"}
        if abs(finite_time(emotion.get("duration"), "Oruk duration") - duration) > 0.1:
            raise ValueError(
                "Oruk's reported duration differs from the decoded recording"
            )
        result = combine(words, emotion)
        result.update(
            {
                "schema_version": 1,
                "created_at": datetime.now(timezone.utc).isoformat(),
                "audio": {
                    "file": args.audio.name,
                    "sha256": audio_hash,
                    "decoded_seconds": duration,
                },
                "whisper_model": args.whisper_model,
                "oruk_source": source,
            }
        )
        encoded = (
            json.dumps(result, ensure_ascii=False, indent=2, allow_nan=False) + "\n"
        )
        with args.output.open("x", encoding="utf-8") as output:
            output.write(encoded)
        print(
            f"Saved {args.output.name}: {len(words['segments'])} Whisper intervals; "
            f"{len(emotion['segments'])} Oruk intervals; {source['kind']}."
        )
        return 0
    except Exception as error:  # noqa: BLE001 — redact third-party exceptions at the CLI boundary.
        # Do not print arbitrary HTTP bodies, request headers, or exception text
        # from remote services. Return a useful status without exposing a key.
        status = getattr(error, "status", None)
        if isinstance(status, int):
            print(
                f"Oruk returned HTTP {status}; no combined result was written. Check your key, "
                "plan, input, and the API status before retrying.",
                file=sys.stderr,
            )
        elif isinstance(error, (ValueError, TypeError, OSError, ImportError)):
            print(f"Cannot complete the example: {error}", file=sys.stderr)
        else:
            print(
                f"Example failed ({type(error).__name__}); check the installed packages, "
                "ffmpeg, model download, and API connectivity.",
                file=sys.stderr,
            )
        return 1


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