#!/usr/bin/env python3 # encoding:utf-8 import json import sys from typing import Dict, Any, Optional, List from src.core.callcenter.constant import get_json_dict from src.core.callcenter.esl.constant.sip_header_constant import sipHeaderCallId, sipHeaderCaller, sipHeaderCalled, \ sipHeaderDeviceId, sipHeaderDeviceType from src.core.callcenter.exception import SipUriSyntaxException from src.core.datasource import SIP_SERVER from src.core.callcenter.esl.constant.esl_constant import SPLIT, SIP_HEADER from src.core.callcenter.enumeration import CallType, DeviceType, Direction, CdrType, NextType class BaseApi: @classmethod def from_json(cls, json_string=None): json_dict = get_json_dict(json_string) return cls(**json_dict) def to_json_string(self): return json.dumps(self, default=lambda x: x.__dict__, ensure_ascii=False) def __repr__(self): return self.to_json_string() class AgentCallRequest(BaseApi): """ 呼叫请求对象 """ def __init__(self, saas_id, agent_id, call_type: CallType, caller, called, caller_display="", called_display="", cti_flow_id=None, uuid1=None, uuid2=None, follow_data: Dict[str, Any] = {}): # 租户id self.saas_id = saas_id # 坐席号 self.agent_id = agent_id # 呼叫类型 self.call_type: CallType = call_type # 主叫，如果没有传，则使用坐席号码 self.caller = caller # 被叫 self.called = called # 主叫显号，接口没有传就用主技能组配置 self.caller_display = caller_display # 被叫显号，接口没传就用主技能配置 self.called_display = called_display self.cti_flow_id = cti_flow_id # uuid1 self.uuid1 = uuid1 # uuid2 self.uuid2 = uuid2 # 随路数据 self.follow_data = follow_data class AgentRequest(BaseApi): def __init__(self, saas_id, agent_number, agent_name, out_id, agent_password, agent_type, phone_number, distribute, agent_state, identity_type): # 租户隔离 self.saas_id = saas_id # 坐席工号, 生成返回 self.agent_number = agent_number # 坐席名称 self.agent_name = agent_name # 外部展示Id self.out_id = out_id # 坐席密码 self.agent_password = agent_password # 坐席类型 0:普通坐席 ;1:组长:2:主管 self.agent_type = agent_type # 分机号 self.phone_number = phone_number # 分配标志 0:不参与排队；1:参与排队 self.distribute = distribute # 账号状态 0:可用;1:禁用 self.agent_state = agent_state # 身份状态 self.identity_type = identity_type class AgentQueryRequest(BaseApi): def __init__(self, saas_id, agent_number, out_id, agent_type, agent_name, page=1, size=10): # 租户隔离 self.saas_id = saas_id # 坐席工号 self.agent_number = agent_number # 坐席名称 self.agent_name = agent_name # 外部展示Id self.out_id = out_id # 坐席类型 0:普通坐席 ;1:组长:2:主管 self.agent_type = agent_type self.page = page, self.size = size class AgentActionRequest(BaseApi): """ 坐席操作 """ def __init__(self, saas_id, agent_id=None, agent_number=None, out_id=None, identity_type=None, scene='manual', service_id=None, **kwargs): self.saas_id =saas_id # 坐席工号 self.agent_id = agent_id # # 坐席工号 self.agent_number = agent_number # 外部展示Id self.out_id = out_id # 身份状态 self.identity_type = identity_type # 场景 manual:手动外呼; robot:机器人外呼; monitor:监听 self.scene = scene self.service_id = service_id # 处理多余的参数 for key, value in kwargs.items(): setattr(self, key, value) class HumanServiceQueryRequest(BaseApi): """人工组查询""" def __init__(self, saas_id, service_id="00000000000000000", agent_state=None, agent_ids: Dict[str, Any] = {}, **kwargs): self.saas_id = saas_id self.serviceId = service_id self.agentState = agent_state self.agentIds = agent_ids # 处理多余的参数 for key, value in kwargs.items(): setattr(self, key, value) class AgentMonitorData(BaseApi): """坐席状态信息""" def __init__(self, saas_id=None, agent_num=None, agent_name=None, out_id=None, check_state=None, online_state=None, check_in_time=None, check_out_time=None, service_state=None, busy_time=None, idle_time=None, call_time=None, hang_time=None, heart_state=None, heart_time=None, update_time=None, check_scene=None): self.saas_id = saas_id self.agent_num = agent_num self.agent_name = agent_name self.out_id = out_id self.check_state = check_state self.online_state = online_state self.check_in_time = check_in_time self.check_out_time = check_out_time self.service_state = service_state self.busy_time = busy_time self.idle_time = idle_time self.call_time = call_time self.hang_time = hang_time self.heart_state = heart_state self.heart_time = heart_time self.update_time = update_time self.check_scene = check_scene class AgentEventData(BaseApi): def __init__(self, app_code, user_id, data: Dict[str, Any] = {}): self.appCode = app_code self.userId = user_id self.data = data class AgentStateData(BaseApi): def __init__(self, _s=None, _t=None, status=None, time=None, assign_time:int=0, phone_num=None): self._s = _s self._t = _t self.status = status self.time = time self.assign_time = assign_time self.phone_num = phone_num class AgentDelayStateData(BaseApi): def __init__(self, saas_id=None, flow_id=None, agent_num=None, service_state=None, scene=None): self.saas_id = saas_id self.flow_id = flow_id self.agent_num = agent_num self.service_state = service_state self.scene = scene class HangupCallRequest(BaseApi): def __init__(self, saas_id=None, flow_id=None, agent_id=None, called=None, call_id=None, scene=None): # saasId(必填) self.saas_id = saas_id self.flow_id = flow_id # 呼叫唯一id(选填) self.call_id = call_id self.called = called # 分机号(必填) self.agent_id = agent_id self.scene = scene class CheckInCallRequest(BaseApi): """检查是否在通话中实体""" def __init__(self, saas_id, agent_number): self.saas_id = saas_id self.agent_number = agent_number class MakeCallContext(BaseApi): def __init__(self, sip_server=SIP_SERVER, route_gateway_name="fxo-gateway", display: Optional[str] = None, caller: Optional[str] = None, called: Optional[str] = None, call_id: Optional[str] = None, device_id: Optional[str] = None, eavesdrop: Optional[str] = None, device_type: Optional[int] = None, timeout: Optional[int] = 60, originate_timeout: Optional[int] = 60, sip_header_map: Optional[Dict[str, str]] = {}, called_prefix: Optional[str] = "", service_id: Optional[str] = None, call_type: Optional[int] = None): # fs 地址 self.sip_server = sip_server # 线路名(非必传) self.route_gateway_name = route_gateway_name # 外显号（必传） self.display = display # 该腿的呼叫者（必传） self.caller = caller # 该腿的被呼叫者（必传） self.called = called # 该腿的callId (必传) self.call_id = call_id # 该腿的uuid（必传） self.device_id = device_id # 是否是监听腿 监听腿必填true，其它不填 self.eavesdrop = eavesdrop # 当前腿的类型（必填 客户 坐席 监听者 机器人） self.device_type = device_type # 超时挂断 self.timeout = timeout # 超时挂断（后面可以考虑删掉） self.originate_timeout = originate_timeout # 额外的sip头添加（非必填） self.sip_header_map = sip_header_map or {} # 被叫前缀 （用户腿必填） self.called_prefix = called_prefix # 任务id（机器人外呼必填） self.service_id = service_id # 呼叫类型（必填） self.call_type = call_type def get_sip_header(self) -> str: headers = [ f"{SIP_HEADER}contact_user={self.display}", "ring_asr=true", f"origination_caller_id_number={self.display}", f"origination_caller_id_name={self.display}", f"origination_uuid={self.device_id}", "absolute_codec_string=^^:PCMU:PCMA", ] if self.originate_timeout is not None: headers.append(f"originate_timeout={self.originate_timeout}") headers += [ "RECORD_STEREO=true", "RECORD_APPEND=true", f"{SIP_HEADER}{sipHeaderCallId}={self.call_id}", f"{SIP_HEADER}{sipHeaderCaller}={self.caller}", f"{SIP_HEADER}{sipHeaderCalled}={self.called}", f"{SIP_HEADER}{sipHeaderDeviceId}={self.device_id}", f"{SIP_HEADER}{sipHeaderDeviceType}={self.device_type}", ] if self.eavesdrop: headers.append(f"{SIP_HEADER}eavesdrop={self.eavesdrop}") if self.service_id: headers.append(f"{SIP_HEADER}service_id={self.service_id}") if self.device_type == DeviceType.CUSTOMER.code: headers += [ "RECORD_STEREO_SWAP=true" ] else: headers += [ "RECORD_STEREO_SWAP=false", "continue_on_fail=true" ] if self.sip_header_map: headers.extend([f"{SIP_HEADER}{k}={v}" for k, v in self.sip_header_map.items()]) return SPLIT.join(headers) def get_called(self) -> str: if self.called_prefix and self.device_type == DeviceType.CUSTOMER.code: return f"{self.called_prefix}{self.called}" return self.called def get_realm(self): sip_uri = SipURI(self.sip_server) return sip_uri.get_host() class SipURI: DEFAULT_PORT = -1 SIP_SCHEME = "sip" SCHEME_SEPARATOR = ':' def __init__(self, sip_uri: str): self.string_representation = sip_uri self.userinfo = None self.host = None self.port = self.DEFAULT_PORT self.uri_parameters = {} scheme = f"{self.SIP_SCHEME}{self.SCHEME_SEPARATOR}" # print('------->', sip_uri, sip_uri.startswith(scheme)) if not sip_uri.startswith(scheme): raise SipUriSyntaxException(f"SIP URI must start with {scheme}") buf = sip_uri[len(scheme):] at_pos = buf.find("@") if at_pos == 0: raise SipUriSyntaxException("userinfo cannot start with a '@'") if at_pos > 0: self.userinfo = buf[:at_pos] buf = buf[at_pos + 1:] end_hostport = buf.find(";") if end_hostport == 0: raise SipUriSyntaxException("hostport not present or it cannot start with ';'") if end_hostport < 0: end_hostport = len(buf) hostport = buf[:end_hostport] buf = buf[end_hostport:] colon_pos = hostport.find(":") if colon_pos > -1: if colon_pos == len(hostport) - 1: raise SipUriSyntaxException("hostport cannot terminate with a ':'") self.port = int(hostport[colon_pos + 1:]) self.host = hostport[:colon_pos] else: self.host = hostport if buf == ";": buf = "" while buf: if buf[0] == ";": buf = buf[1:] next_semicolon = buf.find(";") if next_semicolon < 0: next_semicolon = len(buf) next_equals = buf.find("=") if next_equals < 0 or next_equals > next_semicolon: next_equals = next_semicolon key = buf[:next_equals] value = buf[next_equals + 1:next_semicolon] if next_equals < next_semicolon else "" self.uri_parameters[key] = value buf = buf[next_semicolon:] def __str__(self): return self.string_representation def get_host(self): return self.host def get_port(self): return self.port def get_uri_parameters(self): return self.uri_parameters def get_userinfo(self): return self.userinfo class AgentInfo(BaseApi): def __init__(self, saas_id=None, agent_number=None, realm=None, sip_server=None, call_id=None, device_id=None, real_device_id=None, line_id=None, fs_user=None, domain=None, service_time=0, max_ready_time=0, total_ready_time=0, ready_times=0, not_ready_times=0, total_after_time=0, max_talk_time=0, total_talk_time=0, total_ring_times=0, total_answer_times=0): self.saas_id = saas_id self.agent_number = agent_number self.realm = realm self.sip_server = sip_server self.call_id = call_id self.device_id = device_id # 机器人外呼时真实fs self.real_device_id = real_device_id self.line_id = line_id self.fs_user = fs_user self.domain = domain # 坐席最近的一次服务时间,电话则是振铃时间(秒) self.service_time = service_time # 最大空闲时长 self.max_ready_time = max_ready_time # 累计空闲 self.total_ready_time = total_ready_time # 空闲次数 self.ready_times = ready_times # 忙碌次数 self.not_ready_times = not_ready_times # 累计话后时间长 self.total_after_time = total_after_time # 最大通话时长 self.max_talk_time = max_talk_time # 当日累计通话时长 self.total_talk_time = total_talk_time # 振铃次数 self.total_ring_times = total_ring_times # 当日累计接听次数 self.total_answer_times = total_answer_times def get_realm(self): """ Get the realm by extracting the host from the SIP server if available, otherwise return the current realm. """ try: if self.sip_server: sip_uri = SipURI(self.sip_server) return sip_uri.get_host() if sip_uri.get_host() else self.realm except Exception as ex: print(f"Error parsing SIP server URI: {ex}") return self.realm def is_fs_user(self): """ Determine if the agent is a FreeSWITCH user. Default is True if fs_user is None or True. """ return self.fs_user is None or self.fs_user is True class CallInfo(BaseApi): def __init__(self, core_uuid=None, cti_flow_id=None, call_id=None, conference=None, saas_id=None, group_id=None, hidden_customer=0, caller_display=None, caller=None, called_display=None, called=None, number_location=None, agent_key=None, agent_name=None, login_type=None, ivr_id=None, task_id=None, media_host=None, sip_server=None, client_host=None, record=None, record2=None, record_time=None, call_time=None, call_type=None, direction=None, answer_flag=None, wait_time=None, answer_count=0, hangup_dir=None, sdk_hangup=0, hangup_code=None, answer_time=None, end_time=None, talk_time=None, first_queue_time=None, queue_start_time=None, queue_end_time=None, overflow_count=0, uuid1=None, uuid2=None, cdr_notify_url=None, queue_level=None, transfer_agent=None, bucket_type=None, user_no_answer_end_call=False, device_list=[], device_info_map: Dict[str, Any] = {}, follow_data: Dict[str, Any] = {}, process_data: Dict[str, Any] = {}, next_commands=[], call_details=[]): self.core_uuid = core_uuid # 通话唯一标识 self.cti_flow_id = cti_flow_id self.call_id = call_id # 通话唯一标识 self.conference = conference # 会议号 self.saas_id = saas_id # 企业id self.group_id = group_id # 所在技能组id self.hidden_customer = hidden_customer # 隐藏客户号码(0:不隐藏;1:隐藏) self.caller_display = caller_display # 主叫显号 self.caller = caller # 主叫 self.called_display = called_display # 被叫显号 self.called = called # 被叫 self.number_location = number_location # 号码归属地 self.agent_key = agent_key # 坐席 self.agent_name = agent_name # 坐席名称 self.login_type = login_type # 坐席登录类型 self.ivr_id = ivr_id # ivr self.task_id = task_id # 任务ID self.media_host = media_host # 媒体 self.sip_server = sip_server # 服务地址 self.client_host = client_host # 客户服务地址 self.record = record # 录音地址 self.record2 = record2 # 第三方存储地址 self.record_time = record_time # 录音开始时间 self.call_time = call_time # 呼叫开始时间 self.call_type = call_type # 呼叫类型 (Enum CallType) self.direction = direction # 呼叫方向 (Enum Direction) self.answer_flag = answer_flag # 通话标识 self.wait_time = wait_time # 等待时长 self.answer_count = answer_count # 应答设备数 self.hangup_dir = hangup_dir # 挂机方向 (1主叫挂机, 2:被叫挂机, 3:平台挂机) self.sdk_hangup = sdk_hangup # sdk挂机标志 self.hangup_code = hangup_code # 挂机原因 self.answer_time = answer_time # 接听时间 self.end_time = end_time # 最后一侧电话挂机时间 self.talk_time = talk_time # 通话时长(秒) self.first_queue_time = first_queue_time # 第一次进队列时间 self.queue_start_time = queue_start_time # 进入技能组时间 self.queue_end_time = queue_end_time # 出技能组时间 self.overflow_count = overflow_count # 溢出次数 self.uuid1 = uuid1 # uuid1 self.uuid2 = uuid2 # uuid2 self.cdr_notify_url = cdr_notify_url # 话单通知地址 self.queue_level = queue_level # 排队等级，默认是进队列时间 self.transfer_agent = transfer_agent #是否转人工 self.bucket_type = bucket_type #呼入流量类型 self.user_no_answer_end_call = user_no_answer_end_call #用户未接听挂机 self.device_list = device_list # 当前通话的设备 self.device_info_map = device_info_map self.follow_data = follow_data # 呼叫随路数据(作为落单数据) self.process_data = process_data # 模块流程间数据 self.next_commands = next_commands # 执行下一步命令 self.call_details = call_details # 电话流程 @classmethod def from_json(cls, json_string=None): data = get_json_dict(json_string) device_info_map: Dict[str, Any] = {key: DeviceInfo(**value) for key, value in data.get('device_info_map', {}).items()} follow_data: Dict[str, Any] = {key: value for key, value in data.get('follow_data', {}).items()} process_data: Dict[str, Any] = {key: value for key, value in data.get('process_data', {}).items()} next_commands = [NextCommand(**x) for x in data.get('next_commands', [])] call_details = [CallDetail(**x) for x in data.get('call_details', [])] return cls(core_uuid=data.get('core_uuid'), cti_flow_id=data.get('cti_flow_id'), call_id=data.get('call_id'), conference=data.get('conference'), saas_id=data.get('saas_id'), group_id=data.get('group_id'), hidden_customer=data.get('hidden_customer'), caller_display=data.get('caller_display'), caller=data.get('caller'), called_display=data.get('called_display'), called=data.get('called'), number_location=data.get('number_location'), agent_key=data.get('agent_key'), agent_name=data.get('agent_name'), login_type=data.get('login_type'), ivr_id=data.get('ivr_id'), task_id=data.get('task_id'), media_host=data.get('media_host'), sip_server=data.get('sip_server'), client_host=data.get('client_host'), record=data.get('record'), record2=data.get('record2'), record_time=data.get('record_time'), call_time=data.get('call_time'), call_type=data.get('call_type'), direction=data.get('direction'), answer_flag=data.get('answer_flag'), wait_time=data.get('wait_time'), answer_count=data.get('answer_count'), hangup_dir=data.get('hangup_dir'), sdk_hangup=data.get('sdk_hangup'), hangup_code=data.get('hangup_code'), answer_time=data.get('answer_time'), end_time=data.get('end_time'), talk_time=data.get('talk_time'), first_queue_time=data.get('first_queue_time'), queue_start_time=data.get('queue_start_time'), queue_end_time=data.get('queue_end_time'), overflow_count=data.get('overflow_count'), uuid1=data.get('uuid1'), uuid2=data.get('uuid2'), cdr_notify_url=data.get('cdr_notify_url'), queue_level=data.get('queue_level'), transfer_agent=data.get('transfer_agent'), bucket_type=data.get('bucket_type'), user_no_answer_end_call=data.get('user_no_answer_end_call'), device_list=data.get('device_list', []),device_info_map=device_info_map, follow_data=follow_data, process_data=process_data, next_commands=next_commands, call_details=call_details) def __str__(self): return self.to_json_string() def to_json_string(self): new_dict = { "core_uuid": self.core_uuid, "cti_flow_id": self.cti_flow_id, "call_id": self.call_id, "conference": self.conference, "saas_id": self.saas_id, "group_id": self.group_id, "hidden_customer": self.hidden_customer, "caller_display": self.caller_display, "caller": self.caller, "called_display": self.called_display, "called": self.called, "number_location": self.number_location, "agent_key": self.agent_key, "agent_name": self.agent_name, "login_type": self.login_type, "ivr_id": self.ivr_id, "task_id": self.task_id, "media_host": self.media_host, "sip_server": self.sip_server, "client_host": self.client_host, "record": self.record, "record2": self.record2, "record_time": self.record_time, "call_time": self.call_time, "call_type": self.call_type, "direction": self.direction, "answer_flag": self.answer_flag, "wait_time": self.wait_time, "answer_count": self.answer_count, "hangup_dir": self.hangup_dir, "sdk_hangup": self.sdk_hangup, "hangup_code": self.hangup_code, "answer_time": self.answer_time, "end_time": self.end_time, "talk_time": self.talk_time, "first_queue_time": self.first_queue_time, "queue_start_time": self.queue_start_time, "queue_end_time": self.queue_end_time, "overflow_count": self.overflow_count, "uuid1": self.uuid1, "uuid2": self.uuid2, "cdr_notify_url": self.cdr_notify_url, "queue_level": self.queue_level, "transfer_agent": self.transfer_agent, "bucket_type": self.bucket_type, "user_no_answer_end_call": self.user_no_answer_end_call, "device_list": [x for x in self.device_list], "device_info_map": {key: vars(value) for key, value in self.device_info_map.items()}, "follow_data": {key: vars(value) for key, value in self.follow_data.items()}, "process_data": {key: vars(value) for key, value in self.process_data.items()}, "next_commands": [vars(x) for x in self.next_commands], "call_details": [vars(x) for x in self.call_details], } # print('call_info is :', new_dict, flush=True) return json.dumps(new_dict, default=lambda x: x.__dict__, ensure_ascii=False) class DeviceInfo(BaseApi): def __init__(self, cti_flow_id=None, call_id=None, conference=None, device_id=None, agent_key=None, agent_name=None, device_type=None, cdr_type=0, from_agent=None, caller=None, called=None, display=None, called_location=None, caller_location=None, call_time=None, ring_start_time=None, ring_end_time=None, answer_time=None, bridge_time=None, end_time=None, talk_time=None, sip_protocol=None, channel_name=None, hangup_cause=None, ring_cause=None, sip_status=None, record=None, record_time=None, record_start_time=None, state=None, apparent_number=None, caller_display=None): self.cti_flow_id = cti_flow_id self.call_id = call_id # 通话唯一标识 self.conference = conference # 会议模式 self.device_id = device_id # 设备id self.agent_key = agent_key # 坐席 self.agent_name = agent_name # 坐席 self.device_type = device_type # 1:坐席,2:客户,3:外线 self.cdr_type = cdr_type # 1:呼入,2:外呼,3:内呼,4:转接,5:咨询,6:监听,7:强插,8:耳语 self.from_agent = from_agent # 咨询或转接来源 self.caller = caller # 主叫 self.called = called # 被叫 self.display = display # 显号 self.called_location = called_location # 被叫归属地 self.caller_location = caller_location # 被叫归属地 self.call_time = call_time # 呼叫开始时间 self.ring_start_time = ring_start_time # 振铃开始时间 self.ring_end_time = ring_end_time # 振铃结束时间 self.answer_time = answer_time # 接通时间 self.bridge_time = bridge_time # 桥接时间 self.end_time = end_time # 结束时间 self.talk_time = talk_time # 通话时长 self.sip_protocol = sip_protocol # 信令协议(tcp/udp) self.channel_name = channel_name # 呼叫地址 self.hangup_cause = hangup_cause # 挂机原因 self.ring_cause = ring_cause # 回铃音识别 self.sip_status = sip_status # sip状态 self.record = record # 录音地址 self.record_time = record_time # 录音时长 self.record_start_time = record_start_time # 录音开始时间 self.state = state # 当前设备状态 self.apparent_number = apparent_number # 外显号 self.caller_display = caller_display # 主叫号 class NextCommand(BaseApi): def __init__(self, device_id, next_type, next_value=None): # 记录执行设备 self.device_id = device_id # 下一步执行命令 self.next_type = next_type # 执行参数 self.next_value = next_value class DelayAction(BaseApi): def __init__(self, uuid=None, call_id=None, device_id=None): self.uuid=uuid self.call_id = call_id self.device_id = device_id class CallDetail(BaseApi): def __init__(self, id=None, cts=None, start_time=None, end_time=None, call_id=None, detail_index=None, transfer_type=None, transfer_id=None, reason=None, month=None, ext1=None, ext2=None, status=None): self.id = id # PK self.cts = cts # 创建时间 self.start_time = start_time # 开始时间 self.end_time = end_time # 结束时间 self.call_id = call_id # 通话ID self.detail_index = detail_index # 顺序 self.transfer_type = transfer_type # 类型 self.transfer_id = transfer_id # 转接ID self.reason = reason # 出队列原因 self.month = month # 月份 self.ext1 = ext1 # 扩展字段1 self.ext2 = ext2 # 扩展字段2 self.status = status # 状态 class RouteGateway(BaseApi): def __init__(self, id=None, saas_id=None, cts=None, uts=None, name=None, media_host=None, media_port=None, caller_prefix=None, called_prefix=None, profile='gateway/gateway-fxo', sip_header1='P-LIBRA-CallId=#{[callId]}', sip_header2='P-LIBRA-deviceId=#{[deviceId]}', sip_header3=None, status=None): self.id = id # PK self.saas_id = saas_id self.cts = cts # 创建时间 self.uts = uts # 修改时间 self.name = name # 号码 self.media_host = media_host # 媒体地址 self.media_port = media_port # 媒体端口 self.caller_prefix = caller_prefix # 主叫号码前缀 self.called_prefix = called_prefix # 被叫号码前缀 self.profile = profile # 媒体拨号计划文件 self.sip_header1 = sip_header1 # SIP头1 self.sip_header2 = sip_header2 # SIP头2 self.sip_header3 = sip_header3 # SIP头3 self.status = status # 状态 # 机器人外呼 class BotChatRequest: def __init__(self, nodeId=None, userId=None, sessionId=None, taskId=None, asrText=None, recordId=None, ext: Optional[str] = None): self.nodeId = nodeId # 节点id self.userId = userId # 用户id self.sessionId = sessionId # 会话id self.recordId = recordId # 唯一标识 self.taskId = taskId # 机器人任务id self.asrText = asrText # asr识别的文本 self.ext = ext def to_json_string(self): return json.dumps(self.__dict__, ensure_ascii=False) @classmethod def from_json(cls, json_string): data = json.loads(json_string) return cls(**data) # def __repr__(self): # return (f"OutboundRobotRequestParams(node_id={self.node_id}, user_id={self.user_id}, " # f"session_id={self.session_id}, record_id={self.record_id}, task_id={self.task_id}, " # f"event_type={self.event_type}, asr_text={self.asr_text}, key_input={self.key_input})") class ChatAction: def __init__(self, action_code=None, action_content=None): self.action_code = action_code # normal:正常通话；hang:挂断；transfer：转人工 self.action_content = action_content # 动作内容 def to_json_string(self): return json.dumps(self.__dict__, ensure_ascii=False) @classmethod def from_json(cls, json_data): return cls( action_code=json_data.get("action_code"), action_content=json_data.get("action_content") ) class ChatContent: def __init__(self, content_type=None, content=None, voice_url=None, voice_content=None): self.content_type = content_type # 播放类型 self.content = content # 播放内容 self.voice_url = voice_url # 语音地址 self.voice_content = voice_content # 语音文本 def to_json_string(self): return json.dumps(self.__dict__, ensure_ascii=False) @classmethod def from_json(cls, json_data): return cls( content_type=json_data.get("content_type"), content=json_data.get("content"), voice_url=json_data.get("voice_url"), voice_content=json_data.get("voice_content") ) class ChatMessage: def __init__(self, node_id=None, contents=None, wait_time=None, action=None, inputType=None): self.node_id = node_id # 节点id self.contents = contents if contents is not None else [] # 内容列表 self.wait_time = wait_time # 用户静默时长 self.action = action # 动作代码 self.inputType = inputType # dtmf类型 def to_json_string(self): return json.dumps({ "node_id": self.node_id, "contents": [content.__dict__ for content in self.contents], "wait_time": self.wait_time, "action": self.action.__dict__ if self.action else None, "inputType": self.inputType }, ensure_ascii=False) @classmethod def from_json(cls, json_data): contents = [ChatContent.from_json(item) for item in json_data.get("contents", [])] action = ChatAction.from_json(json_data.get("action", {})) if json_data.get("action") else None return cls( node_id=json_data.get("node_id"), contents=contents, wait_time=json_data.get("wait_time"), action=action, inputType=json_data.get("inputType") ) class ChatResponse: def __init__(self, data=None, message=None, code=None): self.data = data if data is not None else ChatMessage() self.message = message self.code = code def to_json_string(self): return json.dumps({ "data": json.loads(self.data.to_json_string()), "message": self.message, "code": self.code }, ensure_ascii=False) @classmethod def from_json(cls, json_string): data = json.loads(json_string) response_data = ChatMessage.from_json(data.get("data", {})) return cls( data=response_data, message=data.get("message"), code=data.get("code") )