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

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

import json
import sys
import time
import uuid
from datetime import datetime
from src.core.callcenter.constant import *
from src.core.callcenter.api import AgentInfo, CallInfo, RouteGateway
from src.core.callcenter.dao import Agent, Phone
from src.core.datasource import RedisHandler

cacheDay = 7
deviceCall = {}
deviceUserPart = {}
redis_handler = RedisHandler()
print(redis_handler.redis.info())


def get_agent_info(saas_id, agent_number):
    text = redis_handler.get(AGENT_INFO + saas_id + ":" + agent_number)
    print('get_agent_info', saas_id, agent_number, text)
    sys.stdout.flush()  # 强制刷新输出缓冲区
    if text:
        return AgentInfo.from_json(text)
    phone = get_agent_phone(saas_id, agent_number)
    agent_info = AgentInfo(saas_id=saas_id, sip_server=phone.sip_server, agent_number=agent_number)
    print('get_agent_info', saas_id, agent_number, agent_info)
    add_agent_info(agent=agent_info)
    sys.stdout.flush()  # 强制刷新输出缓冲区
    return agent_info


def refresh_agent_token(agent_number, token):
    redis_handler.set(AGENT_TOKEN + token, agent_number, cacheDay * 24 * 60 * 60)


def delete_key(key):
    redis_handler.redis.delete(key)


def get_agent_number(token):
    return redis_handler.get(AGENT_TOKEN + token)


# 缓存坐席
def add_agent_info(call_info: CallInfo = None, agent: AgentInfo = None, call_id=None, device_id=None):
    if call_info and not agent:
        agent = get_agent_info(call_info.saas_id, call_info.agent_key)
    if not agent:
        return
    if call_id:
        agent.call_id = call_id
    if device_id:
        agent.device_id = device_id
    redis_handler.set(AGENT_INFO + agent.saas_id + ":" + agent.agent_number, agent.to_json_string(), cacheDay * 24 * 60 * 60)


# 缓存CALL_INFO
def add_call_info(call: CallInfo):
    for k, v in call.device_info_map.items():
        add_device(k, call.call_id)
    # print('add_call_info', call.call_id, call.to_json_string())
    redis_handler.set(CALL_INFO + call.call_id, call.to_json_string(), cacheDay * 24 * 60 * 60)


def get_call_info_by_device_id(device_id):
    call_id = deviceCall.get(device_id)
    if not call_id:
        return get_call_info(call_id)


# 获取callInfo
def get_call_info(call_id):
    text = None
    if call_id:
        text = redis_handler.get(CALL_INFO + call_id)
        # print('get_call_info', call_id, text)
        # sys.stdout.flush()  # 强制刷新输出缓冲区
    if text:
        return CallInfo.from_json(text)


def remove_call_info(call_id):
    if not call_id:
        return None
    call_info = get_call_info(call_id)
    if call_info and call_info.device_info_map:
        for k, v in call_info.device_info_map.items():
            deviceCall.pop(k)
    delete_key(CALL_INFO + call_id)


def add_device(device_id, call_id):
    if not device_id or not call_id:
        return None
    deviceCall[device_id] = call_id


def get_user_part(device_id):
    return deviceUserPart.get(device_id)


def add_device_user_part(device_id, user_part):
    if not device_id or not user_part:
        return
    deviceUserPart[device_id] = user_part


def get_route_gateway(saas_id):
    return RouteGateway(id=1,
                        saas_id=saas_id,
                        name='63366692',
                        media_host='192.168.20.99',
                        media_port=5060,
                        caller_prefix='',
                        called_prefix='',
                        status=0)


def get_agent_phone(saas_id, agent_num):
    return Phone.query.filter(Phone.saas_id == saas_id, Phone.phone_num == agent_num).first()

def add_delay_message(action, delay_action, timeouts):
    delay_action.uuid = uuid.uuid4()
    key = CTI_ENGINE_DELAY_ACTION % action
    msg = delay_action.to_json_string()
    action_time = datetime.utcnow().timestamp() + timeouts * 1000
    redis_handler.redis.zadd(key, {msg : action_time})

def get_delay_message(action):
    key = CTI_ENGINE_DELAY_ACTION % action
    current_time = int(time.time() * 1000)  # 毫秒级时间戳
    members = redis_handler.redis.zrangebyscore(key, 0, current_time, start=0, num=DELAY_ACTION_BATCH_SIZE, withscores=True)
    if not members:
        return []
    # scored_entries = [{"member": entry[0].decode('utf-8'), "score": entry[1]} for entry in members]
    action_list = [entry[0].decode('utf-8') for entry in members]
    if action_list:
        redis_handler.redis.zrem(key, *action_list)
    return action_list

def lock_delay_action(val):
    key = CTI_ENGINE_DELAY_ACTION_LOCK % val
    return redis_handler.redis.set(key, "1", ex=1000*10, nx=True)