voip语音机器人模块调试,v1

src/core/callcenter/esl/handler/channel_park_handler.py

@@ -1,9 +1,11 @@
 #!/usr/bin/env python3
 # encoding:utf-8
 
+import json
 from src.core.callcenter.esl.annotation import EslEventName
 from src.core.callcenter.esl.constant.event_names import CHANNEL_PARK
 from src.core.callcenter.esl.handler.esl_event_handler import EslEventHandler
+from src.core.voip.bot import BotAgent
 
 
 @EslEventName(CHANNEL_PARK)
@@ -11,6 +13,7 @@ class ChannelParkHandler(EslEventHandler):
 
     def __init__(self, inbound_client, logger):
         super().__init__(inbound_client, logger)
+        self.bot = BotAgent(logger)
 
     def handle(self, address, event, coreUUID):
         service = event.getHeader("variable_service")
@@ -20,12 +23,14 @@ class ChannelParkHandler(EslEventHandler):
 
 
     def process_fxo_calling(self, event):
-        destination = '1001'
+        kwargs = json.loads(event.serialize('json'))
+        destination = self.bot.register(**kwargs)
+        # destination = '1000'
         # 获取通话的 UUID
         call_uuid = event.getHeader("Unique-ID")
         service = event.getHeader("variable_service")
         print(f"Incoming call with UUID: {call_uuid}  {service} is transfer to {destination}")
-        self.inbound_client.con.execute("transfer", "1001 XML default", call_uuid)
+        self.inbound_client.con.execute("transfer", f"{destination} XML default", call_uuid)
         # self.con.execute("answer", "", call_uuid)
         # self.inbound_client.con.execute("bridge", "user/1001", call_uuid)
 

src/core/callcenter/esl/handler/default_esl_event_handler.py

@@ -12,6 +12,6 @@ class DefaultEslEventHandler(EslEventHandler):
         super().__init__(inbound_client, logger)
 
     def handle(self, address, event, coreUUID):
-        self.logger.info(json.loads(event.serialize('json')))
+        # self.logger.info(json.loads(event.serialize('json')))
         pass
 

src/core/callcenter/web.py

@@ -9,7 +9,7 @@ from flask import Flask, render_template_string
 
 from src.core.callcenter.call import CallService
 from src.core.callcenter.model import MakeCallRequest, AgentInfo
-from src.core.voip.hello import Voip
+
 
 dictConfig({
         "version": 1,
@@ -46,7 +46,6 @@ dictConfig({
 app = Flask(__name__)
 app.config['SECRET_KEY'] = ''
 
-voip = Voip(app.logger)
 client = InboundClient(app.logger)
 call_service = CallService(client, app.logger)
 

src/core/voip/hello.py → src/core/voip/bot.py

@@ -35,25 +35,6 @@ class UserStatus(Enum):
     silence = 2
 
 
-# Subclass to extend the Account and get notifications etc.
-class Account(pj.Account):
-    def __init__(self, user_agent):
-        pj.Account.__init__(self)
-        self.user_agent = user_agent
-
-    def onRegState(self, prm):
-        print("***OnRegState: " + prm.reason)
-
-    def onIncomingCall(self, prm):
-        print("***onIncomingCall: ", prm.callId)
-
-        call = MyCall(self, prm.callId)
-        call_op_param = pj.CallOpParam(True)
-        call_op_param.statusCode = pj.PJSIP_SC_OK
-        call.answer(call_op_param)
-        self.user_agent.calls[prm.callId] = call
-
-
 class MyAudioMediaPort(pj.AudioMediaPort):
     def __init__(self, asr=None):
         pj.AudioMediaPort.__init__(self)
@@ -85,11 +66,36 @@ class MyAudioMediaPlayer(pj.AudioMediaPlayer):
         self.stopTransmit(self.sink)
 
 
+# Subclass to extend the Account and get notifications etc.
+class Account(pj.Account):
+    def __init__(self, agent, user_part, **kwargs):
+        pj.Account.__init__(self)
+        self.agent = agent
+        self.user_part = user_part
+        self.calls = {}
+        self.kwargs = kwargs
+
+    def onRegState(self, prm):
+        print("***OnRegState: " + prm.reason)
+
+    def onIncomingCall(self, prm):
+        print("***onIncomingCall: ", prm.callId, json.dumps(self.kwargs))
+
+        call = MyCall(self.agent, self, self.user_part, prm.callId, **self.kwargs)
+        call_op_param = pj.CallOpParam(True)
+        call_op_param.statusCode = pj.PJSIP_SC_OK
+        call.answer(call_op_param)
+        self.calls[prm.callId] = call
+
+
 class MyCall(pj.Call):
 
-    def __init__(self, acc, call_id):
+    def __init__(self, agent, acc, user_part, call_id, **kwargs):
         pj.Call.__init__(self, acc, call_id)
+        self.agent = agent
+        self.user_part = user_part
         self.call_id = call_id
+        self.kwargs = kwargs
         self.audio_port = None
         self.aud_med = None
         self.recorder = None
@@ -121,6 +127,8 @@ class MyCall(pj.Call):
         #     print ('11111111111')
         if call_info.state == pj.PJSIP_INV_STATE_DISCONNECTED:
             print("通话结束")
+            # 远程挂机之后要将分机号回收
+            self.agent.unregister(self.user_part)
 
     def onCallMediaState(self, prm):
         call_info = self.getInfo()
@@ -181,21 +189,26 @@ class MyCall(pj.Call):
             # time.sleep(0.1)  # 每隔1秒更新一次进度
 
 
-class UserAgent:
-    def __init__(self, user_part, host="192.168.124.6", port="5060", password="slibra@#123456"):
-        self.user_part, self.host, self.port, self.password = user_part, host, port, password
+class BotAgent:
+
+    def __init__(self, logger, user_part_range=range(1000, 1010), host="192.168.124.6", port="5060", password="slibra@#123456"):
+        self.logger = logger
+        self.user_part_range, self.host, self.port, self.password = user_part_range, host, port, password
+        self.pool = queue.Queue(maxsize=len(user_part_range))
+        self.accounts = {}
         self.calls = {}
         self.ep = pj.Endpoint()
-        self.acc = Account(self)
         self.is_stopping = False
-
-    def start(self):
         threading.Thread(target=self.create_pjsua2, args=()).start()
 
-    # pjsua2 test function
     def create_pjsua2(self):
         # Create and initialize the library
         ep_cfg = pj.EpConfig()
+        ep_cfg.uaConfig.threadCnt = 1
+        ep_cfg.uaConfig.mainThreadOnly = True
+        ep_cfg.uaConfig.maxCalls = 8
+        ep_cfg.logConfig.level = 5
+        ep_cfg.logConfig.consoleLevel = 5
         self.ep.libCreate()
         self.ep.libInit(ep_cfg)
 
@@ -206,72 +219,71 @@ class UserAgent:
         sipTpConfig = pj.TransportConfig()
         sipTpConfig.port = 30506
         self.ep.transportCreate(pj.PJSIP_TRANSPORT_UDP, sipTpConfig)
-
         # Start the library
         self.ep.libStart()
-        acfg = pj.AccountConfig()
-        acfg.idUri = f"sip:{self.user_part}@{self.host}:{self.port}"
-        acfg.regConfig.registrarUri = f"sip:{self.host}:{self.port}"
-        cred = pj.AuthCredInfo("digest", "*", f"{self.user_part}", 0, self.password)
-        acfg.sipConfig.authCreds.append(cred)
-        # Create the account
-        self.acc.create(acfg)
+
+        for user_part in self.user_part_range:
+            acfg = pj.AccountConfig()
+            acfg.idUri = f"sip:{user_part}@{self.host}:{self.port}"
+            acfg.regConfig.registrarUri = f"sip:{self.host}:{self.port}"
+            cred = pj.AuthCredInfo("digest", "*", f"{user_part}", 0, self.password)
+            acfg.sipConfig.authCreds.append(cred)
+
+            # Create the account
+            acc = Account(self, user_part)
+            acc.create(acfg)
+
+            self.pool.put(user_part)
+            self.accounts[user_part] = acc
 
         while not self.is_stopping:
             self.ep.libHandleEvents(100)
 
-    def hangup(self, reason="NORMAL_CLEARING", **sip_headers):
+    def hangup(self, user_part, reason="NORMAL_CLEARING", **sip_headers):
         call_op_param = pj.CallOpParam(True)
         call_op_param.statusCode = pj.PJSIP_SC_OK
         call_op_param.reason = reason
         for k, v in sip_headers:
             call_op_param.headers.append(pj.SipHeader(f"sip_h_{k}", v))
-        for k, v in self.calls.items():
-            v.hangup(call_op_param)
-
-    def register(self, renew=False):
-        self.acc.setRegistration(renew=renew)
+        acc = self.accounts.get(user_part)
+        if acc:
+            for k, v in acc.calls.items():
+                v.hangup(call_op_param)
+            # 机器人主动挂机回收分机号
+            self.release(user_part)
+
+    def register(self, **kwargs):
+        user_part = self.pool.get()
+        self.logger.info('register, user_part :%d, args: %s', user_part, json.dumps(kwargs))
+        acc = self.accounts.get(user_part)
+        if acc:
+            # acc.setRegistration(renew=True)
+            acc.kwargs = kwargs
+        return user_part
+
+    def unregister(self, user_part):
+        acc = self.accounts.get(user_part)
+        if acc:
+            acc.setRegistration(renew=False)
+        # 用户远程挂机回收分机号
+        self.release(user_part)
+
+    def release(self, user_part):
+        def element_in_queue(q, element):
+            with q.mutex:  # 确保线程安全
+                for item in list(q.queue):  # 将队列转换为列表进行遍历
+                    if item == element:
+                        return True
+            return False
+
+        if element_in_queue(self.pool, user_part):
+            return
+        self.pool.put(user_part)
 
     def destroy(self):
         self.is_stopping = True
         # Destroy the library
         self.ep.libDestroy()
 
-
-class UserAgentPool:
-    def __init__(self, pool_size, user_part_range=range(1000, 1019)):
-        self.pool_size = pool_size
-        self.pool = queue.Queue(maxsize=pool_size)
-
-        # 初始化连接池
-        for i in range(pool_size):
-            conn = self.create(user_part_range[i])
-            self.pool.put(conn)
-
-    def create(self, user_part, host='192.168.124.6', port='6379', password="slibra@#123456"):
-        return UserAgent(user_part, host=host, port=port, password=password)
-
-    def get(self):
-        return self.pool.get()
-
-    def release(self, conn):
-        self.pool.put(conn)
-
-    def destroy(self):
-        while not self.pool.empty():
-            agent = self.pool.get()
-            agent.destroy()
-
-
-class Voip:
-
-    def __init__(self, logger):
-        self.logger = logger
-        # self.pool = UserAgentPool(pool_size=2)
-        agent = UserAgent('1001')
-        agent.start()
-
-
-# if __name__ == "__main__":
-#     voip = Voip()
-#     voip.create_pjsua2()
+    def __del__(self):
+        self.destroy()