#!/usr/bin/env python3
"""Run one explicitly authorized Cell call through the tenant RabbitMQ queue."""

from __future__ import annotations

import argparse
import hashlib
import json
import os
import sys
import time
import wave
from pathlib import Path
from typing import Any

ROOT = Path(__file__).resolve().parents[1]
if str(ROOT) not in sys.path:
    sys.path.insert(0, str(ROOT))


def _data(path: Path) -> dict[str, Any]:
    try:
        value = json.loads(path.read_text(encoding="utf-8"))
    except (OSError, UnicodeDecodeError, json.JSONDecodeError) as exc:
        raise ValueError(f"profile is unavailable: {path}") from exc
    if not isinstance(value, dict):
        raise TypeError("profile must be an object")
    return value


def _path(value: str) -> Path:
    candidate = Path(value)
    return candidate if candidate.is_absolute() else ROOT / candidate


def _integer(value: Any, default: int, field: str, minimum: int = 1) -> int:
    try:
        parsed = int(value)
    except (TypeError, ValueError, OverflowError) as exc:
        raise ValueError(f"{field} must be an integer") from exc
    if parsed < minimum:
        raise ValueError(f"{field} must be at least {minimum}")
    return parsed


def _args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--profile", default="configs/ai-test.bailian.example.yaml")
    parser.add_argument("--prompt-file", default="prompts/test-call.txt")
    parser.add_argument("--callee", required=True)
    parser.add_argument("--tenant-id", default=None)
    parser.add_argument("--tenant-key", default=None)
    parser.add_argument("--route-policy-id", default="route_policy_test")
    parser.add_argument("--caller-profile-id", default="caller_profile_test")
    parser.add_argument("--broker-url", default=None)
    parser.add_argument("--ledger", default="/data/agent-call-cell.sqlite3")
    parser.add_argument("--wait-seconds", type=int, default=240)
    parser.add_argument("--allow-real-call", action="store_true")
    parser.add_argument("--output", default=None)
    return parser.parse_args()


def _command(
    settings: dict[str, Any],
    args: argparse.Namespace,
    tenant_id: str,
    tenant_key: str,
    agent_version_id: str,
) -> dict[str, Any]:
    now_timestamp = time.time()
    issued_at = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime(now_timestamp))
    not_after = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime(now_timestamp + 300))
    execution_id = f"exec_cell_{time.time_ns()}_{os.getpid()}"
    return {
        "schema_version": "1.0",
        "command_type": "call.execute",
        "command_id": f"cmd_cell_{execution_id}",
        "tenant_id": tenant_id,
        "tenant_key": tenant_key,
        "trace_id": f"trace_cell_{execution_id}",
        "issued_at": issued_at,
        "not_after": not_after,
        "payload": {
            "execution_id": execution_id,
            "task_id": str(settings.get("task_id", "task-ai-cell")),
            "task_item_id": f"item_{execution_id}",
            "task_revision": 1,
            "callee": args.callee,
            "route_policy_id": args.route_policy_id,
            "caller_profile_id": args.caller_profile_id,
            "agent_version_id": agent_version_id,
            "variables": {},
            "ring_timeout_ms": 30000,
            "max_call_duration_ms": _integer(
                settings.get("max_duration_ms", 180000),
                180000,
                "max_duration_ms",
            ),
        },
    }


def _recording(path: str) -> dict[str, Any] | None:
    recording = Path(path)
    if not recording.is_file() or recording.is_symlink():
        return None
    try:
        with wave.open(str(recording), "rb") as source:
            evidence = {
                "bytes": recording.stat().st_size,
                "sha256": hashlib.sha256(recording.read_bytes()).hexdigest(),
                "channels": source.getnchannels(),
                "sample_rate_hz": source.getframerate(),
                "frames": source.getnframes(),
                "valid_wav": source.getnframes() > 0 and source.getnchannels() == 1,
            }
    except (OSError, EOFError, wave.Error):
        return None
    return evidence


def _write(report: dict[str, Any], output: str | None) -> None:
    payload = json.dumps(report, ensure_ascii=False, indent=2) + "\n"
    if output:
        destination = _path(output)
        destination.parent.mkdir(parents=True, exist_ok=True)
        destination.write_text(payload, encoding="utf-8")
    print(payload, end="")


def run(args: argparse.Namespace) -> dict[str, Any]:
    if not args.allow_real_call:
        raise ValueError("--allow-real-call is required")
    from agent_call.ai_runtime import ConversationEngine, load_prompt
    from agent_call.bailian import (
        BailianASR,
        BailianLLM,
        BailianTTS,
        build_bailian_config,
    )
    from agent_call.core import PikaBroker
    from agent_call.real_cell import (
        CellCallConfig,
        CellExecutionLedger,
        RealCellCall,
        RealCellWorker,
        load_cell_routes,
    )

    settings = _data(_path(args.profile))
    authorized = {str(value) for value in settings.get("authorized_callees", [])}
    if args.callee not in authorized:
        raise ValueError("callee is not in the profile allow-list")
    tenant_id = str(args.tenant_id or settings.get("tenant_id", "tenant-demo"))
    tenant_key = str(args.tenant_key or settings.get("tenant_key", "tenant-demo-key"))
    agent_version_id = str(settings.get("agent_version_id", "agent_bailian_test_v1"))
    try:
        prompt = load_prompt(
            _path(args.prompt_file),
            _integer(
                settings.get("max_prompt_bytes", 32768), 32768, "max_prompt_bytes"
            ),
        )
    except (OSError, UnicodeDecodeError, ValueError) as exc:
        raise ValueError("prompt is unavailable") from exc
    config = build_bailian_config(
        agent_version_id,
        prompt,
        str(settings.get("model", "qwen-plus")),
        settings.get("tts_model"),
        settings.get("voice"),
        str(settings.get("language", "zh-CN")),
        settings.get("asr_model"),
        str(settings.get("opening", "")),
        min(3, _integer(settings.get("max_turns", 3), 3, "max_turns")),
        _integer(settings.get("max_duration_ms", 180000), 180000, "max_duration_ms"),
    )
    engine = ConversationEngine(
        config,
        llm=BailianLLM.from_env(),
        tts=BailianTTS.from_env(),
        asr=BailianASR.from_env(),
    )
    broker = PikaBroker(args.broker_url or os.environ.get("RABBITMQ_URL", ""))
    ledger = CellExecutionLedger(args.ledger)
    executor = RealCellCall(CellCallConfig.from_env(), engine, BailianASR.from_env())
    worker = RealCellWorker(
        broker,
        tenant_key,
        ledger,
        executor,
        routes=load_cell_routes(os.environ.get("CELL_ROUTE_MAP_JSON")),
    )
    command = _command(settings, args, tenant_id, tenant_key, agent_version_id)
    route = f"agent-call.tenant.{tenant_key}.call.execute"
    broker.publish(
        "agent-call.commands.v1", route, command, message_id=command["command_id"]
    )
    deadline = time.monotonic() + args.wait_seconds
    event: dict[str, Any] | None = None
    while time.monotonic() < deadline:
        event = worker.process_once()
        if event and event.get("event_type") == "call.finished":
            break
        time.sleep(0.2)
    if event is None or event.get("event_type") != "call.finished":
        raise TimeoutError("Cell call did not finish before the bounded wait")
    result = ledger.result(command["payload"]["execution_id"]) or {}
    result_status = event["payload"].get("status")
    recording_path = result.get("recording_path")
    recording = _recording(recording_path) if isinstance(recording_path, str) else None
    passed = (
        result_status == "completed"
        and len(event["payload"].get("turns", [])) >= 3
        and bool(recording and recording.get("valid_wav"))
    )
    report: dict[str, Any] = {
        "schema_version": "1.0",
        "status": "PASS" if passed else "FAIL",
        "mode": "real",
        "call_execution": {
            "execution_id": command["payload"]["execution_id"],
            "tenant_id": tenant_id,
            "tenant_key": tenant_key,
            "callee": args.callee,
            "event_id": event["event_id"],
        },
        "result": event["payload"],
    }
    if recording is not None:
        report["recording"] = recording
    return report


def main() -> int:
    args = _args()
    try:
        report = run(args)
        code = 0 if report.get("status") == "PASS" else 1
    except (
        OSError,
        RuntimeError,
        TimeoutError,
        TypeError,
        ValueError,
        KeyError,
    ) as exc:
        report = {
            "schema_version": "1.0",
            "status": "FAIL",
            "reason_code": type(exc).__name__,
            "message": str(exc),
        }
        code = 1
    _write(report, args.output)
    return code


if __name__ == "__main__":
    raise SystemExit(main())