421 lines
17 KiB
Python
421 lines
17 KiB
Python
from __future__ import annotations
|
|
|
|
from typing import Any
|
|
|
|
from fastapi import HTTPException
|
|
from sqlalchemy import select
|
|
|
|
from services.shared.core import Role, utc_now_iso
|
|
from services.shared.db import get_session
|
|
from services.shared.models import VoiceCallBlindTransferIn, VoiceCallClaimIn, VoiceLiveCallOut
|
|
from services.shared.sql_models import AsteriskCallActionLogRow, AsteriskCallLinkRow
|
|
|
|
|
|
def _bridge_app():
|
|
import services.asterisk_bridge_service.app as bridge_app
|
|
|
|
return bridge_app
|
|
|
|
|
|
def load_live_call_or_404(session, call_id: str) -> AsteriskCallLinkRow:
|
|
bridge = _bridge_app()
|
|
link = bridge._find_call_link(session, call_id)
|
|
if link is None:
|
|
raise HTTPException(status_code=404, detail="Live call not found")
|
|
return link
|
|
|
|
|
|
def call_is_ended(link: AsteriskCallLinkRow) -> bool:
|
|
if str(link.status or "").strip().lower() == "ended":
|
|
return True
|
|
return str(link.telephony_status or "").strip().lower() == "ended"
|
|
|
|
|
|
def resolve_operator_extension(*, actor: dict, requested_extension: str | None = None) -> str:
|
|
bridge = _bridge_app()
|
|
actor_user = str(actor.get("user") or "").strip()
|
|
actor_role = str(actor.get("role") or "").strip().lower()
|
|
mapping = bridge._operator_extension_map()
|
|
mapped = mapping.get(actor_user)
|
|
requested = str(requested_extension or "").strip()
|
|
|
|
if actor_role == Role.OPERATOR.value:
|
|
if not mapped:
|
|
raise HTTPException(status_code=400, detail=f"No extension mapping for operator '{actor_user}'")
|
|
if requested and requested != mapped:
|
|
raise HTTPException(status_code=400, detail="Operator extension override is not allowed")
|
|
return mapped
|
|
|
|
if requested:
|
|
return requested
|
|
if mapped:
|
|
return mapped
|
|
raise HTTPException(status_code=400, detail="operator_extension is required for this role")
|
|
|
|
|
|
def resolve_transfer_extension(*, target_type: str, target_value: str) -> str:
|
|
bridge = _bridge_app()
|
|
raw_value = str(target_value or "").strip()
|
|
if not raw_value:
|
|
raise HTTPException(status_code=400, detail="target_value is required")
|
|
if target_type == "extension":
|
|
return raw_value
|
|
mapped = bridge._transfer_target_map().get(raw_value)
|
|
if not mapped:
|
|
raise HTTPException(status_code=400, detail=f"Unknown transfer queue_code: {raw_value}")
|
|
return mapped
|
|
|
|
|
|
def assert_callcontrol_enabled() -> None:
|
|
bridge = _bridge_app()
|
|
if not bridge._callcontrol_enabled():
|
|
raise HTTPException(status_code=409, detail="Call control is disabled")
|
|
|
|
|
|
def assert_call_control_permissions(*, actor: dict, link: AsteriskCallLinkRow, action_type: str) -> None:
|
|
bridge = _bridge_app()
|
|
role = str(actor.get("role") or "").strip().lower()
|
|
user = str(actor.get("user") or "").strip()
|
|
if role != Role.OPERATOR.value:
|
|
return
|
|
claimed_by = str(link.claimed_by_user or "").strip()
|
|
if action_type == "claim":
|
|
if claimed_by and claimed_by != user:
|
|
raise HTTPException(status_code=403, detail="Call already claimed by another operator")
|
|
return
|
|
mapped_extension = bridge._operator_extension_map().get(user)
|
|
if mapped_extension and str(link.operator_extension or "").strip() == mapped_extension:
|
|
return
|
|
if claimed_by != user:
|
|
raise HTTPException(status_code=403, detail="Operator can control only own claimed call")
|
|
|
|
|
|
def claim_live_call(call_id: str, body: VoiceCallClaimIn, actor: dict) -> VoiceLiveCallOut:
|
|
bridge = _bridge_app()
|
|
assert_callcontrol_enabled()
|
|
session = get_session()
|
|
action: AsteriskCallActionLogRow | None = None
|
|
try:
|
|
link = load_live_call_or_404(session, call_id)
|
|
assert_call_control_permissions(actor=actor, link=link, action_type="claim")
|
|
if call_is_ended(link):
|
|
raise HTTPException(status_code=409, detail="Call is already ended")
|
|
if not link.interaction_id:
|
|
raise HTTPException(status_code=409, detail="Call has no linked interaction")
|
|
|
|
actor_user = str(actor.get("user") or "").strip() or "unknown"
|
|
actor_role = str(actor.get("role") or "").strip() or "unknown"
|
|
operator_extension = resolve_operator_extension(actor=actor, requested_extension=body.operator_extension)
|
|
action = bridge._create_action_log(
|
|
session,
|
|
call_id=call_id,
|
|
interaction_id=link.interaction_id,
|
|
action_type="claim",
|
|
actor_user=actor_user,
|
|
actor_role=actor_role,
|
|
request_payload={"operator_extension": operator_extension},
|
|
)
|
|
|
|
if link.claimed_by_user == actor_user and link.telephony_status in {"claimed", "connected"}:
|
|
bridge._set_action_result(
|
|
session,
|
|
action,
|
|
result_status="ok",
|
|
error="already_claimed_by_actor",
|
|
)
|
|
session.commit()
|
|
session.refresh(link)
|
|
return bridge._live_call_to_out(session, link)
|
|
|
|
snapshot = bridge._build_call_channel_snapshot(session, link, operator_extension=operator_extension)
|
|
operator_channel = snapshot.operator_channel
|
|
connected_channel = snapshot.connected_channel
|
|
already_connected = (
|
|
link.telephony_status == "connected"
|
|
or (connected_channel is not None and bridge._extract_extension_from_channel(connected_channel) == operator_extension)
|
|
)
|
|
channel = operator_channel or connected_channel or bridge._resolve_channel_name(session, link, refresh=True, snapshot=snapshot)
|
|
if not channel:
|
|
raise HTTPException(status_code=409, detail="Unable to resolve active channel for call")
|
|
|
|
ami_result: dict[str, Any] | None = None
|
|
claim_warning: str | None = None
|
|
if already_connected:
|
|
claim_warning = "already_connected_to_operator"
|
|
elif not operator_channel:
|
|
source_channel = None
|
|
for candidate in bridge._candidate_channels_for_call(
|
|
session,
|
|
link,
|
|
operator_extension=operator_extension,
|
|
prefer_operator=False,
|
|
snapshot=snapshot,
|
|
):
|
|
if bridge._extract_extension_from_channel(candidate) != operator_extension:
|
|
source_channel = candidate
|
|
break
|
|
if not source_channel:
|
|
source_channel = bridge._resolve_channel_name(session, link, refresh=True, snapshot=snapshot)
|
|
if not source_channel:
|
|
raise HTTPException(status_code=409, detail="Unable to resolve source channel for claim")
|
|
try:
|
|
ami_result = bridge._ami_action(
|
|
"Redirect",
|
|
{
|
|
"Channel": source_channel,
|
|
"Context": bridge._claim_context(),
|
|
"Exten": operator_extension,
|
|
"Priority": 1,
|
|
},
|
|
)
|
|
except Exception as exc:
|
|
refresh_snapshot = bridge._build_call_channel_snapshot(session, link, operator_extension=operator_extension)
|
|
if refresh_snapshot.operator_channel:
|
|
channel = refresh_snapshot.operator_channel
|
|
elif refresh_snapshot.connected_channel and bridge._extract_extension_from_channel(refresh_snapshot.connected_channel) == operator_extension:
|
|
channel = refresh_snapshot.connected_channel
|
|
else:
|
|
raise
|
|
claim_warning = f"redirect_skipped: {exc}"
|
|
|
|
now = utc_now_iso()
|
|
link.claimed_by_user = actor_user
|
|
link.claimed_at = now
|
|
link.operator_extension = operator_extension
|
|
link.channel_name = channel
|
|
link.telephony_status = "connected" if already_connected else "claimed"
|
|
link.updated_at = now
|
|
|
|
assign_error: str | None = None
|
|
try:
|
|
bridge._assign_interaction(interaction_id=link.interaction_id, assignee=actor_user)
|
|
except Exception as exc:
|
|
assign_error = str(exc)
|
|
action_error = "; ".join(part for part in [claim_warning, assign_error] if part) or None
|
|
|
|
bridge._set_action_result(
|
|
session,
|
|
action,
|
|
result_status="ok",
|
|
ami_action_id=str((ami_result or {}).get("action_id") or "") or None,
|
|
error=action_error,
|
|
)
|
|
session.commit()
|
|
session.refresh(link)
|
|
return bridge._live_call_to_out(session, link)
|
|
except HTTPException as exc:
|
|
if action is not None:
|
|
bridge._set_action_result(session, action, result_status="rejected", error=exc.detail if isinstance(exc.detail, str) else str(exc.detail))
|
|
session.commit()
|
|
raise
|
|
except Exception as exc:
|
|
if action is not None:
|
|
bridge._set_action_result(session, action, result_status="failed", error=str(exc))
|
|
session.commit()
|
|
raise HTTPException(status_code=502, detail=f"Claim failed: {exc}") from exc
|
|
finally:
|
|
session.close()
|
|
|
|
|
|
def hangup_live_call(call_id: str, actor: dict) -> VoiceLiveCallOut:
|
|
bridge = _bridge_app()
|
|
assert_callcontrol_enabled()
|
|
session = get_session()
|
|
action: AsteriskCallActionLogRow | None = None
|
|
try:
|
|
link = load_live_call_or_404(session, call_id)
|
|
assert_call_control_permissions(actor=actor, link=link, action_type="hangup")
|
|
if call_is_ended(link):
|
|
raise HTTPException(status_code=409, detail="Call is already ended")
|
|
|
|
actor_user = str(actor.get("user") or "").strip() or "unknown"
|
|
actor_role = str(actor.get("role") or "").strip() or "unknown"
|
|
action = bridge._create_action_log(
|
|
session,
|
|
call_id=call_id,
|
|
interaction_id=link.interaction_id,
|
|
action_type="hangup",
|
|
actor_user=actor_user,
|
|
actor_role=actor_role,
|
|
request_payload={},
|
|
)
|
|
snapshot = bridge._build_call_channel_snapshot(session, link, operator_extension=link.operator_extension)
|
|
candidates = bridge._candidate_channels_for_call(
|
|
session,
|
|
link,
|
|
operator_extension=link.operator_extension,
|
|
prefer_operator=True,
|
|
snapshot=snapshot,
|
|
)
|
|
if not candidates:
|
|
raise HTTPException(status_code=409, detail="Unable to resolve active channel for call")
|
|
ami_result = None
|
|
channel = None
|
|
successful_channels: list[str] = []
|
|
last_error = None
|
|
for candidate in candidates:
|
|
try:
|
|
result = bridge._ami_action("Hangup", {"Channel": candidate})
|
|
if ami_result is None:
|
|
ami_result = result
|
|
channel = candidate
|
|
successful_channels.append(candidate)
|
|
except Exception as exc:
|
|
last_error = exc
|
|
message = str(exc)
|
|
if "No such channel" in message or "Channel does not exist" in message:
|
|
continue
|
|
raise
|
|
if not successful_channels or ami_result is None or channel is None:
|
|
remaining_channels = bridge._build_call_channel_snapshot(
|
|
session,
|
|
link,
|
|
operator_extension=link.operator_extension,
|
|
).live_channels
|
|
if not remaining_channels:
|
|
now = utc_now_iso()
|
|
link.telephony_status = "ended"
|
|
link.status = "ended"
|
|
link.ended_at = link.ended_at or now
|
|
link.updated_at = now
|
|
bridge._set_action_result(
|
|
session,
|
|
action,
|
|
result_status="ok",
|
|
error="channel_already_closed",
|
|
)
|
|
session.commit()
|
|
session.refresh(link)
|
|
return bridge._live_call_to_out(session, link)
|
|
raise RuntimeError(str(last_error or "No active channel available for hangup"))
|
|
|
|
now = utc_now_iso()
|
|
link.telephony_status = "ended"
|
|
link.status = "ended"
|
|
link.ended_at = link.ended_at or now
|
|
link.channel_name = successful_channels[0]
|
|
link.updated_at = now
|
|
bridge._set_action_result(
|
|
session,
|
|
action,
|
|
result_status="ok",
|
|
ami_action_id=str(ami_result.get("action_id") or "") or None,
|
|
)
|
|
session.commit()
|
|
session.refresh(link)
|
|
return bridge._live_call_to_out(session, link)
|
|
except HTTPException as exc:
|
|
if action is not None:
|
|
bridge._set_action_result(session, action, result_status="rejected", error=exc.detail if isinstance(exc.detail, str) else str(exc.detail))
|
|
session.commit()
|
|
raise
|
|
except Exception as exc:
|
|
if action is not None:
|
|
bridge._set_action_result(session, action, result_status="failed", error=str(exc))
|
|
session.commit()
|
|
raise HTTPException(status_code=502, detail=f"Hangup failed: {exc}") from exc
|
|
finally:
|
|
session.close()
|
|
|
|
|
|
def blind_transfer_live_call(call_id: str, body: VoiceCallBlindTransferIn, actor: dict) -> VoiceLiveCallOut:
|
|
bridge = _bridge_app()
|
|
assert_callcontrol_enabled()
|
|
session = get_session()
|
|
action: AsteriskCallActionLogRow | None = None
|
|
try:
|
|
link = load_live_call_or_404(session, call_id)
|
|
assert_call_control_permissions(actor=actor, link=link, action_type="blind-transfer")
|
|
if call_is_ended(link):
|
|
raise HTTPException(status_code=409, detail="Call is already ended")
|
|
|
|
actor_user = str(actor.get("user") or "").strip() or "unknown"
|
|
actor_role = str(actor.get("role") or "").strip() or "unknown"
|
|
target_extension = resolve_transfer_extension(target_type=body.target_type, target_value=body.target_value)
|
|
action = bridge._create_action_log(
|
|
session,
|
|
call_id=call_id,
|
|
interaction_id=link.interaction_id,
|
|
action_type="blind-transfer",
|
|
actor_user=actor_user,
|
|
actor_role=actor_role,
|
|
request_payload={
|
|
"target_type": body.target_type,
|
|
"target_value": body.target_value,
|
|
"resolved_extension": target_extension,
|
|
},
|
|
)
|
|
|
|
snapshot = bridge._build_call_channel_snapshot(session, link, operator_extension=link.operator_extension)
|
|
channel = bridge._resolve_channel_name(session, link, prefer_operator=True, refresh=True, snapshot=snapshot)
|
|
if not channel:
|
|
raise HTTPException(status_code=409, detail="Unable to resolve active channel for call")
|
|
ami_result = bridge._ami_action(
|
|
"Redirect",
|
|
{
|
|
"Channel": channel,
|
|
"Context": bridge._transfer_context(),
|
|
"Exten": target_extension,
|
|
"Priority": 1,
|
|
},
|
|
)
|
|
|
|
bridge._emit_voice_event(
|
|
event_type="call.transferred",
|
|
call_id=call_id,
|
|
interaction_id=link.interaction_id,
|
|
payload={
|
|
"source": "asterisk",
|
|
"target_type": body.target_type,
|
|
"target_value": body.target_value,
|
|
"target_extension": target_extension,
|
|
"actor_user": actor_user,
|
|
"linked_id": link.linked_id,
|
|
},
|
|
)
|
|
now = utc_now_iso()
|
|
link.status = "ended"
|
|
link.telephony_status = "ended"
|
|
link.ended_at = link.ended_at or now
|
|
link.updated_at = now
|
|
bridge._set_action_result(
|
|
session,
|
|
action,
|
|
result_status="ok",
|
|
ami_action_id=str(ami_result.get("action_id") or "") or None,
|
|
)
|
|
session.commit()
|
|
session.refresh(link)
|
|
return bridge._live_call_to_out(session, link)
|
|
except HTTPException as exc:
|
|
if action is not None:
|
|
bridge._set_action_result(session, action, result_status="rejected", error=exc.detail if isinstance(exc.detail, str) else str(exc.detail))
|
|
session.commit()
|
|
raise
|
|
except Exception as exc:
|
|
if action is not None:
|
|
bridge._set_action_result(session, action, result_status="failed", error=str(exc))
|
|
session.commit()
|
|
raise HTTPException(status_code=502, detail=f"Blind transfer failed: {exc}") from exc
|
|
finally:
|
|
session.close()
|
|
|
|
|
|
def list_live_call_actions(call_id: str, limit: int, actor: dict) -> list[dict]:
|
|
session = get_session()
|
|
bridge = _bridge_app()
|
|
try:
|
|
role = str(actor.get("role") or "").strip().lower()
|
|
if role == Role.OPERATOR.value:
|
|
link = bridge._find_call_link(session, call_id)
|
|
if link and str(link.claimed_by_user or "").strip() not in {"", str(actor.get("user") or "").strip()}:
|
|
raise HTTPException(status_code=403, detail="Operator can view actions only for own or unclaimed call")
|
|
rows = session.execute(
|
|
select(AsteriskCallActionLogRow)
|
|
.where(AsteriskCallActionLogRow.call_id == call_id)
|
|
.order_by(AsteriskCallActionLogRow.id.desc())
|
|
).scalars().all()
|
|
return [bridge._action_to_out(row) for row in rows[: max(limit, 1)]]
|
|
finally:
|
|
session.close()
|