587 Core Rules 1689 Blocked Fix

cdisc_rules_engine/constants/operation_constants.py

@@ -0,0 +1,9 @@
+import re
+
+NO_MATCH_POLICIES = {"keep_original", "set_null", "set_empty", "error"}
+
+REGEX_FLAG_MAP = {
+    "i": re.IGNORECASE,
+    "m": re.MULTILINE,
+    "s": re.DOTALL,
+}

cdisc_rules_engine/models/operation_params.py

@@ -55,6 +55,10 @@ class OperationParams:
     map: List[dict] = None
     original_target: str = None
     regex: str = None
+    find: str = None
+    replace: str = None
+    on_no_match: str = "keep_original"
+    flags: str = ""
     returntype: str = None
     source: str = None
     target: str = None

cdisc_rules_engine/operations/operations_factory.py

@@ -91,6 +91,7 @@
 from cdisc_rules_engine.operations.get_dataset_filtered_variables import (
     GetDatasetFilteredVariables,
 )
+from cdisc_rules_engine.operations.regex_find_replace import RegexFindReplace
 
 
 class OperationsFactory(FactoryInterface):
@@ -146,6 +147,7 @@ class OperationsFactory(FactoryInterface):
         "valid_define_external_dictionary_version": DefineDictionaryVersionValidator,
         "get_dataset_filtered_variables": GetDatasetFilteredVariables,
         "get_xhtml_errors": GetXhtmlErrors,
+        "regex_find_replace": RegexFindReplace,
     }
 
     @classmethod

cdisc_rules_engine/operations/regex_find_replace.py

@@ -0,0 +1,104 @@
+import re
+import pandas as pd
+
+from cdisc_rules_engine.operations.base_operation import BaseOperation
+from cdisc_rules_engine.exceptions.custom_exceptions import OperationError
+from cdisc_rules_engine.constants.operation_constants import (
+    NO_MATCH_POLICIES,
+    REGEX_FLAG_MAP,
+)
+
+
+class RegexFindReplace(BaseOperation):
+    def _execute_operation(self):
+        """
+        Finds and replaces text in a target column using regex pattern matching.
+        Returns a Series with transformed values based on the replace pattern and no_match policy.
+        """
+        operation_id = self.params.operation_id
+        target = self.params.target
+        find = getattr(self.params, "find", None) or getattr(self.params, "regex", None)
+        replace = getattr(self.params, "replace", None)
+        on_no_match = getattr(self.params, "on_no_match", "keep_original")
+        flags_str = getattr(self.params, "flags", "")
+
+        self._validate_required(
+            operation_id, target, find, replace, on_no_match, flags_str
+        )
+
+        if target not in self.evaluation_dataset.columns:
+            raise OperationError(f"Target column not found: {target}")
+
+        flags = self._parse_flags(flags_str)
+        pattern = self._compile_pattern(find, flags)
+
+        source = self.evaluation_dataset[target]
+        transformed = source.map(
+            lambda value: self._transform_value(
+                value=value,
+                pattern=pattern,
+                replace=replace,
+                on_no_match=on_no_match,
+            )
+        )
+
+        return transformed
+
+    def _validate_required(
+        self, operation_id, target, find, replace, on_no_match, flags_str
+    ):
+        if not operation_id:
+            raise OperationError("regex_find_replace requires id (operation_id)")
+        if not target:
+            raise OperationError("regex_find_replace requires name (target)")
+        if not find:
+            raise OperationError("regex_find_replace requires find (or regex)")
+        if replace is None:
+            raise OperationError("regex_find_replace requires replace")
+        if on_no_match not in NO_MATCH_POLICIES:
+            raise OperationError(
+                f"Invalid on_no_match: {on_no_match}. "
+                f"Must be one of {sorted(NO_MATCH_POLICIES)}"
+            )
+        invalid_flags = [f for f in flags_str if f not in REGEX_FLAG_MAP]
+        if invalid_flags:
+            raise OperationError(
+                f"Invalid flags: {''.join(invalid_flags)}. "
+                f"Allowed flags: {''.join(sorted(REGEX_FLAG_MAP.keys()))}"
+            )
+
+    def _parse_flags(self, flags_str):
+        flags = 0
+        for ch in flags_str:
+            flags |= REGEX_FLAG_MAP[ch]
+        return flags
+
+    def _compile_pattern(self, find, flags):
+        try:
+            return re.compile(find, flags)
+        except re.error as exc:
+            raise OperationError(f"Invalid regex pattern '{find}': {exc}") from exc
+
+    def _transform_value(self, value, pattern, replace, on_no_match):
+        if value is None or (isinstance(value, float) and pd.isna(value)):
+            return None
+
+        text = str(value)
+        match = pattern.search(text)
+        if match:
+            try:
+                return pattern.sub(replace, text)
+            except re.error as exc:
+                raise OperationError(
+                    f"Error applying regex pattern '{pattern.pattern}' to value '{text}': {exc}"
+                ) from exc
+
+        if on_no_match == "keep_original":
+            return text
+        if on_no_match == "set_null":
+            return None
+        if on_no_match == "set_empty":
+            return ""
+        raise OperationError(
+            f"No match found for value '{text}' and on_no_match='error'"
+        )

cdisc_rules_engine/utilities/rule_processor.py

@@ -421,6 +421,10 @@ def perform_rule_operations(
                 original_target=original_target,
                 subtract=operation.get("subtract"),
                 regex=operation.get("regex"),
+                find=operation.get("find"),
+                replace=operation.get("replace"),
+                on_no_match=operation.get("on_no_match", "keep_original"),
+                flags=operation.get("flags", ""),
                 returntype=operation.get("returntype"),
                 source=operation.get("source"),
                 standard=standard,