meeting.py 3.3 KB

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586878889909192
  1. from typing import Dict, List, Tuple
  2. from httpx import AsyncClient
  3. from loguru import logger
  4. from app.services.duckling import Duckling
  5. from app.services.tencent_nlp import TencentNLP
  6. class MeetingInfoCatcher:
  7. def __init__(self, nlp_service: TencentNLP, duckling: Duckling):
  8. super(MeetingInfoCatcher, self).__init__()
  9. self.nlp_service = nlp_service
  10. self.duckling = duckling
  11. async def extract_time(self, sentence: str) -> Tuple[str, str, int]:
  12. start_time, end_time, duration = "", "", -1
  13. parsed = await self.duckling.parse(sentence)
  14. for dim in parsed:
  15. if dim["dim"] == "time":
  16. start_time = dim["value"]["from"]["value"]
  17. end_time = dim["value"]["to"]["value"]
  18. if dim["dim"] == "duration":
  19. duration = dim["value"]["normalized"]["value"]
  20. return start_time, end_time, duration
  21. async def extract_room_size(self, sentence: str) -> str:
  22. dp_tokens = await self.nlp_service.get_dependency(sentence)
  23. size = ""
  24. for token in dp_tokens:
  25. if await self.nlp_service.get_word_similarity(token.Word, "会议室") > 0.8:
  26. index = token.Id
  27. for item in dp_tokens:
  28. if item.HeadId == index:
  29. if (
  30. await self.nlp_service.get_word_similarity(item.Word, "小")
  31. > 0.9
  32. ):
  33. size = "small"
  34. if (
  35. await self.nlp_service.get_word_similarity(item.Word, "中")
  36. > 0.9
  37. ):
  38. size = "medium"
  39. if (
  40. await self.nlp_service.get_word_similarity(item.Word, "大")
  41. > 0.9
  42. ):
  43. size = "large"
  44. break
  45. return size
  46. async def extract_topic(self, sentence: str) -> str:
  47. summarization = await self.nlp_service.get_auto_summarization_result(sentence)
  48. return summarization
  49. async def extract_name(self, sentence: str) -> List[str]:
  50. _, ner_tokens = await self.nlp_service.get_lexical_analysis_result(sentence)
  51. name_list = []
  52. if ner_tokens:
  53. for token in ner_tokens:
  54. if token.Type == "PER":
  55. name_list.append(token.Word)
  56. return name_list
  57. async def run(self, sentence: str) -> Tuple:
  58. similarity = await self.nlp_service.get_text_similarity_result(
  59. "我要开会", [sentence]
  60. )
  61. if similarity[-1].Score < 0.5:
  62. return "", "", -1, "", "", []
  63. else:
  64. start_time, end_time, interval = await self.extract_time(sentence)
  65. topic = await self.extract_topic(sentence)
  66. name_list = await self.extract_name(sentence)
  67. room_size = await self.extract_room_size(sentence)
  68. return start_time, end_time, interval, room_size, topic, name_list
  69. @logger.catch()
  70. async def get_caught_result(sentence: str) -> Tuple:
  71. async with AsyncClient() as client:
  72. duckling = Duckling(client)
  73. service = TencentNLP()
  74. catcher = MeetingInfoCatcher(service, duckling)
  75. return await catcher.run(sentence)