AgentCoreService/history_dialogue_service.py

#! /usr/bin/env python
# -*- coding: utf-8 -*-
# vim:fenc=utf-8

import requests
from logging_service import logger
import time

import configs
from message import MessageType


class HistoryDialogueService:
    def __init__(self, base_url: str):
        self.base_url = base_url

    def get_dialogue_history(self, staff_id: str, user_id: str, recent_minutes: int = 1440):
        recent_minutes = min(recent_minutes, 24 * 60)
        time_begin = int(time.time() * 1000) - recent_minutes * 60 * 1000
        url = f"{self.base_url}?sender={staff_id}&receiver={user_id}&time={time_begin}"
        response = requests.post(url, headers={
            'Content-Type': 'application/json'
        })
        if response.status_code != 200:
            raise Exception("Request error [{}]: {}".format(response.status_code, response.text))
        data = response.json()
        if not data.get('success', False):
            raise Exception("Error in response: {}".format(data.get('message', 'no message returned')))
        data = data.get('data', [])
        ret = []
        for record in data:
            sender = record.get('sender')
            if sender == user_id:
                role = 'user'
            elif sender == staff_id:
                role = 'assistant'
            else:
                logger.warning("Unknown sender in dialogue history: {}".format(sender))
                continue
            msg_type = record.get('type', MessageType.TEXT.value)
            message = {
                'role': role,
                'content': record.get('content', ''),
                'timestamp': record.get('sendTime', 0),
                'type': MessageType(msg_type)
            }
            if message['type'] in (MessageType.VOICE_VIDEO_CALL, ):
                logger.warning(f"staff[{staff_id}], user[{user_id}]: skip unsupported message type {message['type']}")
                continue
            ret.append(message)
        ret = sorted(ret, key=lambda x: x['timestamp'])
        return ret


if __name__ == '__main__':
    api_url = configs.get()['storage']['history_dialogue']['api_base_url']
    service = HistoryDialogueService(api_url)
    resp = service.get_dialogue_history(staff_id='1688854492669990', user_id='7881301263964433')
    print(resp)