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