#!/usr/bin/env python3
"""Verify the fixed corpus, run Speakmac, and score a release regression.

This script intentionally uses only the Python standard library. Saved run
files contain CLI stdout and aggregate scoring data only; stderr, local paths,
database identifiers, hardware details, and app-history rows are never saved.
"""

from __future__ import annotations

import argparse
import csv
import hashlib
import json
import os
import plistlib
import re
import string
import subprocess
import sys
import wave
from pathlib import Path


ROOT = Path(__file__).resolve().parents[1]
MANIFEST_PATH = ROOT / "manifest.json"
CHECKSUMS_PATH = ROOT / "checksums.sha256"
HISTORICAL_RESULTS_PATH = ROOT / "paired-outputs.json"
RESULTS_DIR = ROOT / "results"
SCORER_ID = "speakmac-fixed-corpus-wer-v1"
EXPECTED_SAMPLE_COUNT = 27
EXPECTED_REFERENCE_WORDS = 349
EXPECTED_AUDIO_FORMAT = {
    "channels": 1,
    "sample_width_bytes": 2,
    "sample_rate_hz": 16_000,
    "compression": "NONE",
}


def sha256(path: Path) -> str:
    digest = hashlib.sha256()
    with path.open("rb") as handle:
        for chunk in iter(lambda: handle.read(1024 * 1024), b""):
            digest.update(chunk)
    return digest.hexdigest()


def resolve_bundle_path(relative: str) -> Path:
    if relative.startswith("/") or "\\" in relative:
        raise RuntimeError(f"Bundle path must be relative: {relative!r}")
    candidate = (ROOT / relative).resolve()
    if ROOT not in candidate.parents:
        raise RuntimeError(f"Bundle path escapes the corpus directory: {relative!r}")
    return candidate


def verify_checksums() -> int:
    checked = 0
    for line_number, raw_line in enumerate(
        CHECKSUMS_PATH.read_text(encoding="utf-8").splitlines(), start=1
    ):
        line = raw_line.strip()
        if not line or line.startswith("#"):
            continue
        try:
            expected, relative = line.split("  ", 1)
        except ValueError as error:
            raise RuntimeError(
                f"Invalid checksums.sha256 row at line {line_number}"
            ) from error
        path = resolve_bundle_path(relative)
        if not path.is_file():
            raise RuntimeError(f"Checksum target is missing: {relative}")
        actual = sha256(path)
        if actual != expected:
            raise RuntimeError(
                f"Checksum mismatch for {relative}: expected {expected}, got {actual}"
            )
        checked += 1
    if checked == 0:
        raise RuntimeError("checksums.sha256 contains no targets")
    return checked


def normalize_words(text: str) -> list[str]:
    lowered = text.lower().translate(str.maketrans("", "", string.punctuation))
    return re.sub(r"\s+", " ", lowered).strip().split()


def edit_distance(reference: list[str], hypothesis: list[str]) -> int:
    previous = list(range(len(hypothesis) + 1))
    for reference_index, reference_word in enumerate(reference, start=1):
        current = [reference_index]
        for hypothesis_index, hypothesis_word in enumerate(hypothesis, start=1):
            current.append(
                min(
                    current[-1] + 1,
                    previous[hypothesis_index] + 1,
                    previous[hypothesis_index - 1]
                    + (reference_word != hypothesis_word),
                )
            )
        previous = current
    return previous[-1]


def score(reference: str, hypothesis: str) -> dict[str, int | float]:
    reference_words = normalize_words(reference)
    hypothesis_words = normalize_words(hypothesis)
    errors = edit_distance(reference_words, hypothesis_words)
    return {
        "reference_words": len(reference_words),
        "errors": errors,
        "wer": round(errors / len(reference_words), 8),
        "wer_percent": round(errors / len(reference_words) * 100, 1),
    }


def validate_wave(path: Path, sample: dict) -> None:
    with wave.open(str(path), "rb") as audio:
        observed = {
            "channels": audio.getnchannels(),
            "sample_width_bytes": audio.getsampwidth(),
            "sample_rate_hz": audio.getframerate(),
            "compression": audio.getcomptype(),
        }
        duration_seconds = audio.getnframes() / audio.getframerate()
    if observed != EXPECTED_AUDIO_FORMAT:
        raise RuntimeError(
            f"Unexpected WAV format for {sample['id']}: {observed}"
        )
    if abs(duration_seconds - float(sample["duration_seconds"])) > 0.002:
        raise RuntimeError(
            f"Duration mismatch for {sample['id']}: "
            f"manifest={sample['duration_seconds']} wav={duration_seconds:.6f}"
        )


def load_and_validate_manifest() -> tuple[dict, dict[str, str]]:
    manifest = json.loads(MANIFEST_PATH.read_text(encoding="utf-8"))
    if manifest.get("schema_version") != 2:
        raise RuntimeError("manifest.json must use schema_version 2")
    samples = manifest.get("samples")
    if not isinstance(samples, list) or len(samples) != EXPECTED_SAMPLE_COUNT:
        raise RuntimeError(
            f"Expected {EXPECTED_SAMPLE_COUNT} samples, got "
            f"{len(samples) if isinstance(samples, list) else 'invalid'}"
        )

    ids: list[str] = []
    references: dict[str, str] = {}
    audio_hashes: set[str] = set()
    reference_word_total = 0
    for sample in samples:
        sample_id = sample["id"]
        ids.append(sample_id)
        audio_path = resolve_bundle_path(sample["audio"]["path"])
        reference_path = resolve_bundle_path(sample["reference"]["path"])
        if sha256(audio_path) != sample["audio"]["sha256"]:
            raise RuntimeError(f"Audio hash mismatch for {sample_id}")
        if sha256(reference_path) != sample["reference"]["sha256"]:
            raise RuntimeError(f"Reference hash mismatch for {sample_id}")
        validate_wave(audio_path, sample)
        reference = reference_path.read_text(encoding="utf-8").strip()
        reference_words = len(normalize_words(reference))
        if reference_words != sample["reference_words"]:
            raise RuntimeError(
                f"Reference word-count mismatch for {sample_id}: "
                f"manifest={sample['reference_words']} actual={reference_words}"
            )
        references[sample_id] = reference
        audio_hashes.add(sample["audio"]["sha256"])
        reference_word_total += reference_words

    if len(ids) != len(set(ids)):
        raise RuntimeError("Manifest sample IDs are not unique")
    if len(audio_hashes) != EXPECTED_SAMPLE_COUNT:
        raise RuntimeError("Manifest audio hashes are not unique")
    if reference_word_total != EXPECTED_REFERENCE_WORDS:
        raise RuntimeError(
            f"Expected {EXPECTED_REFERENCE_WORDS} reference words, got "
            f"{reference_word_total}"
        )
    return manifest, references


def load_historical_outputs(sample_ids: set[str]) -> dict[str, str]:
    rows = json.loads(HISTORICAL_RESULTS_PATH.read_text(encoding="utf-8"))
    outputs = {
        row["id"]: row["speakmac_final_text"]
        for row in rows
        if row["id"] in sample_ids
    }
    if set(outputs) != sample_ids:
        raise RuntimeError("Historical Speakmac result IDs do not match manifest")
    return outputs


def aggregate(
    samples: list[dict], references: dict[str, str], outputs: dict[str, str]
) -> dict[str, int | float]:
    reference_words = 0
    errors = 0
    for sample in samples:
        result = score(references[sample["id"]], outputs[sample["id"]])
        reference_words += int(result["reference_words"])
        errors += int(result["errors"])
    return {
        "samples": len(samples),
        "reference_words": reference_words,
        "errors": errors,
        "wer": round(errors / reference_words, 8),
        "wer_percent": round(errors / reference_words * 100, 1),
    }


def segment_samples(samples: list[dict]) -> dict[str, list[dict]]:
    return {
        "overall": samples,
        "real_human": [
            sample
            for sample in samples
            if sample["speech_kind"].startswith("real_human")
        ],
        "svarah": [sample for sample in samples if sample["dataset"] == "Svarah"],
        "librispeech": [
            sample for sample in samples if sample["dataset"] == "LibriSpeech"
        ],
    }


def app_identity(cli: Path) -> tuple[str, str]:
    info_path = cli.parent.parent / "Info.plist"
    if not info_path.is_file():
        raise RuntimeError(f"Could not locate the app Info.plist for CLI: {cli}")
    with info_path.open("rb") as handle:
        info = plistlib.load(handle)
    return str(info["CFBundleShortVersionString"]), str(info["CFBundleVersion"])


def read_speakmac_default(key: str) -> str | None:
    completed = subprocess.run(
        ["defaults", "read", "com.kiran.speakmac", key],
        capture_output=True,
        text=True,
        check=False,
        timeout=10,
    )
    if completed.returncode != 0:
        return None
    return completed.stdout.strip()


def validate_runtime_configuration(expected_provider: str) -> dict[str, str]:
    persisted_provider = read_speakmac_default("transcriptionProvider")
    provider = persisted_provider or "fluid_audio"
    if provider != expected_provider:
        raise RuntimeError(
            f"Expected transcription provider {expected_provider!r}; found {provider!r}"
        )
    if read_speakmac_default("gemmaTextIntelligenceModelDownloaded") != "1":
        raise RuntimeError(
            "Clean Up Dictation is not ready; set it up in Speakmac before running"
        )
    return {
        "recognition": (
            "Parakeet (FluidAudio shipping default)"
            if persisted_provider is None
            else "Parakeet (FluidAudio persisted selection)"
        ),
        "cleanup": "on-device Clean Up Dictation (Gemma 4 E2B)",
    }


def validate_output_stem(value: str) -> str:
    if not re.fullmatch(r"[A-Za-z0-9._-]+", value):
        raise argparse.ArgumentTypeError(
            "output stem may contain only letters, numbers, dot, underscore, and dash"
        )
    return value


def run_speakmac(cli: Path, manifest: dict) -> dict[str, str]:
    doctor = subprocess.run(
        [str(cli), "doctor"],
        stdout=subprocess.DEVNULL,
        stderr=subprocess.DEVNULL,
        check=False,
        timeout=30,
    )
    if doctor.returncode != 0:
        raise RuntimeError(f"Speakmac doctor failed with exit code {doctor.returncode}")

    outputs: dict[str, str] = {}
    environment = os.environ.copy()
    environment.setdefault("SPEAKMAC_CLI_TIMEOUT_SECONDS", "7200")
    sample_count = len(manifest["samples"])
    for index, sample in enumerate(manifest["samples"], start=1):
        audio_path = resolve_bundle_path(sample["audio"]["path"])
        print(
            f"[{index:02d}/{sample_count:02d}] {sample['id']}",
            file=sys.stderr,
            flush=True,
        )
        completed = subprocess.run(
            [
                str(cli),
                "transcribe",
                str(audio_path),
                "--stdout",
                "--clean",
                "--no-history",
            ],
            capture_output=True,
            text=True,
            check=False,
            timeout=7300,
            env=environment,
        )
        if completed.returncode != 0:
            raise RuntimeError(
                f"Speakmac failed on {sample['id']} with exit code "
                f"{completed.returncode}; stderr was not saved"
            )
        outputs[sample["id"]] = completed.stdout.strip()
    return outputs


def save_outputs(
    output_stem: str,
    app_version: str,
    app_build: str,
    manifest: dict,
    references: dict[str, str],
    outputs: dict[str, str],
    runtime_configuration: dict[str, str],
) -> tuple[Path, Path, Path]:
    RESULTS_DIR.mkdir(parents=True, exist_ok=True)
    output_path = RESULTS_DIR / f"{output_stem}.json"
    summary_path = RESULTS_DIR / f"{output_stem}-regression-summary.json"
    per_sample_path = RESULTS_DIR / f"{output_stem}-per-sample.csv"
    for path in (output_path, summary_path, per_sample_path):
        if path.exists():
            raise RuntimeError(f"Refusing to overwrite existing result: {path.name}")

    sample_ids = {sample["id"] for sample in manifest["samples"]}
    historical_outputs = load_historical_outputs(sample_ids)
    run_payload = {
        "schema_version": 1,
        "benchmark": manifest["benchmark"],
        "corpus_version": manifest["corpus_version"],
        "application": {
            "name": "Speakmac",
            "version": app_version,
            "build": app_build,
        },
        "configuration": {
            **runtime_configuration,
            "cli_flags": ["--stdout", "--clean", "--no-history"],
        },
        "results": [
            {"id": sample["id"], "stdout": outputs[sample["id"]]}
            for sample in manifest["samples"]
        ],
    }

    segments = {}
    for segment_name, selected in segment_samples(manifest["samples"]).items():
        baseline = aggregate(selected, references, historical_outputs)
        current = aggregate(selected, references, outputs)
        segments[segment_name] = {
            "baseline_speakmac_5_2_1_build_55": baseline,
            "current": current,
            "current_minus_baseline_wer_percentage_points": round(
                float(current["wer"]) * 100 - float(baseline["wer"]) * 100,
                2,
            ),
        }

    historical_overall = segments["overall"]["baseline_speakmac_5_2_1_build_55"]
    historical_real = segments["real_human"]["baseline_speakmac_5_2_1_build_55"]
    if historical_overall["errors"] != 22 or historical_real["errors"] != 22:
        raise RuntimeError(
            "Pinned scorer does not reproduce the published historical baseline"
        )

    empty_outputs = [sample_id for sample_id, text in outputs.items() if not text]
    per_sample_error_deltas = []
    for sample in manifest["samples"]:
        sample_id = sample["id"]
        baseline = score(references[sample_id], historical_outputs[sample_id])
        current = score(references[sample_id], outputs[sample_id])
        per_sample_error_deltas.append(
            int(current["errors"]) - int(baseline["errors"])
        )
    summary_payload = {
        "schema_version": 1,
        "benchmark": manifest["benchmark"],
        "corpus_version": manifest["corpus_version"],
        "scorer": {
            "id": SCORER_ID,
            "normalization": "lowercase, remove ASCII punctuation, collapse whitespace",
            "metric": "micro-averaged word error rate using Levenshtein distance",
        },
        "baseline": {
            "application": "Speakmac 5.2.1 build 55",
            "source": "paired-outputs.json",
        },
        "current": {
            "application": f"Speakmac {app_version} build {app_build}",
            "source": output_path.name,
        },
        "integrity": {
            "sample_count": len(outputs),
            "unique_audio_hashes": len(
                {sample["audio"]["sha256"] for sample in manifest["samples"]}
            ),
            "empty_stdout_outputs": len(empty_outputs),
            "empty_stdout_sample_ids": empty_outputs,
            "saved_stderr": False,
            "saved_local_paths": False,
        },
        "paired_change": {
            "samples_with_changed_error_count": sum(
                delta != 0 for delta in per_sample_error_deltas
            ),
            "samples_improved": sum(delta < 0 for delta in per_sample_error_deltas),
            "samples_regressed": sum(delta > 0 for delta in per_sample_error_deltas),
            "samples_unchanged": sum(delta == 0 for delta in per_sample_error_deltas),
            "net_error_change": sum(per_sample_error_deltas),
        },
        "segments": segments,
        "interpretation_limit": (
            "This is a same-corpus Speakmac release regression, not a fresh "
            "Wispr Flow comparison and not evidence of a universal accuracy winner."
        ),
    }

    rows = []
    for sample in manifest["samples"]:
        sample_id = sample["id"]
        baseline = score(references[sample_id], historical_outputs[sample_id])
        current = score(references[sample_id], outputs[sample_id])
        rows.append(
            {
                "id": sample_id,
                "dataset": sample["dataset"],
                "reference_words": current["reference_words"],
                "baseline_errors": baseline["errors"],
                "baseline_wer_percent": baseline["wer_percent"],
                "current_errors": current["errors"],
                "current_wer_percent": current["wer_percent"],
            }
        )

    output_path.write_text(json.dumps(run_payload, indent=2) + "\n", encoding="utf-8")
    summary_path.write_text(
        json.dumps(summary_payload, indent=2) + "\n", encoding="utf-8"
    )
    with per_sample_path.open("w", newline="", encoding="utf-8") as handle:
        writer = csv.DictWriter(handle, fieldnames=list(rows[0]), lineterminator="\n")
        writer.writeheader()
        writer.writerows(rows)
    return output_path, summary_path, per_sample_path


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument(
        "--cli",
        type=Path,
        default=Path("/Applications/Speakmac.app/Contents/Helpers/speakmac"),
        help="Path to the Speakmac CLI helper",
    )
    parser.add_argument("--expected-version", default="5.5.0")
    parser.add_argument("--expected-build", default="61")
    parser.add_argument("--expected-provider", default="fluid_audio")
    parser.add_argument(
        "--output-stem",
        type=validate_output_stem,
        default="local-speakmac-5.5.0-build-61",
    )
    parser.add_argument(
        "--verify-only",
        action="store_true",
        help="Verify checksums, corpus schema, WAV format, and historical scorer only",
    )
    return parser.parse_args()


def main() -> None:
    args = parse_args()
    checked_files = verify_checksums()
    manifest, references = load_and_validate_manifest()
    historical_outputs = load_historical_outputs(
        {sample["id"] for sample in manifest["samples"]}
    )
    historical_overall = aggregate(manifest["samples"], references, historical_outputs)
    if historical_overall["errors"] != 22:
        raise RuntimeError("Historical scorer verification failed")
    print(
        f"Verified {checked_files} files, {len(manifest['samples'])} unique WAVs, "
        f"and the {historical_overall['wer_percent']}% historical WER baseline.",
        file=sys.stderr,
    )
    if args.verify_only:
        return

    cli = args.cli.resolve()
    if not cli.is_file() or not os.access(cli, os.X_OK):
        raise RuntimeError(f"Speakmac CLI is missing or not executable: {cli}")
    version, build = app_identity(cli)
    if version != args.expected_version or build != args.expected_build:
        raise RuntimeError(
            f"Expected Speakmac {args.expected_version} build {args.expected_build}; "
            f"found {version} build {build}"
        )
    runtime_configuration = validate_runtime_configuration(args.expected_provider)
    outputs = run_speakmac(cli, manifest)
    paths = save_outputs(
        args.output_stem,
        version,
        build,
        manifest,
        references,
        outputs,
        runtime_configuration,
    )
    print("Saved:", file=sys.stderr)
    for path in paths:
        print(f"- {path.relative_to(ROOT)}", file=sys.stderr)


if __name__ == "__main__":
    main()