#!/usr/bin/env python3
"""Test-only replacements for op/curl; no network or real credentials."""

import base64
import json
import os
from pathlib import Path
import subprocess
import sys
import time

log_path = Path(os.environ["TRANSCRIBE_TEST_LOG"])
if Path(sys.argv[0]).name == "op":
    with log_path.with_suffix(".credentials.jsonl").open("a") as log:
        log.write(json.dumps({"args": sys.argv[1:]}) + "\n")
    if os.environ.get("TRANSCRIBE_TEST_OP_FAIL"):
        sys.exit(1)
    print("test-op-api-key")
    sys.exit(0)

args = sys.argv[1:]
output = Path(args[args.index("-o") + 1])
url = next(value for value in args if value.startswith("https://"))
headers = {}
for position, value in enumerate(args):
    if value in ("-H", "--header"):
        value = args[position + 1]
        header_lines = Path(value[1:]).read_text().splitlines() if value.startswith("@") else [value]
        for header_line in header_lines:
            key, _, header_value = header_line.partition(":")
            headers[key.lower()] = header_value.strip()
request = None
request_path = None
if "--data-binary" in args:
    request_path = Path(args[args.index("--data-binary") + 1][1:])
    request = json.loads(request_path.read_text())
mode = os.environ.get("TRANSCRIBE_TEST_MODE", "ok")
events_path = log_path.with_suffix(".events.jsonl")
is_pyannote = "openrouter.ai" not in url
index = -1 if is_pyannote else int(output.stem.split("-")[1])
staging = next(parent for parent in (output.parent, *output.parents)
               if (parent / "chunks.json").is_file())


def event(kind, **extra):
    record = dict(event=kind, index=index, pid=os.getpid(), group=os.getpgrp(),
                  time=time.monotonic(), staging=str(staging), **extra)
    with events_path.open("a") as log:
        log.write(json.dumps(record) + "\n")


def reply(value, status=0, http_status=200):
    if "--write-out" in args:
        print(str(http_status), end="")
    output.write_text(json.dumps(value))
    event("finish")
    sys.exit(status)


event("start")
if is_pyannote:
    pyannote_log = log_path.with_suffix(".pyannote.jsonl")
    previous = [json.loads(line) for line in pyannote_log.read_text().splitlines()] if pyannote_log.exists() else []
    record = {"url": url, "request": request, "headers": headers, "staging": str(staging)}
    if url == "https://api.pyannote.ai/v1/media/input":
        stage = "media"
        assert request["url"].startswith("media://") and request["url"].endswith(".flac")
        response = {"url": "https://storage.test.invalid/upload?signature=fixture-signed-url"}
    elif url.startswith("https://storage.test.invalid/"):
        stage = "upload"
        assert not headers.get("authorization"), "API bearer leaked to signed storage URL"
        upload = Path(args[args.index("--upload-file") + 1])
        info = json.loads(subprocess.check_output([
            "ffprobe", "-v", "error", "-show_entries",
            "format=duration:stream=codec_name,sample_rate,channels", "-of", "json", str(upload),
        ], text=True))
        assert info["streams"][0] == {"codec_name": "flac", "sample_rate": "16000", "channels": 1}
        record["upload_duration"] = float(info["format"]["duration"])
        record["stream"] = info["streams"][0]
        response = {}
    elif url == "https://api.pyannote.ai/v1/diarize":
        stage = "submit"
        assert request["model"] in ("precision-2", "community-1")
        assert request["url"] == next(item["request"]["url"] for item in previous if item["stage"] == "media")
        assert request["exclusive"] is True
        assert request["transcription"] is False
        response = {"jobId": "test-job-42"}
    elif url == "https://api.pyannote.ai/v1/jobs/test-job-42":
        stage = "poll"
        duration = next(item["upload_duration"] for item in previous if item["stage"] == "upload")
        manifest = json.loads((staging / "chunks.json").read_text())
        if mode == "silence":
            turns = []
        elif mode.startswith("reconcile"):
            late_start = round(manifest[0]["end"]) if len(manifest) > 1 else 60
            turns = [(0, min(duration, 10), "voice_a"), (10, min(duration, 16), "voice_b"),
                     (late_start, min(duration, late_start + 3), "voice_a"),
                     (late_start + 4, duration, "voice_b")]
        elif mode in ("speakers", "partial-speakers"):
            boundary = 1.45 if len(manifest) == 1 else 57.3
            turns = [(0, boundary, "voice_a"), (boundary, duration, "voice_b")]
        else:
            turns = [(0, duration, "voice_a")]
        timeline = [{"start": start, "end": end, "speaker": speaker}
                    for start, end, speaker in turns if start < end]
        # Standard diarization intentionally disagrees: clients must select the
        # exclusive timeline requested for word-level speaker attribution.
        response = {"jobId": "test-job-42", "status": "succeeded", "output": {
            "exclusiveDiarization": timeline,
            "diarization": [{"start": 0, "end": duration, "speaker": "wrong_timeline"}],
        }}
        if mode == "reconcile-diagnostics-metadata":
            response["request"] = {"url": "media://fixture-recording.flac", "key": "test-pyannote-key"}
            response["output"]["debug"] = "https://storage.test.invalid/?signature=fixture-signed-url"
            response["output"]["exclusiveDiarization"][0]["request"] = "test-openrouter-key"
            response["output"]["exclusiveDiarization"][0]["confidence"] = {"voice_a": 99.0}
            response["output"]["diarization"][0]["debug"] = "fixture-signed-url"
        poll_count = sum(item["stage"] == "poll" for item in previous)
        if mode == "pyannote-lifecycle" and poll_count < 2:
            response = {"jobId": "test-job-42", "status": ("created", "running")[poll_count]}
        elif mode in ("pyannote-failed", "pyannote-canceled"):
            response = {"jobId": "test-job-42", "status": mode.removeprefix("pyannote-"),
                        "error": "simulated diarization failure"}
        elif mode == "pyannote-missing-timeline":
            response["output"].pop("exclusiveDiarization")
        elif mode == "pyannote-malformed-timeline":
            response["output"]["exclusiveDiarization"][0]["end"] = "not a timestamp"
        elif mode == "pyannote-empty-timeline":
            response["output"]["exclusiveDiarization"] = []
        elif mode == "pyannote-wrong-job":
            response["jobId"] = "different-job"
        elif mode == "pyannote-output-error":
            response["output"]["error"] = "simulated incomplete diarization output"
        elif mode == "pyannote-partial-timeline":
            response["output"]["exclusiveDiarization"] = [{"start": 0, "end": 2, "speaker": "voice_a"}]
    else:
        raise AssertionError(f"Unexpected curl URL: {url}")
    assert args[args.index("--request") + 1] == {
        "media": "POST", "upload": "PUT", "submit": "POST", "poll": "GET",
    }[stage]
    if stage != "upload":
        assert headers.get("authorization") in ("Bearer test-pyannote-key", "Bearer test-op-api-key")
    record["stage"] = stage
    record["response"] = response
    with pyannote_log.open("a") as log:
        log.write(json.dumps(record) + "\n")
    if mode == f"pyannote-{stage}-http-error":
        reply({"error": f"simulated {stage} failure"}, http_status=401)
    if stage == "poll" and mode == "pyannote-transient-poll" and poll_count == 0:
        reply({"error": "simulated transient poll failure"}, status=22)
    if stage == "upload" and mode == "pyannote-transport-error":
        reply({"error": "simulated transport failure"}, status=22)
    if stage == "poll" and mode == "pyannote-bad-json":
        if "--write-out" in args:
            print("200", end="")
        output.write_text("not json")
        event("finish")
        sys.exit(0)
    if stage == "poll" and mode == "pyannote-blocked":
        child = subprocess.Popen([sys.executable, "-c", "import time; time.sleep(30)"])
        event("child", child_pid=child.pid)
        time.sleep(30)
    reply(response)

assert url == "https://openrouter.ai/api/v1/audio/transcriptions"
assert headers.get("authorization") in ("Bearer test-openrouter-key", "Bearer test-op-api-key")
manifest = json.loads((output.parent / "chunks.json").read_text())
chunk = manifest[index]
if mode in ("parallel-error", "blocked") and index == 0:
    child = subprocess.Popen([sys.executable, "-c", "import time; time.sleep(30)"])
    event("child", child_pid=child.pid)
assert request["model"] == "microsoft/mai-transcribe-2"
assert request["input_audio"]["format"] == "mp3"
if len(manifest) > 1:
    assert request["response_format"] == "verbose_json"
    assert request["timestamp_granularities"] == ["segment", "word"]

audio = output.parent / f"uploaded-{index}.mp3"
audio.write_bytes(base64.b64decode(request["input_audio"]["data"], validate=True))
info = json.loads(subprocess.check_output([
    "ffprobe", "-v", "error", "-show_entries",
    "format=duration:stream=codec_name,sample_rate,channels", "-of", "json", str(audio),
], text=True))
assert info["streams"][0] == {"codec_name": "mp3", "sample_rate": "16000", "channels": 1}
assert abs(float(info["format"]["duration"]) - (chunk["end"] - chunk["start"])) < 0.1
with open(os.environ["TRANSCRIBE_TEST_LOG"], "a") as log:
    log.write(json.dumps({
        "index": index, "chunk": chunk, "staging": str(output.parent),
        "work": str(request_path.parent),
        "upload_duration": float(info["format"]["duration"]),
        "provider": request.get("provider"),
        "response_format": request.get("response_format"),
        "timestamp_granularities": request.get("timestamp_granularities"),
        "authorization": headers.get("authorization"),
    }) + "\n")

if mode in ("parallel", "parallel-error"):
    deadline = time.monotonic() + 5
    barrier = int(os.environ.get("TRANSCRIBE_TEST_BARRIER", "1"))
    while True:
        events = [json.loads(line) for line in events_path.read_text().splitlines()]
        enough = sum(e["event"] == "start" for e in events) >= barrier
        child_ready = mode != "parallel-error" or any(e["event"] == "child" for e in events)
        if enough and child_ready:
            break
        if time.monotonic() >= deadline:
            raise RuntimeError("Worker concurrency barrier timed out")
        time.sleep(0.01)
    if mode == "parallel-error" and index == 1:
        output.write_text('{"error":{"message":"simulated concurrent failure"}}')
        event("finish")
        sys.exit(22)
    time.sleep(30 if mode == "parallel-error" else (0.8 if index == 0 else 0.05))
elif mode == "blocked":
    time.sleep(30)

if index == 1:
    if mode == "http-error":
        output.write_text('{"error":{"message":"simulated timeout"}}')
        sys.exit(22)
    if mode == "bad-json":
        output.write_text("not json")
        sys.exit(0)
    if mode == "api-error":
        output.write_text('{"error":{"message":"simulated API error"},"text":""}')
        sys.exit(0)
    if mode == "missing-words":
        output.write_text('{"text":"speech without timestamps"}')
        sys.exit(0)

if "response_format" not in request:
    response = {"text": "Provider's original formatting.\nNext line."}
else:
    words = [
        {"word": text, "start": start - chunk["start"], "end": end - chunk["start"]}
        for text, start, end in [
            ("Hello.", 1, 1.4), ("boundary", 56.7, 57.1),
            ("continues.", 57.4, 57.8), ("Last.", 65, 65.4),
        ] if start >= chunk["start"] and end <= chunk["end"]
    ]
    if mode in ("speakers", "partial-speakers"):
        if len(manifest) == 1:
            words = [
                {"word": "Hello", "start": 1, "end": 1.4},
                {"word": "there", "start": 1.5, "end": 1.9},
            ]
        for speaker, word in enumerate(words):
            if mode == "speakers" or speaker == 0:
                word["speaker"] = (speaker if len(manifest) == 1 else
                                   int(word["start"] + chunk["start"] >= 57.3))
    if mode == "silence":
        words = []
    if mode == "parallel":
        middle = (chunk["end"] - chunk["start"]) / 2
        words = [{"word": f"Part{index}.", "start": middle, "end": middle + 0.4}]
    if mode.startswith("reconcile"):
        # Two labels already describe voice A inside the first request; later
        # chunks reuse the same local numbers for different request identities.
        late_start = round(manifest[0]["end"]) if len(manifest) > 1 else 60
        words = [
            {"word": text, "start": start - chunk["start"], "end": start + 2 - chunk["start"],
             "speaker": speaker}
            for text, start, speaker in [
                ("Alpha.", 1, 7), ("Again.", 6, 8), ("Beta.", 12, 9),
                ("AlphaLate.", late_start, 9), ("BetaLate.", late_start + 5, 7),
            ] if start >= chunk["start"] and start + 2 <= chunk["end"]
        ]
    response = {"text": " ".join(w["word"] for w in words), "words": words}
output.write_text(json.dumps(response))
event("finish")