from __future__ import annotations

import argparse
import asyncio
import base64
import binascii
import csv
import json
import os
import time
import uuid
import wave
from collections import Counter
from dataclasses import asdict, dataclass
from datetime import datetime, timezone
from pathlib import Path
from typing import Any
from urllib.parse import urlencode

import aiohttp
from dotenv import load_dotenv


DEFAULT_TEXT = (
    "Ola! Sou especialista em contas e estou a disposicao. "
    "Como posso te ajudar com a sua fatura hoje?"
)
DEFAULT_OUTPUT_JSONL = Path(".run/xai-tts-load/results.jsonl")
DEFAULT_SUMMARY_CSV = Path(".run/xai-tts-load/summary.csv")
SAMPLE_RATE = 24_000
NUM_CHANNELS = 1
SAMPLE_WIDTH_BYTES = 2


def _now_utc() -> str:
    return datetime.now(timezone.utc).isoformat()


def _elapsed_ms(started_at: float) -> int:
    return max(0, round((time.perf_counter() - started_at) * 1000))


def _env_str(name: str, default: str = "") -> str:
    return str(os.getenv(name, default) or default).strip()


def _bool_query(value: bool) -> str:
    return "true" if value else "false"


def _percentile(values: list[int], percentile: float) -> int | None:
    if not values:
        return None
    ordered = sorted(values)
    index = round((len(ordered) - 1) * percentile)
    return ordered[index]


def _write_wav(path: Path, pcm: bytes) -> Path:
    path.parent.mkdir(parents=True, exist_ok=True)
    with wave.open(str(path), "wb") as handle:
        handle.setnchannels(NUM_CHANNELS)
        handle.setsampwidth(SAMPLE_WIDTH_BYTES)
        handle.setframerate(SAMPLE_RATE)
        handle.writeframes(pcm)
    return path


@dataclass(frozen=True, slots=True)
class Config:
    env_file: Path
    api_key: str
    websocket_url: str
    voice: str
    language: str
    text_normalization: bool
    optimize_streaming_latency: int
    iterations: int
    concurrency: int
    reuse_connection: bool
    reset_on_failure: bool
    prewarm_idle_s: float
    idle_between_requests_s: float
    request_timeout_s: float
    connect_timeout_s: float
    gap_timeout_s: float
    output_jsonl: Path
    summary_csv: Path | None
    audio_dir: Path | None
    texts: tuple[str, ...]


@dataclass(slots=True)
class XAIConnection:
    connection_id: str
    ws: aiohttp.ClientWebSocketResponse
    opened_at: float
    opened_connect_ms: int
    turns_completed: int = 0
    last_outcome: str = ""
    last_audio_bytes: int = 0
    last_audio_chunks: int = 0

    @property
    def closed(self) -> bool:
        return self.ws.closed

    @property
    def age_ms(self) -> int:
        return _elapsed_ms(self.opened_at)


@dataclass(slots=True)
class Result:
    run_id: str
    request_id: str
    worker_id: int
    iteration: int
    started_at: str
    connection_id: str
    connection_age_ms: int
    connection_turn_index: int
    reused_connection: bool
    opened_for_request: bool
    idle_before_turn_ms: int
    previous_outcome: str
    previous_audio_bytes: int
    previous_audio_chunks: int
    text_len: int
    connect_ms: int
    send_ms: int
    first_audio_ms: int | None
    request_to_first_audio_ms: int | None
    total_to_first_audio_ms: int | None
    elapsed_ms: int
    audio_chunks: int
    audio_bytes: int
    empty_audio_deltas: int
    max_gap_ms: int
    gap_timeout_ms: int
    outcome: str
    trace_id: str
    error: str
    audio_path: str


def _build_url(config: Config) -> str:
    params = {
        "voice": config.voice,
        "language": config.language,
        "codec": "pcm",
        "sample_rate": str(SAMPLE_RATE),
        "optimize_streaming_latency": str(config.optimize_streaming_latency),
        "text_normalization": _bool_query(config.text_normalization),
    }
    return f"{config.websocket_url}?{urlencode(params)}"


async def _open_connection(
    session: aiohttp.ClientSession,
    config: Config,
    *,
    worker_id: int,
) -> XAIConnection:
    connection_id = f"direct-{worker_id}-{uuid.uuid4().hex[:8]}"
    started_at = time.perf_counter()
    ws = await asyncio.wait_for(
        session.ws_connect(
            _build_url(config),
            headers={"Authorization": f"Bearer {config.api_key}"},
            heartbeat=20,
        ),
        timeout=config.connect_timeout_s,
    )
    opened_at = time.perf_counter()
    return XAIConnection(
        connection_id=connection_id,
        ws=ws,
        opened_at=opened_at,
        opened_connect_ms=max(0, round((opened_at - started_at) * 1000)),
    )


async def _close_connection(conn: XAIConnection | None, *, reason: str) -> None:
    if conn is None or conn.closed:
        return
    await conn.ws.close(message=reason.encode("utf-8"))


async def _receive_xai_turn(
    *,
    config: Config,
    conn: XAIConnection,
    text: str,
    run_id: str,
    worker_id: int,
    iteration: int,
    job_started_at: float,
    opened_for_request: bool,
    idle_before_turn_ms: int,
    output_audio_dir: Path | None,
) -> Result:
    request_id = uuid.uuid4().hex[:12]
    send_started_at = time.perf_counter()
    await conn.ws.send_str(json.dumps({"type": "text.delta", "delta": text}, ensure_ascii=False))
    await conn.ws.send_str(json.dumps({"type": "text.done"}))
    send_ms = _elapsed_ms(send_started_at)

    turn_started_at = time.perf_counter()
    first_audio_ms: int | None = None
    request_to_first_audio_ms: int | None = None
    total_to_first_audio_ms: int | None = None
    last_audio_at = 0.0
    audio_chunks = 0
    audio_bytes = 0
    empty_audio_deltas = 0
    max_gap_ms = 0
    trace_id = ""
    error = ""
    outcome = "unknown"
    audio = bytearray()
    previous_outcome = conn.last_outcome
    previous_audio_bytes = conn.last_audio_bytes
    previous_audio_chunks = conn.last_audio_chunks
    turn_index = conn.turns_completed + 1
    connection_age_ms = conn.age_ms
    gap_timeout_ms = round(config.gap_timeout_s * 1000)

    while True:
        receive_timeout_s = config.request_timeout_s
        if audio_chunks > 0 and config.gap_timeout_s > 0:
            receive_timeout_s = min(receive_timeout_s, config.gap_timeout_s)

        try:
            msg = await asyncio.wait_for(conn.ws.receive(), timeout=receive_timeout_s)
        except asyncio.TimeoutError:
            if audio_chunks > 0 and config.gap_timeout_s > 0:
                gap_ms = max(0, round((time.perf_counter() - last_audio_at) * 1000))
                max_gap_ms = max(max_gap_ms, gap_ms)
                outcome = "gap_timeout"
                error = f"no audio.delta for {gap_ms}ms"
            else:
                outcome = "no_audio_timeout"
                error = f"no audio.delta before {round(config.request_timeout_s * 1000)}ms"
            break

        now = time.perf_counter()
        if msg.type in (
            aiohttp.WSMsgType.CLOSED,
            aiohttp.WSMsgType.CLOSE,
            aiohttp.WSMsgType.CLOSING,
            aiohttp.WSMsgType.ERROR,
        ):
            outcome = "ws_closed_after_audio" if audio_chunks else "ws_closed_no_audio"
            error = f"websocket message type {msg.type.name}"
            break

        if msg.type != aiohttp.WSMsgType.TEXT:
            continue

        try:
            data = json.loads(str(msg.data))
        except json.JSONDecodeError as exc:
            outcome = "invalid_json"
            error = str(exc)
            break

        msg_type = data.get("type")
        if msg_type == "audio.delta":
            try:
                chunk = base64.b64decode(str(data.get("delta") or ""))
            except (binascii.Error, ValueError) as exc:
                outcome = "invalid_audio_delta"
                error = str(exc)
                break

            if not chunk:
                empty_audio_deltas += 1
                continue

            if audio_chunks == 0:
                first_audio_ms = max(0, round((now - turn_started_at) * 1000))
                request_to_first_audio_ms = max(0, round((now - send_started_at) * 1000))
                total_to_first_audio_ms = max(0, round((now - job_started_at) * 1000))
            else:
                gap_ms = max(0, round((now - last_audio_at) * 1000))
                max_gap_ms = max(max_gap_ms, gap_ms)

            audio_chunks += 1
            audio_bytes += len(chunk)
            last_audio_at = now
            if output_audio_dir is not None:
                audio.extend(chunk)
        elif msg_type == "audio.done":
            trace_id = str(data.get("trace_id") or "")
            outcome = "success" if audio_chunks else "no_audio_done"
            break
        elif msg_type == "audio.clear":
            outcome = "audio_clear"
            break
        elif msg_type == "error":
            trace_id = str(data.get("trace_id") or "")
            outcome = "api_error"
            error = str(data.get("message") or data)
            break

    elapsed_ms = _elapsed_ms(turn_started_at)
    audio_path = ""
    if output_audio_dir is not None and audio:
        audio_path = str(_write_wav(output_audio_dir / f"{iteration:04d}_{request_id}.wav", bytes(audio)))

    result = Result(
        run_id=run_id,
        request_id=request_id,
        worker_id=worker_id,
        iteration=iteration,
        started_at=_now_utc(),
        connection_id=conn.connection_id,
        connection_age_ms=connection_age_ms,
        connection_turn_index=turn_index,
        reused_connection=(not opened_for_request),
        opened_for_request=opened_for_request,
        idle_before_turn_ms=idle_before_turn_ms,
        previous_outcome=previous_outcome,
        previous_audio_bytes=previous_audio_bytes,
        previous_audio_chunks=previous_audio_chunks,
        text_len=len(text),
        connect_ms=conn.opened_connect_ms if opened_for_request else 0,
        send_ms=send_ms,
        first_audio_ms=first_audio_ms,
        request_to_first_audio_ms=request_to_first_audio_ms,
        total_to_first_audio_ms=total_to_first_audio_ms,
        elapsed_ms=elapsed_ms,
        audio_chunks=audio_chunks,
        audio_bytes=audio_bytes,
        empty_audio_deltas=empty_audio_deltas,
        max_gap_ms=max_gap_ms,
        gap_timeout_ms=gap_timeout_ms,
        outcome=outcome,
        trace_id=trace_id,
        error=error,
        audio_path=audio_path,
    )
    conn.turns_completed += 1
    conn.last_outcome = outcome
    conn.last_audio_bytes = audio_bytes
    conn.last_audio_chunks = audio_chunks
    return result


def _compact_result_line(result: Result) -> str:
    return (
        "[xai-tts-load] "
        f"i={result.iteration} worker={result.worker_id} conn={result.connection_id} "
        f"turn={result.connection_turn_index} outcome={result.outcome} "
        f"age_ms={result.connection_age_ms} idle_ms={result.idle_before_turn_ms} "
        f"connect_ms={result.connect_ms} first_audio_ms={result.first_audio_ms} "
        f"request_to_first_audio_ms={result.request_to_first_audio_ms} "
        f"total_to_first_audio_ms={result.total_to_first_audio_ms} "
        f"max_gap_ms={result.max_gap_ms} chunks={result.audio_chunks} bytes={result.audio_bytes} "
        f"trace_id={result.trace_id}"
    )


async def _run_worker(
    *,
    config: Config,
    run_id: str,
    worker_id: int,
    iterations: list[int],
    jsonl_handle: Any,
    write_lock: asyncio.Lock,
) -> list[Result]:
    results: list[Result] = []
    conn: XAIConnection | None = None
    async with aiohttp.ClientSession() as session:
        for iteration in iterations:
            text = config.texts[(iteration - 1) % len(config.texts)]
            job_started_at = time.perf_counter()
            opened_for_request = False
            idle_before_turn_ms = 0
            if conn is None or conn.closed:
                conn = await _open_connection(session, config, worker_id=worker_id)
                opened_for_request = True
                if config.prewarm_idle_s > 0:
                    idle_started_at = time.perf_counter()
                    await asyncio.sleep(config.prewarm_idle_s)
                    idle_before_turn_ms = _elapsed_ms(idle_started_at)
            elif config.idle_between_requests_s > 0:
                idle_started_at = time.perf_counter()
                await asyncio.sleep(config.idle_between_requests_s)
                idle_before_turn_ms = _elapsed_ms(idle_started_at)

            try:
                result = await _receive_xai_turn(
                    config=config,
                    conn=conn,
                    text=text,
                    run_id=run_id,
                    worker_id=worker_id,
                    iteration=iteration,
                    job_started_at=job_started_at,
                    opened_for_request=opened_for_request,
                    idle_before_turn_ms=idle_before_turn_ms,
                    output_audio_dir=config.audio_dir,
                )
            except Exception as exc:
                result = Result(
                    run_id=run_id,
                    request_id=uuid.uuid4().hex[:12],
                    worker_id=worker_id,
                    iteration=iteration,
                    started_at=_now_utc(),
                    connection_id=conn.connection_id if conn is not None else "",
                    connection_age_ms=conn.age_ms if conn is not None else 0,
                    connection_turn_index=(conn.turns_completed + 1) if conn is not None else 0,
                    reused_connection=not opened_for_request,
                    opened_for_request=opened_for_request,
                    idle_before_turn_ms=idle_before_turn_ms,
                    previous_outcome=conn.last_outcome if conn is not None else "",
                    previous_audio_bytes=conn.last_audio_bytes if conn is not None else 0,
                    previous_audio_chunks=conn.last_audio_chunks if conn is not None else 0,
                    text_len=len(text),
                    connect_ms=conn.opened_connect_ms if conn is not None and opened_for_request else 0,
                    send_ms=0,
                    first_audio_ms=None,
                    request_to_first_audio_ms=None,
                    total_to_first_audio_ms=None,
                    elapsed_ms=_elapsed_ms(job_started_at),
                    audio_chunks=0,
                    audio_bytes=0,
                    empty_audio_deltas=0,
                    max_gap_ms=0,
                    gap_timeout_ms=round(config.gap_timeout_s * 1000),
                    outcome="exception",
                    trace_id="",
                    error=f"{type(exc).__name__}: {exc}",
                    audio_path="",
                )

            results.append(result)
            async with write_lock:
                jsonl_handle.write(json.dumps(asdict(result), ensure_ascii=False) + "\n")
                jsonl_handle.flush()
            print(_compact_result_line(result), flush=True)

            should_close = not config.reuse_connection
            if result.outcome != "success" and config.reset_on_failure:
                should_close = True
            if should_close:
                await _close_connection(conn, reason=result.outcome)
                conn = None

        await _close_connection(conn, reason="worker_done")
    return results


def _partition_iterations(iterations: int, concurrency: int) -> list[list[int]]:
    buckets: list[list[int]] = [[] for _ in range(concurrency)]
    for iteration in range(1, iterations + 1):
        buckets[(iteration - 1) % concurrency].append(iteration)
    return buckets


def _write_summary_csv(path: Path, results: list[Result]) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    fields = list(asdict(results[0]).keys()) if results else list(Result.__dataclass_fields__.keys())
    with path.open("w", encoding="utf-8", newline="") as handle:
        writer = csv.DictWriter(handle, fieldnames=fields)
        writer.writeheader()
        for result in sorted(results, key=lambda item: item.iteration):
            writer.writerow(asdict(result))


def _metric_values(results: list[Result], field: str) -> list[int]:
    values: list[int] = []
    for result in results:
        value = getattr(result, field)
        if isinstance(value, int):
            values.append(value)
    return values


def _print_summary(config: Config, results: list[Result], started_at: float) -> None:
    outcomes = Counter(result.outcome for result in results)
    first_audio_values = _metric_values(results, "first_audio_ms")
    request_to_first_values = _metric_values(results, "request_to_first_audio_ms")
    total_to_first_values = _metric_values(results, "total_to_first_audio_ms")
    connect_values = [result.connect_ms for result in results if result.opened_for_request]
    connection_age_values = _metric_values(results, "connection_age_ms")
    gap_values = _metric_values(results, "max_gap_ms")
    no_audio = sum(1 for result in results if result.audio_bytes == 0)
    gap_timeouts = sum(1 for result in results if result.outcome == "gap_timeout")
    total_near_3s = sum(
        1
        for result in results
        if result.total_to_first_audio_ms is not None and result.total_to_first_audio_ms >= 2800
    )
    request_near_3s = sum(
        1
        for result in results
        if result.request_to_first_audio_ms is not None and result.request_to_first_audio_ms >= 2800
    )

    print("", flush=True)
    print("[xai-tts-load] resumo", flush=True)
    print(f"  total={len(results)} duration_ms={_elapsed_ms(started_at)}", flush=True)
    print(
        f"  reuse_connection={int(config.reuse_connection)} concurrency={config.concurrency} "
        f"prewarm_idle_ms={round(config.prewarm_idle_s * 1000)} "
        f"idle_between_requests_ms={round(config.idle_between_requests_s * 1000)}",
        flush=True,
    )
    print(f"  outcomes={dict(outcomes)}", flush=True)
    print(
        f"  no_audio={no_audio} gap_timeouts={gap_timeouts} "
        f"request_to_first_audio_ge_2800ms={request_near_3s} "
        f"total_to_first_audio_ge_2800ms={total_near_3s}",
        flush=True,
    )
    print(
        "  connection_age_ms "
        f"p50={_percentile(connection_age_values, 0.50)} p95={_percentile(connection_age_values, 0.95)} "
        f"max={max(connection_age_values) if connection_age_values else None}",
        flush=True,
    )
    print(
        "  connect_ms "
        f"p50={_percentile(connect_values, 0.50)} p95={_percentile(connect_values, 0.95)} "
        f"max={max(connect_values) if connect_values else None}",
        flush=True,
    )
    print(
        "  first_audio_ms "
        f"p50={_percentile(first_audio_values, 0.50)} p95={_percentile(first_audio_values, 0.95)} "
        f"max={max(first_audio_values) if first_audio_values else None}",
        flush=True,
    )
    print(
        "  request_to_first_audio_ms "
        f"p50={_percentile(request_to_first_values, 0.50)} p95={_percentile(request_to_first_values, 0.95)} "
        f"max={max(request_to_first_values) if request_to_first_values else None}",
        flush=True,
    )
    print(
        "  total_to_first_audio_ms "
        f"p50={_percentile(total_to_first_values, 0.50)} p95={_percentile(total_to_first_values, 0.95)} "
        f"max={max(total_to_first_values) if total_to_first_values else None}",
        flush=True,
    )
    print(
        "  max_gap_ms "
        f"p50={_percentile(gap_values, 0.50)} p95={_percentile(gap_values, 0.95)} "
        f"max={max(gap_values) if gap_values else None}",
        flush=True,
    )


def _load_texts(args: argparse.Namespace) -> tuple[str, ...]:
    texts = [item for item in (args.text or []) if item]
    if args.texts_file:
        for line in Path(args.texts_file).read_text(encoding="utf-8").splitlines():
            stripped = line.strip()
            if stripped and not stripped.startswith("#"):
                texts.append(stripped)
    if not texts:
        texts.append(DEFAULT_TEXT)
    return tuple(texts)


def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
    parser = argparse.ArgumentParser(
        description="Carga direta no WebSocket de TTS xAI, sem LiveKit/TIA.",
    )
    parser.add_argument("--env-file", default=".env.dev")
    parser.add_argument("--websocket-url", default="")
    parser.add_argument("--api-key", default="")
    parser.add_argument("--voice", default="")
    parser.add_argument("--language", default="")
    parser.add_argument("--iterations", type=int, default=20)
    parser.add_argument("--concurrency", type=int, default=1)
    mode = parser.add_mutually_exclusive_group()
    mode.add_argument("--reuse-connection", dest="reuse_connection", action="store_true", default=True)
    mode.add_argument("--new-connection-per-request", dest="reuse_connection", action="store_false")
    parser.add_argument("--keep-connection-after-failure", action="store_true")
    parser.add_argument(
        "--prewarm-idle-ms",
        type=int,
        default=0,
        help="Abre o WebSocket e espera este tempo antes do primeiro texto da conexao.",
    )
    parser.add_argument(
        "--idle-between-requests-ms",
        type=int,
        default=0,
        help="Espera este tempo entre turnos quando a mesma conexao e reutilizada.",
    )
    parser.add_argument("--request-timeout-ms", type=int, default=10_000)
    parser.add_argument("--connect-timeout-ms", type=int, default=10_000)
    parser.add_argument("--gap-timeout-ms", type=int, default=3_000)
    parser.add_argument("--optimize-streaming-latency", type=int, default=1)
    parser.add_argument("--no-text-normalization", action="store_true")
    parser.add_argument("--text", action="append", help="Texto a sintetizar. Pode repetir.")
    parser.add_argument("--texts-file", default="", help="Arquivo com um texto por linha.")
    parser.add_argument("--output-jsonl", default=str(DEFAULT_OUTPUT_JSONL))
    parser.add_argument("--summary-csv", default=str(DEFAULT_SUMMARY_CSV))
    parser.add_argument("--no-summary-csv", action="store_true")
    parser.add_argument("--write-audio-dir", default="")
    return parser.parse_args(argv)


def load_config(args: argparse.Namespace) -> Config:
    env_file = Path(args.env_file)
    if env_file.exists():
        load_dotenv(env_file, override=False)

    api_key = str(args.api_key or _env_str("XAI_API_KEY")).strip()
    websocket_url = str(args.websocket_url or _env_str("XAI_WEBSOCKET_URL")).strip()
    if not api_key:
        raise RuntimeError("Missing XAI_API_KEY. Defina no .env.dev ou passe --api-key.")
    if not websocket_url:
        raise RuntimeError("Missing XAI_WEBSOCKET_URL. Defina no .env.dev ou passe --websocket-url.")

    return Config(
        env_file=env_file,
        api_key=api_key,
        websocket_url=websocket_url,
        voice=str(args.voice or _env_str("XAI_TTS_VOICE", "ara")).strip() or "ara",
        language=str(args.language or _env_str("XAI_TTS_LANGUAGE", "pt-BR")).strip() or "pt-BR",
        text_normalization=not bool(args.no_text_normalization),
        optimize_streaming_latency=max(0, int(args.optimize_streaming_latency)),
        iterations=max(1, int(args.iterations)),
        concurrency=max(1, int(args.concurrency)),
        reuse_connection=bool(args.reuse_connection),
        reset_on_failure=not bool(args.keep_connection_after_failure),
        prewarm_idle_s=max(0.0, int(args.prewarm_idle_ms) / 1000),
        idle_between_requests_s=max(0.0, int(args.idle_between_requests_ms) / 1000),
        request_timeout_s=max(0.1, int(args.request_timeout_ms) / 1000),
        connect_timeout_s=max(0.1, int(args.connect_timeout_ms) / 1000),
        gap_timeout_s=max(0.0, int(args.gap_timeout_ms) / 1000),
        output_jsonl=Path(args.output_jsonl),
        summary_csv=None if args.no_summary_csv else Path(args.summary_csv),
        audio_dir=Path(args.write_audio_dir) if args.write_audio_dir else None,
        texts=_load_texts(args),
    )


async def run(argv: list[str] | None = None) -> int:
    config = load_config(parse_args(argv))
    config.output_jsonl.parent.mkdir(parents=True, exist_ok=True)
    if config.audio_dir is not None:
        config.audio_dir.mkdir(parents=True, exist_ok=True)

    run_id = uuid.uuid4().hex[:12]
    started_at = time.perf_counter()
    print(
        "[xai-tts-load] iniciando "
        f"run_id={run_id} iterations={config.iterations} concurrency={config.concurrency} "
        f"reuse_connection={int(config.reuse_connection)} voice={config.voice} "
        f"language={config.language} prewarm_idle_ms={round(config.prewarm_idle_s * 1000)} "
        f"idle_between_requests_ms={round(config.idle_between_requests_s * 1000)} "
        f"output={config.output_jsonl}",
        flush=True,
    )

    write_lock = asyncio.Lock()
    partitions = _partition_iterations(config.iterations, config.concurrency)
    with config.output_jsonl.open("w", encoding="utf-8") as jsonl_handle:
        tasks = [
            _run_worker(
                config=config,
                run_id=run_id,
                worker_id=index + 1,
                iterations=iterations,
                jsonl_handle=jsonl_handle,
                write_lock=write_lock,
            )
            for index, iterations in enumerate(partitions)
            if iterations
        ]
        nested_results = await asyncio.gather(*tasks)

    results = [result for batch in nested_results for result in batch]
    results.sort(key=lambda item: item.iteration)
    if config.summary_csv is not None:
        _write_summary_csv(config.summary_csv, results)
        print(f"[xai-tts-load] csv={config.summary_csv}", flush=True)
    _print_summary(config, results, started_at)
    return 1 if any(result.outcome != "success" for result in results) else 0


def main() -> None:
    raise SystemExit(asyncio.run(run()))


if __name__ == "__main__":
    main()
