76 lines
2.9 KiB
Python
76 lines
2.9 KiB
Python
import regex
|
|
import re
|
|
|
|
class RegexMoveToCol:
|
|
def __init__(self, match: str, to: str, with_input_iteration: list[str]|None = None):
|
|
self.match = match
|
|
self.to = to
|
|
|
|
if with_input_iteration is not None:
|
|
with_input_iteration = (
|
|
[[with_input_iteration]]
|
|
if isinstance(with_input_iteration, str)
|
|
else [
|
|
x if isinstance(x, list) or isinstance(x, tuple) else [x]
|
|
for x in (with_input_iteration if isinstance(with_input_iteration, (list, tuple)) else [with_input_iteration])
|
|
]
|
|
)
|
|
|
|
self.with_input_iteration = with_input_iteration
|
|
self.__operation_input_depth = self.__find_operation_input_depth()
|
|
|
|
def __find_operation_input_depth(self):
|
|
if not self.with_input_iteration:
|
|
return None
|
|
return max(len(option) for option in self.with_input_iteration)
|
|
|
|
@staticmethod
|
|
def _get_merged_groups(regex_pattern: str, value: str) -> str:
|
|
match = regex.search(regex_pattern, value)
|
|
if not match:
|
|
return None
|
|
|
|
return "".join(g for g in match.groups() if g is not None)
|
|
|
|
@staticmethod
|
|
def _remove_regex_groups(pattern, value):
|
|
regex = re.compile(pattern)
|
|
|
|
def repl(m):
|
|
s = m.group(0)
|
|
# remove all group spans from the full match
|
|
spans = [m.span(i) for i in range(1, len(m.groups()) + 1) if m.span(i) != (-1, -1)]
|
|
spans.sort(reverse=True)
|
|
|
|
for start, end in spans:
|
|
rel_start = start - m.start()
|
|
rel_end = end - m.start()
|
|
s = s[:rel_start] + s[rel_end:]
|
|
|
|
return s
|
|
|
|
return regex.sub(repl, value)
|
|
|
|
def do_changes(self, source_coll: str, used_regex: str, row: dict[str, str], changes={}):
|
|
current_to = changes.get(self.to, row[source_coll].value)
|
|
current_source = changes.get(source_coll, row[source_coll].value)
|
|
|
|
pending_res_to = self._get_merged_groups(used_regex, current_to)
|
|
if pending_res_to is not None:
|
|
changes[self.to] = pending_res_to
|
|
changes[source_coll] = self._remove_regex_groups(used_regex, current_source)
|
|
return changes
|
|
|
|
def execute_on_row(self, row: dict[str, str], source_coll: str = None):
|
|
if not source_coll:
|
|
raise ValueError("source_col required")
|
|
if self.with_input_iteration:
|
|
changes = {}
|
|
for iteration in self.with_input_iteration:
|
|
used_regex = self.match
|
|
for operation_id in range(0, self.__operation_input_depth):
|
|
used_regex = used_regex.replace(f"{{{{% {operation_id} %}}}}", iteration[operation_id])
|
|
changes = self.do_changes(source_coll, used_regex, row, changes)
|
|
return changes
|
|
else:
|
|
return self.do_changes(source_coll, self.match, row) |