#!/usr/bin/env python3
"""Explicit REAL STT smoke using only public/synthetic, repo-external WAV audio.

Sends each fixture once to the configured provider (twice with --compare-prompt).
Never saves audio, changes configuration, contacts an Agent or sends device commands.
"""
import argparse
import asyncio
import io
import json
import os
from pathlib import Path
import re
import shlex
import sys
import time
import unicodedata
import wave

REPO = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(REPO / 'server'))
from device_gateway.voice import MAX_BYTES, MAX_SECONDS, SAMPLE_RATE, configured_stt

TECHNICAL_PROMPT = 'Codex, Pi, session, model, branch, commit, pytest, Gateway, approval, GitHub, Docker, Cloudflare, ESP32'
ENV_NAMES = {'GATEWAY_STT_PROVIDER', 'GATEWAY_STT_BASE_URL', 'GATEWAY_STT_API_KEY',
             'GATEWAY_STT_MODEL', 'GATEWAY_CF_ACCOUNT_ID', 'GATEWAY_CF_API_TOKEN',
             'GROQ_API_KEY', 'GATEWAY_STT_INITIAL_PROMPT'}


def load_env(path):
    path = path.expanduser().resolve()
    if path.is_relative_to(REPO) or path.stat().st_mode & 0o777 != 0o600:
        raise ValueError('env file must be repo-external and mode 0600')
    # Match the supplied file, including absence of keys; no shell execution/expansion.
    values = {}
    for line in path.read_text().splitlines():
        line = line.strip()
        if not line or line.startswith('#') or '=' not in line:
            continue
        key, value = line.split('=', 1)
        if key not in ENV_NAMES:
            continue
        fields = shlex.split(value)
        if len(fields) > 1:
            raise ValueError('quote env values containing spaces')
        values[key] = fields[0] if fields else ''
    for key in ENV_NAMES:
        os.environ.pop(key, None)
    os.environ.update(values)


def read_audio(path):
    path = path.expanduser().resolve()
    if path.is_relative_to(REPO):
        raise ValueError('use a repo-external public/synthetic WAV')
    with path.open('rb') as source:
        audio = source.read(MAX_BYTES + 4097)
    if len(audio) > MAX_BYTES + 4096:
        raise ValueError('audio exceeds the bounded WAV size')
    with wave.open(io.BytesIO(audio), 'rb') as stream:
        if (stream.getframerate(), stream.getsampwidth(), stream.getnchannels(), stream.getcomptype()) != (SAMPLE_RATE, 2, 1, 'NONE'):
            raise ValueError('expected 44.1 kHz / 16-bit / mono PCM WAV; no resampling')
        frames = stream.getnframes()
        duration = frames / SAMPLE_RATE
        if not 0 < duration <= MAX_SECONDS or len(stream.readframes(frames)) != frames * 2:
            raise ValueError('expected complete WAV containing 0 < duration <= 20 seconds')
    return audio, duration


def normalize(text):
    return ''.join(c for c in unicodedata.normalize('NFKC', text).casefold() if c.isalnum())


def term_matches(term, text):
    # Allow punctuation/space/case differences but not "session" inside "sessions".
    if re.fullmatch(r'[\x00-\x7f]+', term):
        words = re.findall(r'[a-z0-9]+', unicodedata.normalize('NFKC', term).casefold())
        if not words:
            return False
        pattern = r'(?<![a-z0-9])' + r'[\W_]*'.join(map(re.escape, words)) + r'(?![a-z0-9])'
        return bool(re.search(pattern, unicodedata.normalize('NFKC', text).casefold()))
    return normalize(term) in normalize(text)


async def main():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument('--env-file', type=Path, help='repo-external 0600 gateway.env; values never printed')
    parser.add_argument('--wav', type=Path, help='public/synthetic audio only, never a private recording')
    parser.add_argument('--expected', help='public test phrase, printed for manual semantic review')
    parser.add_argument('--key-term', action='append', default=[], help='required concept; repeat for each term')
    parser.add_argument('--case', choices=['chinese', 'english', 'mixed', 'technical'], default='mixed')
    parser.add_argument('--compare-prompt', action='store_true', help='explicit A/B: same provider, no prompt vs short technical prompt')
    args = parser.parse_args()
    if args.env_file:
        try:
            load_env(args.env_file)
        except (OSError, ValueError):
            parser.error('cannot load repo-external 0600 env file (use simple KEY=value lines)')
    provider = configured_stt()
    if provider is None:
        print('REAL STT SMOKE: SKIPPED (no configured authorized provider)')
        return 2
    if not args.wav or not args.expected or not args.key_term or any(not normalize(t) for t in args.key_term):
        parser.error('requires --wav, --expected and at least one nonempty --key-term')
    try:
        audio, duration = read_audio(args.wav)
    except (OSError, ValueError, wave.Error, EOFError):
        parser.error('requires repo-external complete 44.1 kHz / 16-bit / mono WAV, <= 20 seconds')
    variants = [('configured', provider.initial_prompt)]
    if args.compare_prompt:
        variants = [('without-prompt', None), ('with-prompt', TECHNICAL_PROMPT)]
    reports = []
    for variant, prompt in variants:
        provider.initial_prompt = prompt
        started = time.monotonic()
        try:
            actual = await provider.transcribe(audio)
        except Exception:
            print(json.dumps({'case': args.case, 'variant': variant, 'expected': args.expected,
                'result': 'FAILED', 'error': 'Transcription failed', 'audio_duration_s': duration,
                'upload_request_duration_s': provider.last_request_seconds,
                'stt_latency_s': round(time.monotonic() - started, 3)}, ensure_ascii=False))
            # Includes 401/403/429: stop, without retrying or running the second variant.
            return 1
        matches = {term: term_matches(term, actual) for term in args.key_term}
        report = {'case': args.case, 'variant': variant, 'provider': provider.provider_name,
            'model': provider.model, 'expected': args.expected, 'actual': actual,
            'key_term_match': matches, 'audio_duration_s': round(duration, 3),
            'upload_request_duration_s': round(provider.last_request_seconds, 3),
            'stt_latency_s': round(time.monotonic() - started, 3),
            'result': 'PASS' if all(matches.values()) else 'REVIEW'}
        reports.append(report)
        print(json.dumps(report, ensure_ascii=False))
    if args.compare_prompt:
        counts = [sum(r['key_term_match'].values()) for r in reports]
        print(json.dumps({'prompt_comparison': {'without': counts[0], 'with': counts[1]},
            'prompt_enabled_automatically': False}))
    # Neither keyword matches nor a supplied expected phrase prove full semantic fidelity.
    print('Review expected/actual for meaning. HTTP duration includes upload and inference; server-only timing is unavailable.')
    return 0 if all(r['result'] == 'PASS' for r in reports) else 1


if __name__ == '__main__':
    raise SystemExit(asyncio.run(main()))