voice-gateway-service/src/core/callcenter/esl/client.py

#!/usr/bin/env python3
# encoding:utf-8

import json
import random
import socketserver

import ESL
import time
import mmh3
import threading
import traceback
import concurrent.futures
import src.core.callcenter.cache as Cache
from src.core.callcenter.constant import SK, EMPTY
from src.core.callcenter.esl.constant.esl_constant import BRIDGE_VARIABLES, BRIDGE, HANGUP, NORMAL_CLEARING, SIP_HEADER, SPACE, SPLIT, SOFIA, \
    ORIGINATE, PARK, SET, EAVESDROP, SMF_ALEG, EXECUTE, PLAYBACK, PAUSE, TRANSFER, UUID_TRANSFER, UUID_BROADCAST, UUID_BREAK, UUID_HOLD, \
    UUID_RECORD, UUID_SETVAR, UUID_GETVAR
import src.core.callcenter.esl.utils.esl_event_util as EslEventUtil
import src.core.callcenter.esl.handler as event_handler
from src.core.callcenter.esl.constant.sip_header_constant import sipHeaderHoldMusic
from src.core.callcenter.enumeration import CallCause
from src.core.callcenter.esl.handler.default_esl_event_handler import DefaultEslEventHandler
from src.core.voip.constant import *


class InboundClient:

    def __init__(self, agent, logger):
        self.con = None
        self.thread_num = 32
        self.is_stopping = False
        self.logger = logger
        self.bot_agent = agent
        self.handler_table = self.scan_esl_event_handlers()
        self.default_event_handler = DefaultEslEventHandler(self, self.bot_agent, self.logger)
        self.host, self.port, self.password = SERVE_HOST, '8021', '4918257983818884358'
        self.executors = {x: concurrent.futures.ThreadPoolExecutor(max_workers=1) for x in range(self.thread_num)}

        threading.Thread(target=self.start, args=()).start()
        # gw = CacheUtil.getRouteGetway(saasId)
        # self.make_call(gw, '63366692', '13241676588', 'C111111111', 'D1111111')

    def scan_esl_event_handlers(self):
        import inspect
        import importlib
        import pkgutil

        classes = []
        # 遍历包中的模块
        for module_info in pkgutil.iter_modules(event_handler.__path__):
            module = importlib.import_module(f"{event_handler.__name__}.{module_info.name}")
            for _name, _cls in inspect.getmembers(module, inspect.isclass):
                if hasattr(_cls, '_esl_event_name'):
                    classes.append(_cls)

        handlers = {}
        for _cls in classes:
            items = handlers.get(_cls._esl_event_name, [])
            items.append(_cls(self, self.bot_agent, self.logger))
            handlers[_cls._esl_event_name] = items
        return handlers

    def start(self):
        self.logger.info('inbound.start')
        self.con = ESL.ESLconnection(self.host, self.port, self.password)
        if self.con.connected():
            self.logger.info('inbound esl connected ... ')
            self.con.events('plain', 'all') #CHANNEL_ORIGINATE,CHANNEL_PROGRESS,CHANNEL_PROGRESS_MEDIA,CHANNEL_ANSWER,CHANNEL_HANGUP,CUSTOM,PLAYBACK_START,PLAYBACK_STOP,DETECTED_TONE
            while not self.is_stopping:
                e = self.con.recvEvent()
                # if e:
                #     self.logger.info(json.loads(e.serialize('json')))
                event_name = e.getHeader("Event-Name")
                if event_name == "SERVER_DISCONNECTED":
                    self.logger.info('come in SERVER_DISCONNECTED case')
                    self.con.disconnect()
                    time.sleep(3)
                    self.start()
                else:
                    # threading.Thread(target=self.process_esl_event, args=(e,)).start()
                    self.choose_thread_pool_executor(e).submit(self.process_esl_event, e)

    def choose_thread_pool_executor(self, e):
        call_id = EslEventUtil.getCallId(e)
        device_id = EslEventUtil.getUniqueId(e)
        wdh_device_id = EslEventUtil.getDeviceId(e)
        random_id = call_id if call_id else device_id
        if random_id:
            random_index = abs(mmh3.hash(random_id)) % len(self.executors)
        else:
            random_index = random.randint(0, len(self.executors) - 1) if self.executors else 0
        print('choose_thread_pool_executor.index=', random_index, call_id, device_id, wdh_device_id) 
        return self.executors.get(random_index)

    def process_esl_event(self, e):
        # print(json.loads(e.serialize('json')))
        event_name = EslEventUtil.getEventName(e)
        coreUUID = EslEventUtil.getCoreUuid(e)
        address = self.host + ':' + self.port
        try:
            if event_name in self.handler_table:
                items = self.handler_table.get(event_name)
                for x in items:
                    try:
                        x.handle(address, e, coreUUID)
                    except:
                        traceback.print_exc()
            else:
                self.default_event_handler.handle(address, e, coreUUID)
        except:
            traceback.print_exc()

    def make_call(self, route_gateway, display, called, call_id, device_id, timeout=30, originate_timeout=30, *sip_headers):

        # called = f"{called}{AT}{route_gateway.media_host}{CO}{route_gateway.media_port}"
        if route_gateway.caller_prefix:
            display = f"{route_gateway.caller_prefix}{display}"
        if route_gateway.called_prefix:
            called = f"{route_gateway.called_prefix}{called}"

        sip_buffer = []
        if sip_headers:
            sip_buffer = [f"{SIP_HEADER}{header}" for header in sip_headers]

        params = {
            "callId": call_id,
            "deviceId": device_id,
            "caller": display,
            "called": called,
        }

        if route_gateway.sip_header1:
            sip_header1 = self.expression(route_gateway.sip_header1, params)
            sip_buffer.append(f"{SIP_HEADER}{sip_header1}")

        if route_gateway.sip_header2:
            sip_header2 = self.expression(route_gateway.sip_header2, params)
            sip_buffer.append(f"{SIP_HEADER}{sip_header2}")

        if route_gateway.sip_header3:
            sip_buffer.append(f"{SIP_HEADER}{route_gateway.sip_header3}")

        builder = [
            "{return_ring_ready=true,",
            f"sip_contact_user={display},",
            "ring_asr=true,",
            "absolute_codec_string=^^:PCMU:PCMA,",  # Assuming codecs is defined somewhere
            f"origination_caller_id_number={display},",
            f"origination_caller_id_name={display},",
            f"origination_uuid={device_id},",
        ]

        if originate_timeout:
            builder.append(f"originate_timeout={originate_timeout},")

        if sip_buffer:
            builder.append(f"{SPLIT}".join(sip_buffer))

        builder.append("}")
        builder.append(f"{SOFIA}{SK}{route_gateway.profile}{SK}{called}{PARK}")
        cmd = "".join(builder)
        print(cmd)
        self.con.bgapi(ORIGINATE, cmd)

    def call_timeout(self, call_id, device_id, timeout):
        """呼叫超时主动挂机"""
        pass

    def send_args(self, device_id, name, arg):
        msg = ESL.ESLevent("sendmsg", device_id)
        msg.addHeader("call-command", EXECUTE)
        msg.addHeader("execute-app-name", name)
        msg.addHeader("execute-app-arg", arg)
        self.con.sendEvent(msg)

    def bridge_call(self, call_id, device_id1, device_id2):
        """桥接电话"""
        self.multi_set_var(device_id1, BRIDGE_VARIABLES)
        self.multi_set_var(device_id2, BRIDGE_VARIABLES)
        self.con.bgapi(BRIDGE, device_id1 + SPACE + device_id2)

    def transfer_call(self, _from, _to):
        """转接"""
        builder = [
            _from,
            "  -both 'set:hangup_after_bridge=false,set:park_after_bridge=true,park:' inline "
        ]
        arg = ''.join(builder)
        self.con.bgapi(TRANSFER, arg)

    def answer(self, device_id):
        """应答"""
        self.con.bgapi('uuid_phone_event', device_id + ' talk')

    def hangup_call(self, call_id, device_id, case_enum=CallCause.DEFAULT):
        """挂机"""
        msg = ESL.ESLevent("sendmsg", device_id)
        msg.addHeader("call-command", EXECUTE)
        msg.addHeader("execute-app-name", HANGUP)
        msg.addHeader("execute-app-arg", NORMAL_CLEARING)
        self.logger.info("hangup_call挂机 hangup call: {}, device: {}, ctiCauseEnum:{}", call_id, device_id, case_enum)
        self.send_args(device_id, SET, EslEventUtil.SIP_H_P_LIBRA_HANGUP_CAUSE + "=" + case_enum.description)
        self.con.sendEvent(msg)

    def broadcast(self, uuid, path, smf):
        builder = [
            UUID_BROADCAST,
            uuid,
            path,
            smf
        ]
        command = ' '.join(builder)
        self.con.bgapi(command, EMPTY)

    def break0(self, uuid, all=False, sync=True):
        builder = [
            UUID_BREAK,
            uuid
        ]
        if all:
            builder.append("all")
        command = ' '.join(builder)
        if sync:
            self.con.api(command, EMPTY)
        else:
            self.con.bgapi(command, EMPTY)

    def hold(self, smf, uuid, display):
        builder = [
            UUID_HOLD,
            smf,
            uuid,
            display
        ]
        if display:
            builder.append("all")
        else:
            builder.append(EMPTY)
        command = ' '.join(builder)
        self.con.bgapi(command, EMPTY)

    def get_var(self, uuid, var):
        builder = [
            UUID_GETVAR,
            uuid,
            var
        ]
        command = ' '.join(builder)
        self.con.bgapi(command, EMPTY)

    def set_var(self, uuid, var, val):
        builder = [
            UUID_SETVAR,
            uuid,
            var,
            val
        ]
        command = ' '.join(builder)
        self.con.bgapi(command, EMPTY)

    def multi_set_var(self, uuid, params):
        builder = [
            "uuid_setvar_multi " + uuid + " "
        ]
        builder1 = []
        for k, v in params.items():
            builder1.append(k + "="+v)
        builder.append(';'.join(builder1))
        command = ''.join(builder)
        self.con.bgapi(command, EMPTY)

    def record(self, uuid, action, path, limit):
        builder = [
            UUID_RECORD,
            uuid,
            action,
            path,
            str(limit)
        ]
        command = ' '.join(builder)
        self.con.bgapi(command, EMPTY)

    def transfer(self, uuid, smf, dest, dialplan, context):
        builder = [
            UUID_TRANSFER,
            uuid,
            smf,
            dest,
            dialplan,
            context
        ]
        command = ' '.join(builder)
        self.con.bgapi(command, EMPTY)

    def insert(self, device_id):
        """强插"""
        builder = [
            device_id,
            "  -both 'set:hangup_after_bridge=false,set:park_after_bridge=true,park:' inline "
        ]
        arg = ''.join(builder)
        self.con.api(TRANSFER, arg)

    def bridge_break(self, device_id):
        """拆线"""
        builder = [
            device_id,
            "  -both 'set:hangup_after_bridge=false,set:park_after_bridge=true,set:" + SIP_HEADER + sipHeaderHoldMusic + "=true,park:' inline "
        ]
        arg = ''.join(builder)
        self.con.api(TRANSFER, arg)

    def play_file(self, call_id, device_id, file, sync):
        """放音"""
        if sync:
            return self.hold_play(device_id, file)
        else:
            msg = ESL.ESLevent("sendmsg", device_id)
            msg.addHeader("call-command", EXECUTE)
            msg.addHeader("execute-app-name", PLAYBACK)
            msg.addHeader("execute-app-arg", file)
            msg.addHeader("async", "true")
            self.con.sendEvent(msg)

    def stop_play(self, device_id):
        """关闭播放音乐"""
        builder = [
            device_id,
            " on"
        ]
        arg = "".join(builder)
        return self.con.api(PAUSE, arg)

    def hold_play(self, device_id, play):
        """向a-leg插播tts音乐(无限播放)"""
        builder = [
            device_id,
            " playback::",
            play,
            " ",
            SMF_ALEG
        ]
        arg = "".join(builder)
        return self.con.api(UUID_BROADCAST, arg)

    def play_timeout(self, call_id, timeout):
        """播放超时主动挂机"""
        pass

    def listen(self, device_id1, device_id2, aleg=True, bleg=True):
        """监听"""
        if aleg:
            self.send_args(device_id1, SET, "eavesdrop_bridge_aleg=true")
        if bleg:
            self.send_args(device_id1, SET, "eavesdrop_bridge_bleg=true")
        self.send_args(device_id1, EAVESDROP, device_id2)

    def show_channel(self, device_id):
        msg = self.con.api("show", " channels like " + device_id + " as json")
        print('show_channel::', msg)
        return msg

    def expression(self, template, params):
        for key, value in params.items():
            template = template.replace("#{["+key+"]}", str(value))
        return template

    def stop(self):
        for k, v in self.executors.items():
            v.shutdown()
        self.con.disconnect()
        self.is_stopping = True


class OutboundClient:

    def __init__(self, agent, logger):
        self.bot_agent = agent
        self.logger = logger
        threading.Thread(target=self.start, args=()).start()

    class ESLRequestHandler(socketserver.BaseRequestHandler):
        def setup(self):
            try:
                self.client.logger.info('%s connected!',self.client_address)
                fd = self.request.fileno()
                con = ESL.ESLconnection(fd)
                self.client.logger.info('Connected: %s', con.connected())
                if con.connected():
                    info = con.getInfo()

                    # print(json.loads(info.serialize('json')))
                    event_name = info.getHeader("Event-Name")
                    device_id = info.getHeader("unique-id")
                    kwargs = json.loads(info.serialize('json'))
                    destination = self.client.bot_agent.register(**kwargs)
                    self.client.logger.info("device_id=%s, destination=%s", device_id, destination)

                    Cache.add_device_user_part(device_id, destination)
                    con.execute("bridge", f"user/{destination}", device_id)

                    # destination = "user/1001"
                    # msg = ESL.ESLevent("sendmsg", uuid)
                    # msg.addHeader("call-command", "execute")
                    # msg.addHeader("execute-app-name", "bridge")
                    # msg.addHeader("execute-app-arg", destination)
                    # # 发送消息以执行 bridge 操作
                    # con.sendEvent(msg)
                    # print(f"Call {uuid} is bridged to {destination}")

                    # con.execute("answer", "", uuid)

                    # con.execute("transfer", "1001 XML default", uuid)
                    # try:
                    #     con.disconnect()
                    # except:
                    #     print('come in ')
                    #     traceback.print_exc()
                else:
                    self.client.logger.info("Failed to connect to FreeSWITCH")
            except:
                traceback.print_exc()

    class CustomTCPServer(socketserver.TCPServer):
        def __init__(self, server_address, RequestHandlerClass, client):
            self.client = client
            socketserver.TCPServer.__init__(server_address, RequestHandlerClass)

    def start(self, HOST='0.0.0.0', PORT=8084):
        # HOST, PORT = "0.0.0.0", 8084
        # 创建一个 TCP 服务器
        with self.CustomTCPServer((HOST, PORT), self.ESLRequestHandler, self) as server:
            self.logger.info(f"ESL server listening on {HOST}:{PORT}")
            server.serve_forever()