normalizer.py 3.7 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114
  1. from typing import List, Optional
  2. import arrow
  3. from loguru import logger
  4. from .helpers.str_common import (
  5. del_keyword,
  6. filter_irregular_expression,
  7. number_translator,
  8. )
  9. from .point import TimePoint
  10. from .result import Result, DeltaType
  11. from .unit import TimeUnit
  12. # 时间表达式识别的主要工作类
  13. class TimeNormalizer:
  14. def __init__(self, isPreferFuture=True, pattern=None):
  15. self.isPreferFuture = isPreferFuture
  16. if pattern is None:
  17. from .resource.pattern import pattern
  18. self.pattern = pattern
  19. def parse(self, target: str, baseTime: arrow.Arrow = None) -> dict:
  20. """
  21. TimeNormalizer的构造方法,baseTime取默认的系统当前时间
  22. :param baseTime: 基准时间点
  23. :param target: 待分析字符串
  24. :return: 时间单元数组
  25. """
  26. if baseTime is None:
  27. baseTime = arrow.now("Asia/Shanghai")
  28. # logger.debug(f"目标字符串: {target}")
  29. self.isTimeDelta = False
  30. self.timeDelta = None # type: Optional[DeltaType]
  31. self.target = target
  32. self.baseTime = baseTime
  33. return self.extract()
  34. def pre(self):
  35. """
  36. 待匹配字符串的清理空白符和语气助词以及大写数字转化的预处理
  37. """
  38. self.target = filter_irregular_expression(self.target)
  39. self.target = del_keyword(self.target, r"\s+") # 清理空白符
  40. self.target = del_keyword(self.target, "[的]+") # 清理语气助词
  41. self.target = number_translator(self.target) # 大写数字转化
  42. # logger.debug(f"清理空白符和语气助词以及大写数字转化的预处理 {self.target}")
  43. def extract(self) -> dict:
  44. """返回 TimeUnit[] 时间表达式类型数组
  45. """
  46. self.pre()
  47. startline = -1
  48. endline = -1
  49. rpointer = 0
  50. temp = []
  51. match = self.pattern.finditer(self.target)
  52. # logger.debug("=======")
  53. # logger.debug("用正则提取关键字:")
  54. for m in match:
  55. # logger.debug(m)
  56. startline = m.start()
  57. if startline == endline:
  58. rpointer -= 1
  59. temp[rpointer] = temp[rpointer] + m.group()
  60. else:
  61. temp.append(m.group())
  62. # logger.debug(f"temp:{temp}")
  63. endline = m.end()
  64. rpointer += 1
  65. # logger.debug("=======")
  66. res: List[TimeUnit] = []
  67. # 时间上下文: 前一个识别出来的时间会是下一个时间的上下文,用于处理:周六3点到5点这样的多个时间的识别,第二个5点应识别到是周六的。
  68. contextTp = TimePoint()
  69. # logger.debug(f"基础时间: {self.baseTime}")
  70. # logger.debug(f"待处理的字段: {temp}")
  71. # logger.debug(f"待处理字段长度: {rpointer}")
  72. for i in range(0, rpointer):
  73. # 这里是一个类嵌套了一个类
  74. res.append(TimeUnit(temp[i], self, contextTp))
  75. contextTp = res[i].tp
  76. # logger.debug(f"全部字段处理后的结果: {res}")
  77. res = self.filter(res)
  78. if self.isTimeDelta and self.timeDelta:
  79. return Result.from_timedelta(self.timeDelta)
  80. if len(res) == 1:
  81. return Result.from_timestamp(res)
  82. if len(res) == 2:
  83. return Result.from_timespan(res)
  84. return Result.from_invalid()
  85. def filter(self, tu_arr: List[TimeUnit]):
  86. """
  87. 过滤掉无效识别。
  88. """
  89. res = []
  90. for tu in tu_arr:
  91. if not tu:
  92. continue
  93. if tu.time.timestamp != 0:
  94. res.append(tu)
  95. # logger.debug(f"过滤无效识别后: {res}")
  96. return res