|
@@ -7,7 +7,7 @@ import traceback
|
|
|
from collections import defaultdict
|
|
|
from concurrent.futures import ThreadPoolExecutor
|
|
|
from typing import List
|
|
|
-
|
|
|
+from datetime import datetime
|
|
|
from apscheduler.schedulers.background import BackgroundScheduler
|
|
|
from sqlalchemy import or_
|
|
|
|
|
@@ -17,11 +17,11 @@ from src.core import with_app_context
|
|
|
from src.core.callcenter.api import AgentActionRequest, AgentQueryRequest, AgentRequest, AgentEventData, \
|
|
|
AgentStateData, HumanServiceQueryRequest, AgentMonitorData, CallInfo, DeviceInfo, AgentDelayStateData
|
|
|
from src.core.callcenter.cache import Cache
|
|
|
-from src.core.callcenter.constant import CENTER_AGENT_HEARTBEAT, SAAS_ID
|
|
|
+from src.core.callcenter.constant import CENTER_AGENT_HEARTBEAT, SAAS_ID, CENTER_AGENT_LIVE_CNT
|
|
|
from src.core.callcenter.dao import *
|
|
|
from src.core.callcenter.data_handler import DataHandleServer
|
|
|
from src.core.callcenter.enumeration import AgentState, AgentCheck, AgentHeartState, AgentServiceState, AgentLogState, \
|
|
|
- AgentScene, BizErrorCode, WorkStatus, DownEvent, HumanState, DeviceType, ServiceDirect
|
|
|
+ AgentScene, BizErrorCode, WorkStatus, DownEvent, HumanState, DeviceType, ServiceDirect, HangupDir
|
|
|
from src.core.callcenter.esl.constant.event_names import *
|
|
|
from src.core.callcenter.exception import BizException
|
|
|
from src.core.callcenter.push import PushHandler
|
|
@@ -80,6 +80,8 @@ class AgentEventService:
|
|
|
|
|
|
def agent_event_channel(self, event, call_info: CallInfo, device_info: DeviceInfo):
|
|
|
event_name = EslEventUtil.getEventName(event)
|
|
|
+ event_timestamp = EslEventUtil.getEventTimestamp(event)
|
|
|
+ event_time = datetime.fromtimestamp(event_timestamp).strftime('%Y-%m-%d %H:%M:%S')
|
|
|
saas_id = call_info.saas_id if call_info else None
|
|
|
flow_id = call_info.cti_flow_id if call_info else None
|
|
|
call_id = call_info.call_id if call_info else None
|
|
@@ -91,18 +93,21 @@ class AgentEventService:
|
|
|
|
|
|
start_time = time.time()
|
|
|
try:
|
|
|
- self.logger.info('agent_event_channel, event_name=%s, agent_num=%s, device_id=%s, is_agent=%s', event_name, agent_num, device_id, is_agent)
|
|
|
+ self.logger.info('agent_event_channel, event_name=%s, call_id=%s, event_time=%s, agent_num=%s, device_id=%s, is_agent=%s, hangup_dir=%s, hangup_count=%s, answer_count=%s', event_name, call_id, event_time, agent_num, device_id, is_agent, call_info.hangup_dir, call_info.hangup_count, call_info.answer_count)
|
|
|
agent = self.data_handle_server.get_agent(saas_id, agent_num)
|
|
|
if not agent:
|
|
|
-
|
|
|
+ self.logger.warn("agent_event_channel:return, agent is null %s %s %s %s %s", saas_id, event_name, event_time, caller, called)
|
|
|
return
|
|
|
agent_monitor = self.data_handle_server.get_agent_monitor(saas_id, agent_num)
|
|
|
if not agent_monitor:
|
|
|
-
|
|
|
+ self.logger.warn("agent_event_channel:return, agentMonitor is null %s %s %s %s %s", saas_id, event_name, event_time, caller, called)
|
|
|
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)
|
|
|
|
|
|
|
|
@@ -144,8 +149,9 @@ class AgentEventService:
|
|
|
if call_id:
|
|
|
self.cache.set_call_is_end(call_id)
|
|
|
self.agent_monitor_service.update_processing(agent_monitor)
|
|
|
- self.logger.info('挂断更新')
|
|
|
- self.reprocessing_idle(AgentDelayStateData(saas_id, flow_id, agent_num, AgentServiceState.REPROCESSING, AgentScene.MANUAL))
|
|
|
+ self.logger.info('挂断更新:%s', agent)
|
|
|
+
|
|
|
+ self.reprocessing_idle(AgentDelayStateData(saas_id, flow_id, call_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)
|
|
@@ -171,14 +177,20 @@ class AgentEventService:
|
|
|
except:
|
|
|
traceback.print_exc()
|
|
|
finally:
|
|
|
- latency = (time.time() - start_time)
|
|
|
+ time_cost = (time.time() - start_time) * 1000
|
|
|
+ registry.ESL_EVENT_CALLBACK_COST.labels(event_name, "agent").observe(time_cost)
|
|
|
+ latency = (time.time() - event_timestamp) * 1000
|
|
|
registry.ESL_EVENT_CALLBACK_LATENCY.labels(event_name, "agent").observe(latency)
|
|
|
+ self.logger.info('agent_event_channel, event_name=%s, time_cost=%s, latency=%s, call_id=%s, event_time=%s, agent_num=%s, device_id=%s, is_agent=%s, hangup_dir=%s, hangup_count=%s, answer_count=%s', event_name, time_cost, latency, call_id, event_time, agent_num, device_id, is_agent, call_info.hangup_dir, call_info.hangup_count, call_info.answer_count)
|
|
|
|
|
|
def bot_event_channel(self, event, call_info, device_info):
|
|
|
event_name = EslEventUtil.getEventName(event)
|
|
|
+ event_timestamp = EslEventUtil.getEventTimestamp(event)
|
|
|
+ event_time = datetime.fromtimestamp(event_timestamp).strftime('%Y-%m-%d %H:%M:%S')
|
|
|
saas_id = call_info.saas_id if call_info else None
|
|
|
flow_id = call_info.cti_flow_id if call_info else None
|
|
|
call_id = call_info.call_id if call_info else None
|
|
|
+ device_id = device_info.device_id if device_info else None
|
|
|
agent_num = device_info.agent_key if device_info else None
|
|
|
is_agent = (device_info and DeviceType.AGENT.code == device_info.device_type) if device_info else False
|
|
|
caller = (device_info.called if is_agent else device_info.caller) if device_info else None
|
|
@@ -187,27 +199,29 @@ class AgentEventService:
|
|
|
|
|
|
start_time = time.time()
|
|
|
try:
|
|
|
- self.logger.info('bot_event_channel, event_name=%s, call_id=%s, is_agent=%s, agent_num=%s', event_name, call_id, is_agent, agent_num)
|
|
|
+ self.logger.info('bot_event_channel, event_name=%s, call_id=%s, device_id=%s, event_time=%s, is_agent=%s, agent_num=%s, hangup_dir=%s, hangup_count=%s, answer_count=%s', event_name, call_id, device_id, event_time, is_agent, agent_num, call_info.hangup_dir, call_info.hangup_count, call_info.answer_count)
|
|
|
agent = self.data_handle_server.get_agent(saas_id, agent_num)
|
|
|
if not agent:
|
|
|
-
|
|
|
-
|
|
|
+ self.logger.warn("bot_event_channel:return, agent is null %s %s %s %s %s %s", saas_id, event_name, event_time, call_id, caller, called)
|
|
|
return
|
|
|
agent_monitor = self.data_handle_server.get_agent_monitor(saas_id, agent_num)
|
|
|
if not agent_monitor:
|
|
|
-
|
|
|
-
|
|
|
+ self.logger.warn("bot_event_channel:return, agentMonitor is null %s %s %s %s %s %s ", saas_id, event_name, event_time, call_id, caller, called)
|
|
|
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)
|
|
|
+ self.data_handle_server.update_record(call_id, transfer_user_id=agent.user_id,transfer_user_name=agent.agent_name)
|
|
|
|
|
|
|
|
|
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, transfer_user_id=agent.user_id,transfer_user_name=agent.agent_name)
|
|
|
+
|
|
|
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")
|
|
@@ -220,7 +234,7 @@ class AgentEventService:
|
|
|
|
|
|
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.reprocessing_idle(AgentDelayStateData(saas_id, flow_id, call_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)
|
|
@@ -233,8 +247,11 @@ class AgentEventService:
|
|
|
except:
|
|
|
traceback.print_exc()
|
|
|
finally:
|
|
|
- latency = (time.time() - start_time)
|
|
|
- registry.ESL_EVENT_CALLBACK_LATENCY.labels(event_name, "agent").observe(latency)
|
|
|
+ time_cost = (time.time() - start_time) * 1000
|
|
|
+ registry.ESL_EVENT_CALLBACK_COST.labels(event_name, "bot").observe(time_cost)
|
|
|
+ latency = (time.time() - event_timestamp) * 1000
|
|
|
+ registry.ESL_EVENT_CALLBACK_LATENCY.labels(event_name, "bot").observe(latency)
|
|
|
+ self.logger.info('bot_event_channel, event_name=%s, time_cost=%s, latency=%s, call_id=%s, device_id=%s, event_time=%s, is_agent=%s, agent_num=%s, hangup_dir=%s, hangup_count=%s, answer_count=%s', event_name, time_cost, latency, call_id, device_id, event_time, is_agent, agent_num, call_info.hangup_dir, call_info.hangup_count, call_info.answer_count)
|
|
|
|
|
|
def reprocessing_idle(self, state_data: AgentDelayStateData):
|
|
|
agent = self.data_handle_server.get_agent(state_data.saas_id, state_data.agent_num)
|
|
@@ -247,7 +264,7 @@ class AgentEventService:
|
|
|
self.agent_state_service.idle(agent.saas_id, agent.out_id, agent.phone_num)
|
|
|
self.logger.info('reprocessing_idle_end')
|
|
|
self.agent_monitor_service.update_idle(agent_monitor)
|
|
|
- self.push_handler.push_on_agent_work_report(state_data.saas_id, state_data.flow_id, state_data.agent_num, "", state_data.scene, WorkStatus.AGENT_HANG_IDLE)
|
|
|
+ self.push_handler.push_on_agent_work_report(state_data.saas_id, state_data.flow_id, state_data.agent_num, state_data.call_id, state_data.scene, WorkStatus.AGENT_HANG_IDLE)
|
|
|
self.agent_actionlog_service.insert_service_state(agent_monitor, AgentServiceState.IDLE, AgentLogState.REPROCESSING_IDLE)
|
|
|
|
|
|
|
|
@@ -266,12 +283,11 @@ class AgentOperService:
|
|
|
self.agent_actionlog_service = AgentActionLogService(app)
|
|
|
self.agent_state_service = AgentStateService(app)
|
|
|
|
|
|
-
|
|
|
self.agent_heartbeat_expire = 30
|
|
|
+ self.agent_serial_live_expire = 60*10
|
|
|
self.agent_heartbeat_job_scheduler = BackgroundScheduler()
|
|
|
self.agent_heartbeat_job_scheduler.add_job(self.agent_heartbeat_daemon, 'interval', seconds=1, max_instances=1, name='agent_heartbeat_daemon')
|
|
|
self.agent_heartbeat_job_scheduler.start()
|
|
|
-
|
|
|
|
|
|
def agent_heartbeat_daemon(self):
|
|
|
def check_out_daemon(_name, key, value):
|
|
@@ -280,22 +296,39 @@ class AgentOperService:
|
|
|
if sec > self.agent_heartbeat_expire:
|
|
|
self.redis_handler.redis.hdel(_name, key)
|
|
|
self.logger.error("agent heartbeat expired, will checkout %s %s", key, value)
|
|
|
- self.checkout(AgentActionRequest(saas_id=SAAS_ID, agent_id=key, agent_number=key))
|
|
|
+
|
|
|
except:
|
|
|
traceback.print_exc()
|
|
|
|
|
|
-
|
|
|
+ def check_agent_live_daemon(_members):
|
|
|
+ _key = CENTER_AGENT_LIVE_CNT % SAAS_ID
|
|
|
+ pre_time = self.redis_handler.redis.get(_key)
|
|
|
+ if not pre_time or not _members or len(_members) == 0:
|
|
|
+ _value = datetime.now().timestamp()
|
|
|
+ self.redis_handler.redis.set(_key, _value, ex=60*60, nx=True)
|
|
|
+ else:
|
|
|
+ _diff = datetime.now().timestamp() - float(pre_time)
|
|
|
+ if _diff > self.agent_serial_live_expire:
|
|
|
+ self.logger.info('check_agent_live_daemon, members=%s, diff=%s, pre_time=%s', (len(_members) if _members else 0), _diff, pre_time)
|
|
|
+ self.logger.warn('WARING::live agent count less than 1 serial ten minutes')
|
|
|
+ self.data_handle_server.create_warning_record(1, '10分钟空岗报警')
|
|
|
+ self.redis_handler.redis.delete(_key)
|
|
|
+
|
|
|
+ if _members and len(_members) > 0:
|
|
|
+ self.redis_handler.redis.delete(_key)
|
|
|
+
|
|
|
name = CENTER_AGENT_HEARTBEAT % SAAS_ID
|
|
|
members = self.redis_handler.redis.hgetall(name)
|
|
|
+ check_agent_live_daemon(members)
|
|
|
if not members:
|
|
|
return
|
|
|
+
|
|
|
+ registry.MANUAL_AGENT_LIVES.set(len(members))
|
|
|
for k,v in members.items():
|
|
|
check_out_daemon(name, k, v)
|
|
|
-
|
|
|
|
|
|
def __del__(self):
|
|
|
self.agent_heartbeat_job_scheduler.shutdown()
|
|
|
-
|
|
|
|
|
|
@with_app_context
|
|
|
def enable(self, req: AgentActionRequest):
|
|
@@ -801,8 +834,8 @@ class AgentStateService:
|
|
|
self.logger = app.logger
|
|
|
self.redis_handler = RedisHandler()
|
|
|
self.assigned_recycle_millisecond = 30 * 1000
|
|
|
- self.state_service_id_data_map = defaultdict(dict)
|
|
|
- self.executor = ThreadPoolExecutor(max_workers=10)
|
|
|
+
|
|
|
+
|
|
|
self.data_handle_server = DataHandleServer(app)
|
|
|
self.agent_monitor_service = AgentMonitorService(app)
|
|
|
self.agent_actionlog_service = AgentActionLogService(app)
|
|
@@ -861,14 +894,39 @@ class AgentStateService:
|
|
|
|
|
|
def assign_agent(self, saas_id, service_id, called=None, ivr_id=None, task_id=None, cbp=None):
|
|
|
choose_phone_num = ''
|
|
|
- self.logger.info("assignAgent %s %s %s"% (saas_id, service_id, called))
|
|
|
- idle_agents = self.idle_agents(saas_id, service_id)
|
|
|
- if len(idle_agents) <= 0:
|
|
|
- return choose_phone_num
|
|
|
- choose_phone_num = self._choose_max_idle_time(idle_agents)
|
|
|
- self.handle_assign_time(saas_id, service_id, choose_phone_num)
|
|
|
+ lock = threading.Lock()
|
|
|
+ try:
|
|
|
+ lock.acquire()
|
|
|
+ self.logger.info("assignAgent %s %s %s"% (saas_id, service_id, called))
|
|
|
+ idle_agents = self.idle_agents(saas_id, service_id)
|
|
|
+ if len(idle_agents) <= 0:
|
|
|
+ return choose_phone_num
|
|
|
+ choose_phone_num = self._choose_max_idle_time(idle_agents)
|
|
|
+ self.handle_assign_time(saas_id, service_id, choose_phone_num)
|
|
|
+ self.handle_lock_agent(choose_phone_num, saas_id, service_id)
|
|
|
+ finally:
|
|
|
+ lock.release()
|
|
|
return choose_phone_num
|
|
|
|
|
|
+ def handle_check_agent_lock(self, choose_phone_num, saas_id, service_id='00000000000000000'):
|
|
|
+ key = self._lock_key(saas_id, service_id, choose_phone_num)
|
|
|
+ res = self.redis_handler.redis.get(key)
|
|
|
+ self.logger.info('checkAgent %s %s %s %s'% (saas_id, service_id, choose_phone_num, res))
|
|
|
+ return False if res else True
|
|
|
+
|
|
|
+ def handle_lock_agent(self, choose_phone_num, saas_id, service_id='00000000000000000'):
|
|
|
+ key = self._lock_key(saas_id, service_id, choose_phone_num)
|
|
|
+ expire = self._get_expire_time()
|
|
|
+ self.redis_handler.redis.set(key, 1, nx=True, ex=expire)
|
|
|
+ res = self.redis_handler.redis.get(key)
|
|
|
+ self.logger.info('lockAgent %s %s %s %s %s'% (saas_id, service_id, choose_phone_num, expire, res))
|
|
|
+
|
|
|
+ def handle_release_agent_lock(self, choose_phone_num, saas_id, service_id='00000000000000000'):
|
|
|
+ key = self._lock_key(saas_id, service_id, choose_phone_num)
|
|
|
+ self.redis_handler.redis.delete(key)
|
|
|
+
|
|
|
+ self.logger.info('releaseAgent %s %s %s'% (saas_id, service_id, choose_phone_num))
|
|
|
+
|
|
|
def handle_assign_time(self, saas_id, service_id, choose_phone_num):
|
|
|
key = self._key(saas_id, service_id)
|
|
|
cache_agent_map = self.get_cache_agent_map(saas_id, service_id)
|
|
@@ -939,13 +997,14 @@ class AgentStateService:
|
|
|
return free_agents
|
|
|
|
|
|
def get_idle_agents(self,cache_agent_list):
|
|
|
- current_time =int(datetime.now().timestamp() * 1000)
|
|
|
+
|
|
|
idle_agents = [
|
|
|
agent for agent in cache_agent_list
|
|
|
- if agent.status == 1 and (
|
|
|
- agent.assign_time == 0 or
|
|
|
- agent.assign_time + self.assigned_recycle_millisecond < current_time
|
|
|
- )
|
|
|
+ if agent.status == 1 and self.handle_check_agent_lock(agent.phone_num, SAAS_ID)
|
|
|
+
|
|
|
+
|
|
|
+
|
|
|
+
|
|
|
]
|
|
|
return idle_agents
|
|
|
|
|
@@ -966,34 +1025,34 @@ class AgentStateService:
|
|
|
busy_agents_size = len(busy_agents)
|
|
|
return busy_agents_size
|
|
|
|
|
|
- def update_report_state(self, saas_id, service_id):
|
|
|
- key = self._key(saas_id, service_id)
|
|
|
-
|
|
|
- data_map = self.state_service_id_data_map[key]
|
|
|
- idle = HumanState.IDLE
|
|
|
- if idle.value not in data_map:
|
|
|
- data_map[idle.code] = threading.Lock()
|
|
|
- self.executor.submit(self.do_report_real_time_human_service_id, saas_id, service_id, idle)
|
|
|
-
|
|
|
- busy = HumanState.BUSY
|
|
|
- if busy.value not in data_map:
|
|
|
- data_map[busy.code] = threading.Lock()
|
|
|
- self.executor.submit(self.do_report_real_time_human_service_id, saas_id, service_id, busy)
|
|
|
-
|
|
|
-
|
|
|
- def do_report_real_time_human_service_id(self, saas_id, service_id, human_state):
|
|
|
- name = "cti_center_real_time_human_service_state"
|
|
|
- tag_list = {
|
|
|
- "vcc_id": saas_id,
|
|
|
- "service_id": service_id,
|
|
|
- "state": human_state.code,
|
|
|
- }
|
|
|
- if human_state == HumanState.IDLE:
|
|
|
-
|
|
|
- self.meter_registry.gauge(name, tag_list, self, lambda ctx: ctx.get_agent_service_busy_size(saas_id, service_id))
|
|
|
- elif human_state == HumanState.BUSY:
|
|
|
- self.meter_registry.gauge(name, tag_list, self, lambda ctx: ctx.get_agent_service_idle_size(saas_id, service_id))
|
|
|
- return 0
|
|
|
+
|
|
|
+
|
|
|
+
|
|
|
+
|
|
|
+
|
|
|
+
|
|
|
+
|
|
|
+
|
|
|
+
|
|
|
+
|
|
|
+
|
|
|
+
|
|
|
+
|
|
|
+
|
|
|
+
|
|
|
+
|
|
|
+
|
|
|
+
|
|
|
+
|
|
|
+
|
|
|
+
|
|
|
+
|
|
|
+
|
|
|
+
|
|
|
+
|
|
|
+
|
|
|
+
|
|
|
+
|
|
|
|
|
|
def _check_in_key(self, saas_id):
|
|
|
return "CTI:%s:HUMAN:AGENT"%(saas_id.upper())
|
|
@@ -1001,11 +1060,15 @@ class AgentStateService:
|
|
|
def _key(self, saas_id, service_id):
|
|
|
return "CTI:%s:HUMAN:%s"%(saas_id.upper(), service_id)
|
|
|
|
|
|
+ def _lock_key(self, saas_id, service_id, choose_phone_num):
|
|
|
+ return "CTI:%s:HUMAN:%s:%s:LOCK"%(saas_id.upper(), service_id, choose_phone_num)
|
|
|
+
|
|
|
def _get_expire_time(self):
|
|
|
- now = datetime.now()
|
|
|
- end_of_day = now.replace(hour=23, minute=59, second=59, microsecond=0)
|
|
|
- expire_time = (end_of_day - now).total_seconds() * 1000
|
|
|
- return int(expire_time)
|
|
|
+
|
|
|
+
|
|
|
+
|
|
|
+
|
|
|
+ return 60*60*24*30
|
|
|
|
|
|
def _choose_max_idle_time(self, idle_agents: List[AgentStateData]) -> str:
|
|
|
idle_agents = sorted(idle_agents, key=lambda agent: agent.assign_time, reverse=False)
|