runable

.gitignore

@@ -0,0 +1 @@
+.venv

.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"
+        }
+    ]
+}

.vscode/settings.json

@@ -0,0 +1,3 @@
+{
+    "nixEnvSelector.nixFile": "${workspaceFolder}/shell.nix"
+}

__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")

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"]: {
+    ("Art",): {
-        "command": {
+        "command": "FIND_AND_REPLACE",
-            "name": "FIND_AND_REPLACE",
+        "map": SPICIES
-            "map": CATCHED_BY
-        }
     },
-    ["Ort"]: {
+    ("Ort",): {
-        "command": {
+        "command": "REGEX_MOVE_TO_COL",
-            "name": "REGEX_MOVE_TO_COL",
+        "with_input_iteration": CATCHED_BY,  # Moves matching group {{% 1 %}} to "to"
-            "with_input_iteration": flatten_dict(CATCHED_BY),  # Moves matching group {{% 1 %}} to "to"
         "match": r"^(?:[^ ]* +)*({{% 1 %}}),?(?: .*+)*$",
         "to": "Fangart"
-        }
     },
-    ["Fangart"]: {
+    ("Fangart",): {
-        "command": {
+        "command": "REGEX_EDIT_COL",
-            "name": "REGEX_EDIT_COL",
         "match": r"^ *$",
-            "to": "1"
+        "replace_with": "1"
-        }
     },
 }
 

instructions/ring_formater.py

@@ -21,16 +21,16 @@ EXPECTED_OUTPUT_TABLE_SCHEMA = EXPECTED_INPUT_TABLE_SCHEMA
 
 COL_INSTRUCTIONS = {
     ["Geschlecht"]: {
-        "command": {
         "name": "REGEX_MOVE_TO_COL",
+        "command": {
             "with_input_iteration": flatten_dict(REPRODUCTION),  # Moves matching group {{% 1 %}} to "to"
             "match": r"^(?:[^ ]* +)*({{% 1 %}}),?(?: .*+)*$",
             "to": "Reproduktionsstatus"
         }
     },
     ["Geschlecht"]: {
-        "command": {
         "name": "REGEX_MOVE_TO_COL",
+        "command": {
             "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
-        "command": {
         "name": "REGEX_MOVE_TO_COL",
+        "command": {
             "match": r"^(?:[^ ]* +)*m(a),?(?: .*+)*$",
             "to": "Alter"
         }
     },
     ["Ringnummer"]: {
-        "command": {
         "name": "REGEX_EDIT_COL",
+        "command": {
             "match": r"^[^a-zA-Z\d]*([a-zA-Z]*)[^a-zA-Z\d]*(\d+)$",
             "to": "{{% 1 %}}{{% 2 %}}"
         }
     },
     ["Art"]: {
-        "command": {
         "name": "FIND_AND_REPLACE",
+        "command": {
             "map": SPICIES
         }
     },
     ["Ort"]: {
-        "command": {
         "name": "REGEX_MOVE_TO_COL",
+        "command": {
             "with_input_iteration": flatten_dict(CATCHED_BY),  # Moves matching group {{% 1 %}} to "to"
             "match": r"^(?:[^ ]* +)*({{% 1 %}}),?(?: .*+)*$",
             "to": "Fangart"

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

shell.nix

@@ -0,0 +1,16 @@
+let
+  pkgs = import <nixpkgs> {
+    overlays = [ ];
+  };
+in
+pkgs.mkShell {
+  buildInputs = with pkgs; [
+    python3
+  ];
+  shellHook = ''
+    if [ ! -d .venv ]; then
+      python -m venv .venv
+    fi
+    source .venv/bin/activate
+  '';
+}

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]
+        }

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)

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))

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)

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

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