diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..b694934 --- /dev/null +++ b/.gitignore @@ -0,0 +1 @@ +.venv \ No newline at end of file diff --git a/.vscode/launch.json b/.vscode/launch.json new file mode 100644 index 0000000..1f0c528 --- /dev/null +++ b/.vscode/launch.json @@ -0,0 +1,16 @@ +{ + // Use IntelliSense to learn about possible attributes. + // Hover to view descriptions of existing attributes. + // For more information, visit: https://go.microsoft.com/fwlink/?linkid=830387 + "version": "0.2.0", + "configurations": [ + + { + "name": "Python Debugger: Current File", + "type": "debugpy", + "request": "launch", + "program": "${workspaceFolder}/__init__.py", + "console": "integratedTerminal" + } + ] +} \ No newline at end of file diff --git a/.vscode/settings.json b/.vscode/settings.json new file mode 100644 index 0000000..786d4d3 --- /dev/null +++ b/.vscode/settings.json @@ -0,0 +1,3 @@ +{ + "nixEnvSelector.nixFile": "${workspaceFolder}/shell.nix" +} \ No newline at end of file diff --git a/__init__.py b/__init__.py new file mode 100644 index 0000000..93a44a0 --- /dev/null +++ b/__init__.py @@ -0,0 +1,17 @@ + +from table_handler.serializer import TableSerializer +from pprint import pprint as print + + +if __name__ == "__main__": + # Example usage + import instructions.countings_formater as countings_formater + + file = "data/zählungen_short.xlsx" + serializer = TableSerializer(file, countings_formater.EXPECTED_INPUT_TABLE_SCHEMA, countings_formater.EXPECTED_OUTPUT_TABLE_SCHEMA) + serializer.run( + countings_formater.COL_INSTRUCTIONS, + countings_formater.ROW_INSTRUCTION + ) + print(serializer.serialized_output) + serializer.write_output("zählungen_test_out.xlsx") \ No newline at end of file diff --git a/data/zählungen_short.xlsx b/data/zählungen_short.xlsx new file mode 100644 index 0000000..c1a6939 Binary files /dev/null and b/data/zählungen_short.xlsx differ diff --git a/instructions/__init__.py b/instructions/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/instructions/__pycache__/__init__.cpython-312.pyc b/instructions/__pycache__/__init__.cpython-312.pyc new file mode 100644 index 0000000..122812d Binary files /dev/null and b/instructions/__pycache__/__init__.cpython-312.pyc differ diff --git a/instructions/__pycache__/countings_formater.cpython-312.pyc b/instructions/__pycache__/countings_formater.cpython-312.pyc new file mode 100644 index 0000000..62b3816 Binary files /dev/null and b/instructions/__pycache__/countings_formater.cpython-312.pyc differ diff --git a/instructions/__pycache__/mappings.cpython-312.pyc b/instructions/__pycache__/mappings.cpython-312.pyc new file mode 100644 index 0000000..01a9437 Binary files /dev/null and b/instructions/__pycache__/mappings.cpython-312.pyc differ diff --git a/instructions/countings_formater.py b/instructions/countings_formater.py index ce5e8d3..678790a 100644 --- a/instructions/countings_formater.py +++ b/instructions/countings_formater.py @@ -1,4 +1,5 @@ -from mappings import * +from .mappings import * +from copy import deepcopy EXPECTED_INPUT_TABLE_SCHEMA = [ "Anzahl", @@ -12,29 +13,23 @@ EXPECTED_INPUT_TABLE_SCHEMA = [ "Bemerkung" ] -EXPECTED_OUTPUT_TABLE_SCHEMA = EXPECTED_INPUT_TABLE_SCHEMA.remove("Geschlecht") +EXPECTED_OUTPUT_TABLE_SCHEMA = deepcopy(EXPECTED_INPUT_TABLE_SCHEMA).remove("Geschlecht") COL_INSTRUCTIONS = { - ["Fangart"]: { - "command": { - "name": "FIND_AND_REPLACE", - "map": CATCHED_BY - } + ("Art",): { + "command": "FIND_AND_REPLACE", + "map": SPICIES }, - ["Ort"]: { - "command": { - "name": "REGEX_MOVE_TO_COL", - "with_input_iteration": flatten_dict(CATCHED_BY), # Moves matching group {{% 1 %}} to "to" - "match": r"^(?:[^ ]* +)*({{% 1 %}}),?(?: .*+)*$", - "to": "Fangart" - } + ("Ort",): { + "command": "REGEX_MOVE_TO_COL", + "with_input_iteration": CATCHED_BY, # Moves matching group {{% 1 %}} to "to" + "match": r"^(?:[^ ]* +)*({{% 1 %}}),?(?: .*+)*$", + "to": "Fangart" }, - ["Fangart"]: { - "command": { - "name": "REGEX_EDIT_COL", - "match": r"^ *$", - "to": "1" - } + ("Fangart",): { + "command": "REGEX_EDIT_COL", + "match": r"^ *$", + "replace_with": "1" }, } diff --git a/instructions/ring_formater.py b/instructions/ring_formater.py index bde2d55..560bb3f 100644 --- a/instructions/ring_formater.py +++ b/instructions/ring_formater.py @@ -21,16 +21,16 @@ EXPECTED_OUTPUT_TABLE_SCHEMA = EXPECTED_INPUT_TABLE_SCHEMA COL_INSTRUCTIONS = { ["Geschlecht"]: { + "name": "REGEX_MOVE_TO_COL", "command": { - "name": "REGEX_MOVE_TO_COL", "with_input_iteration": flatten_dict(REPRODUCTION), # Moves matching group {{% 1 %}} to "to" "match": r"^(?:[^ ]* +)*({{% 1 %}}),?(?: .*+)*$", "to": "Reproduktionsstatus" } }, ["Geschlecht"]: { + "name": "REGEX_MOVE_TO_COL", "command": { - "name": "REGEX_MOVE_TO_COL", "with_input_iteration": flatten_dict(AGE), # Moves matching group {{% 1 %}} to "to" "match": r"^(?:[^ ]* +)*({{% 1 %}}),?(?: .*+)*$", "to": "Alter" @@ -38,28 +38,28 @@ COL_INSTRUCTIONS = { }, ["Geschlecht"]: { # Special case ma -> Geschlecht a | Alter a + "name": "REGEX_MOVE_TO_COL", "command": { - "name": "REGEX_MOVE_TO_COL", "match": r"^(?:[^ ]* +)*m(a),?(?: .*+)*$", "to": "Alter" } }, ["Ringnummer"]: { + "name": "REGEX_EDIT_COL", "command": { - "name": "REGEX_EDIT_COL", "match": r"^[^a-zA-Z\d]*([a-zA-Z]*)[^a-zA-Z\d]*(\d+)$", "to": "{{% 1 %}}{{% 2 %}}" } }, ["Art"]: { + "name": "FIND_AND_REPLACE", "command": { - "name": "FIND_AND_REPLACE", "map": SPICIES } }, ["Ort"]: { + "name": "REGEX_MOVE_TO_COL", "command": { - "name": "REGEX_MOVE_TO_COL", "with_input_iteration": flatten_dict(CATCHED_BY), # Moves matching group {{% 1 %}} to "to" "match": r"^(?:[^ ]* +)*({{% 1 %}}),?(?: .*+)*$", "to": "Fangart" diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..4294cd5 --- /dev/null +++ b/requirements.txt @@ -0,0 +1,7 @@ +et_xmlfile==2.0.0 +numpy==2.4.2 +openpyxl==3.1.5 +pandas==3.0.1 +python-dateutil==2.9.0.post0 +regex==2026.2.28 +six==1.17.0 diff --git a/shell.nix b/shell.nix new file mode 100644 index 0000000..22acaa4 --- /dev/null +++ b/shell.nix @@ -0,0 +1,16 @@ +let + pkgs = import { + overlays = [ ]; + }; +in +pkgs.mkShell { + buildInputs = with pkgs; [ + python3 + ]; + shellHook = '' + if [ ! -d .venv ]; then + python -m venv .venv + fi + source .venv/bin/activate + ''; +} \ No newline at end of file diff --git a/table_handler/__init__.py b/table_handler/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/table_handler/__pycache__/__init__.cpython-312.pyc b/table_handler/__pycache__/__init__.cpython-312.pyc new file mode 100644 index 0000000..a8121bd Binary files /dev/null and b/table_handler/__pycache__/__init__.cpython-312.pyc differ diff --git a/table_handler/__pycache__/serializer.cpython-312.pyc b/table_handler/__pycache__/serializer.cpython-312.pyc new file mode 100644 index 0000000..0a3b328 Binary files /dev/null and b/table_handler/__pycache__/serializer.cpython-312.pyc differ diff --git a/table_handler/commands/__pycache__/copy.cpython-312.pyc b/table_handler/commands/__pycache__/copy.cpython-312.pyc new file mode 100644 index 0000000..581bd36 Binary files /dev/null and b/table_handler/commands/__pycache__/copy.cpython-312.pyc differ diff --git a/table_handler/commands/__pycache__/find_and_replace.cpython-312.pyc b/table_handler/commands/__pycache__/find_and_replace.cpython-312.pyc new file mode 100644 index 0000000..117c77d Binary files /dev/null and b/table_handler/commands/__pycache__/find_and_replace.cpython-312.pyc differ diff --git a/table_handler/commands/__pycache__/regex_edit_col.cpython-312.pyc b/table_handler/commands/__pycache__/regex_edit_col.cpython-312.pyc new file mode 100644 index 0000000..9230af2 Binary files /dev/null and b/table_handler/commands/__pycache__/regex_edit_col.cpython-312.pyc differ diff --git a/table_handler/commands/__pycache__/regex_move_to_col.cpython-312.pyc b/table_handler/commands/__pycache__/regex_move_to_col.cpython-312.pyc new file mode 100644 index 0000000..278d5d1 Binary files /dev/null and b/table_handler/commands/__pycache__/regex_move_to_col.cpython-312.pyc differ diff --git a/table_handler/commands/copy.py b/table_handler/commands/copy.py new file mode 100644 index 0000000..66be82e --- /dev/null +++ b/table_handler/commands/copy.py @@ -0,0 +1,13 @@ + + +class Copy: + def __init__(self, source: str, target: str, replace = False): + self.source = source + self.target = target + self.replace = replace + + def execute_on_row(self, row=dict[str, str], source_col=None): + return { + **({} if not self.replace else {source_col: ""}), + self.target: row[source_col] + } \ No newline at end of file diff --git a/table_handler/commands/find_and_replace.py b/table_handler/commands/find_and_replace.py new file mode 100644 index 0000000..daf4427 --- /dev/null +++ b/table_handler/commands/find_and_replace.py @@ -0,0 +1,11 @@ +from copy import deepcopy + +class FindAndReplace: + def __init__(self, map: dict[str, dict[str]]): + self.map = map + + def execute_on_cell(self, entry) -> str: + for species, alternatives in self.map.items(): + for old_version in alternatives: + if old_version in entry.value: + return deepcopy(entry.value).replace(old_version, species) \ No newline at end of file diff --git a/table_handler/commands/regex_edit_col.py b/table_handler/commands/regex_edit_col.py new file mode 100644 index 0000000..42dfcfe --- /dev/null +++ b/table_handler/commands/regex_edit_col.py @@ -0,0 +1,11 @@ +import re + +class RegexEditCol: + def __init__(self, match, replace_with): + self.match = match + self.replace_with = replace_with + + def execute_on_cell(self, value) -> str: + if value is None: + return value + return re.sub(self.match, self.replace_with, str(value)) \ No newline at end of file diff --git a/table_handler/commands/regex_move_to_col.py b/table_handler/commands/regex_move_to_col.py new file mode 100644 index 0000000..10a8301 --- /dev/null +++ b/table_handler/commands/regex_move_to_col.py @@ -0,0 +1,62 @@ +import regex + +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 and type(with_input_iteration) != list: + with_input_iteration = list(map([e] for e in 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.__iteration_input_depth()) + + @staticmethod + def _get_merged_groups(regex_pattern: str, entry: str) -> str: + match = regex.search(regex_pattern, entry.value) + if not match: + return "" + + return "".join(g for g in match.groups() if g is not None) + + @staticmethod + def _remove_regex_groups(regex_pattern: str, entry: str) -> str: + match = regex.search(regex_pattern, entry.value) + if not match: + return entry.value + + spans = [s for s in match.spans()[1:] if s != (-1, -1)] + + result = [] + last = 0 + + for start, end in sorted(spans): + result.append(entry.value[last:start]) + last = end + + result.append(entry.value[last:]) + return "".join(result) + + def do_changes(self, source_coll: str, used_regex: str, row: dict[str, str], changes={}): + changes[self.to] = self._get_merged_groups(used_regex, row[source_coll]) + changes[source_coll] = self._remove_regex_groups(used_regex, row[source_coll]) + 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 = None + used_regex = self.match + for iteration in self.with_input_iteration: + for operation_id in range(0, self.__operation_input_depth-1): + used_regex.replace(f"{{{{ {operation_id} }}}}", iteration.get(operation_id)) + changes = self.do_changes(source_coll, used_regex, row, changes) + return changes + else: + return self.do_changes(source_coll, self.match, row) \ No newline at end of file diff --git a/table_handler/conditions/__pycache__/regex_cell.cpython-312.pyc b/table_handler/conditions/__pycache__/regex_cell.cpython-312.pyc new file mode 100644 index 0000000..f1c8aa8 Binary files /dev/null and b/table_handler/conditions/__pycache__/regex_cell.cpython-312.pyc differ diff --git a/table_handler/conditions/regex_cell.py b/table_handler/conditions/regex_cell.py new file mode 100644 index 0000000..07d8db6 --- /dev/null +++ b/table_handler/conditions/regex_cell.py @@ -0,0 +1,12 @@ +import regex + +class RegexCellCondition: + def __init__(self, map: dict[str, str]): + self.map = map + + def check_on_row(self, row: dict[str, any]): + for apply_coll, condition in self.map.items(): + entry_value = str(row.get(apply_coll).value) + if not regex.match(condition, entry_value if entry_value else ""): + return False + return True \ No newline at end of file diff --git a/table_handler/serializer.py b/table_handler/serializer.py new file mode 100644 index 0000000..33048fe --- /dev/null +++ b/table_handler/serializer.py @@ -0,0 +1,147 @@ + +from openpyxl import Workbook, load_workbook +from dataclasses import dataclass +from pprint import pprint as print + +import instructions.mappings as mappings + +from .commands.find_and_replace import FindAndReplace +from .commands.regex_edit_col import RegexEditCol +from .commands.regex_move_to_col import RegexMoveToCol +from .commands.copy import Copy + +from .conditions.regex_cell import RegexCellCondition + + +@dataclass +class Entry: + value: str|int|float|bool|None + bg_color: tuple[str, str, str] + fg_color: tuple[str, str, str] + + + +class TableSerializer: + def __init__(self, file: str, expected_input_schema: list[str], expected_output_schema: list[str]): + self.wb: Workbook = load_workbook(file, read_only=True) + self.ws = self.wb.active + if not self.ws: + raise ValueError("No active worksheet found in the workbook.") + + self.data = self._serialize_table(expected_input_schema) + self.serialized_output = {} + self.seperator = ", " + self.expected_output_schema = expected_output_schema or expected_input_schema + self.implicit_copy = True + self.output_header = True + + + def _serialize_table(self, expected_cols: list[str]) -> dict[int, list[Entry]]: + data = {col: {} for col in expected_cols} + map_expected_with_ws = {} + for idx, cell in enumerate(self.ws[1]): + cell_value = cell.value if cell.value else "" + if cell_value in expected_cols: + map_expected_with_ws[cell_value] = idx + else: + print(f"Skipping unexpected Col {cell_value}") + + if set(map_expected_with_ws.keys()) != set(expected_cols): + raise ValueError("Missing Col in input") + + for rid, row in enumerate(self.ws.iter_rows(min_row=2, values_only=False), start=2): + for expected_col, ws_idx in map_expected_with_ws.items(): + cell = row[ws_idx] + data[expected_col][rid] = Entry( + value=cell.value, + bg_color=cell.fill.start_color.rgb if cell.fill and cell.fill.start_color else "00000000", + fg_color=cell.font.color.rgb if cell.font and cell.font.color else "00000000" + ) + return data + + def write_output(self, destination_file: str): + if not self.serialized_output: + raise ValueError("There is nothing to write, serialize first") + + # write header + owb = load_workbook(destination_file) + ows = owb.active + + if self.output_header: + for col_i, col_name in enumerate(self.expected_output_schema, start=1): + ows.cell(row=1, column=col_i, value=col_name) + + # write rows + for row_i, (row_id, row_data) in enumerate(self.serialized_output.items(), start=2 if self.output_header else 1): + for col_i, col_name in enumerate(self.expected_output_schema, start=1): + ows.cell(row=row_i, column=col_i, value=row_data.get(col_name)) + + owb.save(destination_file) + + + def _check_conditions(self, instruct: dict[str, str], row: dict[str, any]): + when_regex_condition_map = instruct.get("when_regex") + if when_regex_condition_map: + return RegexCellCondition(when_regex_condition_map).check_on_row(row) + return True + + def _merge_change_to_cel(self, value, changes): + if not value and changes: + value = changes + elif changes: + value = f"{value}{self.seperator}{changes}" + return value + + def _extecute_instruct(self, source_cell: str, instruct: dict, row: dict[str, any], working_row: dict[str, any]): + # execute changes + res = "" + res_row = {} + match instruct.get("command"): + case "FIND_AND_REPLACE": + res = FindAndReplace(instruct["map"]).execute_on_cell(row[source_cell]) + case "REGEX_MOVE_TO_COL": + res_row = RegexMoveToCol(instruct["match"], instruct["to"], instruct.get("with_iteration_input")) \ + .execute_on_row(row, source_cell) + case "REGEX_EDIT_COL": + res = RegexEditCol(instruct["match"], instruct["replace_with"])\ + .execute_on_cell(row[source_cell]) + case "CONDITIONAL_COPY": + res_row = Copy(instruct["source"], instruct["target"], replace=instruct.get("replace"))\ + .execute_on_row(row) + + # Merge row_changes to working_row + for target_cel, change in res_row.items(): + working_row[target_cel] = self._merge_change_to_cel(working_row.get(target_cel), change) + + # Merge changes in working row + if source_cell: + working_row[source_cell] = self._merge_change_to_cel(working_row.get(source_cell), res) + return working_row + + + @property + def vertical_len(self): + return max(len(data_row) for data_row in self.data.values()) + + def get_row_as_dict(self, rid: int): + return {col: rows.get(rid) for col, rows in self.data.items()} + + def run(self, col_instructions: dict[tuple[str], dict], row_instructions: dict[tuple[str], dict]): + self.serialized_output = {} + for rid in range(2, self.vertical_len + 1): + working_row = {} + source_row = self.get_row_as_dict(rid) + for source_cols, instruct in list(col_instructions.items()) + [ + ((None,), instr) for instr in row_instructions + ]: + for col in source_cols: + if not self._check_conditions(instruct, source_row): + continue + working_row = self._extecute_instruct(col, instruct, source_row, working_row) + if not self.implicit_copy: + continue + for col in self.expected_output_schema: + if col in working_row: + continue + working_row[col] = source_row.get(col) + self.serialized_output[rid] = working_row