env_regulation_api/app/controllers/nlp/meeting.py

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[str, str, int, str, str, list[str]]:
        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[str, str, int, str, str, list[str]]:
    async with AsyncClient() as client:
        duckling = Duckling(client)
        service = TencentNLP()

        catcher = MeetingInfoCatcher(service, duckling)
        return await catcher.run(sentence)