decoder.py 1.8 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869
  1. from .message import message
  2. from .message_builder import message_builder
  3. from .message_step_builder import message_step_builder
  4. from .protocol import protocol
  5. class decoder(protocol):
  6. def __init__(self, content: str) -> None:
  7. self.__content = content
  8. self.__builder = message_step_builder()
  9. def process(self) -> object:
  10. if self.__content is None:
  11. return self
  12. for line in self._lines:
  13. self.__parse_line(line)
  14. self.__content = None
  15. return self
  16. def result(self, message_type: type | None = None) -> message:
  17. if self.__content is not None:
  18. self.process()
  19. if message_type is None:
  20. return self._builder.build()
  21. return message_type(self._builder.build())
  22. @property
  23. def _lines(self) -> str:
  24. return self.__content.split(self._line_separator)
  25. @property
  26. def _builder(self) -> message_step_builder:
  27. return self.__builder
  28. def __parse_line(self, line: str) -> None:
  29. line = line.strip()
  30. if self.__is_section(line):
  31. self.__parse_section(line)
  32. return
  33. self.__parse_key_value(line)
  34. def __is_section(self, line: str) -> bool:
  35. if len(line) < 1:
  36. return False
  37. return line[0] == self._section_buckles[0]
  38. def __parse_key_value(self, line: str) -> None:
  39. position = line.find("=")
  40. if position == -1:
  41. return
  42. key = line[:position].strip()
  43. position = position + 1
  44. value = line[position:].strip()
  45. self._builder.add(key, value)
  46. def __parse_section(self, line: str) -> None:
  47. for char in self._section_buckles:
  48. line = line.replace(char, "")
  49. self._builder.change_section(line)