CMVR-AI-ANALYSIS/tools/import_audio_references.py

import argparse
import csv
import hashlib
import json
import shutil
import subprocess
from collections import Counter
from pathlib import Path
from urllib.parse import urlparse

import httpx


SUPPORTED_LABELS = {
    "POWER_ON",
    "POWER_OFF",
    "ARMED",
    "DISARMED",
    "FIND_VEHICLE_HORN",
}


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


def audio_duration(path: Path) -> float:
    process = subprocess.run(
        [
            "ffprobe",
            "-v",
            "error",
            "-show_entries",
            "format=duration",
            "-of",
            "default=noprint_wrappers=1:nokey=1",
            str(path),
        ],
        capture_output=True,
        text=True,
        check=False,
    )
    if process.returncode != 0:
        detail = process.stderr.strip()[-500:]
        raise ValueError(f"ffprobe failed: {detail or process.returncode}")
    return float(process.stdout.strip())


def download(client: httpx.Client, url: str, destination: Path) -> None:
    with client.stream("GET", url) as response:
        response.raise_for_status()
        with destination.open("wb") as stream:
            for chunk in response.iter_bytes():
                stream.write(chunk)


def media_suffix(url: str) -> str:
    suffix = Path(urlparse(url).path).suffix.lower()
    return suffix if suffix and len(suffix) <= 10 else ".audio"


def existing_reference_hashes(reference_root: Path) -> dict[str, str]:
    hashes: dict[str, str] = {}
    for label_dir in sorted(reference_root.iterdir()):
        if not label_dir.is_dir():
            continue
        for path in sorted(label_dir.iterdir()):
            if path.is_file():
                hashes.setdefault(file_sha256(path), label_dir.name)
    return hashes


def import_references(
    manifest: Path,
    profile_dir: Path,
    import_dir: Path,
    minimum_duration: float,
    commit: bool,
) -> dict:
    reference_root = profile_dir / "references"
    download_dir = import_dir / "downloads"
    accepted_dir = import_dir / "accepted"
    download_dir.mkdir(parents=True, exist_ok=True)
    accepted_dir.mkdir(parents=True, exist_ok=True)

    known_hashes = existing_reference_hashes(reference_root)
    batch_hashes: dict[str, str] = {}
    audit_rows: list[dict[str, str]] = []
    counts: Counter[str] = Counter()

    with manifest.open("r", encoding="utf-8-sig", newline="") as stream:
        rows = list(csv.DictReader(stream, delimiter="\t"))

    timeout = httpx.Timeout(120, connect=10)
    with httpx.Client(timeout=timeout, follow_redirects=True) as client:
        for row in rows:
            log_id = str(row.get("log_id") or "").strip()
            label = str(row.get("trusted_label") or "").strip().upper()
            url = str(row.get("media_url") or "").strip()
            audit = dict(row)
            audit.update({"duration_seconds": "", "sha256": "", "status": ""})

            if label not in SUPPORTED_LABELS:
                audit["status"] = "SKIPPED_NO_TRUSTED_LABEL"
                counts[audit["status"]] += 1
                audit_rows.append(audit)
                continue
            if not url.startswith(("http://", "https://")):
                audit["status"] = "SKIPPED_INVALID_URL"
                counts[audit["status"]] += 1
                audit_rows.append(audit)
                continue

            downloaded = download_dir / f"{log_id}{media_suffix(url)}"
            try:
                if not downloaded.exists():
                    download(client, url, downloaded)
                duration = audio_duration(downloaded)
                digest = file_sha256(downloaded)
                audit["duration_seconds"] = f"{duration:.3f}"
                audit["sha256"] = digest
            except (httpx.HTTPError, OSError, ValueError) as exc:
                audit["status"] = f"FAILED:{type(exc).__name__}:{str(exc)[:200]}"
                counts["FAILED"] += 1
                audit_rows.append(audit)
                continue

            if duration < minimum_duration:
                audit["status"] = "SKIPPED_TOO_SHORT"
            elif digest in known_hashes:
                audit["status"] = f"SKIPPED_EXISTING_REFERENCE:{known_hashes[digest]}"
            elif digest in batch_hashes:
                audit["status"] = f"SKIPPED_DUPLICATE_IMPORT:{batch_hashes[digest]}"
            else:
                batch_hashes[digest] = label
                staged = accepted_dir / label / downloaded.name
                staged.parent.mkdir(parents=True, exist_ok=True)
                if not staged.exists():
                    shutil.copy2(downloaded, staged)
                if commit:
                    destination = reference_root / label / downloaded.name
                    destination.parent.mkdir(parents=True, exist_ok=True)
                    if not destination.exists():
                        shutil.copy2(staged, destination)
                    audit["status"] = "IMPORTED"
                else:
                    audit["status"] = "READY"

            counts[audit["status"].split(":", 1)[0]] += 1
            audit_rows.append(audit)

    audit_path = import_dir / "audit.csv"
    fieldnames = list(audit_rows[0].keys()) if audit_rows else ["status"]
    with audit_path.open("w", encoding="utf-8-sig", newline="") as stream:
        writer = csv.DictWriter(stream, fieldnames=fieldnames)
        writer.writeheader()
        writer.writerows(audit_rows)

    ready_by_label = Counter(
        row.get("trusted_label", "")
        for row in audit_rows
        if row["status"] in {"READY", "IMPORTED"}
    )
    return {
        "manifestRows": len(rows),
        "minimumDurationSeconds": minimum_duration,
        "committed": commit,
        "statusCounts": dict(sorted(counts.items())),
        "acceptedByLabel": dict(sorted(ready_by_label.items())),
        "auditPath": str(audit_path),
    }


def main() -> None:
    parser = argparse.ArgumentParser(
        description="Import trusted audio references from an exported log manifest"
    )
    parser.add_argument("--manifest", required=True, type=Path)
    parser.add_argument("--profile-dir", required=True, type=Path)
    parser.add_argument("--import-dir", required=True, type=Path)
    parser.add_argument("--minimum-duration", type=float, default=4.0)
    parser.add_argument("--commit", action="store_true")
    args = parser.parse_args()
    result = import_references(
        args.manifest,
        args.profile_dir,
        args.import_dir,
        args.minimum_duration,
        args.commit,
    )
    print(json.dumps(result, ensure_ascii=False, indent=2))


if __name__ == "__main__":
    main()