openrvdas.logger.transforms.strip_transform

 1#!/usr/bin/env python3
 2"""Strip undesired characters out of a record."""
 3
 4import logging
 5
 6from logger.transforms.transform import Transform  # noqa: E402
 7
 8
 9################################################################################
10class StripTransform(Transform):
11    def __init__(self, chars=None, unprintable=False,
12                 strip_prefix=False, strip_suffix=False, **kwargs):
13        """
14        ```
15        chars     A string of characters that are to be stripped out of the
16                  record. By default, the transform will remove them regardless
17                  of where in the record they occur. If omitted, will strip standard
18                  whitespace characters.
19
20        unprintable - As an alternative to specifying 'chars', setting unprintable
21                  to True will strip out all unprintable characters.
22
23        strip_prefix - If true, strip characters at the start of the record,
24                  rather than everywhere (compatible with strip_suffix).
25
26        strip_suffix - If true, strip characters at the end of the record,
27                  rather than everywhere (compatible with strip_prefix).
28        ```
29        """
30        super().__init__(**kwargs)  # processes 'quiet' and type hints
31
32        if chars and unprintable:
33            raise ValueError('Can not specify both "chars" and "unprintable".')
34        self.chars = chars or ' \t\v\r\n\f'
35        self.unprintable = unprintable
36        self.strip_prefix = strip_prefix
37        self.strip_suffix = strip_suffix
38
39    ############################
40    def transform(self, record: str) -> str:
41        """Strip and return only the requested fields."""
42
43        # See if it's something we can process, and if not, try digesting
44        if not self.can_process_record(record):  # inherited from BaseModule()
45            return self.digest_record(record)  # inherited from BaseModule()
46
47        if not type(record) is str:
48            logging.warning('SplitTransform received non-string input: %s', record)
49            return None
50
51        # If we're stripping unprintables
52        if self.unprintable:
53            if not self.strip_prefix and not self.strip_suffix:
54                record = ''.join([c for c in record if c.isprintable()])
55            if self.strip_prefix:
56                while record and not record[0].isprintable():
57                    record = record[1:]
58            if self.strip_suffix:
59                while record and not record[-1].isprintable():
60                    record = record[:-1]
61
62        # If we're working with a specified character set
63        else:
64            if not self.strip_prefix and not self.strip_suffix:
65                record = ''.join([c for c in record if c not in self.chars])
66            if self.strip_prefix:
67                record = record.lstrip(self.chars)
68            if self.strip_suffix:
69                record = record.rstrip(self.chars)
70
71        return record