3cx-telefonie-middleware/app/service.py

import asyncio
import json
import logging
from datetime import datetime, timezone
import logging
from app.call_context import CallContextService
from app.call_context_repository import CallContextRepository
from app.kontor_mcp import KontorMCPClient

logger = logging.getLogger(__name__)

from .config import Settings
from .repository import Repository
from .threecx import ThreeCXClient

log = logging.getLogger(__name__)

class TelephonyService:
    def __init__(self, settings: Settings):
        self.settings = settings

        self.repo = Repository(
            settings.database_path
        )

        self.cx = ThreeCXClient(
            settings
        )

        # Long-lived Kontor MCP client.
        self.kontor_mcp = KontorMCPClient(
            url=settings.kontor_mcp_url,
            token=settings.kontor_mcp_token,
            timeout=settings.kontor_mcp_timeout,
        )

        # Generic customer/business context service.
        self.call_context = CallContextService(
            self.kontor_mcp
        )

        # Persisted enrichment snapshots.
        self._call_context_repository = (
            CallContextRepository()
        )

        # One enrichment task per (callid, legid).
        self._call_context_tasks = {}

        self._active_participants = []
        self._active_observed_at = None
        self._last_logged_active_state = None

    async def start(self):
        await self.repo.init()

        # Initialize the long-lived MCP session once.
        await self.kontor_mcp.initialize()

        asyncio.create_task(
            self._snapshot_loop()
        )

        asyncio.create_task(
            self.cx.run_websocket(
                self._on_event
            )
        )

    async def _on_event(self, event):
        log.debug("3CX event: %s", event)
        await self._process_queue()

    async def _snapshot_loop(self):
        while True:
            try:
                await self._process_queue()
            except Exception:
                log.exception("Queue snapshot failed")

            try:
                await self._process_outbound()
            except Exception:
                log.exception("Outbound tracking failed")

            await asyncio.sleep(2)

    def get_active_participants(self):
        return list(self._active_participants)


    async def _process_outbound(self):
        """
        Verfolgt ausschließlich von der Outbound-API gestartete Calls.

        Ein einzelner Participant-Request wird pro source_dn ausgeführt.
        Fehler bei 3CX werden NICHT als Call-Ende interpretiert.
        """

        calls = await self.repo.list_active_outbound_calls()

        if not calls:
            return

        now = datetime.now(timezone.utc)
        now_iso = now.isoformat()

        # Nur die tatsächlich benötigten DNs abfragen.
        dns = sorted({
            str(call["source_dn"])
            for call in calls
            if call.get("source_dn")
        })

        participants_by_dn = {}

        for dn in dns:
            try:
                participants_by_dn[dn] = await self.cx.get_participants(dn)
            except Exception:
                # Ganz wichtig:
                # Bei einem 3CX-/Netzwerkfehler niemals den Call
                # fälschlich als beendet markieren.
                log.exception(
                    "Outbound participant lookup failed for DN %s",
                    dn,
                )

        for call in calls:
            source_dn = str(call["source_dn"])
            participants = participants_by_dn.get(source_dn)

            # Wenn die Abfrage für diesen DN fehlgeschlagen ist,
            # bleibt der Call unverändert.
            if participants is None:
                continue

            callid = str(call["callid"])
            legid = str(call["legid"]) if call["legid"] is not None else None

            participant = None

            for p in participants:
                if str(p.get("callid")) != callid:
                    continue

                if legid is not None and str(p.get("legid")) != legid:
                    continue

                participant = p
                break

            if participant is None:
                # Call ist nicht mehr bei 3CX aktiv.
                started = None
                answered = None

                try:
                    if call.get("started_at"):
                        started = datetime.fromisoformat(
                            call["started_at"].replace("Z", "+00:00")
                        )

                    if call.get("answered_at"):
                        answered = datetime.fromisoformat(
                            call["answered_at"].replace("Z", "+00:00")
                        )
                except Exception:
                    log.exception(
                        "Could not parse timestamps for outbound call %s",
                        call["id"],
                    )

                duration = None

                if answered:
                    duration = max(
                        0,
                        int((now - answered).total_seconds()),
                    )

                final_status = "ended" if answered else "failed"

                await self.repo.finalize_outbound_call(
                    call_id=call["id"],
                    ended_at=now_iso,
                    status=final_status,
                    duration_seconds=duration,
                )

                log.info(
                    "OUTBOUND_CALL_ENDED callid=%s legid=%s status=%s duration=%s",
                    callid,
                    legid,
                    final_status,
                    duration,
                )

                continue

            status = str(participant.get("status") or "").lower()

            if status == "connected":
                if not call.get("answered_at"):
                    await self.repo.update_outbound_connected(
                        call_id=call["id"],
                        answered_at=now_iso,
                        last_seen_at=now_iso,
                    )

                    log.info(
                        "OUTBOUND_CALL_CONNECTED callid=%s legid=%s source=%s destination=%s",
                        callid,
                        legid,
                        source_dn,
                        call.get("destination"),
                    )
                else:
                    await self.repo.update_outbound_seen(
                        call_id=call["id"],
                        status="connected",
                        last_seen_at=now_iso,
                    )

            elif status:
                await self.repo.update_outbound_seen(
                    call_id=call["id"],
                    status=status,
                    last_seen_at=now_iso,
                )

    async def _enrich_and_persist_call_context(
        self,
        participant,
    ):
        """
        Enrich one external inbound call with Kontor MCP context
        and persist the resulting snapshot.

        This task is deliberately detached from the 3CX event loop.
        """

        callid = participant.get("callid")
        legid = participant.get("legid")
        raw_phone = participant.get("party_caller_id")

        if not callid or not raw_phone:
            return

        key = (
            str(callid),
            str(legid) if legid is not None else None,
        )

        try:
            # Existing middleware phone normalization.
            e164 = self.repo.normalize_phone(
                raw_phone
            )

            if not e164:
                logger.warning(
                    "CALL_CONTEXT invalid phone callid=%s legid=%s raw=%s",
                    callid,
                    legid,
                    raw_phone,
                )
                return

            context = await self.call_context.enrich_phone(
                e164
            )

            self._call_context_repository.upsert(
                callid=str(callid),
                legid=(
                    str(legid)
                    if legid is not None
                    else None
                ),
                phone_e164=e164,
                context=context,
            )

            logger.info(
                "CALL_CONTEXT persisted callid=%s legid=%s phone=%s status=%s",
                callid,
                legid,
                e164,
                context.get("status"),
            )

        except Exception:
            logger.exception(
                "CALL_CONTEXT enrichment failed callid=%s legid=%s",
                callid,
                legid,
            )

        finally:
            self._call_context_tasks.pop(
                key,
                None,
            )

    async def _process_queue(self):
        data = await self.cx.get_dn(self.settings.threecx_queue_dn)
        participants = data.get("participants", [])
        self._active_participants = participants
        self._active_observed_at = asyncio.get_running_loop().time()

        external = [
            p for p in participants
            if p.get("party_dn_type") == "Wexternalline"
        ]

        # Enrich each new external call exactly once.
        for participant in external:
            callid = participant.get("callid")
            legid = participant.get("legid")

            if not callid:
                continue

            key = (
                str(callid),
                str(legid) if legid is not None else None,
            )

            if key in self._call_context_tasks:
                continue

            task = asyncio.create_task(
                self._enrich_and_persist_call_context(
                    participant
                )
            )

            self._call_context_tasks[key] = task

        if external:
            state = []
            for p in external:
                state.append({
                    "callId": p.get("callid"),
                    "legId": p.get("legid"),
                    "status": p.get("status"),
                    "caller": p.get("party_caller_id"),
                    "partyDn": p.get("party_dn"),
                    "deviceId": p.get("device_id"),
                })

            state.sort(
                key=lambda x: (
                    str(x.get("callId")),
                    str(x.get("legId")),
                )
            )

            fingerprint = json.dumps(
                state,
                sort_keys=True,
                ensure_ascii=False,
                separators=(",", ":"),
            )

            if fingerprint != self._last_logged_active_state:
                logger.info(
                    "CALL_STATE %s",
                    json.dumps(
                        {
                            "queue": self.settings.threecx_queue_dn,
                            "calls": state,
                        },
                        ensure_ascii=False,
                        separators=(",", ":"),
                    ),
                )
                self._last_logged_active_state = fingerprint
        else:
            self._last_logged_active_state = None

        active = await self.repo.observe_queue(
            participants,
            self.settings.threecx_queue_dn,
            self.settings.threecx_monitored_extensions,
        )
        await self.repo.finalize_disappeared(self.settings.threecx_queue_dn, active)