Structure parsing works, needs more testing

Author: Sven Siegmund

src/rtfparse/entities.py

@@ -4,8 +4,10 @@
 import io
 import logging
 import re
+from itertools import count
 from rtfparse import re_patterns
 from rtfparse import utils
+from rtfparse import errors
 from rtfparse.enums import Bytestring_Type
 
 
@@ -14,73 +16,152 @@
 
 
 # Constants, number of bytes to read when creating entities
-CHARACTER = BACKSLASH = len(b"\\")
-IGNORABLE = BACKSLASH + len(rb"*")
-GROUP_START = len(rb"x") + IGNORABLE # x = "}" cannot have a rogue brace for vim's auto-indent's sake
-DELIMITER = len(rb" ")
+CHARACTER = BACKSLASH = DELIMITER = MINUS = GROUP_END = len(b"\\")
+SYMBOL = IGNORABLE = BACKSLASH + CHARACTER
+GROUP_START = BACKSLASH + IGNORABLE
 MAX_CW_LETTERS = 32
 INTEGER_MAGNITUDE = 32
-CONTROL_WORD = BACKSLASH + MAX_CW_LETTERS + len(rb"-") + len(str((1 << INTEGER_MAGNITUDE) // 2)) + DELIMITER
+PLAIN_TEXT = CONTROL_WORD = BACKSLASH + MAX_CW_LETTERS + MINUS + len(str((1 << INTEGER_MAGNITUDE) // 2)) + DELIMITER
 
 
 class Entity:
     @classmethod
     def probe(cls, pattern: re_patterns.Bytes_Regex, file: io.BufferedReader) -> Bytestring_Type:
         logger.debug(f"in Entity.probed")
         original_position = file.tell()
-        probed = file.read(len(re_patterns.probe_pattern))
-        logger.debug(f"{probed = }")
-        file.seek(original_position - len(probed))
-        if (match := re_patterns.group_start.match(probed)):
-            result = Bytestring_Type.GROUP_START
-        elif (match := re_patterns.group_end.match(probed)):
-            result = Bytestring_Type.GROUP_END
-        elif (match := re_patterns.control_word.match(probed)):
-            result = Bytestring_Type.CONTROL_WORD
-        elif (match := re_patterns.control_symbol.match(probed)):
-            result = Bytestring_Type.CONTROL_SYMBOL
-        else:
-            result = Bytestring_Type.PLAIN_TEXT
+        while True:
+            probed = file.read(len(re_patterns.probe_pattern))
+            logger.debug(f"{probed = }")
+            file.seek(original_position)
+            logger.debug(f"Probe returned to position {file.tell()}")
+            if (match := re_patterns.group_start.match(probed)):
+                result = Bytestring_Type.GROUP_START
+            elif (match := re_patterns.group_end.match(probed)):
+                result = Bytestring_Type.GROUP_END
+            elif (match := re_patterns.control_word.match(probed)):
+                result = Bytestring_Type.CONTROL_WORD
+            elif (match := re_patterns.control_symbol.match(probed)):
+                result = Bytestring_Type.CONTROL_SYMBOL
+            elif (match := re_patterns.plain_text.match(probed)):
+                result = Bytestring_Type.PLAIN_TEXT
+            else:
+                logger.debug(f"This does not match anything, it's probably a newline, moving on")
+                original_position += 1
+                file.seek(original_position)
+                logger.debug(f"Probe moved to position {file.tell()}")
+                if not probed:
+                    logger.warning(f"Reached unexpected end of file.")
+                    raise errors.UnexpectedEndOfFileError(f"at position {file.tell()}")
+                continue
+            break
         logger.debug(f"{result = }")
+        logger.debug(f"Probe leaving file at position {file.tell()}")
         return result
 
 
 class Control_Word(Entity):
     def __init__(self, file: io.BufferedReader) -> None:
         logger.debug(f"Control_Word.__init__")
+        self.control_name = "missing"
+        self.parameter = ""
         self.start_position = file.tell()
         logger.debug(f"Starting at file position {self.start_position}")
         probe = file.read(CONTROL_WORD)
         if (match := re_patterns.control_word.match(probe)):
-            self.control_name = match.group("control_name")
+            self.control_name = match.group("control_name").decode("ascii")
             logger.debug(f"{self.control_name = }")
-            self.parameter = match.group("parameter")
-            file.seek(self.start_position + match.span()[1])
+            parameter = match.group("parameter")
+            if parameter:
+                self.parameter = int(parameter.decode("ascii"))
+                logger.debug(f"{self.parameter = }")
+            target_position = self.start_position + match.span()[1]
+            if match.group("other"):
+                logger.debug(f"Delimiter is {match.group('other').decode('ascii')}, len: {len(match.group('delimiter'))}")
+                target_position -= len(match.group("delimiter"))
+            file.seek(target_position)
+        else:
+            logger.warning(f"Missing Control Word")
+            file.seek(self.start_position)
+    def __repr__(self) -> str:
+        name = self.control_name
+        return f"<{self.__class__.__name__}: {name}{self.parameter}>"
+
+
+class Control_Symbol(Entity):
+    def __init__(self, file: io.BufferedReader) -> None:
+        self.start_position = file.tell()
+        logger.debug(f"Starting at file position {self.start_position}")
+        self.text = file.read(SYMBOL)[-1].decode("ascii")
+    def __repr__(self) -> str:
+        return f"<{self.__class__.__name__}: {self.text}>"
+
+
+class Plain_Text(Entity):
+    def __init__(self, file: io.BufferedReader) -> None:
+        self.start_position = file.tell()
+        logger.debug(f"Starting at file position {self.start_position}")
+        self.text = ""
+        while True:
+            read = file.read(PLAIN_TEXT)
+            logger.debug(f"Read file up to position {file.tell()}")
+            logger.debug(f"Read: {read}")
+            # see if we have read all the plain text there is:
+            if (match := re_patterns.plain_text.match(read)):
+                logger.debug(f"This matches the plain text pattern")
+                _text = match.group("text").decode("ascii")
+                logger.debug(f"{_text = }")
+                self.text = "".join((self.text, _text))
+                logger.debug(f"{self.text = }")
+                if len(_text) == PLAIN_TEXT:
+                    continue
+                else:
+                    file.seek(self.start_position + len(self.text))
+                    logger.debug(f"Returned to position {file.tell()}")
+                    break
+            else:
+                break
+    def __repr__(self) -> str:
+        return f"<{self.__class__.__name__}: {self.text}>"
 
 
 class Destination_Group(Entity):
     def __init__(self, file: io.BufferedReader) -> None:
         logger.debug(f"Destination_Group.__init__")
-        logger.debug(f"Creating destination group from {file.name}")
         self.known = False
         self.name = "unknown"
         self.ignorable = False
+        self.structure = list()
+        logger.debug(f"Creating destination group from {file.name}")
         self.start_position = file.tell()
         logger.debug(f"Starting at file position {self.start_position}")
         probe = file.read(GROUP_START)
-        logger.debug(f"Read file up to position {file.tell()}")
+        logger.debug(f"Read file up to position {file.tell()}, read {probe = }")
         if (match := re_patterns.group_start.match(probe)):
             self.known = bool(match.group("group_start"))
             self.ignorable = bool(match.group("ignorable"))
             if not self.ignorable:
                 file.seek(-IGNORABLE, io.SEEK_CUR)
                 logger.debug(f"Returned to position {file.tell()}")
-            self.cw = Control_Word(file)
-            self.name = self.cw.control_name
         else:
-            logger.warning(utils.warn(f"Expected group has no start. Creating unknown group"))
-        probed = self.probe(re_patterns.probe, file)
-        self.content = list()
+            logger.warning(utils.warn(f"Expected a group but found no group start. Creating unknown group"))
+            file.seek(self.start_position)
+        self.cw = Control_Word(file)
+        self.name = self.cw.control_name
+        while True:
+            probed = self.probe(re_patterns.probe, file)
+            if probed is Bytestring_Type.CONTROL_WORD:
+                self.structure.append(Control_Word(file))
+            elif probed is Bytestring_Type.GROUP_END:
+                file.read(GROUP_END)
+                break
+            elif probed is Bytestring_Type.GROUP_START:
+                self.structure.append(Destination_Group(file))
+            elif probed is Bytestring_Type.CONTROL_SYMBOL:
+                self.structure.append(Control_Symbol(file))
+            else:
+                self.structure.append(Plain_Text(file))
+    def __repr__(self) -> str:
+        return f"<{self.__class__.__name__}: {self.cw.control_name}{self.cw.parameter}>"
 
 
 if __name__ == "__main__":

src/rtfparse/errors.py

@@ -16,3 +16,7 @@ class WrongConfiguration(Error):
     def __init__(self, message, payload):
         self.message = message
         self.payload = payload
+
+
+class UnexpectedEndOfFileError(Error):
+    pass

src/rtfparse/parser.py

@@ -7,6 +7,7 @@
 from collections import OrderedDict
 from rtfparse import re_patterns
 from rtfparse import entities
+from rtfparse import errors
 
 
 # Setup logging
@@ -59,6 +60,8 @@ def parse_file(self, file: io.BufferedReader) -> None:
         logger.debug(f"Parsing file {file.name}")
         try:
             self.parsed = entities.Destination_Group(file)
+        except errors.UnexpectedEndOfFileError as err:
+            logger.error(f"{err}")
         except Exception as err:
             logger.exception(err)
         finally:

src/rtfparse/re_patterns.py

@@ -33,8 +33,12 @@ def no_capture(content: bytes) -> bytes:
 
 
 _control_characters = rb"\\\{\}"
+_newline = b"\\" + rb"r" + b"\\" + rb"n"
 control_character = group(_control_characters)
 not_control_character = group(rb"^" + _control_characters)
+_control_characters_or_newline = _control_characters + _newline
+control_character_or_newline = group(_control_characters + _newline)
+not_control_character_or_newline = group(rb"^" + _control_characters_or_newline)
 rtf_backslash = named_regex_group("backslash", not_preceded_by(rb"\\", rb"\\"))
 unnamed_rtf_backslash = not_preceded_by(rb"\\", rb"\\")
 _letters = rb"a-zA-Z"
@@ -50,17 +54,17 @@ def no_capture(content: bytes) -> bytes:
 hdigit = named_regex_group("hdigit", group(_hdigits))
 minus = named_regex_group("minus", rb"-?")
 # int16 = minus + digit + rb"{1,5}"
-parameter = named_regex_group("parameter", minus + digit + rb"{1,10}")
+parameter_pattern = named_regex_group("parameter", minus + digit + rb"{1,10}")
 space = named_regex_group("space", rb" ")
 other = named_regex_group("other", group(rb"^" + _letters + _digits))
 
 
 ascii_letter_sequence = named_regex_group("control_name", ascii_letters)
-delimiter = named_regex_group("delimiter", rb"|".join((rb" ", parameter, other, rb"$")))
+delimiter = named_regex_group("delimiter", rb"|".join((space, parameter_pattern, other, rb"$")))
 symbol = named_regex_group("symbol", other)
 control_word_pattern = named_regex_group("control_word", rtf_backslash + ascii_letter_sequence + delimiter)
 pcdata_delimiter = no_capture(rb"|".join((rtf_brace_open, rtf_brace_close, control_word_pattern)))
-plain_text_pattern = named_regex_group("text", not_control_character + rb"+") + no_capture(control_character)
+plain_text_pattern = named_regex_group("text", not_control_character_or_newline + rb"+") + no_capture(rb"|".join((control_character_or_newline, rb"$")))
 probe_pattern = rb".."
 
 
@@ -73,9 +77,9 @@ def regex101(self) -> None:
         print(self.pattern_bytes.decode("ascii"))
 
 
-probe = Bytes_Regex(named_regex_group("probe", probe_pattern))
 meaningful_bs = Bytes_Regex(rtf_backslash)
-# control_word = Bytes_Regex(rtf_backslash + ascii_letter_sequence + delimiter)
+probe = Bytes_Regex(named_regex_group("probe", probe_pattern), flags=re.DOTALL)
+parameter = Bytes_Regex(parameter_pattern)
 control_word = Bytes_Regex(control_word_pattern)
 control_symbol = Bytes_Regex(rtf_backslash + symbol)
 group_start = Bytes_Regex(rtf_brace_open)

src/rtfparse/version.py

@@ -1,4 +1,4 @@
 #!/usr/bin/env python
 
 
-version = "0.2.2"
+version = "0.3.1"