#!/usr/bin/env python3
"""Run one explicit, non-retrying SIP probe through the current ECS host."""

from __future__ import annotations

import argparse
import base64
import json
import os
import re
import sys
import time
from pathlib import Path
from typing import Any

try:
    from .aliyun_host import AliyunCLI, CloudError, plan
except ImportError:
    from aliyun_host import AliyunCLI, CloudError, plan

ROOT = Path(__file__).resolve().parents[1]
DEFAULT_CONFIG = ROOT / ".local/agent-call-persistent/aliyun-real.json"
DEFAULT_OUTPUT_DIR = ROOT / ".local/agent-call-persistent"
ANSI = re.compile(r"\x1b\[[0-?]*[ -/]*[@-~]")
SIP_STATUS = re.compile(r"SIP/2\.0\s+([1-6]\d{2})(?:\s+([^\r\n]+))?")
TERMINAL = {"Finished", "Success", "Failed", "Stopped", "PartialFailed"}


def validate_number(number: str) -> str:
    if not re.fullmatch(r"[0-9]{3,32}", number):
        raise ValueError(
            "number must contain only 3-32 digits; the original value is preserved"
        )
    return number


def clean(text: str) -> str:
    return ANSI.sub("", text)


def parse_int(value: str) -> int | None:
    try:
        return int(value)
    except (TypeError, ValueError):
        return None


def remote_command(number: str, target: str, wait_seconds: int) -> str:
    pattern = (
        f"{target}|INVITE sip:|ACK sip:|BYE sip:|"
        r"SIP/2.0 (100|180|183|200|4[0-9][0-9]|5[0-9][0-9]|6[0-9][0-9])|"
        r"Reason:|m=audio|a=rtpmap|Called|Executing.*outbound|"
        r"Everyone is busy|Hangup|Spawn extension"
    )
    return f"""set +e
docker exec agent-call-asterisk asterisk -rx 'pjsip set logger on' >/dev/null
start=$(date -u +%Y-%m-%dT%H:%M:%SZ)
echo "CALL_START $start number={number} target={target} codec=PCMA logger=on"
docker exec agent-call-asterisk asterisk -rx 'channel originate Local/{target}@outbound application Wait {wait_seconds}'
rc=$?
echo "ORIGINATE_RC $rc"
sleep {wait_seconds + 2}
echo "CALL_END $(date -u +%Y-%m-%dT%H:%M:%SZ)"
echo '== channels =='
docker exec agent-call-asterisk asterisk -rx 'core show channels count'
echo '== filtered SIP/Asterisk evidence =='
docker logs --since "$start" agent-call-asterisk 2>&1 | grep -E '{pattern}' || true
"""


def output_text(row: dict[str, Any]) -> str:
    value = row.get("Output", "")
    if not isinstance(value, str):
        return ""
    if "CALL_START" in value:
        return value
    try:
        decoded = base64.b64decode(value, validate=True).decode("utf-8", "replace")
    except (ValueError, UnicodeDecodeError):
        return value
    return decoded if "CALL_START" in decoded else value


def summarize(text: str, target: str) -> dict[str, Any]:
    text = clean(text)
    marker = text.find(f"INVITE sip:{target}")
    relevant = text[marker:] if marker >= 0 else text
    responses = []
    for match in SIP_STATUS.finditer(relevant):
        code = parse_int(match.group(1))
        if code is not None:
            responses.append({"code": code, "phrase": (match.group(2) or "").strip()})
    final = next((item for item in responses if item["code"] >= 200), None)
    reason_match = re.search(r"Reason:\s*([^\r\n]+)", relevant, re.IGNORECASE)
    origin_match = re.search(r"ORIGINATE_RC\s+(-?\d+)", text)
    channels_match = re.search(r"(\d+) active channels", text)
    if final is None:
        result = "provisional" if responses else "no_response"
    else:
        result = "connected" if final["code"] < 300 else "rejected"
    return {
        "result": result,
        "origin_rc": parse_int(origin_match.group(1)) if origin_match else None,
        "responses": responses,
        "final_response": final,
        "reason": reason_match.group(1).strip() if reason_match else None,
        "active_channels": parse_int(channels_match.group(1))
        if channels_match
        else None,
    }


def first_invocation_result(data: dict[str, Any]) -> dict[str, Any] | None:
    rows = (
        data.get("Invocation", {})
        .get("InvocationResults", {})
        .get("InvocationResult", [])
    )
    if not isinstance(rows, list) or not rows or not isinstance(rows[0], dict):
        return None
    return rows[0]


def save_evidence(path: Path, evidence: dict[str, Any]) -> None:
    if path.is_symlink():
        raise ValueError("output path must not be a symlink")
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text(
        json.dumps(evidence, ensure_ascii=False, indent=2), encoding="utf-8"
    )
    os.chmod(path, 0o600)


def run(args: argparse.Namespace) -> tuple[int, Path, dict[str, Any]]:
    number = validate_number(args.number)
    config_path = Path(args.config).expanduser()
    try:
        cfg = json.loads(config_path.read_text(encoding="utf-8"))
    except (OSError, UnicodeDecodeError, json.JSONDecodeError) as exc:
        raise ValueError(f"config file is unavailable: {config_path}") from exc
    api = AliyunCLI(cfg.get("profile"))
    current = plan(cfg, api)
    if current.get("action") != "reuse" or current.get("instance_status") != "Running":
        raise CloudError(
            "fixed-EIP host is not a running reusable instance; no call sent"
        )
    instance_id = current.get("instance_id")
    if not isinstance(instance_id, str) or not instance_id:
        raise CloudError("reusable instance ID is unavailable; no call sent")
    target = "7089" + number
    command_name = (
        f"agent-call-sip-probe-{number}-{time.strftime('%Y%m%d%H%M%S', time.gmtime())}"
    )
    run_result = api(
        "ecs",
        "RunCommand",
        **{
            "InstanceId.1": instance_id,
            "Type": "RunShellScript",
            "Name": command_name,
            "CommandContent": remote_command(number, target, args.wait_seconds),
            "ContentEncoding": "PlainText",
            "Timeout": max(30, args.wait_seconds + 12),
            "RepeatMode": "Once",
            "KeepCommand": "false",
        },
    )
    command_id = run_result.get("CommandId")
    invoke_id = run_result.get("InvokeId")
    if not isinstance(command_id, str) or not isinstance(invoke_id, str):
        raise CloudError(
            "RunCommand returned no command/invocation ID; inspect cloud state before retry"
        )

    result: dict[str, Any] | None = None
    row: dict[str, Any] | None = None
    deadline = time.monotonic() + args.timeout_seconds
    while time.monotonic() < deadline:
        result = api(
            "ecs",
            "DescribeInvocationResults",
            InvokeId=invoke_id,
            ContentEncoding="PlainText",
            MaxResults=10,
        )
        candidate = first_invocation_result(result)
        if candidate is not None:
            row = candidate
            status = candidate.get("InvokeRecordStatus") or candidate.get(
                "InvocationStatus"
            )
            if status in TERMINAL:
                break
        time.sleep(args.poll_seconds)
    if row is None:
        raise CloudError(
            f"invocation {invoke_id} did not finish before timeout; do not redial automatically"
        )
    status = row.get("InvokeRecordStatus") or row.get("InvocationStatus")
    if status not in TERMINAL:
        raise CloudError(
            f"invocation {invoke_id} did not finish before timeout; do not redial automatically"
        )

    raw_output = output_text(row)
    probe = summarize(raw_output, target)
    stamp = time.strftime("%Y%m%dT%H%M%SZ", time.gmtime())
    output_path = (
        Path(args.output)
        if args.output
        else DEFAULT_OUTPUT_DIR / f"sip-probe-{number}-{stamp}.json"
    )
    evidence = {
        "instance_id": instance_id,
        "public_ip": current.get("public_ip"),
        "number": number,
        "target": target,
        "dial_prefix": "7089",
        "codec": "PCMA",
        "logger": "on",
        "command_id": command_id,
        "invoke_id": invoke_id,
        "invocation_status": row.get("InvocationStatus"),
        "invoke_record_status": row.get("InvokeRecordStatus"),
        "exit_code": row.get("ExitCode"),
        "probe": probe,
        "raw_output": clean(raw_output),
    }
    save_evidence(output_path, evidence)
    summary = {key: value for key, value in evidence.items() if key != "raw_output"}
    summary["evidence_file"] = str(output_path)
    return (
        0 if row.get("ExitCode") == 0 and probe["result"] != "no_response" else 1,
        output_path,
        summary,
    )


def main() -> int:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument(
        "--number", required=True, help="original callee number; one call, no retry"
    )
    parser.add_argument("--config", default=str(DEFAULT_CONFIG))
    parser.add_argument("--output", help="0600 evidence JSON path")
    parser.add_argument("--wait-seconds", type=int, default=5)
    parser.add_argument("--poll-seconds", type=float, default=1.0)
    parser.add_argument("--timeout-seconds", type=int, default=60)
    args = parser.parse_args()
    if not 3 <= args.wait_seconds <= 30:
        parser.error("--wait-seconds must be 3-30")
    if args.poll_seconds <= 0 or args.timeout_seconds <= 0:
        parser.error("poll and timeout must be positive")
    try:
        status, _, summary = run(args)
    except (CloudError, OSError, ValueError, json.JSONDecodeError) as exc:
        print(f"ERROR: {exc}", file=sys.stderr)
        return 1
    print(json.dumps(summary, ensure_ascii=False, indent=2))
    return status


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