spl str value manager

Author: oleksandr.volha

uncoder-core/app/translator/platforms/base/spl/escape_manager.py

@@ -6,9 +6,7 @@
 
 
 class SplEscapeManager(EscapeManager):
-    escape_map: ClassVar[dict[str, list[EscapeDetails]]] = {
-        ValueType.value: [EscapeDetails(pattern='("|(?<!\\\\)\\\\(?![*?\\\\]))')]
-    }
+    escape_map: ClassVar[dict[str, list[EscapeDetails]]] = {ValueType.value: [EscapeDetails(pattern=r"([<>=\"'\|\\])")]}
 
 
 spl_escape_manager = SplEscapeManager()

uncoder-core/app/translator/platforms/base/spl/renders/spl.py

@@ -20,55 +20,67 @@
 from typing import Union
 
 from app.translator.const import DEFAULT_VALUE_TYPE
+from app.translator.core.custom_types.values import ValueType
 from app.translator.core.exceptions.render import UnsupportedRenderMethod
 from app.translator.core.render import BaseFieldValueRender, PlatformQueryRender
-from app.translator.platforms.base.spl.escape_manager import spl_escape_manager
+from app.translator.core.str_value_manager import StrValue
+from app.translator.platforms.base.spl.str_value_manager import spl_str_value_manager
 
 
 class SplFieldValueRender(BaseFieldValueRender):
-    escape_manager = spl_escape_manager
+    str_value_manager = spl_str_value_manager
+
+    @staticmethod
+    def _wrap_str_value(value: str) -> str:
+        return f'"{value}"'
+
+    def _pre_process_value(
+        self, field: str, value: Union[int, str, StrValue], value_type: str = ValueType.value, wrap_str: bool = False
+    ) -> Union[int, str]:
+        value = super()._pre_process_value(field, value, value_type=value_type, wrap_str=wrap_str)
+        return self._wrap_str_value(str(value)) if not isinstance(value, str) else value
 
     def equal_modifier(self, field: str, value: DEFAULT_VALUE_TYPE) -> str:
         if isinstance(value, list):
             return f"({self.or_token.join([self.equal_modifier(field=field, value=v) for v in value])})"
-        return f'{field}="{self.apply_value(value)}"'
+        return f"{field}={self._pre_process_value(field, value, wrap_str=True)}"
 
     def less_modifier(self, field: str, value: Union[int, str]) -> str:
-        return f'{field}<"{self.apply_value(value)}"'
+        return f"{field}<{self._pre_process_value(field, value, wrap_str=True)}"
 
     def less_or_equal_modifier(self, field: str, value: Union[int, str]) -> str:
-        return f'{field}<="{self.apply_value(value)}"'
+        return f"{field}<={self._pre_process_value(field, value, wrap_str=True)}"
 
     def greater_modifier(self, field: str, value: Union[int, str]) -> str:
-        return f'{field}>"{self.apply_value(value)}"'
+        return f"{field}>{self._pre_process_value(field, value, wrap_str=True)}"
 
     def greater_or_equal_modifier(self, field: str, value: Union[int, str]) -> str:
-        return f'{field}>="{self.apply_value(value)}"'
+        return f"{field}>={self._pre_process_value(field, value, wrap_str=True)}"
 
     def not_equal_modifier(self, field: str, value: DEFAULT_VALUE_TYPE) -> str:
         if isinstance(value, list):
             return f"({self.or_token.join([self.not_equal_modifier(field=field, value=v) for v in value])})"
-        return f'{field}!="{self.apply_value(value)}"'
+        return f"{field}!={self._pre_process_value(field, value, wrap_str=True)}"
 
     def contains_modifier(self, field: str, value: DEFAULT_VALUE_TYPE) -> str:
         if isinstance(value, list):
             return f"({self.or_token.join([self.contains_modifier(field=field, value=v) for v in value])})"
-        return f'{field}="*{self.apply_value(value)}*"'
+        return f'{field}="*{self._pre_process_value(field, value)}*"'
 
     def endswith_modifier(self, field: str, value: DEFAULT_VALUE_TYPE) -> str:
         if isinstance(value, list):
             return f"({self.or_token.join([self.endswith_modifier(field=field, value=v) for v in value])})"
-        return f'{field}="*{self.apply_value(value)}"'
+        return f'{field}="*{self._pre_process_value(field, value)}"'
 
     def startswith_modifier(self, field: str, value: DEFAULT_VALUE_TYPE) -> str:
         if isinstance(value, list):
             return f"({self.or_token.join([self.startswith_modifier(field=field, value=v) for v in value])})"
-        return f'{field}="{self.apply_value(value)}*"'
+        return f'{field}="{self._pre_process_value(field, value)}*"'
 
     def keywords(self, field: str, value: DEFAULT_VALUE_TYPE) -> str:
         if isinstance(value, list):
             return f"({self.or_token.join(self.keywords(field=field, value=v) for v in value)})"
-        return f'"{self.apply_value(value)}"'
+        return f"{self._pre_process_value(field, value, wrap_str=True)}"
 
     def regex_modifier(self, field: str, value: DEFAULT_VALUE_TYPE) -> str:  # noqa: ARG002
         raise UnsupportedRenderMethod(platform_name=self.details.name, method="Regex Expression")

uncoder-core/app/translator/platforms/base/spl/str_value_manager.py

@@ -0,0 +1,55 @@
+"""
+Uncoder IO Community Edition License
+-----------------------------------------------------------------
+Copyright (c) 2023 SOC Prime, Inc.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+-----------------------------------------------------------------
+"""
+from typing import ClassVar
+
+from app.translator.core.str_value_manager import BaseSpecSymbol, StrValue, StrValueManager, UnboundLenWildCard
+from app.translator.platforms.base.spl.escape_manager import spl_escape_manager
+
+
+class SplStrValueManager(StrValueManager):
+    escape_manager = spl_escape_manager
+    str_spec_symbols_map: ClassVar[dict[str, type[BaseSpecSymbol]]] = {"*": UnboundLenWildCard}
+
+    def from_str_to_container(self, value: str) -> StrValue:
+        split = []
+        prev_char = None
+        for char in value:
+            if char == "\\":
+                if prev_char == "\\":
+                    split.append("\\")
+                    prev_char = None
+                    continue
+            elif char in self.str_spec_symbols_map:
+                if prev_char == "\\":
+                    split.append(char)
+                else:
+                    split.append(self.str_spec_symbols_map[char]())
+            elif char in ('"', "=", "|", "<", ">"):
+                split.append(char)
+            else:
+                if prev_char == "\\":
+                    split.append(prev_char)
+                split.append(char)
+
+            prev_char = char
+
+        return StrValue(self.escape_manager.remove_escape(value), self._concat(split))
+
+
+spl_str_value_manager = SplStrValueManager()

uncoder-core/app/translator/platforms/base/spl/tokenizer.py

@@ -29,7 +29,7 @@
 from app.translator.platforms.base.spl.const import NO_QUOTES_VALUES_PATTERN as NO_Q_V_PATTERN
 from app.translator.platforms.base.spl.const import NUM_VALUE_PATTERN as N_V_PATTERN
 from app.translator.platforms.base.spl.const import SINGLE_QUOTES_VALUE_PATTERN as S_Q_V_PATTERN
-from app.translator.platforms.base.spl.escape_manager import spl_escape_manager
+from app.translator.platforms.base.spl.str_value_manager import spl_str_value_manager
 from app.translator.tools.utils import get_match_group
 
 
@@ -57,7 +57,7 @@ class SplTokenizer(QueryTokenizer, ANDLogicOperatorMixin):
 
     wildcard_symbol = "*"
 
-    escape_manager = spl_escape_manager
+    str_value_manager = spl_str_value_manager
 
     def get_operator_and_value(
         self, match: re.Match, mapped_operator: str = OperatorType.EQ, operator: Optional[str] = None
@@ -66,13 +66,13 @@ def get_operator_and_value(
             return mapped_operator, num_value
 
         if (no_q_value := get_match_group(match, group_name=ValueType.no_quotes_value)) is not None:
-            return mapped_operator, no_q_value
+            return mapped_operator, self.str_value_manager.from_str_to_container(no_q_value)
 
         if (d_q_value := get_match_group(match, group_name=ValueType.double_quotes_value)) is not None:
-            return mapped_operator, self.escape_manager.remove_escape(d_q_value)
+            return mapped_operator, self.str_value_manager.from_str_to_container(d_q_value)
 
         if (s_q_value := get_match_group(match, group_name=ValueType.single_quotes_value)) is not None:
-            return mapped_operator, self.escape_manager.remove_escape(s_q_value)
+            return mapped_operator, self.str_value_manager.from_str_to_container(s_q_value)
 
         return super().get_operator_and_value(match, mapped_operator, operator)
 

uncoder-core/app/translator/platforms/logscale/tokenizer.py

@@ -57,7 +57,7 @@ def get_operator_and_value(
             return mapped_operator, num_value
 
         if (d_q_value := get_match_group(match, group_name=ValueType.double_quotes_value)) is not None:
-            return mapped_operator, d_q_value
+            return mapped_operator, self.escape_manager.remove_escape(d_q_value)
 
         if (re_value := get_match_group(match, group_name=ValueType.regex_value)) is not None:
             return OperatorType.REGEX, re_value