voice-gateway-service/src/core/callcenter/model.py

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

import json
from urllib.parse import urlparse
from typing import Dict, Any, Optional

from src.core.callcenter.enumeration import CallType


class MakeCallRequest:
    """
    呼叫请求对象
    """

    def __init__(self, saas_id, call_type: CallType, caller, called, caller_display="", called_display="",
                 uuid1=None, uuid2=None, follow_data: Dict[str, Any] = {}):
        # 租户id
        self.saas_id = saas_id
        # 呼叫类型
        self.call_type: CallType = call_type
        # 主叫，如果没有传，则使用坐席号码
        self.caller = caller
        # 被叫
        self.called = called
        # 主叫显号，接口没有传就用主技能组配置
        self.caller_display = caller_display
        # 被叫显号，接口没传就用主技能配置
        self.called_display = called_display
        # uuid1
        self.uuid1 = uuid1
        # uuid2
        self.uuid2 = uuid2
        # 随路数据
        self.follow_data = follow_data

    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)


class AgentRequest:
    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:
    def __init__(self, saas_id, agent_number, out_id, agent_type, agent_name):
        # 租户隔离
        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


class AgentActionRequest:
    """
    坐席操作
    """

    def __init__(self, saas_id, agent_id, agent_number, out_id, identity_type, scene):
        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

    def from_json(self, data):
        return self(**data)


class AgentMonitor:
    def __init__(self, id=None, saas_id=None, agent_num=None, out_id=None,
                 check_state=1, check_scene=None, check_in_time =None,
                 check_out_time=None, service_state=0, busy_time=None,
                 idle_time=None, call_time=None, hang_time=None,
                 heart_state=None, heart_time=None, session_id = None,
                 is_delete=0, update_time=None, create_time=None):
        self.id = id
        # 租户隔离
        self.saas_id = saas_id
        # 坐席工号
        self.agent_num = agent_num
        # 使用方id
        self.out_id = out_id
        # 是否签入 0:是 1: 否  默认未签入
        self.check_state = check_state
        # 迁入时scene
        self.check_scene = check_scene
        # 签入时间
        self.check_in_time = check_in_time
        # 签出时间
        self.check_out_time = check_out_time
        # 坐席服务状态 0:签出 1:忙碌 2:空闲
        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.session_id = session_id
        # 删除标识
        self.is_delete = is_delete
        # 更新时间
        self.update_time = update_time
        # 创建时间
        self.create_time = create_time


class HangupCallRequest:
    def __init__(self, saas_id, call_id, agent_number):
        # saasId(必填)
        self.saas_id = saas_id
        # 呼叫唯一id(选填)
        self.call_id = call_id
        # 分机号(必填)
        self.agent_number = agent_number


class CheckInCallRequest:
    """检查是否在通话中实体"""

    def __init__(self, saas_id, agent_number):
        self.saas_id = saas_id
        self.agent_number = agent_number


class AgentInfo:
    def __init__(self, sass_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.sass_id = sass_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:
                parsed_uri = urlparse(self.sip_server)
                return parsed_uri.hostname if parsed_uri.hostname 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

    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)


class CallInfo:
    def __init__(self, core_uuid=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, cti_host=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,
                 frist_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, device_list=[], device_info_map={},
                 follow_data={}, process_data={}, next_commands=[], call_details=[]):
        self.core_uuid = core_uuid  # 通话唯一标识
        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.cti_host = cti_host  # 服务地址
        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.frist_queue_time = frist_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.device_list = device_list  # 当前通话的设备
        self.device_info_map = device_info_map,  # K-V
        self.follow_data = follow_data  # 呼叫随路数据(作为落单数据)
        self.process_data = process_data  # 模块流程间数据
        self.next_commands = next_commands  # 执行下一步命令
        self.call_details = call_details  # 电话流程

    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"CallInfo(core_uuid={self.core_uuid}, call_id={self.call_id}, conference={self.conference}, "
    #             f"company_id={self.company_id}, group_id={self.group_id}, hidden_customer={self.hidden_customer}, "
    #             f"caller_display={self.caller_display}, caller={self.caller}, called_display={self.called_display}, "
    #             f"called={self.called}, number_location={self.number_location}, agent_key={self.agent_key}, "
    #             f"agent_name={self.agent_name}, login_type={self.login_type}, ivr_id={self.ivr_id}, task_id={self.task_id}, "
    #             f"media_host={self.media_host}, cti_host={self.cti_host}, client_host={self.client_host}, record={self.record}, "
    #             f"record2={self.record2}, record_time={self.record_time}, call_time={self.call_time}, call_type={self.call_type}, "
    #             f"direction={self.direction}, answer_flag={self.answer_flag}, wait_time={self.wait_time}, "
    #             f"answer_count={self.answer_count}, hangup_dir={self.hangup_dir}, sdk_hangup={self.sdk_hangup}, "
    #             f"hangup_code={self.hangup_code}, answer_time={self.answer_time}, end_time={self.end_time}, "
    #             f"talk_time={self.talk_time}, frist_queue_time={self.frist_queue_time}, queue_start_time={self.queue_start_time}, "
    #             f"queue_end_time={self.queue_end_time}, overflow_count={self.overflow_count}, uuid1={self.uuid1}, uuid2={self.uuid2}, "
    #             f"cdr_notify_url={self.cdr_notify_url}, queue_level={self.queue_level})")


class DeviceInfo:
    def __init__(self, call_id=None, conference=None, device_id=None, agent_key=None, agent_name=None, device_type=None,
                 cdr_type=None, 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):
        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  # 当前设备状态

    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"DeviceInfo(call_id={self.call_id}, conference={self.conference}, device_id={self.device_id}, "
    #             f"agent_key={self.agent_key}, agent_name={self.agent_name}, device_type={self.device_type}, "
    #             f"cdr_type={self.cdr_type}, from_agent={self.from_agent}, caller={self.caller}, called={self.called}, "
    #             f"display={self.display}, called_location={self.called_location}, caller_location={self.caller_location}, "
    #             f"call_time={self.call_time}, ring_start_time={self.ring_start_time}, ring_end_time={self.ring_end_time}, "
    #             f"answer_time={self.answer_time}, bridge_time={self.bridge_time}, end_time={self.end_time}, "
    #             f"talk_time={self.talk_time}, sip_protocol={self.sip_protocol}, channel_name={self.channel_name}, "
    #             f"hangup_cause={self.hangup_cause}, ring_cause={self.ring_cause}, sip_status={self.sip_status}, "
    #             f"record={self.record}, record_time={self.record_time}, record_start_time={self.record_start_time}, "
    #             f"state={self.state})")


class NextCommand:

    def __init__(self, device_id, next_type, next_value):
        # 记录执行设备
        self.device_id = device_id
        # 下一步执行命令
        self.next_type = next_type
        # 执行参数
        self.next_value = next_value

    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)


class CallDetail:
    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  # 状态

    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"CallDetail(id={self.id}, cts={self.cts}, start_time={self.start_time}, "
    #             f"end_time={self.end_time}, call_id={self.call_id}, detail_index={self.detail_index}, "
    #             f"transfer_type={self.transfer_type}, transfer_id={self.transfer_id}, reason={self.reason}, "
    #             f"month={self.month}, ext1={self.ext1}, ext2={self.ext2}, status={self.status})")


class RouteGateway:
    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  # 状态

    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"RouteGateway(id={self.id}, cts={self.cts}, uts={self.uts}, name={self.name}, "
    #             f"media_host={self.media_host}, media_port={self.media_port}, caller_prefix={self.caller_prefix}, "
    #             f"called_prefix={self.called_prefix}, profile={self.profile}, "
    #             f"sip_header1={self.sip_header1}, sip_header2={self.sip_header2}, "
    #             f"sip_header3={self.sip_header3}, status={self.status})")







#机器人外呼
class BotChatRequest:
    def __init__(self, node_id=None, user_id=None, session_id=None, record_id=None,
                 task_id=None, asr_text=None, key_input=None, ext:Dict={}):
        self.node_id = node_id  # 节点id
        self.user_id = user_id  # 用户id
        self.session_id = session_id  # 会话id
        self.record_id = record_id  # 唯一标识
        self.task_id = task_id  # 机器人任务id
        self.event_type = None  # 1:电话接通,2:用户语音asr结果上传,3:用户按键输入,4:用户挂断电话,5:读取外呼结果(可在外呼过程中调用，不一定在结束之后),6:用户不说话超时,7:TTS或录音文件播放完毕或被打断,8:ASR错误导致挂断电话,9:TTS错误导致挂断电话,10:其它系统错误导致挂断电话
        self.asr_text = asr_text  # 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, interruptable=None, wait_time=None,
                 action=None, talk_time_out=None,dtmf_type=None):
        self.node_id = node_id   # 节点id
        self.contents = contents if contents is not None else []   # 内容列表
        self.interruptable = interruptable   # 是否可打断
        self.wait_time = wait_time   # 用户静默时长
        self.action = action        # 动作代码
        self.talk_time_out = talk_time_out     # 用户通话时长
        self.dtmf_type = dtmf_type   # dtmf类型

    def to_json_string(self):
        return json.dumps({
            "node_id": self.node_id,
            "contents": [content.__dict__ for content in self.contents],
            "interruptable": self.interruptable,
            "wait_time": self.wait_time,
            "action": self.action.__dict__ if self.action else None,
            "talk_time_out": self.talk_time_out,
            "dtmf_type": self.dtmf_type
        }, 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,
            interruptable=json_data.get("interruptable"),
            wait_time=json_data.get("wait_time"),
            action=action,
            talk_time_out=json_data.get("talk_time_out"),
            dtmf_type=json_data.get("dtmf_type")
        )


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")
        )