feat: facilitate comparison of multiple PIDs stored in separate xlsx tabs (#86)

* Facilitate comparison of different PIDs across multiple FVs

* Store multiple comparison in separate xlsx tabs

* Rename `main` command to `compare`

* Add PIDs to column headers

* Change list to List type

Co-authored-by: konstantin <[email protected]>

* Outsource xlsx header formatting into separate helper functions

* Remove unused imports

* Cache PID file locations to scan through input directory only once

* Comment

* Use type `Path` instead of `str` for paths

* Refactor `_set_sheet_name` function to improve readability

* Add example worksheet/tab names as comment

* Use `strict: bool = True` to make sure there is no length mismatch

---------

Co-authored-by: konstantin <[email protected]>

Author: OLILHR

.github/workflows/cli_test.yml

@@ -22,4 +22,4 @@ jobs:
       - name: Try to install the script, then run the cli
         run: |
           pip install .
-          ahlbatross -i data/machine-readable_anwendungshandbuecher -o data/output
+          ahlbatross compare -i data/machine-readable_anwendungshandbuecher -o data/output

src/ahlbatross/core/ahb_multicomparison.py

@@ -0,0 +1,184 @@
+"""
+Interactive CLI PID comparison.
+Output = xlsx only.
+Multiple comparisons PID_A <-> PID_B, PID_A <-> PID_C, PID_A <-> PID_D, ... are merged in separate tabs.
+"""
+
+import logging
+import sys
+from pathlib import Path
+
+import typer
+from rich.console import Console
+from rich.prompt import Prompt
+
+from ahlbatross.core.ahb_comparison import align_ahb_rows
+from ahlbatross.core.ahb_processing import _get_formatversion_dirs, _get_nachrichtenformat_dirs
+from ahlbatross.formats.csv import get_csv_files, load_csv_files
+from ahlbatross.formats.xlsx import export_to_xlsx_multicompare
+
+logger = logging.getLogger(__name__)
+console = Console()
+
+_FORMATVERSION_PID_CACHE: dict[str, dict[str, tuple[Path, str]]] = {}
+
+
+def find_pid(root_dir: Path, formatversion: str, pruefid: str) -> tuple[Path, str] | None:
+    """
+    Find a PID file across all nachrichtenformat directories in a given FV.
+    """
+    # Store the locations of all PIDs after the initial scan/prompt of a FV directory
+    if formatversion not in _FORMATVERSION_PID_CACHE:
+        formatversion_dir = root_dir / formatversion
+        if not formatversion_dir.exists():
+            return None
+
+        nachrichtenformat_dirs = _get_nachrichtenformat_dirs(formatversion_dir)
+        _FORMATVERSION_PID_CACHE[formatversion] = {}
+
+        for nf_dir in nachrichtenformat_dirs:
+            csv_dir = nf_dir / "csv"
+            if not csv_dir.exists():
+                continue
+
+            for file in get_csv_files(csv_dir):
+                _FORMATVERSION_PID_CACHE[formatversion][file.stem] = (file, nf_dir.name)
+
+    return _FORMATVERSION_PID_CACHE[formatversion].get(pruefid)
+
+
+def get_pids(root_dir: Path, formatversion: str) -> list[str]:
+    """
+    Get all available PIDs across all nachrichtenformat directories for a given FV.
+    The result is sorted and contains every PID once at max.
+    """
+    if formatversion not in _FORMATVERSION_PID_CACHE:
+        find_pid(root_dir, formatversion, "")
+
+    return sorted(list(_FORMATVERSION_PID_CACHE.get(formatversion, {}).keys()))
+
+
+# pylint:disable=too-many-locals, too-many-branches, too-many-statements
+def multicompare_command(
+    input_dir: Path = typer.Option(..., "--input-dir", "-i", help="Directory containing AHB <PID>.json files."),
+    output_dir: Path = typer.Option(
+        ..., "--output-dir", "-o", help="Destination path to output directory containing merged xlsx files."
+    ),
+) -> None:
+    """
+    Interactive command to compare two PIDs across different FVs.
+    """
+    try:
+        if not input_dir.exists():
+            logger.error("❌ Input directory does not exist: %s", input_dir.absolute())
+            sys.exit(1)
+
+        formatversions = _get_formatversion_dirs(input_dir)
+        if not formatversions:
+            logger.error("❌ No format versions found in input directory")
+            sys.exit(1)
+
+        # show available FVs
+        formatversions_list = ", ".join(str(fv) for fv in formatversions)
+        console.print(f"\nAVAILABLE FVs: {formatversions_list}")
+
+        # get first FV
+        while True:
+            first_fv = Prompt.ask("\nSELECT FV")
+            if first_fv in [str(fv) for fv in formatversions]:
+                break
+            console.print("❌ Invalid FV.")
+
+        # get first PID
+        first_available_pids = get_pids(input_dir, first_fv)
+        if not first_available_pids:
+            logger.error("❌ No PIDs found in format version %s", first_fv)
+            sys.exit(1)
+
+        # show available PIDs
+        first_pids_list = ", ".join(first_available_pids)
+        console.print(f"\nAVAILABLE PIDs: {first_pids_list}")
+
+        while True:
+            first_pruefid = Prompt.ask("\nSELECT PID #1")
+            if first_pruefid in first_available_pids:
+                break
+            console.print("❌ Invalid PID.")
+
+        first_file = find_pid(input_dir, first_fv, first_pruefid)
+        if not first_file:
+            logger.error("❌ Could not find PID file for %s in %s", first_pruefid, first_fv)
+            sys.exit(1)
+
+        first_file_path, _ = first_file
+
+        comparison_groups = []
+        comparison_names = []
+
+        comparison_number = 2
+        while True:
+            # show available FVs
+            formatversions_list = ", ".join(str(fv) for fv in formatversions)
+            console.print(f"\nAVAILABLE FVs (🏁 PRESS ENTER TO FINISH): {formatversions_list}")
+
+            next_fv = Prompt.ask(f"\nSELECT FV #{comparison_number}", default="")
+            if not next_fv:
+                # hitting enter aborts the process.
+                break
+
+            if next_fv not in [str(fv) for fv in formatversions]:
+                console.print("❌ Invalid FV.")
+                continue
+
+            next_available_pids = get_pids(input_dir, next_fv)
+            if not next_available_pids:
+                logger.error("❌ No PIDs found for format version %s", next_fv)
+                continue
+
+            # show available PIDs
+            next_pids_list = ", ".join(next_available_pids)
+            console.print(f"\nAVAILABLE PIDs (FV{next_fv}): {next_pids_list}")
+
+            while True:
+                next_pruefid = Prompt.ask(f"\nSELECT PID #{comparison_number}")
+                if next_pruefid == first_pruefid and next_fv == first_fv:
+                    console.print("❌ Cannot compare identical PIDs of the same format version.")
+                elif next_pruefid in next_available_pids:
+                    break
+                else:
+                    console.print("❌ Invalid PID.")
+
+            next_file = find_pid(input_dir, next_fv, next_pruefid)
+            if not next_file:
+                logger.error("❌ Could not find PID file for %s in %s", next_pruefid, next_fv)
+                continue
+
+            next_file_path, _ = next_file
+
+            try:
+                first_rows, next_rows = load_csv_files(first_file_path, next_file_path, first_fv, next_fv)
+                comparisons = align_ahb_rows(first_rows, next_rows)
+
+                comparison_groups.append(comparisons)
+                comparison_names.append(f"{first_pruefid}_{next_pruefid}")
+
+                comparison_number += 1
+            except (OSError, IOError, ValueError) as e:
+                logger.error(
+                    "❌ Error comparing %s/%s with %s/%s: %s", first_fv, first_pruefid, next_fv, next_pruefid, str(e)
+                )
+                continue
+
+        if not comparison_groups:
+            sys.exit(1)
+
+        output_dir.mkdir(parents=True, exist_ok=True)
+
+        xlsx_path = output_dir / f"{first_pruefid}_comparisons.xlsx"
+        export_to_xlsx_multicompare(comparison_groups, comparison_names, Path(xlsx_path))
+
+        logger.info("✅ Successfully processed: %s", xlsx_path)
+
+    except (OSError, IOError, ValueError, TypeError) as e:
+        logger.exception("❌ Error: %s", str(e))
+        sys.exit(1)

src/ahlbatross/core/ahb_processing.py

@@ -48,7 +48,7 @@ def _get_nachrichtenformat_dirs(formatversion_dir: Path) -> list[Path]:
     Fetch all <nachrichtenformat> directories that contain actual csv files.
     """
     if not formatversion_dir.exists():
-        raise FileNotFoundError(f"❌ Formatversion directory not found: {formatversion_dir.absolute()}")
+        raise FileNotFoundError(f"❌ FV directory not found: {formatversion_dir.absolute()}")
 
     return [d for d in formatversion_dir.iterdir() if d.is_dir() and (d / "csv").exists() and (d / "csv").is_dir()]
 
@@ -77,15 +77,15 @@ def get_formatversion_pairs(root_dir: Path) -> list[tuple[EdifactFormatVersion,
 
         if is_subsequent_empty or is_previous_empty:
             logger.warning(
-                "❗️Skipping empty consecutive formatversions: %s -> %s",
+                "❗️Skipping empty consecutive FVs: %s -> %s",
                 subsequent_formatversion,
                 previous_formatversion,
             )
             continue
 
         consecutive_formatversions.append((subsequent_formatversion, previous_formatversion))
 
-    logger.debug("Consecutive formatversions: %s", consecutive_formatversions)
+    logger.debug("Consecutive FVs: %s", consecutive_formatversions)
     return consecutive_formatversions
 
 
@@ -100,7 +100,7 @@ def get_matching_csv_files(
     subsequent_formatversion_dir = root_dir / subsequent_formatversion
 
     if not all(d.exists() for d in [previous_formatversion_dir, subsequent_formatversion_dir]):
-        logger.error("❌ At least one formatversion directory does not exist.")
+        logger.error("❌ At least one FV directory does not exist.")
         return []
 
     matching_files = []
@@ -140,13 +140,11 @@ def process_ahb_files(input_dir: Path, output_dir: Path) -> None:
 
     consecutive_formatversions = get_formatversion_pairs(input_dir)
     if not consecutive_formatversions:
-        logger.warning("❗️ No valid consecutive formatversion subdirectories found to compare.")
+        logger.warning("❗️ No valid consecutive FVs subdirectories found to compare.")
         return
 
     for subsequent_formatversion, previous_formatversion in consecutive_formatversions:
-        logger.info(
-            "⌛ Processing consecutive formatversions: %s -> %s", subsequent_formatversion, previous_formatversion
-        )
+        logger.info("⌛ Processing consecutive FVs: %s -> %s", subsequent_formatversion, previous_formatversion)
 
         try:
             matching_files = get_matching_csv_files(input_dir, previous_formatversion, subsequent_formatversion)
@@ -184,7 +182,7 @@ def process_ahb_files(input_dir: Path, output_dir: Path) -> None:
 
         except (OSError, IOError, ValueError) as e:
             logger.error(
-                "❌ Error processing formatversions %s -> %s: %s",
+                "❌ Error processing FVs %s -> %s: %s",
                 subsequent_formatversion,
                 previous_formatversion,
                 str(e),