Merge pull request #12 from Code-Partners/feat-text-protocol

Feat text protocol

Author: SergeyCP

formatters/formatters.py common/__init__.pyformatters/formatters.py renamed to common/__init__.py

@@ -9,3 +9,7 @@ class Level(Enum):
     ERROR = 4
     FATAL = 5
     CONTROL = 6
+
+    def __str__(self):
+        return "%s" % self._name_
+

common/level.py

@@ -0,0 +1,117 @@
+import logging
+import typing
+
+from typing import List
+
+from common.tokens.token_abc import Token
+from common.tokens.token_factory import TokenFactory
+from packets.log_entry.log_entry import LogEntry, LogEntryType
+
+logger = logging.getLogger(__name__)
+
+
+class PatternParser:
+    _SPACES: str = " " * 3
+
+    def __init__(self) -> None:
+        self._tokens: List[Token] = list()
+        self._buffer: list = []
+        self._pattern: str = str()
+        self._indent_level: int = 0
+        self._indent: bool = False
+        self._position = 0
+
+    def expand(self, log_entry: LogEntry) -> str:
+        size = len(self._tokens)
+
+        if size == 0:
+            return ""
+
+        self._buffer.clear()
+        if log_entry.log_entry_type == LogEntryType.LEAVE_METHOD:
+            if self._indent_level > 0:
+                self._indent_level -= 1
+                logger.debug("Decreased indent level when leaving method, new indent level is %d" % self._indent_level)
+
+        for token in self._tokens:
+            token: Token
+            if self.indent and token.indent:
+                for i in range(self._indent_level):
+                    self._buffer.append(self._SPACES)
+
+            expanded = token.expand(log_entry)
+            width = token.width
+
+            if width < 0:
+                # left-aligned
+                self._buffer.append(expanded)
+                pad = -width - len(expanded)
+                for i in range(pad):
+                    self._buffer.append(" ")
+            elif width > 0:
+                pad = width - len(expanded)
+                for i in range(pad):
+                    self._buffer.append(" ")
+                # right-aligned
+                self._buffer.append(expanded)
+            else:
+                self._buffer.append(expanded)
+        if log_entry.log_entry_type == LogEntryType.ENTER_METHOD:
+            self._indent_level += 1
+            logger.debug("Added indent level when entering method, new indent level is %d" % self._indent_level)
+
+        return "".join(self._buffer)
+
+    def _next(self) -> typing.Optional[Token]:
+        length = len(self._pattern)
+        if self._position < length:
+            is_variable = False
+            pos: int = self._position
+
+            if self._pattern[pos] == "$":
+                is_variable = True
+                pos += 1
+
+            while pos < length:
+                if self._pattern[pos] == "$":
+                    if is_variable:
+                        pos += 1
+                    break
+                pos += 1
+
+            value = self._pattern[self._position: pos]
+            self._position = pos
+
+            return TokenFactory.get_token(value)
+        else:
+            return None
+
+    def _parse(self) -> None:
+        self._tokens.clear()
+        token = self._next()
+        while token is not None:
+            self._tokens.append(token)
+            token = self._next()
+
+    @property
+    def pattern(self) -> str:
+        return self._pattern
+
+    @pattern.setter
+    def pattern(self, pattern: str) -> None:
+        if not isinstance(pattern, str):
+            raise TypeError("pattern must be an str")
+        self._position = 0
+        self._pattern = pattern.strip() if pattern else ""
+        self._parse()
+
+    @property
+    def indent(self) -> bool:
+        return self._indent
+
+    @indent.setter
+    def indent(self, indent: bool) -> None:
+        if not isinstance(indent, bool):
+            raise TypeError("indent must be a bool")
+
+        self._indent = indent

common/pattern_parser.py

@@ -0,0 +1,48 @@
+from abc import ABC, abstractmethod
+
+from packets.log_entry.log_entry import LogEntry
+
+
+class Token(ABC):
+    _value: str
+    _options: str
+    _width: int
+
+    @staticmethod
+    @abstractmethod
+    def expand(log_entry: LogEntry) -> str:
+        ...
+
+    @property
+    def value(self) -> str:
+        return self._value
+
+    @value.setter
+    def value(self, value: str) -> None:
+        if not isinstance(value, str):
+            raise TypeError("value must be an str")
+        self._value = value
+
+    @property
+    def options(self) -> str:
+        return self._options
+
+    @options.setter
+    def options(self, options: str) -> None:
+        if not isinstance(options, str):
+            raise TypeError("options must be an str")
+        self._options = options
+
+    @property
+    def indent(self) -> bool:
+        return False
+
+    @property
+    def width(self) -> int:
+        return self._width
+
+    @width.setter
+    def width(self, width: int) -> None:
+        if not isinstance(width, int):
+            raise TypeError("width must be an int")
+        self._width = width

common/tokens/__init__.py

@@ -0,0 +1,94 @@
+from .tokens import *
+from .token_abc import Token
+
+
+class TokenFactory:
+    tokens = {
+        "$appname$": AppNameToken,
+        "$session$": SessionToken,
+        "$hostname$": HostNameToken,
+        "$title$": TitleToken,
+        "$timestamp$": TimestampToken,
+        "$level$": LevelToken,
+        "$color$": ColorToken,
+        "$logentrytype$": LogEntryTypeToken,
+        "$viewerid$": ViewerIdToken,
+        "$thread$": ThreadIdToken,
+        "$process$": ProcessIdToken,
+    }
+
+    @staticmethod
+    def _create_literal(value: str) -> Token:
+        if not isinstance(value, str):
+            raise TypeError("value must be an str")
+        token = LiteralToken()
+        token.options = ""
+        token.value = value
+        token.width = 0
+        return token
+
+    @classmethod
+    def get_token(cls, value: str) -> Token:
+        if not isinstance(value, str):
+            raise TypeError("value must be an str")
+
+        length = len(value)
+
+        if length <= 2:
+            return cls._create_literal(value)
+
+        if value[0] != "$" or value[-1] != "$":
+            return cls._create_literal(value)
+
+        original = value
+        options = ""
+
+        # extract the token options: $token{options}$
+        if value[-2] == "}":
+            idx = value.find("{")
+
+            if idx > -1:
+                idx += 1
+                options = value[idx: -2]
+                value = value[:idx - 1] + value[-1]
+
+        width = ""
+        idx = value.find(",")
+
+        # extract the token width: $token, width$
+        if idx > -1:
+            idx += 1
+            width = value[idx: -1]
+            value = value[: idx - 1] + value[-1]
+
+        value = value.lower()
+        impl = cls.tokens.get(value)
+        if impl is None:
+            return cls._create_literal(original)
+
+        # noinspection PyBroadException
+        try:
+            token: Token = impl()
+            token.options = options
+            token.value = original
+            token.width = cls._parse_width(width)
+        except Exception:
+            return cls._create_literal(original)
+
+        return token
+
+    @staticmethod
+    def _parse_width(value: str) -> int:
+        if not isinstance(value, str):
+            raise TypeError("value must be an str")
+
+        value = value.strip()
+        if len(value) == 0:
+            return 0
+
+        try:
+            width = int(value)
+        except ValueError:
+            width = 0
+
+        return width

common/tokens/token_abc.py

@@ -0,0 +1,121 @@
+import logging
+from datetime import datetime
+
+from common.tokens.token_abc import Token
+from packets.log_entry.log_entry import LogEntry
+from session.session import Session
+
+logger = logging.getLogger(__name__)
+
+
+class AppNameToken(Token):
+    @staticmethod
+    def expand(log_entry: LogEntry) -> str:
+        return log_entry.appname
+
+
+class SessionToken(Token):
+    @staticmethod
+    def expand(log_entry: LogEntry) -> str:
+        return log_entry.session_name
+
+
+class HostNameToken(Token):
+    @staticmethod
+    def expand(log_entry: LogEntry) -> str:
+        return log_entry.hostname
+
+
+class TitleToken(Token):
+    @staticmethod
+    def expand(log_entry: LogEntry) -> str:
+        return log_entry.title
+
+    @property
+    def indent(self) -> bool:
+        return True
+
+
+class TimestampToken(Token):
+    _FORMAT: str = "%Y-%m-%d %H:%M:%S.%f"
+
+    @staticmethod
+    def _get_timestamp(log_entry: LogEntry) -> datetime:
+        # convert timestamp value stored in LogEntry to seconds
+        timestamp = log_entry.timestamp / 1_000_000
+
+        # convert time to utc
+        offset_from_utc = datetime.fromtimestamp(timestamp).astimezone().utcoffset().total_seconds()
+        timestamp -= offset_from_utc
+        return datetime.fromtimestamp(timestamp)
+
+    def expand(self, log_entry: LogEntry) -> str:
+        timestamp = self._get_timestamp(log_entry)
+
+        options = self.options
+        if options is not None and len(options) > 0:
+            fmt = options
+            logger.debug("Timestamp format is set by options string:  %s " % fmt)
+            return timestamp.strftime(fmt)
+
+        logger.debug("Timestamp format is default: %s " % self._FORMAT)
+        return timestamp.strftime(self._FORMAT)
+
+
+class LevelToken(Token):
+
+    @staticmethod
+    def expand(log_entry: LogEntry) -> str:
+        return str(log_entry.level)
+
+
+class ColorToken(Token):
+    _CHAR_MAP: str = "0123456789ABCDEF"
+
+    def _append_hex(self, string_buffer: list, value: int) -> list:
+        value &= 0xff
+        string_buffer.append(self._CHAR_MAP[value & 0xf])
+        string_buffer.append(self._CHAR_MAP[value >> 4])
+        return string_buffer
+
+    def expand(self, log_entry: LogEntry) -> str:
+        color = log_entry.color
+
+        if color is not None and color != Session.DEFAULT_COLOR:
+            string_buffer = ["0x"]
+            self._append_hex(string_buffer, color.get_red())
+            self._append_hex(string_buffer, color.get_green())
+            self._append_hex(string_buffer, color.get_blue())
+
+            return "".join(string_buffer)
+        else:
+            return "<default>"
+
+
+class LogEntryTypeToken(Token):
+    @staticmethod
+    def expand(log_entry: LogEntry) -> str:
+        return str(log_entry.log_entry_type)
+
+
+class ViewerIdToken(Token):
+    @staticmethod
+    def expand(log_entry: LogEntry) -> str:
+        return str(log_entry.viewer_id)
+
+
+class ThreadIdToken(Token):
+    @staticmethod
+    def expand(log_entry: LogEntry) -> str:
+        return str(log_entry.thread_id)
+
+
+class ProcessIdToken(Token):
+    @staticmethod
+    def expand(log_entry: LogEntry) -> str:
+        return str(log_entry.process_id)
+
+
+class LiteralToken(Token):
+    def expand(self, log_entry: LogEntry) -> str:
+        return self.value

common/tokens/token_factory.py

@@ -23,3 +23,6 @@ class ViewerId(Enum):
     JPEG = 401
     ICON = 402
     METAFILE = 403
+
+    def __str__(self):
+        return "%s" % self._name_