call-center/services/asterisk_bridge_service/ami.py

from __future__ import annotations

from dataclasses import dataclass
import json
import socket
import sys
import time
from typing import Any

from sqlalchemy import select

from services.shared.core import new_id, utc_now_iso
from services.shared.sql_models import AsteriskCallLinkRow, AsteriskEventLogRow


def _bridge_app():
    return sys.modules["services.asterisk_bridge_service.app"]


def first_non_empty(*values: Any) -> str | None:
    for value in values:
        text = str(value or "").strip()
        if text:
            return text
    return None


def extract_call_id(payload: dict[str, Any]) -> str | None:
    return first_non_empty(
        payload.get("CallID"),
        payload.get("CallId"),
        payload.get("Callid"),
        payload.get("LinkedID"),
        payload.get("LinkedId"),
        payload.get("Linkedid"),
        payload.get("UniqueID"),
        payload.get("UniqueId"),
        payload.get("Uniqueid"),
    )


def extract_linked_id(payload: dict[str, Any], call_id: str | None = None) -> str | None:
    return first_non_empty(
        payload.get("LinkedID"),
        payload.get("LinkedId"),
        payload.get("Linkedid"),
        payload.get("CallID"),
        payload.get("CallId"),
        payload.get("Callid"),
        call_id,
    )


def resolve_channel_from_payload(payload: dict[str, Any]) -> str | None:
    return first_non_empty(
        payload.get("Channel"),
        payload.get("channel"),
        payload.get("InboundChannel"),
        payload.get("Inboundchannel"),
    )


def read_ami_frame(handle) -> dict[str, str] | None:
    fields: dict[str, str] = {}
    while True:
        try:
            raw = handle.readline()
        except TimeoutError:
            return {}
        except socket.timeout:
            return {}
        if raw == b"":
            return fields or None
        line = raw.decode("utf-8", errors="replace").strip("\r\n")
        if not line:
            return fields or {}
        if ":" not in line:
            continue
        key, value = line.split(":", 1)
        fields[key.strip()] = value.strip()


def send_ami_action(handle, fields: dict[str, str]) -> None:
    chunks = [f"{key}: {value}\r\n".encode("utf-8") for key, value in fields.items()]
    chunks.append(b"\r\n")
    handle.write(b"".join(chunks))
    handle.flush()


def ami_login(handle) -> None:
    bridge = _bridge_app()
    send_ami_action(
        handle,
        {
            "Action": "Login",
            "Username": bridge._ami_username(),
            "Secret": bridge._ami_secret(),
            "Events": "on",
        },
    )
    while True:
        frame = read_ami_frame(handle)
        if frame is None:
            raise RuntimeError("AMI connection closed during login")
        if not frame:
            continue
        if frame.get("Response"):
            if frame.get("Response", "").lower() != "success":
                raise RuntimeError(frame.get("Message", "AMI login failed"))
            return


def list_channels_via_coreshowchannels(*, call_id: str, linked_id: str | None) -> list[str] | None:
    bridge = _bridge_app()
    sock = None
    handle = None
    action_id = new_id("ami")
    try:
        sock = socket.create_connection((bridge._ami_host(), bridge._ami_port()), timeout=10)
        sock.settimeout(bridge._callcontrol_action_timeout_seconds())
        handle = sock.makefile("rwb")
        ami_login(handle)
        send_ami_action(
            handle,
            {
                "Action": "CoreShowChannels",
                "ActionID": action_id,
            },
        )

        candidates: list[str] = []
        started_at = time.time()
        while True:
            if (time.time() - started_at) > bridge._callcontrol_action_timeout_seconds():
                break
            frame = read_ami_frame(handle)
            if frame is None:
                break
            if not frame:
                continue
            if frame.get("Event") == "CoreShowChannel":
                uniqueid = first_non_empty(frame.get("Uniqueid"), frame.get("UniqueID"))
                frame_linked = first_non_empty(frame.get("Linkedid"), frame.get("LinkedID"))
                channel = first_non_empty(frame.get("Channel"))
                if not channel:
                    continue
                if uniqueid == call_id or (linked_id and frame_linked == linked_id) or frame_linked == call_id:
                    candidates.append(channel)
            if frame.get("Event") == "CoreShowChannelsComplete":
                break

        return candidates
    except Exception:
        return None
    finally:
        try:
            if handle is not None:
                handle.close()
        except Exception:
            pass
        try:
            if sock is not None:
                sock.close()
        except Exception:
            pass


def resolve_channel_via_coreshowchannels(*, call_id: str, linked_id: str | None) -> str | None:
    candidates = list_channels_via_coreshowchannels(call_id=call_id, linked_id=linked_id)
    if not candidates:
        return None
    for channel in candidates:
        if channel.startswith("PJSIP/"):
            return channel
    return candidates[0]


def extract_extension_from_channel(channel: str | None) -> str | None:
    value = str(channel or "").strip()
    if not value or "/" not in value:
        return None
    body = value.split("/", 1)[1]
    if "-" in body:
        body = body.split("-", 1)[0]
    return body.strip() or None


def operator_channel_for_extension(channels: list[str], operator_extension: str | None) -> str | None:
    prefix = str(operator_extension or "").strip()
    if not prefix:
        return None
    needle = f"PJSIP/{prefix}-"
    for channel in channels:
        if channel.startswith(needle):
            return channel
    return None


@dataclass(slots=True)
class CallChannelSnapshot:
    live_channels: list[str]
    lookup_available: bool
    operator_extension: str | None
    cached_channel: str | None
    operator_channel: str | None
    connected_channel: str | None
    started_channel: str | None


def latest_event_channel(session, *, call_id: str, event_name: str) -> str | None:
    row = session.execute(
        select(AsteriskEventLogRow)
        .where(AsteriskEventLogRow.call_id == call_id)
        .where(AsteriskEventLogRow.ami_event_name == event_name)
        .order_by(AsteriskEventLogRow.id.desc())
        .limit(1)
    ).scalar_one_or_none()
    if row is None:
        return None
    try:
        payload = json.loads(row.payload_json or "{}")
    except Exception:
        payload = {}
    return resolve_channel_from_payload(payload)


def build_call_channel_snapshot(
    session,
    link: AsteriskCallLinkRow,
    *,
    operator_extension: str | None = None,
) -> CallChannelSnapshot:
    bridge = _bridge_app()
    normalized_operator_extension = str(operator_extension or link.operator_extension or "").strip() or None
    resolved_live_channels = bridge._list_channels_via_coreshowchannels(call_id=link.call_id, linked_id=link.linked_id)
    live_channels = resolved_live_channels or []
    return CallChannelSnapshot(
        live_channels=live_channels,
        lookup_available=resolved_live_channels is not None,
        operator_extension=normalized_operator_extension,
        cached_channel=str(link.channel_name or "").strip() or None,
        operator_channel=operator_channel_for_extension(live_channels, normalized_operator_extension),
        connected_channel=latest_event_channel(session, call_id=link.call_id, event_name=f"{bridge._ami_prefix()}OperatorConnected"),
        started_channel=latest_event_channel(session, call_id=link.call_id, event_name=f"{bridge._ami_prefix()}CallStarted"),
    )


def candidate_channels_for_call(
    session,
    link: AsteriskCallLinkRow,
    *,
    operator_extension: str | None = None,
    prefer_operator: bool = False,
    snapshot: CallChannelSnapshot | None = None,
) -> list[str]:
    candidates: list[str] = []

    def _add(channel: str | None) -> None:
        value = str(channel or "").strip()
        if value and value not in candidates:
            candidates.append(value)

    snapshot = snapshot or build_call_channel_snapshot(session, link, operator_extension=operator_extension)
    operator_extension = snapshot.operator_extension

    if prefer_operator:
        _add(snapshot.operator_channel)
        if extract_extension_from_channel(snapshot.connected_channel) == operator_extension:
            _add(snapshot.connected_channel)

    _add(snapshot.cached_channel)
    _add(snapshot.operator_channel)
    _add(snapshot.connected_channel)
    _add(snapshot.started_channel)
    for channel in snapshot.live_channels:
        _add(channel)
    return candidates


def resolve_channel_name(
    session,
    link: AsteriskCallLinkRow,
    *,
    prefer_operator: bool = False,
    refresh: bool = False,
    snapshot: CallChannelSnapshot | None = None,
) -> str | None:
    bridge = _bridge_app()
    snapshot = snapshot or build_call_channel_snapshot(session, link)
    cached_channel = snapshot.cached_channel
    live_channels = snapshot.live_channels
    lookup_available = snapshot.lookup_available
    operator_channel = snapshot.operator_channel
    connected_channel = snapshot.connected_channel

    if not refresh and cached_channel and (not lookup_available or cached_channel in live_channels):
        if prefer_operator and operator_channel:
            cached_channel = operator_channel
        if cached_channel != link.channel_name:
            link.channel_name = cached_channel
            link.updated_at = utc_now_iso()
        return cached_channel

    connected_channel = latest_event_channel(session, call_id=link.call_id, event_name=f"{bridge._ami_prefix()}OperatorConnected")
    if connected_channel and (not lookup_available or connected_channel in live_channels):
        cached_channel = connected_channel
    elif prefer_operator and operator_channel:
        cached_channel = operator_channel
    elif cached_channel and lookup_available and live_channels and cached_channel in live_channels:
        pass
    elif operator_channel:
        cached_channel = operator_channel
    else:
        started_channel = latest_event_channel(session, call_id=link.call_id, event_name=f"{bridge._ami_prefix()}CallStarted")
        if started_channel and (not lookup_available or started_channel in live_channels):
            cached_channel = started_channel
        elif lookup_available and live_channels:
            cached_channel = bridge._resolve_channel_via_coreshowchannels(call_id=link.call_id, linked_id=link.linked_id)

    if cached_channel:
        link.channel_name = cached_channel
        link.updated_at = utc_now_iso()
    return cached_channel


def ami_action(action: str, params: dict[str, Any]) -> dict[str, Any]:
    bridge = _bridge_app()
    if not (bridge._ami_host() and bridge._ami_username() and bridge._ami_secret()):
        raise RuntimeError("AMI host/credentials are not configured")

    action_id = new_id("ami")
    sock = None
    handle = None
    try:
        sock = socket.create_connection((bridge._ami_host(), bridge._ami_port()), timeout=10)
        sock.settimeout(bridge._callcontrol_action_timeout_seconds())
        handle = sock.makefile("rwb")
        ami_login(handle)

        frame_fields = {
            "Action": action,
            "ActionID": action_id,
        }
        for key, value in params.items():
            if value is None:
                continue
            frame_fields[str(key)] = str(value)
        send_ami_action(handle, frame_fields)

        started_at = time.time()
        while True:
            if (time.time() - started_at) > bridge._callcontrol_action_timeout_seconds():
                raise RuntimeError(f"AMI action timeout: {action}")
            frame = read_ami_frame(handle)
            if frame is None:
                raise RuntimeError("AMI action connection closed")
            if not frame:
                continue
            if frame.get("Response"):
                if frame.get("ActionID") and frame.get("ActionID") != action_id:
                    continue
                if frame.get("Response", "").lower() != "success":
                    raise RuntimeError(frame.get("Message", f"AMI {action} failed"))
                return {
                    "action_id": action_id,
                    "response": frame,
                }
    finally:
        try:
            if handle is not None:
                handle.close()
        except Exception:
            pass
        try:
            if sock is not None:
                sock.close()
        except Exception:
            pass


def ami_loop(stop_event=None) -> None:
    bridge = _bridge_app()
    active_stop_event = stop_event or bridge._background_stop_event()
    while not active_stop_event.is_set():
        if not bridge._bridge_enabled():
            bridge._STATE.set_connected(False)
            if active_stop_event.wait(bridge._poll_interval()):
                break
            continue

        if not (bridge._ami_host() and bridge._ami_username() and bridge._ami_secret()):
            bridge._STATE.set_connected(False)
            bridge._STATE.set_error("AMI host/credentials are not configured")
            if active_stop_event.wait(max(bridge._poll_interval(), 2)):
                break
            continue

        sock = None
        handle = None
        try:
            sock = socket.create_connection((bridge._ami_host(), bridge._ami_port()), timeout=10)
            sock.settimeout(10)
            handle = sock.makefile("rwb")
            ami_login(handle)
            bridge._STATE.set_connected(True)

            while not active_stop_event.is_set():
                if bridge._STATE.take_reconnect():
                    raise RuntimeError("Reconnect requested")
                frame = read_ami_frame(handle)
                if frame is None:
                    raise RuntimeError("AMI connection closed")
                if not frame:
                    continue
                event_name = frame.get("Event")
                if event_name == "UserEvent":
                    user_event = str(frame.get("UserEvent") or "").strip()
                    if not user_event.startswith(bridge._ami_prefix()):
                        continue
                    bridge._STATE.set_last_event()
                    bridge._record_ami_payload(frame)
                elif event_name in {"DialEnd", "Hangup"}:
                    bridge._STATE.set_last_event()
                    bridge._record_ami_payload(frame, event_name=event_name)
                else:
                    continue
        except Exception as exc:
            bridge._STATE.set_connected(False)
            bridge._STATE.set_error(str(exc))
            if active_stop_event.wait(max(bridge._poll_interval(), 2)):
                break
        finally:
            try:
                if handle is not None:
                    handle.close()
            except Exception:
                pass
            try:
                if sock is not None:
                    sock.close()
            except Exception:
                pass