Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
132 changes: 96 additions & 36 deletions ahb_diff/main.py
Original file line number Diff line number Diff line change
Expand Up @@ -191,7 +191,7 @@ def create_row(
"""
fills rows for all columns that belong to one dataframe depending on whether previous/subsequent segments exist.
"""
row = {f"Segmentname_{previous_formatversion}": "", "diff": "", f"Segmentname_{subsequent_formatversion}": ""}
row = {f"Segmentname_{previous_formatversion}": "", "Änderung": "", f"Segmentname_{subsequent_formatversion}": ""}

if previous_df is not None:
for col in previous_df.columns:
Expand Down Expand Up @@ -239,7 +239,7 @@ def align_columns(
column_order = (
[f"Segmentname_{previous_formatversion}"]
+ [f"{col}_{previous_formatversion}" for col in columns_of_previous_formatversion]
+ ["diff"]
+ ["Änderung"]
+ [f"Segmentname_{subsequent_formatversion}"]
+ [f"{col}_{subsequent_formatversion}" for col in columns_of_subsequent_formatversion]
)
Expand All @@ -259,7 +259,7 @@ def align_columns(
for i in range(len(df_of_previous_formatversion))
]
for row in result_rows:
row["diff"] = "REMOVED"
row["Änderung"] = "ENTFÄLLT"
result_df = pd.DataFrame(result_rows)
return result_df[column_order]

Expand All @@ -275,7 +275,7 @@ def align_columns(
for j in range(len(df_of_subsequent_formatversion))
]
for row in result_rows:
row["diff"] = "NEW"
row["Änderung"] = "NEU"
result_df = pd.DataFrame(result_rows)
return result_df[column_order]

Expand All @@ -298,7 +298,7 @@ def align_columns(
previous_formatversion=previous_formatversion,
subsequent_formatversion=subsequent_formatversion,
)
row["diff"] = "NEW"
row["Änderung"] = "NEU"
result_rows.append(row)
j += 1
elif j >= len(segments_of_subsequent_formatversion):
Expand All @@ -309,7 +309,7 @@ def align_columns(
previous_formatversion=previous_formatversion,
subsequent_formatversion=subsequent_formatversion,
)
row["diff"] = "REMOVED"
row["Änderung"] = "ENTFÄLLT"
result_rows.append(row)
i += 1
elif segments_of_previous_formatversion[i] == segments_of_subsequent_formatversion[j]:
Expand All @@ -321,7 +321,7 @@ def align_columns(
previous_formatversion=previous_formatversion,
subsequent_formatversion=subsequent_formatversion,
)
row["diff"] = ""
row["Änderung"] = ""
result_rows.append(row)
i += 1
j += 1
Expand All @@ -337,7 +337,7 @@ def align_columns(
previous_formatversion=previous_formatversion,
subsequent_formatversion=subsequent_formatversion,
)
row["diff"] = "NEW"
row["Änderung"] = "NEU"
result_rows.append(row)
j += 1
continue
Expand All @@ -350,7 +350,7 @@ def align_columns(
previous_formatversion=previous_formatversion,
subsequent_formatversion=subsequent_formatversion,
)
row["diff"] = "REMOVED"
row["Änderung"] = "ENTFÄLLT"
result_rows.append(row)
i += 1

Expand All @@ -364,13 +364,21 @@ def export_to_excel(df: DataFrame, output_path_xlsx: str) -> None:
"""
exports the merged dataframe to .xlsx with highlighted differences.
"""
sheet_name = Path(output_path_xlsx).stem # excel sheet name = <pruefid>

# add column for indexing through all rows
df = df.reset_index()
df["index"] = df["index"] + 1
df = df.rename(columns={"index": "#"})
# remove duplicate columns that index through the rows
df_filtered = df[[col for col in df.columns if not col.startswith("Unnamed:")]]

with pd.ExcelWriter(output_path_xlsx, engine="xlsxwriter") as writer:
df_filtered.to_excel(writer, sheet_name="AHB-Diff", index=False)
df_filtered.to_excel(writer, sheet_name=sheet_name, index=False)
sheet_name = Path(output_path_xlsx).stem

workbook = writer.book
worksheet = writer.sheets["AHB-Diff"]
worksheet = writer.sheets[sheet_name]

# sticky table header
worksheet.freeze_panes(1, 0)
Expand All @@ -389,14 +397,14 @@ def export_to_excel(df: DataFrame, output_path_xlsx: str) -> None:

# formatting highlighted/changed cells.
diff_formats: dict[str, Format] = {
"NEW": workbook.add_format({"bg_color": "#C6EFCE", "border": 1, "text_wrap": True}),
"REMOVED": workbook.add_format({"bg_color": "#FFC7CE", "border": 1, "text_wrap": True}),
"NEU": workbook.add_format({"bg_color": "#C6EFCE", "border": 1, "text_wrap": True}),
"ENTFÄLLT": workbook.add_format({"bg_color": "#FFC7CE", "border": 1, "text_wrap": True}),
"": workbook.add_format({"border": 1, "text_wrap": True}),
}

# formatting diff column.
diff_text_formats: dict[str, Format] = {
"NEW": workbook.add_format(
"NEU": workbook.add_format(
{
"bold": True,
"color": "#7AAB8A",
Expand All @@ -406,7 +414,7 @@ def export_to_excel(df: DataFrame, output_path_xlsx: str) -> None:
"text_wrap": True,
}
),
"REMOVED": workbook.add_format(
"ENTFÄLLT": workbook.add_format(
{
"bold": True,
"color": "#E94C74",
Expand All @@ -422,18 +430,7 @@ def export_to_excel(df: DataFrame, output_path_xlsx: str) -> None:
for col_num, value in enumerate(df_filtered.columns.values):
worksheet.write(0, col_num, value, header_format)

diff_idx = df_filtered.columns.get_loc("diff")

def _try_convert_to_number(cell: str) -> int | float | str:
"""
tries to format cell values to numbers where appropriate.
"""
try:
if cell.isdigit():
return int(cell)
return float(cell)
except ValueError:
return cell
diff_idx = df_filtered.columns.get_loc("Änderung")

previous_formatversion = None
subsequent_formatversion = None
Expand All @@ -451,19 +448,82 @@ def _try_convert_to_number(cell: str) -> int | float | str:
diff_value = str(row_data[diff_idx])

for col_num, (value, col_name) in enumerate(zip(row_data, df_filtered.columns)):
converted_value = _try_convert_to_number(str(value)) if value != "" else ""
value = str(value) if value != "" else ""

if col_name == "diff":
if col_name == "Änderung":
worksheet.write(row_num, col_num, value, diff_text_formats[diff_value])
elif diff_value == "REMOVED" and previous_formatversion and col_name.endswith(previous_formatversion):
worksheet.write(row_num, col_num, converted_value, diff_formats["REMOVED"])
elif diff_value == "NEW" and subsequent_formatversion and col_name.endswith(subsequent_formatversion):
worksheet.write(row_num, col_num, converted_value, diff_formats["NEW"])
elif diff_value == "ENTFÄLLT" and previous_formatversion and col_name.endswith(previous_formatversion):
worksheet.write(row_num, col_num, value, diff_formats["ENTFÄLLT"])
elif diff_value == "NEU" and subsequent_formatversion and col_name.endswith(subsequent_formatversion):
worksheet.write(row_num, col_num, value, diff_formats["NEU"])
else:
worksheet.write(row_num, col_num, converted_value, base_format)
worksheet.write(row_num, col_num, value, base_format)

# format first row each time the `Segmentname` changes
diff_formats["segmentname_changed"] = workbook.add_format(
{"bg_color": "#D9D9D9", "border": 1, "text_wrap": True}
)
segment_name_bold = workbook.add_format({"bold": True, "border": 1, "text_wrap": True, "bg_color": "#D9D9D9"})

previous_segmentname = None

for row_num, row in enumerate(df_filtered.itertuples(index=False), start=1):
row_data = list(row)
diff_value = str(row_data[diff_idx])

current_segmentname = None
segmentname_col = None
for col_name in df_filtered.columns:
if col_name.startswith("Segmentname_"):
idx = df_filtered.columns.get_loc(col_name)
value = str(row_data[idx])
if value:
current_segmentname = value
segmentname_col = col_name
break

is_new_segment = current_segmentname and current_segmentname != previous_segmentname
previous_segmentname = current_segmentname

# apply formatting only when `Segmentname` is not affected by "NEU"/"ENTFÄLLT" styling
for col_num, (value, col_name) in enumerate(zip(row_data, df_filtered.columns)):
value = str(value) if value != "" else ""

if col_name == "Änderung":
worksheet.write(row_num, col_num, value, diff_text_formats[diff_value])
elif diff_value == "ENTFÄLLT" and previous_formatversion and col_name.endswith(previous_formatversion):
worksheet.write(row_num, col_num, value, diff_formats["ENTFÄLLT"])
elif diff_value == "NEU" and subsequent_formatversion and col_name.endswith(subsequent_formatversion):
worksheet.write(row_num, col_num, value, diff_formats["NEU"])
else:
if is_new_segment and diff_value == "":
if col_name == segmentname_col:
worksheet.write(row_num, col_num, value, segment_name_bold)
else:
worksheet.write(row_num, col_num, value, diff_formats["segmentname_changed"])
else:
worksheet.write(row_num, col_num, value, base_format)

column_widths = {
"#": 25,
"Segmentname_": 175,
"Segmentgruppe_": 100,
"Segment_": 100,
"Datenelement_": 100,
"Segment ID_": 100,
"Code_": 100,
"Qualifier_": 100,
"Beschreibung_": 150,
"Bedingungsausdruck_": 100,
"Bedingung_": 150,
}

for col_num in range(len(df_filtered.columns)):
worksheet.set_column(col_num, col_num, min(150 / 7, 21)) # cell width = 150 px.
for col_num, col_name in enumerate(df_filtered.columns):
width_px = next(
(width for prefix, width in column_widths.items() if col_name.startswith(prefix)), 150
) # default = 150 px
excel_width = width_px / 7
worksheet.set_column(col_num, col_num, excel_width)

logger.info("✅successfully exported XLSX file to: %s", {output_path_xlsx})

Expand Down
2 changes: 1 addition & 1 deletion unittests/test_export.py
Original file line number Diff line number Diff line change
Expand Up @@ -52,7 +52,7 @@ def test_empty_dataframe_export() -> None:
subsequent_formatversion = "FV2504"

df = pd.DataFrame(
columns=[f"Segmentname_{previous_formatversion}", "diff", f"Segmentname_{subsequent_formatversion}"]
columns=[f"Segmentname_{previous_formatversion}", "Änderung", f"Segmentname_{subsequent_formatversion}"]
)

with tempfile.TemporaryDirectory() as temp_dir:
Expand Down
26 changes: 13 additions & 13 deletions unittests/test_process_csv.py
Original file line number Diff line number Diff line change
Expand Up @@ -36,7 +36,7 @@ def test_align_columns(self) -> None:
"9",
"10",
],
"diff": ["", "", "", "REMOVED", "", "", "NEW", "NEW", "", ""],
"Änderung": ["", "", "", "ENTFÄLLT", "", "", "NEU", "NEU", "", ""],
f"Segmentname_{self.formatversions.subsequent_formatversion}": [
"1",
"2",
Expand Down Expand Up @@ -67,7 +67,7 @@ def test_align_columns_empty_dataframes(self) -> None:
expected_output: DataFrame = pd.DataFrame(
{
f"Segmentname_{self.formatversions.previous_formatversion}": pd.Series([], dtype="float64"),
"diff": pd.Series([], dtype="float64"),
"Änderung": pd.Series([], dtype="float64"),
f"Segmentname_{self.formatversions.subsequent_formatversion}": pd.Series([], dtype="float64"),
}
)
Expand All @@ -87,7 +87,7 @@ def test_align_columns_one_empty_dataframe(self) -> None:
expected_output: DataFrame = pd.DataFrame(
{
f"Segmentname_{self.formatversions.previous_formatversion}": ["1", "2", "3"],
"diff": ["REMOVED", "REMOVED", "REMOVED"],
"Änderung": ["ENTFÄLLT", "ENTFÄLLT", "ENTFÄLLT"],
f"Segmentname_{self.formatversions.subsequent_formatversion}": ["", "", ""],
}
)
Expand All @@ -107,7 +107,7 @@ def test_align_columns_full_offset(self) -> None:
expected_output: DataFrame = pd.DataFrame(
{
f"Segmentname_{self.formatversions.previous_formatversion}": ["1", "2", "3", "", "", ""],
"diff": ["REMOVED", "REMOVED", "REMOVED", "NEW", "NEW", "NEW"],
"Änderung": ["ENTFÄLLT", "ENTFÄLLT", "ENTFÄLLT", "NEU", "NEU", "NEU"],
f"Segmentname_{self.formatversions.subsequent_formatversion}": ["", "", "", "4", "5", "6"],
}
)
Expand All @@ -127,7 +127,7 @@ def test_align_columns_duplicate_segments(self) -> None:
expected_output: DataFrame = pd.DataFrame(
{
f"Segmentname_{self.formatversions.previous_formatversion}": ["1", "2", "2", ""],
"diff": ["", "", "REMOVED", "NEW"],
"Änderung": ["", "", "ENTFÄLLT", "NEU"],
f"Segmentname_{self.formatversions.subsequent_formatversion}": ["1", "2", "", "4"],
}
)
Expand All @@ -147,7 +147,7 @@ def test_align_columns_repeating_segments(self) -> None:
expected_output: DataFrame = pd.DataFrame(
{
f"Segmentname_{self.formatversions.previous_formatversion}": ["1", "2", "3", "3", "2", ""],
"diff": ["", "", "", "REMOVED", "REMOVED", "NEW"],
"Änderung": ["", "", "", "ENTFÄLLT", "ENTFÄLLT", "NEU"],
f"Segmentname_{self.formatversions.subsequent_formatversion}": ["1", "2", "3", "", "", "4"],
}
)
Expand Down Expand Up @@ -212,7 +212,7 @@ def test_align_columns(self) -> None:
"g",
"h",
],
"diff": ["", "", "", "REMOVED", "", "", "NEW", "NEW", "", ""],
"Änderung": ["", "", "", "ENTFÄLLT", "", "", "NEU", "NEU", "", ""],
f"Segmentname_{self.formatversions.subsequent_formatversion}": [
"1",
"2",
Expand Down Expand Up @@ -256,7 +256,7 @@ def test_align_columns_empty_dataframes(self) -> None:
{
f"Segmentname_{self.formatversions.previous_formatversion}": [],
f"Segmentgruppe_{self.formatversions.previous_formatversion}": [],
"diff": [],
"Änderung": [],
f"Segmentname_{self.formatversions.subsequent_formatversion}": [],
f"Segmentgruppe_{self.formatversions.subsequent_formatversion}": [],
}
Expand All @@ -278,7 +278,7 @@ def test_align_columns_one_empty_dataframe(self) -> None:
{
f"Segmentname_{self.formatversions.previous_formatversion}": ["1", "2", "3"],
f"Segmentgruppe_{self.formatversions.previous_formatversion}": ["a", "b", "c"],
"diff": ["REMOVED", "REMOVED", "REMOVED"],
"Änderung": ["ENTFÄLLT", "ENTFÄLLT", "ENTFÄLLT"],
f"Segmentname_{self.formatversions.subsequent_formatversion}": ["", "", ""],
f"Segmentgruppe_{self.formatversions.subsequent_formatversion}": ["", "", ""],
}
Expand All @@ -300,7 +300,7 @@ def test_align_columns_full_offset(self) -> None:
{
f"Segmentname_{self.formatversions.previous_formatversion}": ["1", "2", "3", "", "", ""],
f"Segmentgruppe_{self.formatversions.previous_formatversion}": ["a", "b", "c", "", "", ""],
"diff": ["REMOVED", "REMOVED", "REMOVED", "NEW", "NEW", "NEW"],
"Änderung": ["ENTFÄLLT", "ENTFÄLLT", "ENTFÄLLT", "NEU", "NEU", "NEU"],
f"Segmentname_{self.formatversions.subsequent_formatversion}": ["", "", "", "4", "5", "6"],
f"Segmentgruppe_{self.formatversions.subsequent_formatversion}": ["", "", "", "d", "e", "f"],
}
Expand All @@ -322,7 +322,7 @@ def test_align_columns_duplicate_segments(self) -> None:
{
f"Segmentname_{self.formatversions.previous_formatversion}": ["1", "2", "2", ""],
f"Segmentgruppe_{self.formatversions.previous_formatversion}": ["a", "b", "c", ""],
"diff": ["", "", "REMOVED", "NEW"],
"Änderung": ["", "", "ENTFÄLLT", "NEU"],
f"Segmentname_{self.formatversions.subsequent_formatversion}": ["1", "2", "", "4"],
f"Segmentgruppe_{self.formatversions.subsequent_formatversion}": ["a", "b", "", "d"],
}
Expand All @@ -348,7 +348,7 @@ def test_align_columns_repeating_segments(self) -> None:
{
f"Segmentname_{self.formatversions.previous_formatversion}": ["1", "2", "3", "3", "2", ""],
f"Segmentgruppe_{self.formatversions.previous_formatversion}": ["a", "b", "c", "d", "e", ""],
"diff": ["", "", "", "REMOVED", "REMOVED", "NEW"],
"Änderung": ["", "", "", "ENTFÄLLT", "ENTFÄLLT", "NEU"],
f"Segmentname_{self.formatversions.subsequent_formatversion}": ["1", "2", "3", "", "", "4"],
f"Segmentgruppe_{self.formatversions.subsequent_formatversion}": ["a", "b", "c", "", "", "d"],
}
Expand Down Expand Up @@ -386,7 +386,7 @@ def test_align_columns_different_column_sets(self) -> None:
f"Segmentgruppe_{self.formatversions.previous_formatversion}": ["a", "b", ""],
f"Datenelement_{self.formatversions.previous_formatversion}": ["x", "y", ""],
f"Qualifier_{self.formatversions.previous_formatversion}": ["XY", "YZ", ""],
"diff": ["REMOVED", "", "NEW"],
"Änderung": ["ENTFÄLLT", "", "NEU"],
f"Segmentname_{self.formatversions.subsequent_formatversion}": ["", "2", "3"],
f"Segmentgruppe_{self.formatversions.subsequent_formatversion}": ["", "b", "c"],
f"Datenelement_{self.formatversions.subsequent_formatversion}": ["", "m", "n"],
Expand Down