Added default encoding to config, encapsulated HTML parser started

Author: Sven Siegmund

README.md

@@ -12,3 +12,4 @@ If you find a working official Microsoft link to the RTF specification and add i
 
 * [Swissmains Link to RTF Spec 1.9.1](https://manuals.swissmains.com/pages/viewpage.action?pageId=1376332&preview=%2F1376332%2F10620104%2FWord2007RTFSpec9.pdf)
 * [Webarchive Link to RTF Spec 1.9.1](https://web.archive.org/web/20190708132914/http://www.kleinlercher.at/tools/Windows_Protocols/Word2007RTFSpec9.pdf)
+* [RTF Extensions, MS-OXRTFEX](https://docs.microsoft.com/en-us/openspecs/exchange_server_protocols/ms-oxrtfex/411d0d58-49f7-496c-b8c3-5859b045f6cf)

requirements.txt

@@ -1 +1,2 @@
 argcomplete
+extract-msg

src/rtfparse/bytestring_types.py

@@ -38,16 +38,22 @@ def __init__(self, cfg_path: pathlib.Path, autoconfig: bool) -> None:
         self.path_to_config_file = cfg_path
         self.path_to_home = utils.provide_dir(self.path_to_config_file.parent)
         self.path_to_pyrtfparse_home = pathlib.Path.home() / utils.home_dir_name
-        self._subdir_dir = Preconfigured_Path(
-                internal_name="subdir_dir",
-                path=self.path_to_pyrtfparse_home / "subdir",
-                comment="some subdir",
+        self._email_rtf = Preconfigured_Path(
+                internal_name="email_rtf",
+                path=self.path_to_pyrtfparse_home / "email_rtf",
+                comment="Directory for RTF extracted from MS Outlook emails",
             )
+        self._html = Preconfigured_Path(
+                internal_name="html",
+                path=self.path_to_pyrtfparse_home / "html",
+                comment="Directory for HTML extracted from the email",
+            )
+        self.check_paths = (self._email_rtf,
+                            self._html,
+                           )
         self._wizard_has_run = False
         self.autoconfig = autoconfig
         self.read_config_file()
-        self.check_paths = (self._subdir_dir,
-                           )
         self.integrity_check()
     def __enter__(self):
         return self
@@ -75,7 +81,7 @@ def integrity_check(self) -> None:
             for preconf_path in self.check_paths:
                 path_to_check = preconf_path.path
                 assert path_to_check.exists()
-        except AssertionError as e:
+        except AssertionError as err:
             logger.debug(f"Path not found, starting wizard")
             self.wizard(errors.WrongConfiguration(f"{self.path_to_config_file.name}: '{str(path_to_check)}', path does not exist!", preconf_path), autoconfig=self.autoconfig)
     def create_config_file(self) -> None:
@@ -89,11 +95,15 @@ def create_config_file(self) -> None:
         self.config_parser.set("Paths", "# or a '\\' at the end of the final directory of a Windows path")
         self.config_parser.set("Paths", "# does not interfere with the path parser.")
         self.config_parser.set("Paths", "")
-        for preconf_path in (
-                self._subdir_dir,
-                ):
+        for preconf_path in self.check_paths:
             self.config_parser.set("Paths", f"# {preconf_path.comment[0].capitalize()}{preconf_path.comment[1:]}")
             self.config_parser.set("Paths", f"{preconf_path.internal_name}", f"{preconf_path.path}")
+        self.config_parser.add_section("Encoding")
+        self.config_parser.set("Encoding", "# Set default expected RTF encoding here")
+        self.config_parser.set("Encoding", "# RTF usually uses ANSI by defaul, but that's not specific enough.")
+        self.config_parser.set("Encoding", "# This specifies what exactly we mean by ANSI, e.g. cp1252 (Western Europe)")
+        self.config_parser.set("Encoding", "")
+        self.config_parser.set("Encoding", "default", "cp1252")
         with open(self.path_to_config_file, mode="w", encoding="utf-8") as configfh:
             self.config_parser.write(configfh)
         self.read_config_file()
@@ -121,7 +131,9 @@ def parse(self) -> None:
         Parses the configuration files into usable attributes
         """
         try:
-            self.subdir_dir = self.getpath("Paths", "subdir_dir")
+            self.email_rtf = self.getpath("Paths", "email_rtf")
+            self.html = self.getpath("Paths", "html")
+            self.default_encoding = self.config_parser.get("Encoding", "default")
         except ValueError:
             exc_type, exc_value, exc_traceback = sys.exc_info()
             lines = traceback.format_exc().splitlines()

src/rtfparse/config_loader.py

@@ -4,10 +4,11 @@
 import io
 import logging
 import re
-from itertools import count
+# Own modules
 from rtfparse import re_patterns
 from rtfparse import utils
 from rtfparse import errors
+from rtfparse import config_loader
 from rtfparse.enums import Bytestring_Type
 
 
@@ -19,15 +20,17 @@
 CHARACTER = BACKSLASH = DELIMITER = MINUS = GROUP_END = len(b"\\")
 SYMBOL = IGNORABLE = BACKSLASH + CHARACTER
 GROUP_START = BACKSLASH + IGNORABLE
-MAX_CW_LETTERS = 32
-INTEGER_MAGNITUDE = 32
+MAX_CW_LETTERS = 32 # As specified in RTF Spec
+INTEGER_MAGNITUDE = 32 # As specified in RTF Spec
 PLAIN_TEXT = CONTROL_WORD = BACKSLASH + MAX_CW_LETTERS + MINUS + len(str((1 << INTEGER_MAGNITUDE) // 2)) + DELIMITER
 
 
 class Entity:
+    def __init__(self) -> None:
+        self.text = ""
     @classmethod
     def probe(cls, pattern: re_patterns.Bytes_Regex, file: io.BufferedReader) -> Bytestring_Type:
-        logger.debug(f"in Entity.probed")
+        logger.debug(f"Probing file at position {file.tell()}")
         original_position = file.tell()
         while True:
             probed = file.read(len(re_patterns.probe_pattern))
@@ -54,84 +57,101 @@ def probe(cls, pattern: re_patterns.Bytes_Regex, file: io.BufferedReader) -> Byt
                     raise errors.UnexpectedEndOfFileError(f"at position {file.tell()}")
                 continue
             break
-        logger.debug(f"{result = }")
+        logger.debug(f"Probe {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__")
+    def __init__(self, config: config_loader.Config, file: io.BufferedReader) -> None:
+        super().__init__()
+        self.config = config
+        logger.debug(f"Reading Control Word at file position {file.tell()}")
         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").decode("ascii")
-            logger.debug(f"{self.control_name = }")
+            self.control_name = match.group("control_name").decode(self.config.default_encoding)
+            logger.debug(f"Preliminary {self.control_name = }")
             parameter = match.group("parameter")
-            if parameter:
-                self.parameter = int(parameter.decode("ascii"))
+            if parameter is not None:
+                self.parameter = int(parameter.decode(self.config.default_encoding))
                 logger.debug(f"{self.parameter = }")
+                self.control_name = self.control_name.removesuffix(str(self.parameter))
+                logger.debug(f"Final {self.control_name = }")
             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'))}")
+                logger.debug(f"Delimiter is {match.group('other').decode(self.config.default_encoding)}, 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}>"
+        return f"<{self.__class__.__name__}: {self.control_name}{self.parameter}>"
 
 
 class Control_Symbol(Entity):
-    def __init__(self, file: io.BufferedReader) -> None:
+    def __init__(self, config: config_loader.Config, file: io.BufferedReader) -> None:
+        super().__init__()
+        self.config = config
         self.start_position = file.tell()
-        logger.debug(f"Starting at file position {self.start_position}")
-        self.text = file.read(SYMBOL)[-1].decode("ascii")
+        logger.debug(f"Reading Symbol at file position {self.start_position}")
+        self.char = ""
+        self.text = chr(file.read(SYMBOL)[-1])
+        if self.text == "'":
+            self.char = file.read(SYMBOL).decode(self.config.default_encoding)
+            self.text = bytes((int(self.char, base=16), )).decode(self.config.default_encoding)
+            logger.debug(f"Encountered escaped ANSI character, read two more bytes: {self.char}, character: {self.text}")
+            if self.text in "\\{}":
+                file.seek(file.tell() - SYMBOL)
     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}")
+    def __init__(self, config: config_loader.Config, file: io.BufferedReader) -> None:
+        super().__init__()
+        self.config = config
         self.text = ""
+        logger.debug(f"Constructing Plain_Text")
         while True:
+            self.start_position = file.tell()
             read = file.read(PLAIN_TEXT)
-            logger.debug(f"Read file up to position {file.tell()}")
-            logger.debug(f"Read: {read}")
+            logger.debug(f"Read file from {self.start_position} to position {file.tell()}, 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")
+                _text = match.group("text").decode(self.config.default_encoding)
                 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()}")
+                    file.seek(self.start_position + len(_text))
                     break
             else:
+                file.seek(self.start_position)
                 break
+        logger.debug(f"Returned to position {file.tell()}")
     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__")
+class Group(Entity):
+    def __init__(self, config: config_loader.Config, file: io.BufferedReader) -> None:
+        super().__init__()
+        logger.debug(f"Group.__init__")
+        self.config = config
         self.known = False
         self.name = "unknown"
         self.ignorable = False
         self.structure = list()
-        logger.debug(f"Creating destination group from {file.name}")
+        parsed_object = utils.what_is_being_parsed(file)
+        logger.debug(f"Creating destination group from {parsed_object}")
         self.start_position = file.tell()
         logger.debug(f"Starting at file position {self.start_position}")
         probe = file.read(GROUP_START)
@@ -145,23 +165,24 @@ def __init__(self, file: io.BufferedReader) -> None:
         else:
             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))
+                self.structure.append(Control_Word(self.config, 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))
+                self.structure.append(Group(self.config, file))
             elif probed is Bytestring_Type.CONTROL_SYMBOL:
-                self.structure.append(Control_Symbol(file))
+                self.structure.append(Control_Symbol(self.config, file))
             else:
-                self.structure.append(Plain_Text(file))
+                self.structure.append(Plain_Text(self.config, file))
     def __repr__(self) -> str:
-        return f"<{self.__class__.__name__}: {self.cw.control_name}{self.cw.parameter}>"
+        cwinfo = ""
+        if isinstance(self.structure[0], Control_Word):
+            cwinfo = f" {self.structure[0].control_name}"
+        return f"<Group{cwinfo}>"
 
 
 if __name__ == "__main__":

src/rtfparse/entities.py

@@ -7,6 +7,7 @@
 import logging.config
 import argparse
 import argcomplete
+import io
 from argcomplete.completers import EnvironCompleter as EC
 from itertools import filterfalse
 # Own modules
@@ -46,14 +47,32 @@ def argument_parser() -> argparse.ArgumentParser:
     parser.add_argument("-v", "--version", action="store_true", help="print out rtfparse version and exit").completer = EC
     parser.add_argument("--autoconfig", action="store_true", help="Configure rtfparse automatically").completer = EC
     parser.add_argument("-f", "--file", action="store", metavar="PATH", type=pathlib.Path, help="path to the rtf file").completer = EC
+    parser.add_argument("-m", "--msg", action="store", metavar="PATH", type=pathlib.Path, help="Parse RTF from MS Outlook's .msg file").completer = EC
     return parser
 
 
 def run(config: config_loader.Config) -> None:
     if config.cli_args.file and config.cli_args.file.exists():
         rp = Rtf_Parser()
         with open(config.cli_args.file, mode="rb") as rtf_file:
-            rp.parse_file(rtf_file)
+            rp.parse_file(config, rtf_file)
+    elif config.cli_args.msg:
+        import extract_msg as em
+        import compressed_rtf as cr
+        msg = em.openMsg(f"{config.cli_args.msg}")
+        decompressed_rtf = cr.decompress(msg.compressedRtf)
+        with open((config.email_rtf / config.cli_args.msg.name).with_suffix(".rtf"), mode="wb") as email_rtf:
+            email_rtf.write(decompressed_rtf)
+        with io.BytesIO(decompressed_rtf) as rtf_file:
+            rp = Rtf_Parser()
+            rp.parse_file(config, rtf_file)
+        from rtfparse.renderers import encapsulated_html
+        renderer = encapsulated_html.Encapsulated_HTML()
+        with open((config.html / config.cli_args.msg.name).with_suffix(".html"), mode="w") as htmlfile:
+            logger.info(f"Rendering the encapsulated HTML")
+            renderer.render(rp.parsed, htmlfile)
+            logger.info(f"Encapsulated HTML rendered")
+
 
 
 def cli_start(version) -> None:

src/rtfparse/entry.py

@@ -53,7 +53,7 @@ def create_dict_config(directory: pathlib.Path, all_log: str, info_log: str, err
 
     root_console_handler_conf = {
         "class": "logging.StreamHandler",
-        "level": "DEBUG",
+        "level": "INFO",
         "formatter": "console_formatter",
         "stream": "ext://sys.stdout",
     }