from typing import Dict, List, Tuple from httpx import AsyncClient from loguru import logger from app.services.duckling import Duckling from app.services.tencent_nlp import TencentNLP class MeetingInfoCatcher: def __init__(self, nlp_service: TencentNLP, duckling: Duckling): super(MeetingInfoCatcher, self).__init__() self.nlp_service = nlp_service self.duckling = duckling async def extract_time(self, sentence: str) -> Tuple[str, str, int]: start_time, end_time, duration = '', '', -1 parsed = await self.duckling.parse(sentence) for dim in parsed: if dim['dim'] == 'time': start_time = dim['value']['from']['value'] end_time = dim['value']['to']['value'] if dim['dim'] == 'duration': duration = dim['value']['normalized']['value'] return start_time, end_time, duration async def extract_room_size(self, sentence: str) -> str: dp_tokens = await self.nlp_service.get_dependency(sentence) size = '' for token in dp_tokens: if await self.nlp_service.get_word_similarity(token.Word, '会议室') > 0.8: index = token.Id for item in dp_tokens: if item.HeadId == index: if await self.nlp_service.get_word_similarity(item.Word, '小') > 0.9: size = 'small' if await self.nlp_service.get_word_similarity(item.Word, '中') > 0.9: size = 'medium' if await self.nlp_service.get_word_similarity(item.Word, '大') > 0.9: size = 'large' break return size async def extract_topic(self, sentence: str) -> str: summarization = await self.nlp_service.get_auto_summarization_result(sentence) return summarization async def extract_name(self, sentence: str) -> List[str]: _, ner_tokens = await self.nlp_service.get_lexical_analysis_result(sentence) name_list = [] if ner_tokens: for token in ner_tokens: if token.Type == 'PER': name_list.append(token.Word) return name_list async def run(self, sentence: str) -> Tuple: similarity = await self.nlp_service.get_text_similarity_result('我要开会', [sentence]) if similarity[-1].Score < 0.5: return '', '', -1, '', '', [] else: start_time, end_time, interval = await self.extract_time(sentence) topic = await self.extract_topic(sentence) name_list = await self.extract_name(sentence) room_size = await self.extract_room_size(sentence) return start_time, end_time, interval, room_size, topic, name_list @logger.catch() async def get_caught_result(sentence: str) -> Tuple: async with AsyncClient() as client: duckling = Duckling(client) service = TencentNLP() catcher = MeetingInfoCatcher(service, duckling) return await catcher.run(sentence)