debug

src/core/callcenter/agent.py

@@ -2,12 +2,15 @@
 # encoding:utf-8
 
 import threading
+import time
+import traceback
 from collections import defaultdict
 from concurrent.futures import ThreadPoolExecutor
 from typing import List
 
 from sqlalchemy import or_
 
+from src.core.callcenter import registry
 import src.core.callcenter.esl.utils.esl_event_util as EslEventUtil
 from src.core import with_app_context
 from src.core.callcenter.api import AgentActionRequest, AgentQueryRequest, AgentRequest, AgentEventData, \
@@ -84,81 +87,87 @@ class AgentEventService:
         called = device_info.called if device_info else None
         is_agent = (device_info and DeviceType.AGENT.code == device_info.device_type) if device_info else False
 
-        self.logger.info('agent_event_channel, event_name=%s, call_id=%s, device_id=%s, is_agent=%s', event_name, call_id, device_id, is_agent)
-        agent = self.data_handle_server.get_agent(saas_id, agent_num)
-        if not agent:
-            # self.logger.warn("event service channel agent is null %s %s %s %s %s", saas_id, event_name, caller, called, json.loads(event.serialize('json')))
-            return
-        agent_monitor = self.data_handle_server.get_agent_monitor(saas_id, agent_num)
-        if not agent_monitor:
-            # self.logger.warn("event service channel agentMonitor is null %s %s %s %s %s", saas_id, event_name, caller, called, json.loads(event.serialize('json')))
-            return
-
-        # 信道发起事件，触发完成发起（或桥）&& 坐席侧
-        if CHANNEL_ORIGINATE == event_name and is_agent:
-            self.push_handler.push_on_agent_work_report(saas_id, flow_id, agent_num, call_id, AgentScene.MANUAL, WorkStatus.AGENT_RINGING,phone=call_info.caller)
-
-        # 进度事件，外呼时对方提醒。或者入呼时提醒 && 坐席侧
-        if CHANNEL_PROGRESS == event_name and is_agent:
-            self.push_handler.push_on_agent_report(saas_id, agent_num, AgentScene.MANUAL, AgentServiceState.DIALING)
-            self.push_handler.push_on_agent_work_report(saas_id, flow_id, agent_num, call_id, AgentScene.MANUAL, WorkStatus.AGENT_CALLING)
-
-        # 媒体进度事件，外呼时对方提醒。或者入呼时提醒 && 用户侧
-        if CHANNEL_PROGRESS_MEDIA == event_name and not is_agent:
-            self.push_handler.push_on_agent_work_report(saas_id, flow_id, agent_num, call_id, AgentScene.MANUAL, WorkStatus.AGENT_CALLING_RINGING)
-
-        #应答
-        if CHANNEL_ANSWER == event_name:
-            if call_id:
-                if self.cache.get_call_is_end(call_id):
-                    # self.logger.warn("event service channel call is end {} {} {} {} {} {}", saas_id, event_name, caller, called, call_id, json.dumps(event.serialize('json')))
-                    return
-
-            self.agent_state_service.busy(saas_id, agent.agent_num, agent.phone_num)
-            self.data_handle_server.update_record(call_id, status=1)
-            if is_agent:
-                # 坐席接起
-                self.cache.set_call_is_answer(saas_id, flow_id)
-                self.push_handler.push_on_agent_work_report(saas_id, flow_id, agent_num, call_id, AgentScene.MANUAL, WorkStatus.ANSWER_COMPENSATE)
-                self.push_handler.push_answer_call(saas_id, flow_id, agent_num, call_id, AgentScene.MANUAL, ServiceDirect.MANUAL_CALL.service_direct, WorkStatus.AGENT_HANG_REPROCESSING)
-
-                self.agent_actionlog_service.insert_service_state(agent_monitor, AgentServiceState.CALLING, AgentLogState.CHANNEL_TURN_ON)
-            else:
-                # 用户侧接起
-                self.agent_monitor_service.update_calling(agent_monitor)
-                self.push_handler.push_on_agent_work_report(saas_id, flow_id, agent_num, call_id, AgentScene.MANUAL, WorkStatus.AGENT_ANSWER_OUTGOING)
-                self.push_handler.push_on_agent_report(saas_id, agent_num, AgentScene.MANUAL, AgentServiceState.CALLING)
-
-        #挂断
-        if CHANNEL_HANGUP == event_name:
-            # 坐席侧挂断
-            if is_agent:
-                if call_id:
-                    self.cache.set_call_is_end(call_id)
-            self.agent_monitor_service.update_processing(agent_monitor)
-            self.reprocessing_idle(AgentDelayStateData(saas_id, flow_id, agent_num, AgentServiceState.REPROCESSING, AgentScene.MANUAL))
-            self.push_handler.push_on_call_end(saas_id, flow_id, agent_num, AgentScene.MANUAL, ServiceDirect.MANUAL_CALL.service_direct, '0')
-            self.push_handler.push_on_agent_work_report(saas_id, flow_id, agent_num, call_id, AgentScene.MANUAL, WorkStatus.AGENT_HANG_REPROCESSING)
-            self.push_handler.push_on_agent_report(saas_id, agent_num, AgentScene.MANUAL, AgentServiceState.REPROCESSING)
-
-            self.agent_actionlog_service.insert_service_state(agent_monitor, AgentServiceState.REPROCESSING, AgentLogState.CHANNEL_HANG_UP)
-
-            self.data_handle_server.update_record(call_id, time_end=datetime.now())
-            # 同步处理后处理置闲
-            # reprocessingIdle(statusDto);
-            # agentProducer.pushDelayedStatus(statusDto, reprocessingTimeout);
+        start_time = time.time()
+        try:
+            self.logger.info('agent_event_channel, event_name=%s, call_id=%s, device_id=%s, is_agent=%s', event_name, call_id, device_id, is_agent)
+            agent = self.data_handle_server.get_agent(saas_id, agent_num)
+            if not agent:
+                # self.logger.warn("event service channel agent is null %s %s %s %s %s", saas_id, event_name, caller, called, json.loads(event.serialize('json')))
+                return
+            agent_monitor = self.data_handle_server.get_agent_monitor(saas_id, agent_num)
+            if not agent_monitor:
+                # self.logger.warn("event service channel agentMonitor is null %s %s %s %s %s", saas_id, event_name, caller, called, json.loads(event.serialize('json')))
+                return
 
-        if (CHANNEL_BRIDGE == event_name or PLAYBACK_START == event_name) and is_agent:
-            self.push_handler.push_on_ring_start(saas_id, flow_id, agent_num, AgentScene.MANUAL, call_id)
-            # self.push_handler.push_on_agent_work_report(saas_id, flow_id, agent_num, call_id, AgentScene.MANUAL, WorkStatus.AGENT_ANSWER_OUTGOING)
-            # self.push_handler.push_on_agent_report(saas_id, agent_num, AgentScene.MANUAL, AgentServiceState.CALLING)
+            # 信道发起事件，触发完成发起（或桥）&& 坐席侧
+            if CHANNEL_ORIGINATE == event_name and is_agent:
+                self.push_handler.push_on_agent_work_report(saas_id, flow_id, agent_num, call_id, AgentScene.MANUAL, WorkStatus.AGENT_RINGING,phone=call_info.caller)
 
-        if DETECTED_TONE == event_name and not is_agent:
-            self.push_handler.push_on_detected_tone(saas_id, flow_id, call_id, AgentScene.MANUAL, call_id)
+            # 进度事件，外呼时对方提醒。或者入呼时提醒 && 坐席侧
+            if CHANNEL_PROGRESS == event_name and is_agent:
+                self.push_handler.push_on_agent_report(saas_id, agent_num, AgentScene.MANUAL, AgentServiceState.DIALING)
+                self.push_handler.push_on_agent_work_report(saas_id, flow_id, agent_num, call_id, AgentScene.MANUAL, WorkStatus.AGENT_CALLING)
 
-        if (CHANNEL_UNBRIDGE == event_name or PLAYBACK_STOP == event_name) and is_agent:
-            self.push_handler.push_on_ring_end(saas_id, flow_id, call_id, AgentScene.MANUAL, call_id)
+            # 媒体进度事件，外呼时对方提醒。或者入呼时提醒 && 用户侧
+            if CHANNEL_PROGRESS_MEDIA == event_name and not is_agent:
+                self.push_handler.push_on_agent_work_report(saas_id, flow_id, agent_num, call_id, AgentScene.MANUAL, WorkStatus.AGENT_CALLING_RINGING)
 
+            #应答
+            if CHANNEL_ANSWER == event_name:
+                if call_id:
+                    if self.cache.get_call_is_end(call_id):
+                        # self.logger.warn("event service channel call is end {} {} {} {} {} {}", saas_id, event_name, caller, called, call_id, json.dumps(event.serialize('json')))
+                        return
+
+                self.agent_state_service.busy(saas_id, agent.agent_num, agent.phone_num)
+                self.data_handle_server.update_record(call_id, status=1)
+                if is_agent:
+                    # 坐席接起
+                    self.cache.set_call_is_answer(saas_id, flow_id)
+                    self.push_handler.push_on_agent_work_report(saas_id, flow_id, agent_num, call_id, AgentScene.MANUAL, WorkStatus.ANSWER_COMPENSATE)
+                    self.push_handler.push_answer_call(saas_id, flow_id, agent_num, call_id, AgentScene.MANUAL, ServiceDirect.MANUAL_CALL.service_direct, WorkStatus.AGENT_HANG_REPROCESSING)
+
+                    self.agent_actionlog_service.insert_service_state(agent_monitor, AgentServiceState.CALLING, AgentLogState.CHANNEL_TURN_ON)
+                else:
+                    # 用户侧接起
+                    self.agent_monitor_service.update_calling(agent_monitor)
+                    self.push_handler.push_on_agent_work_report(saas_id, flow_id, agent_num, call_id, AgentScene.MANUAL, WorkStatus.AGENT_ANSWER_OUTGOING)
+                    self.push_handler.push_on_agent_report(saas_id, agent_num, AgentScene.MANUAL, AgentServiceState.CALLING)
+
+            #挂断
+            if CHANNEL_HANGUP == event_name:
+                # 坐席侧挂断
+                if is_agent:
+                    if call_id:
+                        self.cache.set_call_is_end(call_id)
+                self.agent_monitor_service.update_processing(agent_monitor)
+                self.reprocessing_idle(AgentDelayStateData(saas_id, flow_id, agent_num, AgentServiceState.REPROCESSING, AgentScene.MANUAL))
+                self.push_handler.push_on_call_end(saas_id, flow_id, agent_num, AgentScene.MANUAL, ServiceDirect.MANUAL_CALL.service_direct, '0')
+                self.push_handler.push_on_agent_work_report(saas_id, flow_id, agent_num, call_id, AgentScene.MANUAL, WorkStatus.AGENT_HANG_REPROCESSING)
+                self.push_handler.push_on_agent_report(saas_id, agent_num, AgentScene.MANUAL, AgentServiceState.REPROCESSING)
+
+                self.agent_actionlog_service.insert_service_state(agent_monitor, AgentServiceState.REPROCESSING, AgentLogState.CHANNEL_HANG_UP)
+
+                self.data_handle_server.update_record(call_id, time_end=datetime.now())
+                # 同步处理后处理置闲
+                # reprocessingIdle(statusDto);
+                # agentProducer.pushDelayedStatus(statusDto, reprocessingTimeout);
+
+            if (CHANNEL_BRIDGE == event_name or PLAYBACK_START == event_name) and is_agent:
+                self.push_handler.push_on_ring_start(saas_id, flow_id, agent_num, AgentScene.MANUAL, call_id)
+                # self.push_handler.push_on_agent_work_report(saas_id, flow_id, agent_num, call_id, AgentScene.MANUAL, WorkStatus.AGENT_ANSWER_OUTGOING)
+                # self.push_handler.push_on_agent_report(saas_id, agent_num, AgentScene.MANUAL, AgentServiceState.CALLING)
+
+            if DETECTED_TONE == event_name and not is_agent:
+                self.push_handler.push_on_detected_tone(saas_id, flow_id, call_id, AgentScene.MANUAL, call_id)
+
+            if (CHANNEL_UNBRIDGE == event_name or PLAYBACK_STOP == event_name) and is_agent:
+                self.push_handler.push_on_ring_end(saas_id, flow_id, call_id, AgentScene.MANUAL, call_id)
+        except:
+            traceback.print_exc()
+        finally:
+            latency = (time.time() - start_time)
+            registry.ESL_EVENT_CALLBACK_LATENCY.labels(event_name, "agent").observe(latency)
 
     def bot_event_channel(self, event, call_info, device_info):
         event_name = EslEventUtil.getEventName(event)
@@ -171,50 +180,56 @@ class AgentEventService:
         called = (device_info.caller if is_agent else device_info.called) if device_info else None
         human_service_id = '00000000000000000'
 
-        if CHANNEL_ORIGINATE == event_name:
-            self.logger.info('bot_event_channel, event_name=%s, call_id=%s, is_agent=%s, agent_num=%s, call_info=%s', event_name, call_id, is_agent, agent_num, call_info)
+        start_time = time.time()
+        try:
+            if CHANNEL_ORIGINATE == event_name:
+                self.logger.info('bot_event_channel, event_name=%s, call_id=%s, is_agent=%s, agent_num=%s, call_info=%s', event_name, call_id, is_agent, agent_num, call_info)
 
-        agent = self.data_handle_server.get_agent(saas_id, agent_num)
-        if not agent:
-            # self.logger.warn("bot event service channel agent is null %s %s %s %s %s", saas_id, event_name, caller, called,
-            #                  json.dumps(event.serialize('json')))
-            return
-        agent_monitor = self.data_handle_server.get_agent_monitor(saas_id, agent_num)
-        if not agent_monitor:
-            # self.logger.warn("bot event service channel agentMonitor is null %s %s %s %s %s", saas_id, event_name, caller,
-            #                  called, json.dumps(event.serialize('json')))
-            return
-
-        # 信道发起事件，触发完成发起（或桥）&& 坐席侧
-        if CHANNEL_ORIGINATE == event_name and is_agent:
-            self.push_handler.push_on_call_ring(saas_id, flow_id, agent_num, AgentScene.ROBOT, call_id, ServiceDirect.ROBOT_CALL.service_direct, called, caller, human_service_id)
-            self.push_handler.push_on_agent_work_report(saas_id, flow_id, agent_num, call_id, AgentScene.ROBOT, WorkStatus.AGENT_RINGING,phone=call_info.caller)
-
-
-        if CHANNEL_ANSWER == event_name:
-            self.agent_state_service.busy(saas_id, agent.agent_num, agent.phone_num)
-            self.data_handle_server.update_record(call_id, status=1)
-            if is_agent:
-                self.agent_monitor_service.update_calling(agent_monitor)
-                self.push_handler.push_on_agent_work_report(saas_id, flow_id, agent_num, call_id, AgentScene.ROBOT, WorkStatus.AGENT_ANSWER_INCOMING, "座席接通呼入电话! internal")
-                self.push_handler.push_on_agent_report(saas_id, agent_num, AgentScene.ROBOT, AgentServiceState.CALLING)
-                self.push_handler.push_answer_call(saas_id, flow_id, agent_num, call_id, AgentScene.ROBOT, ServiceDirect.ROBOT_CALL.service_direct, WorkStatus.AGENT_HANG_REPROCESSING)
-
-                self.agent_actionlog_service.insert_service_state(agent_monitor, AgentServiceState.CALLING, AgentLogState.CHANNEL_TURN_ON, service_id=human_service_id)
-            else:
-                self.push_handler.push_on_agent_work_report(saas_id, flow_id, agent_num, call_id, AgentScene.ROBOT, WorkStatus.AGENT_ANSWER_INCOMING, "座席接通呼入电话! external")
-
-        if CHANNEL_HANGUP == event_name and is_agent:
-            self.agent_monitor_service.update_processing(agent_monitor)
-            self.reprocessing_idle(AgentDelayStateData(saas_id, flow_id, agent_num, AgentServiceState.REPROCESSING, AgentScene.ROBOT))
-            self.push_handler.push_on_call_end(saas_id, flow_id, agent_num, AgentScene.ROBOT, ServiceDirect.ROBOT_CALL.service_direct, "0")
-            self.push_handler.push_on_agent_work_report(saas_id, flow_id, agent_num, call_id, AgentScene.ROBOT, WorkStatus.AGENT_HANG_REPROCESSING)
-            self.push_handler.push_on_agent_report(saas_id, agent_num, AgentScene.ROBOT, AgentServiceState.REPROCESSING)
-
-            self.agent_actionlog_service.insert_service_state(agent_monitor, AgentServiceState.REPROCESSING,
-                                                  AgentLogState.CHANNEL_HANG_UP, service_id=human_service_id)
-            self.data_handle_server.update_record(call_id, time_end=datetime.now())
+            agent = self.data_handle_server.get_agent(saas_id, agent_num)
+            if not agent:
+                # self.logger.warn("bot event service channel agent is null %s %s %s %s %s", saas_id, event_name, caller, called,
+                #                  json.dumps(event.serialize('json')))
+                return
+            agent_monitor = self.data_handle_server.get_agent_monitor(saas_id, agent_num)
+            if not agent_monitor:
+                # self.logger.warn("bot event service channel agentMonitor is null %s %s %s %s %s", saas_id, event_name, caller,
+                #                  called, json.dumps(event.serialize('json')))
+                return
 
+            # 信道发起事件，触发完成发起（或桥）&& 坐席侧
+            if CHANNEL_ORIGINATE == event_name and is_agent:
+                self.push_handler.push_on_call_ring(saas_id, flow_id, agent_num, AgentScene.ROBOT, call_id, ServiceDirect.ROBOT_CALL.service_direct, called, caller, human_service_id)
+                self.push_handler.push_on_agent_work_report(saas_id, flow_id, agent_num, call_id, AgentScene.ROBOT, WorkStatus.AGENT_RINGING,phone=call_info.caller)
+
+
+            if CHANNEL_ANSWER == event_name:
+                self.agent_state_service.busy(saas_id, agent.agent_num, agent.phone_num)
+                self.data_handle_server.update_record(call_id, status=1)
+                if is_agent:
+                    self.agent_monitor_service.update_calling(agent_monitor)
+                    self.push_handler.push_on_agent_work_report(saas_id, flow_id, agent_num, call_id, AgentScene.ROBOT, WorkStatus.AGENT_ANSWER_INCOMING, "座席接通呼入电话! internal")
+                    self.push_handler.push_on_agent_report(saas_id, agent_num, AgentScene.ROBOT, AgentServiceState.CALLING)
+                    self.push_handler.push_answer_call(saas_id, flow_id, agent_num, call_id, AgentScene.ROBOT, ServiceDirect.ROBOT_CALL.service_direct, WorkStatus.AGENT_HANG_REPROCESSING)
+
+                    self.agent_actionlog_service.insert_service_state(agent_monitor, AgentServiceState.CALLING, AgentLogState.CHANNEL_TURN_ON, service_id=human_service_id)
+                else:
+                    self.push_handler.push_on_agent_work_report(saas_id, flow_id, agent_num, call_id, AgentScene.ROBOT, WorkStatus.AGENT_ANSWER_INCOMING, "座席接通呼入电话! external")
+
+            if CHANNEL_HANGUP == event_name and is_agent:
+                self.agent_monitor_service.update_processing(agent_monitor)
+                self.reprocessing_idle(AgentDelayStateData(saas_id, flow_id, agent_num, AgentServiceState.REPROCESSING, AgentScene.ROBOT))
+                self.push_handler.push_on_call_end(saas_id, flow_id, agent_num, AgentScene.ROBOT, ServiceDirect.ROBOT_CALL.service_direct, "0")
+                self.push_handler.push_on_agent_work_report(saas_id, flow_id, agent_num, call_id, AgentScene.ROBOT, WorkStatus.AGENT_HANG_REPROCESSING)
+                self.push_handler.push_on_agent_report(saas_id, agent_num, AgentScene.ROBOT, AgentServiceState.REPROCESSING)
+
+                self.agent_actionlog_service.insert_service_state(agent_monitor, AgentServiceState.REPROCESSING,
+                                                      AgentLogState.CHANNEL_HANG_UP, service_id=human_service_id)
+                self.data_handle_server.update_record(call_id, time_end=datetime.now())
+        except:
+            traceback.print_exc()
+        finally:
+            latency = (time.time() - start_time)
+            registry.ESL_EVENT_CALLBACK_LATENCY.labels(event_name, "agent").observe(latency)
 
     def reprocessing_idle(self, state_data: AgentDelayStateData):
         agent = self.data_handle_server.get_agent(state_data.saas_id, state_data.agent_num)

src/core/callcenter/registry.py

@@ -16,4 +16,5 @@ CALL_BOT_ANSWER_REQUESTS = Counter('call_bot_answer_requests', '机器人接听
 CALL_BOT_TRANSFER_REQUESTS = Counter('call_bot_transfer_requests', '机器转人量', ['bucket'])
 
 # esl时间耗时
-ESL_EVENT_LATENCY = Histogram('esl_event_latency', 'Esl Event latency in milliseconds', ['eventName'])
+ESL_EVENT_LATENCY = Histogram('esl_event_latency', 'Esl Event latency in seconds', ['eventName'])
+ESL_EVENT_CALLBACK_LATENCY = Histogram('esl_event_callback_latency', 'Esl Event callback latency in seconds', ['eventName','callType'])