pyxecm Helper Classes

ASSOC

Assoc Module to implement functions to read / write from so called "Assoc" data structures in Content Server.

Right now this module is used to tweak settings in XML-based transport packages that include Assoc structures inside some of the XML elements.

Assoc

Class Assoc is used to handle Extended ECM Assoc data structures.

Source code in packages/pyxecm/src/pyxecm/helper/assoc.py

class Assoc:
    """Class Assoc is used to handle Extended ECM Assoc data structures."""

    @classmethod
    def is_unicode_escaped(cls, assoc_string: str) -> bool:
        """Determine if a string is unicode escaped.

        Args:
            assoc_string (str): string with the Assoc data

        Returns:
            bool: True if string is in Unicode, False otherwise

        """

        pattern = r"\\u[0-9a-fA-F]{4}"
        matches = re.findall(pattern, assoc_string)

        return len(matches) > 0

    # end method definition

    @classmethod
    def escape_unicode(cls, assoc_string: str) -> str:
        """Escape / Encode a given string in Unicode.

        Args:
            assoc_string (str):
                The source string to escape.

        Returns:
            str:
                The escaped string.

        """

        encoded_string = assoc_string.encode("unicode_escape")  # .decode()

        return encoded_string

    # end method definition

    @classmethod
    def unescape_unicode(cls, assoc_string: str) -> str:
        """Unescape / Decode a given string.

        Args:
            assoc_string (str):
                The source string to unescape.

        Returns:
            str:
                The unescaped string.

        """
        try:
            decoded_string = bytes(assoc_string, "utf-8").decode("unicode_escape")
        except UnicodeDecodeError:
            return assoc_string

        return decoded_string

    # end method definition

    @classmethod
    def is_html_escaped(cls, assoc_string: str) -> bool:
        """Check if an Assoc String is HTML escaped.

        Args:
            assoc_string (str):
                The string to test for HTML escaping.

        Returns:
            bool:
                True = string is HTML escaped, False if now

        """

        decoded_string = html.unescape(assoc_string)

        return assoc_string != decoded_string

    # end method definition

    @classmethod
    def unescape_html(cls, assoc_string: str) -> str:
        """HTML unescape a a string.

        Args:
            assoc_string (str): the string to unescape.

        Returns:
            str: unescaped string

        """

        decoded_string = html.unescape(assoc_string)
        return decoded_string

    # end method definition

    @classmethod
    def string_to_dict(cls, assoc_string: str) -> dict:
        """Convert an Assoc string to a Python dict.

        Each comma-separated element of the Assoc will become a dict element.

        Args:
            assoc_string (str):
                The source Assoc string to convert.

        Returns:
            dict: Python dict with the Assoc elements

        """

        if cls.is_html_escaped(assoc_string):
            assoc_string = cls.unescape_html(assoc_string)
        if cls.is_unicode_escaped(assoc_string):
            assoc_string = cls.unescape_unicode(assoc_string)

        # Split the string using regex pattern
        pieces = re.split(r",(?=(?:[^']*'[^']*')*[^']*$)", assoc_string)

        # Trim any leading/trailing spaces from each piece
        pieces = [piece.strip() for piece in pieces]

        # Split the last pieces from the assoc close tag
        last_piece = pieces[-1].split(">")[0]

        # Remove the first two and last pieces from the list
        # the first two are mostly "1" and "?"
        pieces = pieces[2:-1]

        # Insert the last pieces separately
        pieces.append(last_piece)

        assoc_dict: dict = {}

        for piece in pieces:
            name = piece.split("=")[0]
            if name[0] == "'":
                name = name[1:]
            if name[-1] == "'":
                name = name[:-1]
            value = piece.split("=")[1]
            if value[0] == "'":
                value = value[1:]
            if value[-1] == "'":
                value = value[:-1]
            assoc_dict[name] = value

        return assoc_dict

    # end method definition

    @classmethod
    def dict_to_string(cls, assoc_dict: dict) -> str:
        """Convert a Python dict to an Assoc string.

        Args:
            assoc_dict (dict):
                The source dictionary to convert.

        Returns:
            str:
                The resulting Assoc string.

        """

        assoc_string: str = "A<1,?,"

        for item in assoc_dict.items():
            assoc_string += "\u0027" + item[0] + "\u0027"
            assoc_string += "="
            # Extended ECM's XML is a bit special in cases.
            # If the value is empty set (curly braces) it does
            # not put it in quotes. As Extended ECM is also very
            # picky about XML syntax we better produce it exactly like that.
            if item[1] == "{}":
                assoc_string += item[1] + ","
            else:
                assoc_string += "\u0027" + item[1] + "\u0027,"

        if assoc_dict.items():
            assoc_string = assoc_string[:-1]
        assoc_string += ">"
        return assoc_string

    # end method definition

    @classmethod
    def extract_substring(
        cls,
        input_string: str,
        start_sequence: str,
        stop_sequence: str,
    ) -> str | None:
        """Extract a substring that is delimited by a start and stop sequence.

        Args:
            input_string (str):
                Input string to search the delimited substring in.
            start_sequence (str):
                Start esequence of characters.
            stop_sequence (str):
                Stop sequence of characters

        Returns:
            str | None:
                The deliminated substring or None if not found.

        """

        start_index = input_string.find(start_sequence)
        if start_index == -1:
            return None

        end_index = input_string.find(stop_sequence, start_index)
        if end_index == -1:
            return None

        end_index += len(stop_sequence)
        return input_string[start_index:end_index]

    # end method definition

    @classmethod
    def extract_assoc_string(cls, input_string: str, is_escaped: bool = False) -> str:
        """Extract an Assoc from a string.

        The assoc is deliminated by A< ... >.

        Args:
            input_string (str):
                Input string that includes the Assoc as a substring.
            is_escaped (bool, optional):
                Whether or not the input string includes the
                assoc escaped or not.

        Returns:
            str:
                The assoc string.

        """

        if is_escaped:
            assoc_string = cls.extract_substring(input_string, "A&lt;", "&gt;")
        else:
            assoc_string = cls.extract_substring(input_string, "A<", ">")
        return assoc_string

dict_to_string(assoc_dict) classmethod

Convert a Python dict to an Assoc string.

Parameters:

Name	Type	Description	Default
assoc_dict	dict	The source dictionary to convert.	required

Returns:

Name	Type	Description
str	str	The resulting Assoc string.

Source code in packages/pyxecm/src/pyxecm/helper/assoc.py

@classmethod
def dict_to_string(cls, assoc_dict: dict) -> str:
    """Convert a Python dict to an Assoc string.

    Args:
        assoc_dict (dict):
            The source dictionary to convert.

    Returns:
        str:
            The resulting Assoc string.

    """

    assoc_string: str = "A<1,?,"

    for item in assoc_dict.items():
        assoc_string += "\u0027" + item[0] + "\u0027"
        assoc_string += "="
        # Extended ECM's XML is a bit special in cases.
        # If the value is empty set (curly braces) it does
        # not put it in quotes. As Extended ECM is also very
        # picky about XML syntax we better produce it exactly like that.
        if item[1] == "{}":
            assoc_string += item[1] + ","
        else:
            assoc_string += "\u0027" + item[1] + "\u0027,"

    if assoc_dict.items():
        assoc_string = assoc_string[:-1]
    assoc_string += ">"
    return assoc_string

escape_unicode(assoc_string) classmethod

Escape / Encode a given string in Unicode.

Parameters:

Name	Type	Description	Default
assoc_string	str	The source string to escape.	required

Returns:

Name	Type	Description
str	str	The escaped string.

Source code in packages/pyxecm/src/pyxecm/helper/assoc.py

@classmethod
def escape_unicode(cls, assoc_string: str) -> str:
    """Escape / Encode a given string in Unicode.

    Args:
        assoc_string (str):
            The source string to escape.

    Returns:
        str:
            The escaped string.

    """

    encoded_string = assoc_string.encode("unicode_escape")  # .decode()

    return encoded_string

extract_assoc_string(input_string, is_escaped=False) classmethod

Extract an Assoc from a string.

The assoc is deliminated by A< ... >.

Parameters:

Name	Type	Description	Default
input_string	str	Input string that includes the Assoc as a substring.	required
is_escaped	bool	Whether or not the input string includes the assoc escaped or not.	False

Returns:

Name	Type	Description
str	str	The assoc string.

Source code in packages/pyxecm/src/pyxecm/helper/assoc.py

@classmethod
def extract_assoc_string(cls, input_string: str, is_escaped: bool = False) -> str:
    """Extract an Assoc from a string.

    The assoc is deliminated by A< ... >.

    Args:
        input_string (str):
            Input string that includes the Assoc as a substring.
        is_escaped (bool, optional):
            Whether or not the input string includes the
            assoc escaped or not.

    Returns:
        str:
            The assoc string.

    """

    if is_escaped:
        assoc_string = cls.extract_substring(input_string, "A&lt;", "&gt;")
    else:
        assoc_string = cls.extract_substring(input_string, "A<", ">")
    return assoc_string

extract_substring(input_string, start_sequence, stop_sequence) classmethod

Extract a substring that is delimited by a start and stop sequence.

Parameters:

Name	Type	Description	Default
input_string	str	Input string to search the delimited substring in.	required
start_sequence	str	Start esequence of characters.	required
stop_sequence	str	Stop sequence of characters	required

Returns:

Type	Description
str | None	str | None: The deliminated substring or None if not found.

Source code in packages/pyxecm/src/pyxecm/helper/assoc.py

@classmethod
def extract_substring(
    cls,
    input_string: str,
    start_sequence: str,
    stop_sequence: str,
) -> str | None:
    """Extract a substring that is delimited by a start and stop sequence.

    Args:
        input_string (str):
            Input string to search the delimited substring in.
        start_sequence (str):
            Start esequence of characters.
        stop_sequence (str):
            Stop sequence of characters

    Returns:
        str | None:
            The deliminated substring or None if not found.

    """

    start_index = input_string.find(start_sequence)
    if start_index == -1:
        return None

    end_index = input_string.find(stop_sequence, start_index)
    if end_index == -1:
        return None

    end_index += len(stop_sequence)
    return input_string[start_index:end_index]

is_html_escaped(assoc_string) classmethod

Check if an Assoc String is HTML escaped.

Parameters:

Name	Type	Description	Default
assoc_string	str	The string to test for HTML escaping.	required

Returns:

Name	Type	Description
bool	bool	True = string is HTML escaped, False if now

Source code in packages/pyxecm/src/pyxecm/helper/assoc.py

@classmethod
def is_html_escaped(cls, assoc_string: str) -> bool:
    """Check if an Assoc String is HTML escaped.

    Args:
        assoc_string (str):
            The string to test for HTML escaping.

    Returns:
        bool:
            True = string is HTML escaped, False if now

    """

    decoded_string = html.unescape(assoc_string)

    return assoc_string != decoded_string

is_unicode_escaped(assoc_string) classmethod

Determine if a string is unicode escaped.

Parameters:

Name	Type	Description	Default
assoc_string	str	string with the Assoc data	required

Returns:

Name	Type	Description
bool	bool	True if string is in Unicode, False otherwise

Source code in packages/pyxecm/src/pyxecm/helper/assoc.py

@classmethod
def is_unicode_escaped(cls, assoc_string: str) -> bool:
    """Determine if a string is unicode escaped.

    Args:
        assoc_string (str): string with the Assoc data

    Returns:
        bool: True if string is in Unicode, False otherwise

    """

    pattern = r"\\u[0-9a-fA-F]{4}"
    matches = re.findall(pattern, assoc_string)

    return len(matches) > 0

string_to_dict(assoc_string) classmethod

Convert an Assoc string to a Python dict.

Each comma-separated element of the Assoc will become a dict element.

Parameters:

Name	Type	Description	Default
assoc_string	str	The source Assoc string to convert.	required

Returns:

Name	Type	Description
dict	dict	Python dict with the Assoc elements

Source code in packages/pyxecm/src/pyxecm/helper/assoc.py

@classmethod
def string_to_dict(cls, assoc_string: str) -> dict:
    """Convert an Assoc string to a Python dict.

    Each comma-separated element of the Assoc will become a dict element.

    Args:
        assoc_string (str):
            The source Assoc string to convert.

    Returns:
        dict: Python dict with the Assoc elements

    """

    if cls.is_html_escaped(assoc_string):
        assoc_string = cls.unescape_html(assoc_string)
    if cls.is_unicode_escaped(assoc_string):
        assoc_string = cls.unescape_unicode(assoc_string)

    # Split the string using regex pattern
    pieces = re.split(r",(?=(?:[^']*'[^']*')*[^']*$)", assoc_string)

    # Trim any leading/trailing spaces from each piece
    pieces = [piece.strip() for piece in pieces]

    # Split the last pieces from the assoc close tag
    last_piece = pieces[-1].split(">")[0]

    # Remove the first two and last pieces from the list
    # the first two are mostly "1" and "?"
    pieces = pieces[2:-1]

    # Insert the last pieces separately
    pieces.append(last_piece)

    assoc_dict: dict = {}

    for piece in pieces:
        name = piece.split("=")[0]
        if name[0] == "'":
            name = name[1:]
        if name[-1] == "'":
            name = name[:-1]
        value = piece.split("=")[1]
        if value[0] == "'":
            value = value[1:]
        if value[-1] == "'":
            value = value[:-1]
        assoc_dict[name] = value

    return assoc_dict

unescape_html(assoc_string) classmethod

HTML unescape a a string.

Parameters:

Name	Type	Description	Default
assoc_string	str	the string to unescape.	required

Returns:

Name	Type	Description
str	str	unescaped string

Source code in packages/pyxecm/src/pyxecm/helper/assoc.py

@classmethod
def unescape_html(cls, assoc_string: str) -> str:
    """HTML unescape a a string.

    Args:
        assoc_string (str): the string to unescape.

    Returns:
        str: unescaped string

    """

    decoded_string = html.unescape(assoc_string)
    return decoded_string

unescape_unicode(assoc_string) classmethod

Unescape / Decode a given string.

Parameters:

Name	Type	Description	Default
assoc_string	str	The source string to unescape.	required

Returns:

Name	Type	Description
str	str	The unescaped string.

Source code in packages/pyxecm/src/pyxecm/helper/assoc.py

@classmethod
def unescape_unicode(cls, assoc_string: str) -> str:
    """Unescape / Decode a given string.

    Args:
        assoc_string (str):
            The source string to unescape.

    Returns:
        str:
            The unescaped string.

    """
    try:
        decoded_string = bytes(assoc_string, "utf-8").decode("unicode_escape")
    except UnicodeDecodeError:
        return assoc_string

    return decoded_string

Data

Data Module leveraging Pandas to manipulte data sets read for bulk generation of Content Server items.

See: https://pandas.pydata.org

This code implements a class called "Data" which is a wrapper to Pandas data frame.

Data

Used to automate data loading for the customizer.

Source code in packages/pyxecm/src/pyxecm/helper/data.py

class Data:
    """Used to automate data loading for the customizer."""

    logger: logging.Logger = default_logger

    _df: pd.DataFrame
    _lock: threading.Lock = threading.Lock()

    def __init__(
        self,
        init_data: pd.DataFrame | list = None,
        logger: logging.Logger = default_logger,
    ) -> None:
        """Initialize the Data object.

        Args:
            init_data (pd.DataFrame | list, optional):
                Data to initialize the data frame. Can either be
                another data frame (that gets copied) or a list of dictionaries.
                Defaults to None.
            logger (logging.Logger, optional):
                Pass a special logging object. This is optional. If not provided,
                the default logger is used.

        """

        if logger != default_logger:
            self.logger = logger.getChild("data")
            for logfilter in logger.filters:
                self.logger.addFilter(logfilter)

        if init_data is not None:
            # if a data frame is passed to the constructor we
            # copy its content to the new Data object

            if isinstance(init_data, pd.DataFrame):
                self._df: pd.DataFrame = init_data.copy()
            elif isinstance(init_data, Data):
                if init_data.get_data_frame() is not None:
                    self._df: pd.DataFrame = init_data.get_data_frame().copy()
            elif isinstance(init_data, list):
                self._df: pd.DataFrame = pd.DataFrame(init_data)
            elif isinstance(init_data, dict):
                # it is important to wrap the dict in a list to avoid that more than 1 row is created
                self._df: pd.DataFrame = pd.DataFrame([init_data])
            else:
                self.logger.error("Illegal initialization data for 'Data' class!")
                self._df = None
        else:
            self._df = None

    # end method definition

    def __len__(self) -> int:
        """Return lenght of the embedded Pandas data frame object.

        This is basically a convenience method.

        Returns:
            int:
                Lenght of the data frame.

        """

        if self._df is not None:
            return len(self._df)
        return 0

    # end method definition

    def __str__(self) -> str:
        """Print the Pandas data frame object.

        Returns:
            str:
                String representation.

        """

        # if data frame is initialized we return
        # the string representation of pd.DataFrame
        if self._df is not None:
            return str(self._df)

        return str(self)

    # end method definition

    def __getitem__(self, column: str) -> pd.Series:
        """Return the column corresponding to the key from the data frame.

        Args:
            column (str): The name of the data frame column.

        Returns:
            pd.Series: The column of the data frame with the given name.

        """

        return self._df[column]

    # end method definition

    def lock(self) -> threading.Lock:
        """Return the threading lock object.

        Returns:
            threading.Lock: The threading lock object.

        """

        return self._lock

    # end method definition

    def get_data_frame(self) -> pd.DataFrame:
        """Get the Pandas data frame object.

        Returns:
            pd.DataFrame: The Pandas data frame object.

        """

        return self._df

    # end method definition

    def set_data_frame(self, df: pd.DataFrame) -> None:
        """Set the Pandas data frame object.

        Args:
            df (pd.DataFrame): The new Pandas data frame object.

        """

        self._df = df

    # end method definition

    def get_columns(self) -> list | None:
        """Get the list of column names of the data frame.

        Returns:
            list | None:
                The list of column names in the data frame.

        """

        if self._df is None:
            return None

        return self._df.columns

    # end method definition

    def print_info(
        self,
        show_size: bool = True,
        show_info: bool = False,
        show_columns: bool = False,
        show_first: bool = False,
        show_last: bool = False,
        show_sample: bool = False,
        show_statistics: bool = False,
        row_num: int = 10,
    ) -> None:
        """Log information about the data frame.

        Args:
            show_size (bool, optional):
                Show size of data frame. Defaults to True.
            show_info (bool, optional):
                Show information for data frame. Defaults to False.
            show_columns (bool, optional):
                Show columns of data frame. Defaults to False.
            show_first (bool, optional):
                Show first N items. Defaults to False. N is defined
                by the row_num parameter.
            show_last (bool, optional):
                Show last N items. Defaults to False. N is defined
                by the row_num parameter.
            show_sample (bool, optional):
                Show N sample items. Defaults to False. N is defined
                by the row_num parameter.
            show_statistics (bool, optional):
                Show data frame statistics. Defaults to False.
            row_num (int, optional):
                Used as the number of rows printed using show_first,
                show_last, show_sample. Default is 10.

        """

        if self._df is None:
            self.logger.warning("Data frame is not initialized!")
            return

        if show_size:
            self.logger.info(
                "Data frame has %s row(s) and %s column(s)",
                self._df.shape[0],
                self._df.shape[1],
            )

        if show_info:
            # df.info() can not easily be embedded into a string
            self._df.info()

        if show_columns:
            self.logger.info("Columns:\n%s", self._df.columns)
            self.logger.info(
                "Columns with number of NaN values:\n%s",
                self._df.isna().sum(),
            )
            self.logger.info(
                "Columns with number of non-NaN values:\n%s",
                self._df.notna().sum(),
            )

        if show_first:
            # the default for head is n = 5:
            self.logger.info("First %s rows:\n%s", str(row_num), self._df.head(row_num))

        if show_last:
            # the default for tail is n = 5:
            self.logger.info("Last %s rows:\n%s", str(row_num), self._df.tail(row_num))

        if show_sample:
            # the default for sample is n = 1:
            self.logger.info(
                "%s Sample rows:\n%s",
                str(row_num),
                self._df.sample(n=row_num),
            )

        if show_statistics:
            self.logger.info(
                "Description of statistics for data frame:\n%s",
                self._df.describe(),
            )
            self.logger.info(
                "Description of statistics for data frame (transformed):\n%s",
                self._df.describe().T,
            )
            self.logger.info(
                "Description of statistics for data frame (objects):\n%s",
                self._df.describe(include="object"),
            )

    # end method definition

    def append(self, add_data: pd.DataFrame | list | dict) -> bool:
        """Append additional data to the data frame.

        Args:
            add_data (pd.DataFrame | list | dict):
                Additional data. Can be pd.DataFrame or list of dicts (or Data).

        Returns:
            bool:
                True = Success, False = Error

        """

        # Does the data frame has already content?
        # Then we need to concat / append. Otherwise
        # we just initialize self._df
        if self._df is not None:
            if isinstance(add_data, pd.DataFrame):
                self._df = pd.concat([self._df, add_data], ignore_index=True)
                return True
            elif isinstance(add_data, Data):
                df = add_data.get_data_frame()
                if df is not None and not df.empty:
                    self._df = pd.concat([self._df, df], ignore_index=True)
                return True
            elif isinstance(add_data, list):
                if add_data:
                    df = Data(add_data, logger=self.logger)
                    self._df = pd.concat(
                        [self._df, df.get_data_frame()],
                        ignore_index=True,
                    )
                return True
            elif isinstance(add_data, dict):
                if add_data:
                    # it is important to wrap the dict in a list to avoid that more than 1 row is created
                    df = Data([add_data], logger=self.logger)
                    self._df = pd.concat(
                        [self._df, df.get_data_frame()],
                        ignore_index=True,
                    )
                return True
            else:
                self.logger.error("Illegal data type -> '%s'", type(add_data))
                return False
        elif isinstance(add_data, pd.DataFrame):
            self._df = add_data
            return True
        elif isinstance(add_data, Data):
            self._df = add_data.get_data_frame()
            return True
        elif isinstance(add_data, list):
            self._df = pd.DataFrame(add_data)
            return True
        elif isinstance(add_data, dict):
            # it is important to wrap the dict in a list to avoid that more than 1 row is created
            self._df = pd.DataFrame([add_data])
            return True
        else:
            self.logger.error("Illegal data type -> '%s'", type(add_data))
            return False

    # end method definition

    def merge(
        self,
        merge_data: pd.DataFrame,
        on: str | list[str] | None = None,
        how: str = "inner",
        left_on: str | list[str] | None = None,
        right_on: str | list[str] | None = None,
        left_index: bool = False,
        right_index: bool = False,
        suffixes: tuple[str, str] = ("_x", "_y"),
        indicator: bool = False,
        validate: str | None = None,
    ) -> pd.DataFrame | None:
        """Merge the current DataFrame (_df) with another DataFrame.

        Args:
            merge_data (pd.DataFrame | Data):
                The DataFrame to merge with.
            on (str | list[str]):
                Column(s) to merge on. Defaults to None.
            how (str, optional):
                Type of merge ('inner', 'outer', 'left', 'right', 'cross'). Defaults to 'inner'.
            left_on (str | list[str] | None, optional):
                Column(s) from self._df to merge on. Defaults to None.
            right_on (str | list[str] | None, optional):
                Column(s) from other DataFrame to merge on. Defaults to None.
            left_index (str | list[str], optional):
                 Whether to merge on the index of self._df. Defaults to False.
            right_index (bool, optional):
                Whether to merge on the index of other. Defaults to False.
            suffixes (tuple[str, str]):
                Suffixes for overlapping column names. Defaults to ('_x', '_y').
            indicator (bool, optional):
                If True, adds a column showing the merge source. Defaults to False.
            validate ():
                If provided, checks merge integrity
                ('one_to_one', 'one_to_many', 'many_to_one', 'many_to_many'). Defaults to None.

        Returns:
            The merged DataFrame or None in case of an error.

        Exceptions:
            ValueError: If `other` is not a DataFrame.
            KeyError: If required columns for merging are missing.
            ValueError: If `validate` check fails.

        """

        if self._df is None or self._df.empty:
            self._df = merge_data

        if isinstance(merge_data, Data):
            merge_data = merge_data.get_data_frame()  # Extract DataFrame from Data instance

        try:
            return self._df.merge(
                merge_data,
                how=how,
                on=on,
                left_on=left_on,
                right_on=right_on,
                left_index=left_index,
                right_index=right_index,
                suffixes=suffixes,
                indicator=indicator,
                validate=validate,
            )
        except KeyError:
            self.logger.error("Column(s) not found for merging!")
        except ValueError:
            self.logger.error("Invalid merge operation!")

        return None

    # end method definition

    def strip(self, columns: list | None = None, inplace: bool = True) -> pd.DataFrame:
        """Strip leading and trailing spaces from specified columns in a data frame.

        Args:
            columns (list | None):
                The list of column names to strip. If None, it strips
                leading and trailing spaces from _all_ string columns.
            inplace (bool, optional):
                If True, the data modification is done in place, i.e.
                modifying the existing data frame of the object.
                If False, the data frame is copied and the copy is modified
                and returned.

        Returns:
            pd.DataFrame:
                The modified data frame with stripped columns.

        """

        df = self._df.copy() if not inplace else self._df

        if columns is None:
            # Strip spaces from all string columns
            df = df.apply(lambda x: x.str.strip() if x.dtype == "object" else x)
        else:
            # Strip spaces from specified columns
            for col in columns:
                if col in df.columns and df[col].dtype == "object":  # Check if the column exists and is of string type
                    df[col] = df[col].str.strip()

        if inplace:
            self._df = df

        return df

    # end method definition

    def load_json_data(
        self,
        json_path: str,
        convert_dates: bool = False,
        index_column: str | None = None,
        compression: str | None = None,
    ) -> bool:
        """Load JSON data into a Pandas data frame.

        Args:
            json_path (str):
                The path to the JSON file.
            convert_dates (bool, optional):
                Defines whether or not dates should be converted.
                The default is False = dates are NOT converted.
            index_column (str | None, optional):
                The Name of the column (i.e. JSON data field) that should
                become the index in the loaded data frame.
            compression (str | None):
                Remove a compression:
                * gzip (.gz)
                * bz2 (.bz2)
                * zip (.zip)
                * xz (.xz)
                The value for compression should not include the dot.
                Default is None = no compression.

        Returns:
            bool: False in case an error occured, True otherwise.

        """

        if not json_path:
            self.logger.error(
                "You have not specified a JSON path!",
            )
            return False

        # If compression is enabled the file path should have
        # the matching file name extension:
        if compression:
            compression = compression.lstrip(".")  # remove a dot prefix if present
            suffix = "." + compression if compression != "gzip" else "gz"
            if not json_path.endswith(suffix):
                json_path += suffix

        if not os.path.exists(json_path):
            self.logger.error(
                "Missing JSON file - you have not specified a valid path -> '%s'.",
                json_path,
            )
            return False

        # Load data from JSON file
        try:
            df = pd.read_json(
                path_or_buf=json_path,
                convert_dates=convert_dates,
                compression=compression,
            )

            if index_column and index_column not in df.columns:
                self.logger.error(
                    "Specified index column -> '%s' not found in the JSON data.",
                    index_column,
                )
                return False

            if index_column:
                df = df.set_index(keys=index_column)
            if self._df is None:
                self._df = df
            else:
                self._df = pd.concat([self._df, df])
            self.logger.info(
                "After loading JSON file -> '%s', the data frame has %s row(s) and %s column(s)",
                json_path,
                self._df.shape[0],
                self._df.shape[1],
            )
        except FileNotFoundError:
            self.logger.error(
                "JSON file -> '%s' not found. Please check the file path.",
                json_path,
            )
            return False
        except PermissionError:
            self.logger.error(
                "Missing permission to access the JSON file -> '%s'.",
                json_path,
            )
            return False
        except OSError:
            self.logger.error("An I/O error occurred!")
            return False
        except json.JSONDecodeError:
            self.logger.error(
                "Unable to decode JSON file -> '%s'",
                json_path,
            )
            return False
        except ValueError:
            self.logger.error("Invalid JSON input -> %s", json_path)
            return False
        except AttributeError:
            self.logger.error("Unexpected JSON data structure in file -> %s", json_path)
            return False
        except TypeError:
            self.logger.error("Unexpected JSON data type in file -> %s", json_path)
            return False
        except KeyError:
            self.logger.error("Missing key in JSON data in file -> %s", json_path)
            return False

        return True

    # end method definition

    def save_json_data(
        self,
        json_path: str,
        orient: str = "records",
        preserve_index: bool = False,
        index_column: str = "index",
        compression: str | None = None,
    ) -> bool:
        """Save JSON data from data frame to file.

        Args:
            json_path (str): The path to where the JSON file should be safed.
            orient (str, optional):
                The structure of the JSON. Possible values:
                * "records" (this is the default)
                * "columns"
                * "index"
                * "table"
                * "split"
            preserve_index (bool, optional):
                Defines if the index column of the data frame should be exported as well.
                The default is False (index is not exported).
            index_column (str, optional):
                The Name of the column (i.e. JSON data field) that should
                become the index in the loaded data frame. The default is "index".
            compression (str | None):
                Apply a compression:
                * gzip (.gz)
                * bz2 (.bz2)
                * zip (.zip)
                * xz (.xz)

        Returns:
            bool:
                False in case an error occured, True otherwise.

        """

        if not json_path:
            self.logger.error(
                "You have not specified a JSON path!",
            )
            return False

        # If compression is enabled the file path should have
        # the matching file name extension:
        if compression:
            suffix = "." + compression if compression != "gzip" else ".gz"
            if not json_path.endswith(suffix):
                json_path += suffix

        # Save data to JSON file
        try:
            if self._df is not None:
                if not os.path.exists(os.path.dirname(json_path)):
                    os.makedirs(os.path.dirname(json_path), exist_ok=True)

                # index parameter is only allowed if orient has one of the following values:
                if orient in ("columns", "index", "table", "split"):
                    self._df.to_json(
                        path_or_buf=json_path,
                        index=preserve_index,
                        orient=orient,
                        indent=2,
                        compression=compression,
                        date_format="iso",
                    )
                # In this case we cannot use the index parameter as this would give this error:
                # Value Error -> 'index=True' is only valid when 'orient' is 'split', 'table', 'index', or 'columns'
                # So we create a new column that preserves the original row IDs from the index. The nasme

                elif preserve_index:
                    df_with_index = self._df.reset_index(
                        names=index_column,
                        inplace=False,
                    )
                    df_with_index.to_json(
                        path_or_buf=json_path,
                        orient=orient,
                        indent=2,
                        compression=compression,
                        date_format="iso",
                    )
                else:
                    self._df.to_json(
                        path_or_buf=json_path,
                        orient=orient,
                        indent=2,
                        compression=compression,
                        date_format="iso",
                    )
            else:
                self.logger.warning(
                    "Data frame is empty. Cannot write it to JSON file -> '%s'.",
                    json_path,
                )
                return False
        except FileNotFoundError:
            self.logger.error(
                "File -> '%s' not found. Please check the file path.",
                json_path,
            )
            return False
        except PermissionError:
            self.logger.error(
                "Permission denied to access the file -> '%s'.",
                json_path,
            )
            return False
        except OSError:
            self.logger.error("An I/O error occurred accessing file -> %s", json_path)
            return False
        except ValueError:
            self.logger.error("Value error!")
            return False

        return True

    # end method definition

    def load_excel_data(
        self,
        xlsx_path: str,
        sheet_names: str | list | None = 0,
        usecols: str | list | None = None,
        skip_rows: int | None = None,
        header: int | None = 0,
        names: list | None = None,
        na_values: list | None = None,
    ) -> bool:
        """Load Excel (xlsx) data into Pandas data frame.

        Supports xls, xlsx, xlsm, xlsb, odf, ods and odt file extensions
        read from a local filesystem or URL. Supports an option to read a
        single sheet or a list of sheets.

        Args:
            xlsx_path (str):
                The path to the Excel file to load.
            sheet_names (list | str | int, optional):
                Name or Index of the sheet in the Excel workbook to load.
                If 'None' then all sheets will be loaded.
                If 0 then first sheet in workbook will be loaded (this is the Default).
                If string then this is interpreted as the name of the sheet to load.
                If a list is passed, this can be a list of index values (int) or
                a list of strings with the sheet names to load.
            usecols (list | str, optional):
                A list of columns to load, specified by general column names in Excel,
                e.g. usecols='B:D', usecols=['A', 'C', 'F']
            skip_rows (int, optional):
                List of rows to skip on top of the sheet (e.g. to not read headlines)
            header (int | None, optional):
                Excel Row (0-indexed) to use for the column labels of the parsed data frame.
                If file contains no header row, then you should explicitly pass header=None.
                Default is 0.
            names (list, optional):
                A list of column names to use. Default is None.
            na_values (list, optional):
                A list of values in the Excel that should become the Pandas NA value.

        Returns:
            bool:
                False in case an error occured, True otherwise.

        """

        if xlsx_path is not None and os.path.exists(xlsx_path):
            # Load data from Excel file
            try:
                df = pd.read_excel(
                    io=xlsx_path,
                    sheet_name=sheet_names,
                    usecols=usecols,
                    skiprows=skip_rows,
                    header=header,
                    names=names,
                    na_values=na_values,
                )
                # If multiple sheets from an Excel workbook are loaded,
                # then read_excel() returns a dictionary. The keys are
                # the names of the sheets and the values are the data frames.
                # As this class can only handle one data frame per object,
                # We handle this case by concatenating the different sheets.
                # If you don't want this make sure your Excel workbook has only
                # one sheet or use the "sheet_name" parameter to select the one(s)
                # you want to load.
                if isinstance(df, dict):
                    self.logger.info("Loading multiple Excel sheets from the workbook!")
                    multi_sheet_df = pd.DataFrame()
                    for sheet in df:
                        multi_sheet_df = pd.concat(
                            [multi_sheet_df, df[sheet]],
                            ignore_index=True,
                        )
                    df = multi_sheet_df
                if self._df is None:
                    self._df = df
                else:
                    self._df = pd.concat([self._df, df], ignore_index=True)
            except FileNotFoundError:
                self.logger.error(
                    "Excel file -> '%s' not found. Please check the file path.",
                    xlsx_path,
                )
                return False
            except PermissionError:
                self.logger.error(
                    "Missing permission to access the Excel file -> '%s'.",
                    xlsx_path,
                )
                return False
            except OSError:
                self.logger.error(
                    "An I/O error occurred while reading the Excel file -> '%s'",
                    xlsx_path,
                )
                return False
            except ValueError:
                self.logger.error(
                    "Invalid Excel input in file -> '%s'",
                    xlsx_path,
                )
                return False
            except AttributeError:
                self.logger.error("Unexpected data structure in file -> %s", xlsx_path)
                return False
            except TypeError:
                self.logger.error("Unexpected data type in file -> %s", xlsx_path)
                return False
            except KeyError:
                self.logger.error("Missing key in Excel data in file -> %s", xlsx_path)
                return False

        else:
            self.logger.error(
                "Missing Excel file -> '%s'. You have not specified a valid path!",
                xlsx_path,
            )
            return False

        return True

    # end method definition

    def save_excel_data(
        self,
        excel_path: str,
        sheet_name: str = "Pandas Export",
        index: bool = False,
        columns: list | None = None,
    ) -> bool:
        """Save the data frame to an Excel file, with robust error handling and logging.

        Args:
            excel_path (str):
                The file path to save the Excel file.
            sheet_name (str):
                The sheet name where data will be saved. Default is 'Sheet1'.
            index (bool, optional):
                Whether to write the row names (index). Default is False.
            columns (list | None, optional):
                A list of column names to write into the excel file.

        Returns:
            bool:
                True = success, False = error.

        """

        try:
            # Check if the directory exists
            directory = os.path.dirname(excel_path)
            if directory and not os.path.exists(directory):
                os.makedirs(directory)

            # Validate columns if provided
            if columns:
                existing_columns = [col for col in columns if col in self._df.columns]
                missing_columns = set(columns) - set(existing_columns)
                if missing_columns:
                    self.logger.warning(
                        "The following columns do not exist in the data frame and cannot be saved to Excel -> %s",
                        ", ".join(missing_columns),
                    )
                columns = existing_columns

            # Attempt to save the data frame to Excel:
            if self._df is None:
                self.logger.error(
                    "Cannot write Excel file -> '%s' from empty / non-initialized data frame!", excel_path
                )
            self._df.to_excel(
                excel_path,
                sheet_name=sheet_name,
                index=index,
                columns=columns or None,  # Pass None if no columns provided
            )
            self.logger.info(
                "Data frame saved successfully to Excel file -> '%s'.",
                excel_path,
            )

        except FileNotFoundError as fnf_error:
            self.logger.error("Cannot write data frame to Excel file -> '%s'; error -> %s", excel_path, str(fnf_error))
            return False
        except PermissionError as pe:
            self.logger.error("Cannot write data frame to Excel file -> '%s'; error -> %s", excel_path, str(pe))
            return False
        except ValueError as ve:
            self.logger.error("Cannot write data frame to Excel file -> '%s'; error -> %s", excel_path, str(ve))
            return False
        except OSError as ose:
            self.logger.error("Cannot write data frame to Excel file -> '%s'; error -> %s", excel_path, str(ose))
            return False

        return True

    # end method definition

    def load_csv_data(
        self,
        csv_path: str,
        delimiter: str = ",",
        names: list | None = None,
        header: int | None = 0,
        usecols: list | None = None,
        encoding: str = "utf-8",
    ) -> bool:
        """Load CSV (Comma separated values) data into data frame.

        Args:
            csv_path (str):
                The path to the CSV file.
            delimiter (str, optional, length = 1):
                The character used to delimit values. Default is "," (comma).
            names (list | None, optional):
                The list of column names. This is useful if file does not have a header line
                but just the data.
            header (int | None, optional):
                The index of the header line. Default is 0 (first line). None indicates
                that the file does not have a header line
            usecols (list | None, optional):
                There are three possible list values types:
                1. int:
                    These values are treated as column indices for columns to keep
                    (first column has index 0).
                2. str:
                    The names of the columns to keep. For this to work the file needs
                    either a header line (i.e. 'header != None') or the 'names'
                    parameter must be specified.
                3. bool:
                    The length of the list must match the number of columns. Only
                    columns that have a value of True are kept.
            encoding (str, optional):
                The encoding of the file. Default = "utf-8".

        Returns:
            bool:
                False in case an error occured, True otherwise.

        """

        if csv_path.startswith("http"):
            # Download file from remote location specified by the packageUrl
            # this must be a public place without authentication:
            self.logger.debug("Download CSV file from URL -> '%s'.", csv_path)

            try:
                response = requests.get(url=csv_path, timeout=1200)
                response.raise_for_status()
            except requests.exceptions.HTTPError:
                self.logger.error("HTTP error with -> %s", csv_path)
                return False
            except requests.exceptions.ConnectionError:
                self.logger.error("Connection error with -> %s", csv_path)
                return False
            except requests.exceptions.Timeout:
                self.logger.error("Timeout error with -> %s", csv_path)
                return False
            except requests.exceptions.RequestException:
                self.logger.error("Request error with -> %s", csv_path)
                return False

            self.logger.debug(
                "Successfully downloaded CSV file -> %s; status code -> %s",
                csv_path,
                response.status_code,
            )

            # Convert bytes to a string using utf-8 and create a file-like object
            csv_file = StringIO(response.content.decode(encoding))

        elif os.path.exists(csv_path):
            self.logger.debug("Using local CSV file -> '%s'.", csv_path)
            csv_file = csv_path

        else:
            self.logger.error(
                "Missing CSV file -> '%s' you have not specified a valid path!",
                csv_path,
            )
            return False

        # Load data from CSV file or buffer
        try:
            df = pd.read_csv(
                filepath_or_buffer=csv_file,
                delimiter=delimiter,
                names=names,
                header=header,
                usecols=usecols,
                encoding=encoding,
                skipinitialspace=True,
            )
            if self._df is None:
                self._df = df
            else:
                self._df = pd.concat([self._df, df])
        except FileNotFoundError:
            self.logger.error(
                "CSV file -> '%s' not found. Please check the file path.",
                csv_path,
            )
            return False
        except PermissionError:
            self.logger.error(
                "Permission denied to access the CSV file -> '%s'.",
                csv_path,
            )
            return False
        except OSError:
            self.logger.error("An I/O error occurred!")
            return False
        except ValueError:
            self.logger.error("Invalid CSV input in file -> %s", csv_path)
            return False
        except AttributeError:
            self.logger.error("Unexpected data structure in file -> %s", csv_path)
            return False
        except TypeError:
            self.logger.error("Unexpected data type in file -> %s", csv_path)
            return False
        except KeyError:
            self.logger.error("Missing key in CSV data -> %s", csv_path)
            return False

        return True

    # end method definition

    def load_xml_data(
        self,
        xml_path: str,
        xpath: str | None = None,
        xslt_path: str | None = None,
        encoding: str = "utf-8",
    ) -> bool:
        """Load XML data into a Pandas data frame.

        Args:
            xml_path (str):
                The path to the XML file to load.
            xpath (str, optional):
                An XPath to the elements we want to select.
            xslt_path (str, optional):
                An XSLT transformation file to convert the XML data.
            encoding (str, optional):
                The encoding of the file. Default is UTF-8.

        Returns:
            bool:
                False in case an error occured, True otherwise.

        """

        if xml_path.startswith("http"):
            # Download file from remote location specified by the packageUrl
            # this must be a public place without authentication:
            self.logger.debug("Download XML file from URL -> '%s'.", xml_path)

            try:
                response = requests.get(url=xml_path, timeout=1200)
                response.raise_for_status()
            except requests.exceptions.HTTPError:
                self.logger.error("HTTP error with -> %s", xml_path)
                return False
            except requests.exceptions.ConnectionError:
                self.logger.error("Connection error with -> %s", xml_path)
                return False
            except requests.exceptions.Timeout:
                self.logger.error("Timeout error with -> %s", xml_path)
                return False
            except requests.exceptions.RequestException:
                self.logger.error("Request error with -> %s", xml_path)
                return False

            self.logger.debug(
                "Successfully downloaded XML file -> '%s'; status code -> %s",
                xml_path,
                response.status_code,
            )
            # Convert bytes to a string using utf-8 and create a file-like object
            xml_file = StringIO(response.content.decode(encoding))

        elif os.path.exists(xml_path):
            self.logger.debug("Using local XML file -> '%s'.", xml_path)
            xml_file = xml_path

        else:
            self.logger.error(
                "Missing XML file -> '%s'. You have not specified a valid path or URL!",
                xml_path,
            )
            return False

        # Load data from XML file or buffer
        try:
            df = pd.read_xml(
                path_or_buffer=xml_file,
                xpath=xpath,
                stylesheet=xslt_path,
                encoding=encoding,
            )
            # Process the loaded data as needed
            if self._df is None:
                self._df = df
            else:
                self._df = pd.concat([self._df, df])
            self.logger.info("XML file -> '%s' loaded successfully!", xml_path)
        except FileNotFoundError:
            self.logger.error("XML file -> '%s' not found.", xml_path)
            return False
        except PermissionError:
            self.logger.error(
                "Missing permission to access the XML file -> '%s'.",
                xml_path,
            )
            return False
        except OSError:
            self.logger.error("An I/O error occurred loading from -> %s", xml_path)
            return False
        except ValueError:
            self.logger.error("Invalid XML data in file -> %s", xml_path)
            return False
        except AttributeError:
            self.logger.error("Unexpected data structure in XML file -> %s", xml_path)
            return False
        except TypeError:
            self.logger.error("Unexpected data type in XML file -> %s", xml_path)
            return False
        except KeyError:
            self.logger.error("Missing key in XML file -> %s", xml_path)
            return False

        return True

    # end method definition

    def load_directory(self, path_to_root: str) -> bool:
        """Load directory structure into Pandas data frame.

        Args:
            path_to_root (str):
                Path to the root element of the directory structure.

        Returns:
            bool: True = Success, False = Failure

        """

        try:
            # Check if the provided path is a directory
            if not os.path.isdir(path_to_root):
                self.logger.error(
                    "The provided path -> '%s' is not a valid directory.",
                    path_to_root,
                )
                return False

            # Initialize a list to hold file information
            data = []

            # Walk through the directory
            for root, _, files in os.walk(path_to_root):
                for file in files:
                    file_path = os.path.join(root, file)
                    file_size = os.path.getsize(file_path)
                    relative_path = os.path.relpath(file_path, path_to_root)
                    path_parts = relative_path.split(os.sep)

                    # Create a dictionary with the path parts and file details
                    entry = {"level {}".format(i): part for i, part in enumerate(path_parts[:-1], start=1)}

                    entry.update(
                        {
                            "filename": path_parts[-1],
                            "size": file_size,
                            "path": path_parts[1:-1],
                            "relative_path": relative_path,
                            "download_dir": root,
                        },
                    )
                    data.append(entry)

            # Create data frame from list of dictionaries:
            self._df = pd.DataFrame(data)

            # Determine the maximum number of levels
            max_levels = max((len(entry) - 2 for entry in data), default=0)

            # Ensure all entries have the same number of levels:
            for entry in data:
                for i in range(1, max_levels + 1):
                    entry.setdefault("level {}".format(i), "")

            # Convert to data frame again to make sure all columns are consistent:
            self._df = pd.DataFrame(data)

        except NotADirectoryError:
            self.logger.error(
                "Provided path -> '%s' is not a directory!",
                path_to_root,
            )
            return False
        except FileNotFoundError:
            self.logger.error(
                "Provided path -> '%s' does not exist in file system!",
                path_to_root,
            )
            return False
        except PermissionError:
            self.logger.error(
                "Permission error accessing path -> '%s'!",
                path_to_root,
            )
            return False

        return True

    # end method definition

    def load_xml_directory(
        self,
        path_to_root: str,
        xpath: str | None = None,
        xml_files: list | None = None,
    ) -> bool:
        """Load XML files from a directory structure into Pandas data frame.

        Args:
            path_to_root (str):
                Path to the root element of the directory structure.
            xpath (str, optional):
                XPath to the XML elements we want to select.
            xml_files (list | None, optional):
                Names of the XML files to load from the directory.

        Returns:
            bool:
                True = Success, False = Failure

        """

        # Establish a default if None is passed via the parameter:
        if not xml_files:
            xml_files = ["docovw.xml"]

        try:
            # Check if the provided path is a directory
            if not os.path.isdir(path_to_root):
                self.logger.error(
                    "The provided path -> '%s' is not a valid directory.",
                    path_to_root,
                )
                return False

            # Walk through the directory
            for root, _, files in os.walk(path_to_root):
                for file in files:
                    file_path = os.path.join(root, file)
                    file_size = os.path.getsize(file_path)
                    file_name = os.path.basename(file_path)

                    if file_name in xml_files:
                        self.logger.info(
                            "Load XML file -> '%s' of size -> %s from -> '%s'...",
                            file_name,
                            file_size,
                            file_path,
                        )
                        success = self.load_xml_data(file_path, xpath=xpath)
                        if success:
                            self.logger.info(
                                "Successfully loaded XML file -> '%s'.",
                                file_path,
                            )

        except NotADirectoryError:
            self.logger.error(
                "Provided path -> '%s' is not a directory",
                path_to_root,
            )
            return False
        except FileNotFoundError:
            self.logger.error(
                "Provided path -> '%s' does not exist in file system!",
                path_to_root,
            )
            return False
        except PermissionError:
            self.logger.error(
                "Missing permission to access path -> '%s'",
                path_to_root,
            )
            return False

        return True

    # end method definition

    def load_web_links(
        self,
        url: str,
        common_data: dict | None = None,
        pattern: str = r"",
    ) -> list | None:
        """Get all linked file URLs on a given web page (url) that are following a given pattern.

        Construct a list of dictionaries based on this. This method is a helper method for load_web() below.

        Args:
            url (str):
                The web page URL.
            common_data (dict | None, optional):
                Fields that should be added to each dictionary item. Defaults to None.
            pattern (str, optional):
                Regular Expression. Defaults to r"".

        Returns:
            list | None:
                List of links on the web page that are complying with the given regular expression.

        """

        try:
            response = requests.get(url, timeout=300)
            response.raise_for_status()
        except requests.RequestException:
            self.logger.error("Failed to retrieve page at %s", url)
            return []

        # Find all file links (hyperlinks) on the page (no file extension assumed)
        # Example filename pattern: "al022023.public.005"
        file_links = re.findall(r'href="([^"]+)"', response.text)
        if not file_links:
            self.logger.warning("No file links found on the web page -> %s", url)
            return []

        result_list = []
        base_url = url if url.endswith("/") else url + "/"

        for link in file_links:
            data = common_data.copy() if common_data else {}

            # Construct the full URL
            full_url = base_url + link.lstrip("/")

            if pattern:
                # Filter by expected naming pattern for links
                match = re.search(pattern, link)
                if not match:
                    continue

                # Extract and assign groups if they exist
                # TODO(mdiefenb): these names are currently hard-coded
                # for the National Hurricane Center Dataset (NHC)
                if len(match.groups()) >= 1:
                    data["Code"] = match.group(1).upper()
                if len(match.groups()) >= 2:
                    data["Type"] = match.group(2)
                if len(match.groups()) >= 3:
                    data["Message ID"] = match.group(3)

            data["URL"] = full_url
            data["Filename"] = link

            result_list.append(data)

        return result_list

    # end method definition

    def load_web(
        self,
        values: list,
        value_name: str,
        url_templates: list,
        special_values: list | None = None,
        special_url_templates: dict | None = None,
        pattern: str = r"",
    ) -> bool:
        """Traverse years and bulletin types to collect all bulletin URLs.

        Args:
            values (list):
                List of values to travers over
            value_name (str):
                Dictionary key to construct an item in combination with a value from values
            url_templates (list):
                URLs to travers per value. The URLs should contain one {} that is
                replace by the current value.
            special_values (list | None, optional):
                List of vales (a subset of the other values list)
                that we want to handle in a special way. Defaults to None.
            special_url_templates (dict | None, optional):
                URLs for the special values. Defaults to None.
                The dictionary keys are the special values. The
                dictionary values are lists of special URLs with placeholders.
            pattern (str, optional):
                Regular expression to find the proper links on the page. Defaults to r"".

        Returns:
            bool:
                True for success, False in case of an error.

        """

        result_list = []

        # We have two nested for loops below. The out traverses over all placeholder values.
        # These could be the calendar years, e.g. [2003,...,2024]
        # The inner for loop traverses over the list of specified URLs. We can have multiple for
        # each value.

        # Do we have a list of placeholder values we want to iterate over?
        if values:
            # Traverse all values in the values list:
            for value in values:
                # Do we want a special treatment for this value (e.g. the current year)
                if value in special_values:
                    self.logger.debug("Processing special value -> '%s'...", value)
                    if value not in special_url_templates and str(value) not in special_url_templates:
                        self.logger.error(
                            "Cannot find key -> '%s' in special URL templates dictionary -> %s! Skipping...",
                            value,
                            str(special_url_templates),
                        )
                        continue
                    # If the dictionary uses string keys then we need to convert the value
                    # to a string as well to avoid key errors:
                    if str(value) in special_url_templates:
                        value = str(value)
                    special_url_template_list = special_url_templates[value]
                    for special_url_template in special_url_template_list:
                        # Now the value is inserted into the placeholder in the URL:
                        special_url = special_url_template.format(value)
                        common_data = {value_name: value} if value_name else None
                        result_list += self.load_web_links(
                            url=special_url,
                            common_data=common_data,
                            pattern=pattern,
                        )
                else:  # normal URLs
                    self.logger.info("Processing value -> '%s'...", value)
                    for url_template in url_templates:
                        # Now the value is inserted into the placeholder in the URL:
                        url = url_template.format(value)
                        common_data = {value_name: value} if value_name else None
                        result_list += self.load_web_links(
                            url=url,
                            common_data=common_data,
                            pattern=pattern,
                        )
        else:
            for url_template in url_templates:
                url = url_template.format(value)
                result_list += self.load_web_links(
                    url=url,
                    common_data=None,
                    pattern=pattern,
                )

        # Add the data list to the data frame:
        self.append(result_list)

        return True

    # end method definition

    def partitionate(self, number: int) -> list:
        """Partition a data frame into equally sized partitions.

        Args:
            number (int):
                The number of desired partitions.

        Returns:
            list:
                A list of created partitions.

        """

        # Calculate the approximate size of each partition
        size = len(self._df)

        if size >= number:
            partition_size = size // number
            remainder = size % number
        else:
            partition_size = size
            number = 1
            remainder = 0

        self.logger.info(
            "Data frame has -> %s elements. We split it into -> %s partitions with -> %s row%s and remainder -> %s...",
            str(size),
            str(number),
            str(partition_size),
            "s" if partition_size > 1 else "",
            str(remainder),
        )

        # Initialize a list to store partitions:
        partitions = []
        start_index = 0

        # Slice the data frame into equally sized partitions:
        for i in range(number):
            # Calculate the end index for this partition
            end_index = start_index + partition_size + (1 if i < remainder else 0)
            partition = self._df.iloc[start_index:end_index]
            partitions.append(partition)
            start_index = end_index

        return partitions

    # end method definition

    def partitionate_by_column(self, column_name: str) -> list | None:
        """Partition a data frame based on equal values in a specified column.

        Args:
            column_name (str):
                The column name to partition by.

        Returns:
            list | None:
                List of partitions or None in case of an error (e.g. column name does not exist).

        """

        if column_name not in self._df.columns:
            self.logger.error(
                "Cannot partitionate by column -> '%s'. Column does not exist in the data frame. Data frame has these columns -> %s",
                column_name,
                str(self._df.columns),
            )
            return None

        # Separate rows with NaN or None values in the specified column:
        nan_partitions = self._df[self._df[column_name].isna()]

        # Keep only rows where the specified column has valid (non-NaN) values:
        non_nan_df = self._df.dropna(subset=[column_name])

        # Group the non-NaN DataFrame by the specified column's values:
        grouped = non_nan_df.groupby(column_name)

        # Create a list of partitions (DataFrames) for each unique value in the column:
        partitions = [group for _, group in grouped]

        # Add each row with NaN/None as its own partition
        # iterrows() returns each row as a Series. To convert it back to a DataFrame:
        # 1. .to_frame() turns the Series into a DataFrame, but with the original column names as rows.
        # 2. .T (transpose) flips it back, turning the original row into a proper DataFrame row.
        # This ensures that even rows with NaN values are treated as DataFrame partitions.
        partitions.extend([row.to_frame().T for _, row in nan_partitions.iterrows()])

        self.logger.info(
            "Data frame has been partitioned into -> %s partitions based on the values in column -> '%s'...",
            str(len(partitions)),
            column_name,
        )

        return partitions

    # end method definition

    def deduplicate(self, unique_fields: list, inplace: bool = True) -> pd.DataFrame:
        """Remove dupclicate rows that have all fields in unique_fields in common.

        Args:
            unique_fields (list):
                Defines the fields for which we want a unique combination for.
            inplace (bool, optional):
                True if the deduplication happens in-place. Defaults to True.

        Returns:
            pd.DataFrame:
                If inplace is False than a new deduplicatd data frame is returned.
                Otherwise the object is modified in place and self._df is returned.

        """

        if inplace:
            self._df.drop_duplicates(subset=unique_fields, inplace=True)
            self._df.reset_index(drop=True, inplace=True)
            return self._df
        else:
            df = self._df.drop_duplicates(subset=unique_fields, inplace=False)
            df = df.reset_index(drop=True, inplace=False)
            return df

    # end method definition

    def sort(self, sort_fields: list, inplace: bool = True) -> pd.DataFrame | None:
        """Sort the data frame based on one or multiple fields.

        Sorting can be either in place or return it as a new data frame
        (e.g. not modifying self._df).

        Args:
            sort_fields (list):
                The columns / fields to be used for sorting.
            inplace (bool, optional):
                If the sorting should be inplace, i.e. modifying self._df.
                Defaults to True.

        Returns:
            pd.DataFrame | None:
                New data frame (if inplace = False) or self._df (if inplace = True).
                None in case of an error.

        """

        if self._df is None:
            return None

        if not all(sort_field in self._df.columns for sort_field in sort_fields):
            self.logger.warning(
                "Not all of the given sort fields -> %s do exist in the data frame.",
                str(sort_fields),
            )
            # Reduce the sort fields to those that really exist in the data frame:
            sort_fields = [sort_field for sort_field in sort_fields if sort_field in self._df.columns]
            self.logger.warning(
                "Only these given sort fields -> %s do exist as columns in the data frame.",
                str(sort_fields),
            )

        if inplace:
            self._df.sort_values(by=sort_fields, inplace=True)
            self._df.reset_index(drop=True, inplace=True)
            return self._df
        else:
            df = self._df.sort_values(by=sort_fields, inplace=False)
            df = df.reset_index(drop=True, inplace=False)
            return df

    # end method definition

    def flatten(self, parent_field: str, flatten_fields: list, concatenator: str = "_") -> None:
        """Flatten a sub-dictionary by copying selected fields to the parent dictionary.

        This is e.g. useful for then de-duplicate a data frame.
        To flatten a data frame makes sense in situation when a column used
        to have a list of dictionaries and got "exploded" (see explode_and_flatten()
        method below). In this case the column as dictionary values that then can
        be flattened.

        Args:
            parent_field (str):
                Name prefix of the new column in the data frame. The flattened field
                names are added with a leading underscore.
            flatten_fields (list):
                Fields in the dictionary of the source column that are copied
                as new columns into the data frame.
            concatenator (str, optional):
                Character or string used to concatenate the parent field with the flattened field
                to create a unique name.

        """

        # First do a sanity check if the data frame is not yet initialized.
        if self._df is None:
            self.logger.error(
                "The data frame is not initialized or empty. Cannot flatten field(s) -> '%s' in the data frame.",
                flatten_fields,
            )
            return

        if parent_field not in self._df.columns:
            self.logger.warning(
                "The parent field -> '%s' cannot be flattened as it doesn't exist as column in the data frame!",
                parent_field,
            )
            return

        for flatten_field in flatten_fields:
            flat_field = parent_field + concatenator + flatten_field
            # The following expression generates a new column in the
            # data frame with the name of 'flat_field'.
            # In the lambda function x is a dictionary that includes the subvalues
            # and it returns the value of the given flatten field
            # (if it exists, otherwise None). So x is self._df[parent_field], i.e.
            # what the lambda function gets 'applied' on.
            self._df[flat_field] = self._df[parent_field].apply(
                lambda x, sub_field=flatten_field: (x.get(sub_field, None) if isinstance(x, dict) else None),
            )

    # end method definition

    def explode_and_flatten(
        self,
        explode_fields: str | list,
        flatten_fields: list | None = None,
        make_unique: bool = False,
        reset_index: bool = False,
        split_string_to_list: bool = False,
        separator: str = ";,",
    ) -> pd.DataFrame | None:
        """Explode a substructure in the Pandas data frame.

        Args:
            explode_fields (str | list):
                Field(s) to explode. Each field to explode should have a list structure.
                Exploding multiple columns at once is possible. This delivers
                a very different result compared to exploding one column after the other!
            flatten_fields (list):
                Fields in the exploded substructure to include
                in the main dictionaries for easier processing.
            make_unique (bool, optional):
                If True, deduplicate the exploded data frame.
            reset_index (bool, False):
                If True, then the index is reset, False = Index is not reset.
            split_string_to_list (bool, optional):
                If True flatten the exploded data frame.
            separator (str, optional):
                Characters used to split the string values in the given column into a list.

        Returns:
            pd.DataFrame | None:
                Pointer to the Pandas data frame.

        """

        def update_column(row: pd.Series, sub: str) -> str:
            """Extract the value of a sub-column from a nested dictionary within a Pandas Series.

            Args:
                row (pd.Series):
                    A row from the data frame.
                sub (str):
                    The sub-column name to extract.

            Returns:
                str:
                    The value of the sub-column, or an empty string if not found.

            """

            if isinstance(row, dict) and sub in row:
                return row[sub]
            return ""

        # end def update_column()

        def string_to_list(value: str) -> list:
            """Convert a string to a list by splitting it using a specified separator.

            If the input is already a list, it is returned as-is. If the input is `None` or a missing value,
            an empty list is returned. Otherwise, the string is split into a list of substrings using
            the given separator. Leading and trailing spaces in the resulting substrings are removed.

            Args:
                value (str):
                    The input string to be converted into a list. Can also be a list, `None`,
                    or a missing value (e.g., NaN).

            Returns:
                list:
                    A list of substrings if the input is a string, or an empty list if the input
                    is `None` or a missing value. If the input is already a list, it is returned unchanged.

            """

            # Check if the value is already a list; if so, return it directly
            if isinstance(value, list):
                return value

            # If the value is None or a missing value (e.g., NaN), return an empty list
            if not value or pd.isna(value):
                return []

            # Use a regular expression to split the string by the separator
            # and remove leading/trailing spaces from each resulting substring
            return_list = re.split(rf"[{separator}]\s*", str(value))

            return return_list

        # end def string_to_list()

        #
        # Start of main method:
        #

        # First do a sanity check if the data frame is not yet initialized.
        if self._df is None:
            self.logger.error(
                "The data frame is not initialized or empty. Cannot explode data frame.",
            )
            return None

        # Next do a sanity check for the given explode_field. It should
        # either be a string (single column name) or a list (multiple column names):
        if isinstance(explode_fields, list):
            self.logger.info("Exploding list of columns -> %s", str(explode_fields))
        elif isinstance(explode_fields, str):
            self.logger.info("Exploding single column -> '%s'", explode_fields)
        else:
            self.logger.error(
                "Illegal explode field(s) data type -> %s. Explode field must either be a string or a list of strings.",
                type(explode_fields),
            )
            return self._df

        # Ensure explode_fields is a list for uniform processing:
        if isinstance(explode_fields, str):
            explode_fields = [explode_fields]

        # Process nested field names with '.'
        processed_fields = []
        for field in explode_fields:
            # The "." indicates that the column has dictionary values:
            if "." in field:
                main, sub = field.split(".", 1)
                if main not in self._df.columns:
                    self.logger.error(
                        "The column -> '%s' does not exist in the data frame! Cannot explode it. Data frame has these columns -> %s",
                        main,
                        str(self._df.columns.tolist()),
                    )
                    continue

                # Use update_column to extract the dictionary key specified by the sub value:
                self.logger.info(
                    "Extracting dictionary value for key -> '%s' from column -> '%s'.",
                    sub,
                    main,
                )
                self._df[main] = self._df[main].apply(update_column, args=(sub,))
                processed_fields.append(main)
            else:
                processed_fields.append(field)

        # Verify all processed fields exist in the data frame:
        missing_columns = [col for col in processed_fields if col not in self._df.columns]
        if missing_columns:
            self.logger.error(
                "The following columns are missing in the data frame and cannot be exploded -> %s. Data frame has these columns -> %s",
                missing_columns,
                str(self._df.columns.tolist()),
            )
            return self._df

        # Handle splitting strings into lists if required:
        if split_string_to_list:
            for field in processed_fields:
                self.logger.info(
                    "Splitting strings in column -> '%s' into lists using separator -> '%s'",
                    field,
                    separator,
                )
                # Apply the function to convert the string values in the column (give by the name in explode_field) to lists
                # The string_to_list() sub-method above also considers the separator parameter.
                self._df[field] = self._df[field].apply(string_to_list)

        # Explode all specified columns at once.
        # explode() can either take a string field or a list of fields.
        # # It is VERY important to do the explosion of multiple columns together -
        # otherwise we get combinatorial explosion. Explosion of multiple columns 1-by-1
        # is VERY different from doing the explosion together!
        self.logger.info("Validated column(s) to explode -> %s", processed_fields)
        try:
            self._df = self._df.explode(
                column=processed_fields,
                ignore_index=reset_index,
            )
        except ValueError:
            self.logger.error(
                "Error exploding columns -> %s",
                processed_fields,
            )
            return self._df

        if flatten_fields:
            # Ensure that flatten() is called for each exploded column
            for field in processed_fields:
                self.flatten(parent_field=field, flatten_fields=flatten_fields)

            # Deduplicate rows if required
            if make_unique:
                self._df.drop_duplicates(subset=flatten_fields, inplace=True)

        # Reset index explicitly if not handled during explode
        if reset_index:
            self._df.reset_index(drop=True, inplace=True)

        return self._df

    # end method definition

    def drop_columns(self, column_names: list, inplace: bool = True) -> pd.DataFrame:
        """Drop selected columns from the Pandas data frame.

        Args:
            column_names (list):
                The list of column names to drop.
            inplace (bool, optional):
                Whether or not the dropping should be inplace, i.e. modifying self._df.
                Defaults to True.

        Returns:
            pd.DataFrame:
                New data frame (if inplace = False) or self._df (if inplace = True)

        """

        if not all(column_name in self._df.columns for column_name in column_names):
            # Reduce the column names to those that really exist in the data frame:
            column_names = [column_name for column_name in column_names if column_name in self._df.columns]
            self.logger.info(
                "Drop columns -> %s from the data frame.",
                str(column_names),
            )

        if inplace:
            self._df.drop(column_names, axis=1, inplace=True)
            return self._df
        else:
            df = self._df.drop(column_names, axis=1, inplace=False)
            return df

    # end method definition

    def keep_columns(self, column_names: list, inplace: bool = True) -> pd.DataFrame:
        """Keep only selected columns in the data frame. Drop the rest.

        Args:
            column_names (list):
                A list of column names to keep.
            inplace (bool, optional):
                If the keeping should be inplace, i.e. modifying self._df.
                Defaults to True.

        Returns:
            pd.DataFrame:
                New data frame (if inplace = False) or self._df (if inplace = True).

        """

        if not all(column_name in self._df.columns for column_name in column_names):
            # Reduce the column names to those that really exist in the data frame:
            column_names = [column_name for column_name in column_names if column_name in self._df.columns]
            self.logger.info(
                "Reduce columns to keep to these columns -> %s that do exist in the data frame.",
                column_names,
            )

        if inplace:
            # keep only those columns which are in column_names:
            if column_names != []:
                self._df = self._df[column_names]
            return self._df
        else:
            # keep only those columns which are in column_names:
            if column_names != []:
                df = self._df[column_names]
                return df
            return None

    # end method definition

    def rename_column(self, old_column_name: str, new_column_name: str) -> bool:
        """Rename a data frame column.

        Args:
            old_column_name (str):
                The old name of the column.
            new_column_name (str):
                The new name of the column.

        Returns:
            bool:
                True = Success, False = Error

        """

        if self._df is None:
            return False

        if old_column_name not in self._df.columns:
            self.logger.error(
                "Cannot rename column -> '%s'. It does not exist in the data frame! Data frame has these columns -> %s",
                old_column_name,
                str(self._df.columns),
            )
            return False

        if new_column_name in self._df.columns:
            self.logger.error(
                "Cannot rename column -> '%s' to -> '%s'. New name does already exist as column in the data frame! Data frame has these columns -> %s",
                old_column_name,
                new_column_name,
                str(self._df.columns),
            )
            return False

        self._df.rename(columns={old_column_name: new_column_name}, inplace=True)

        return True

    # end method definition

    def is_dict_column(self, column: pd.Series, threshold: float = 0.5) -> bool:
        """Safely checks if a column predominantly contains dictionary-like objects.

        Args:
            column (pd.Series):
                The pandas Series (column) to check.
            threshold (float, optional):
                0.0 < threshold <= 1.0. Float representation of the percentage.
                Default = 0.5 (50%).

        Returns:
            bool:
                True if the column contains mostly dictionary-like objects, False otherwise.

        """

        if not isinstance(column, pd.Series):
            self.logger.error(
                "Expected Pandas series, but got -> %s",
                str(type(column)),
            )
            return False
        if not 0.0 < threshold <= 1.0:
            self.logger.error(
                "Threshold must be between 0.0 and 1.0, but got -> %s",
                str(threshold),
            )
            return False

        # Drop null values (NaN or None) and check types of remaining values
        non_null_values = column.dropna()
        dict_count = non_null_values.apply(lambda x: isinstance(x, dict)).sum()

        # If more than threshold % of non-null values are dictionaries, return True.
        # Else return False.
        return dict_count / len(non_null_values) > threshold if len(non_null_values) > 0 else False

    # end method definition

    def is_list_column(self, column: pd.Series, threshold: float = 0.5) -> bool:
        """Safely checks if a column predominantly contains list-like objects.

        Args:
            column (pd.Series):
                The pandas Series (column) to check.
            threshold (float, optional):
                0.0 < threshold <= 1.0. Float representation of the percentage. Default = 0.5 (50%).

        Returns:
            bool:
                True if the column contains list-like objects, False otherwise.

        """

        if not isinstance(column, pd.Series):
            self.logger.error(
                "Expected pandas series, but got -> %s",
                str(type(column)),
            )
            return False
        if not 0.0 < threshold <= 1.0:
            self.logger.error(
                "Threshold must be between 0.0 and 1.0, but got -> %s",
                str(threshold),
            )
            return False

        # Drop null values (NaN or None) and check types of remaining values
        non_null_values = column.dropna()
        list_count = non_null_values.apply(lambda x: isinstance(x, list)).sum()

        # If more than threshold % of non-null values are lists, return True.
        # Else return False.
        return list_count / len(non_null_values) > threshold if len(non_null_values) > 0 else False

    # end method definition

    def is_string_column(self, column: pd.Series) -> bool:
        """Determine if a Pandas series predominantly contains string values, ignoring NaN values.

        Args:
            column (pd.Series):
                The Pandas Series to check.

        Returns:
            bool:
                True if all non-NaN values in the column are strings, False otherwise.

        """

        # Drop NaN values and check if remaining values are strings
        return column.dropna().map(lambda x: isinstance(x, str)).all()

    # end method definition

    def cleanse(self, cleansings: dict) -> None:
        """Cleanse data with regular expressions and upper/lower case conversions.

        Args:
            cleansings (dict):
                Dictionary with keys that equal the column names.
                The dictionary values are dictionaries themselves with
                these fields:
                * replacements (dict): name of a column in the data frame
                * upper (bool, optional, default = False): change the value to uppercase
                * lower (bool, optional, default = False): change the value to lowercase
                * capitalize (bool, optional, default = False) - first character upper case, rest lower-case
                * title (bool, optional, default = False) - first character of each word upper case
                * length (int, optional, default = 0): truncate to max length

        """

        # Iterate over each column in the cleansing dictionary
        for column, cleansing in cleansings.items():
            # Read the cleansing parameters:
            replacements = cleansing.get("replacements", {})
            upper = cleansing.get("upper", False)
            lower = cleansing.get("lower", False)
            capitalize = cleansing.get("capitalize", False)
            title = cleansing.get("title", False)
            length = cleansing.get("length", 0)

            # Handle dict columns - we expect the column name to seperate
            # main field from sub field using a dot syntax (e.g., "column.subfield")
            if "." in column:
                column, dict_key = column.split(".")
                if column not in self._df.columns:
                    self.logger.error(
                        "Cannot cleanse column -> '%s'. It does not exist in the data frame! Data frame has these columns -> %s",
                        column,
                        str(self._df.columns),
                    )
                    continue
                # Apply cleansing to dictionary values in the main column
                self.logger.debug(
                    "Cleansing for column -> '%s' has a subfield -> '%s' configured. Do cleansing for dictionary items with key -> '%s'...",
                    column,
                    dict_key,
                    dict_key,
                )
                self._df[column] = self._df[column].apply(
                    lambda x,
                    dict_key=dict_key,
                    replacements=replacements,
                    upper=upper,
                    lower=lower,
                    capitalize=capitalize,
                    title=title,
                    length=length: self._cleanse_subfield(
                        data=x,
                        dict_key=dict_key,
                        replacements=replacements,
                        upper=upper,
                        lower=lower,
                        capitalize=capitalize,
                        title=title,
                        length=length,
                    ),
                )
            # end if "." in column
            else:  # the else case handles strings and list columns
                if column not in self._df.columns:
                    self.logger.error(
                        "Cannot cleanse column -> '%s'. It does not exist in the data frame! Data frame has these columns -> %s",
                        column,
                        str(self._df.columns),
                    )
                    continue

                # Handle string columns:
                if self.is_string_column(self._df[column]):
                    # Apply cleansing operations on string column
                    self.logger.debug(
                        "Column -> '%s' has string values. Do cleansing for string values...",
                        column,
                    )
                    self._df[column] = self._df[column].apply(
                        lambda x,
                        replacements=replacements,
                        upper=upper,
                        lower=lower,
                        capitalize=capitalize,
                        title=title,
                        length=length: (
                            self._apply_string_cleansing(
                                value=x,
                                replacements=replacements,
                                upper=upper,
                                lower=lower,
                                capitalize=capitalize,
                                title=title,
                                length=length,
                            )
                            if isinstance(x, str)
                            else x
                        ),
                    )

                # Handle list columns:
                elif self.is_list_column(self._df[column]):
                    # Handle list-like columns for this we iterate over each list item
                    # and apply the cleansing by calling _apply_string_cleansing() for item:
                    self.logger.debug(
                        "Column -> '%s' has list values. Do cleansing for each list item...",
                        column,
                    )
                    self._df[column] = self._df[column].apply(
                        lambda x,
                        replacements=replacements,
                        upper=upper,
                        lower=lower,
                        capitalize=capitalize,
                        title=title,
                        length=length: (
                            [
                                (
                                    self._apply_string_cleansing(
                                        value=item,
                                        replacements=replacements,
                                        upper=upper,
                                        lower=lower,
                                        capitalize=capitalize,
                                        title=title,
                                        length=length,
                                    )
                                    if isinstance(
                                        item,
                                        str,
                                    )  # we just change string list items
                                    else item
                                )
                                for item in x
                            ]
                            if isinstance(x, list)
                            else x
                        ),
                    )

                else:
                    self.logger.error(
                        "Column -> '%s' is not a string, list, or dict-like column. Skipping cleansing...",
                        column,
                    )
            # end else handling strings and lists
        # for column, cleansing in cleansings.items()

    # end method definition

    def _cleanse_dictionary(
        self,
        data: dict,
        dict_key: str,
        replacements: dict[str, str],
        upper: bool,
        lower: bool,
        capitalize: bool = False,
        title: bool = False,
        length: int = 0,
    ) -> dict:
        """Cleanse dictionary data within a single column value that has a given key.

        Args:
            data (dict):
                The column dictionary value.
            dict_key (str):
                The dictionary key whose value should be cleansed in the row to cleanse.
            replacements (dict):
                Dictionary of regex replacements to apply to the subfield value.
            upper (bool):
                If True, convert value in subfield to upper-case.
            lower (bool):
                If True, convert value in subfield to lower-case.
            capitalize (bool, optional):
                If True, capitalize the first letter of the subfield value.
            title (bool, optional):
                If True, title-case the subfield value.
            length (int, optional):
                The maximum length for the subfield value.

        Returns:
            dict:
                The updated data with the cleansing applied to the dictionary item with the given key.

        """

        if pd.isna(data):
            return data

        if dict_key not in data:
            self.logger.warning(
                "The dictionary key -> '%s' (field) is not in the data frame row! Cleansing skipped!",
                dict_key,
            )
            return data

        # 1. Read the value to be cleansed from the data dict:
        value = data[dict_key]

        # 2. Apply string operations based on the type of the value (str, list, or dict)

        if isinstance(value, str):
            # If the value is a string, apply the string operations directly
            value: str = self._apply_string_cleansing(
                value=value,
                replacements=replacements,
                upper=upper,
                lower=lower,
                capitalize=capitalize,
                title=title,
                length=length,
            )
        elif isinstance(value, list):
            # If the value is a list, apply string operations to each element
            value: list = [
                (
                    self._apply_string_cleansing(
                        value=item,
                        replacements=replacements,
                        upper=upper,
                        lower=lower,
                        capitalize=capitalize,
                        title=title,
                        length=length,
                    )
                    if isinstance(item, str)
                    else item
                )
                for item in value
            ]
        elif isinstance(value, dict):
            # If the value is a dictionary, apply string operations to each value
            value: dict = {
                k: (
                    self._apply_string_cleansing(
                        value=v,
                        replacements=replacements,
                        upper=upper,
                        lower=lower,
                        capitalize=capitalize,
                        title=title,
                        length=length,
                    )
                    if isinstance(v, str)
                    else v
                )
                for k, v in value.items()
            }

        # 3. Write back the cleansed value to the data dict:
        data[dict_key] = value

        return data

    # end method definition

    def _cleanse_subfield(
        self,
        data: dict | list,
        dict_key: str,
        replacements: dict[str, str],
        upper: bool,
        lower: bool,
        capitalize: bool = False,
        title: bool = False,
        length: int = 0,
    ) -> dict | list:
        """Cleanse subfield data within a single column value.

        This is NOT a pd.Series but either a dictionary or a list of dictionaries.

        Args:
            data (dict | list):
                The column value. Can be a dictionary or a list of dictionaries
            dict_key (str):
                The dictionary key whose value should be cleansed in the data to cleanse.
            replacements (dict):
                Dictionary of regex replacements to apply to the subfield value.
            upper (bool):
                If True, convert value in subfield to upper-case.
            lower (bool):
                If True, convert value in subfield to lower-case.
            capitalize (bool, optional):
                If True, capitalize the first letter of the subfield value.
            title (bool, optional):
                If True, title-case the subfield value.
            length (int, optional):
                The maximum length for the subfield value.

        Returns:
            dict | list:
                The updated data with the cleansing applied to the subfield.

        """

        if isinstance(data, list):
            data = [
                (
                    self._cleanse_dictionary(
                        data=item,
                        dict_key=dict_key,
                        replacements=replacements,
                        upper=upper,
                        lower=lower,
                        capitalize=capitalize,
                        title=title,
                        length=length,
                    )
                    if item is not None and dict_key in item and not pd.isna(item[dict_key])
                    else item
                )
                for item in data
            ]
        elif isinstance(data, dict):
            data = self._cleanse_dictionary(
                data=data,
                dict_key=dict_key,
                replacements=replacements,
                upper=upper,
                lower=lower,
                capitalize=capitalize,
                title=title,
                length=length,
            )

        return data

    # end method definition

    def _apply_string_cleansing(
        self,
        value: str,
        replacements: dict[str, str],
        upper: bool,
        lower: bool,
        capitalize: bool,
        title: bool,
        length: int,
    ) -> str | None:
        """Apply string operations (upper, lower, capitalize, title-case, replacements) to a string.

        Args:
            value (str):
                The string value to which the operations will be applied.
            replacements (dict[str, str]):
                A dictionary of regular expression patterns (keys) and replacement strings (values) to apply to the string.
            upper (bool):
                If True, convert the string to uppercase.
            lower (bool):
                If True, convert the string to lowercase.
            capitalize (bool):
                If True, capitalize the first letter of the string and lowercase the rest. Default is False.
            title (bool):
                If True, convert the string to title-case (first letter of each word is capitalized). Default is False.
            length (int):
                If greater than 0, truncate the string to this length. Default is 0 (no truncation).

        Returns:
            str | None:
                The updated string with all the applied operations. None in case an error occured.

        Example:
            value = "hello world"
            replacements = {r"world": "there"}
            upper = True
            length = 5

            result = _apply_string_cleansing(value, replacements, upper, length=length)
            # result would be "HELLO"

        """

        if not isinstance(
            value,
            str,
        ):  # Only apply string operations if the value is a string
            return None

        if upper:
            value = value.upper()
        if lower:
            value = value.lower()
        if capitalize:
            value = value.capitalize()
        if title:
            value = value.title()

        # Handle regex replacements
        for regex_pattern, replacement in replacements.items():
            if regex_pattern:
                # Check if the pattern does NOT contain any regex special characters
                # (excluding dot and ampersand) and ONLY then use \b ... \b
                # Special regexp characters include: ^ $ * + ? ( ) | [ ] { } \
                if not re.search(r"[\\^$*+?()|[\]{}]", regex_pattern):
                    # Wrap with word boundaries for whole-word matching
                    # \b is a word boundary anchor in regular expressions.
                    # It matches a position where one side is a word character
                    # (like a letter or digit) and the other side is a non-word character
                    # (like whitespace or punctuation). It's used to match whole words.
                    # We want to have this to e.g. not replace "INT" with "INTERNATIONAL"
                    # if the word is already "INTERNATIONAL". It is important
                    # that the \b ... \b enclosure is ONLY used if regex_pattern is NOT
                    # a regular expression but just a normal string.
                    # TODO: we may reconsider if re.escape() is required or not:
                    regex_pattern = re.escape(regex_pattern)
                    regex_pattern = rf"\b{regex_pattern}\b"
                try:
                    value = re.sub(regex_pattern, replacement, value)
                except re.error:
                    self.logger.error(
                        "Invalid regex pattern -> '%s' in replacement processing!",
                        regex_pattern,
                    )
                    continue

        # Truncate to the specified length, starting from index 0
        if 0 < length < len(value):
            value = value[:length]

        return value

    # end method definition

    def filter(
        self,
        conditions: list,
        inplace: bool = True,
        reset_index: bool = True,
    ) -> pd.DataFrame | None:
        """Filter the data frame based on (multiple) conditions.

        Args:
            conditions (list):
                Conditions are a list of dictionaries with 3 items:
                * field (str): The name of a column in the data frame
                * value (str or list):
                    Expected value (filter criterium).
                    If it is a list then one of the list elements must match the field value (OR)
                * equal (bool):
                    Whether to test for equal or non-equal. If not specified equal is treated as True.
                * regex (bool):
                    This flag controls if the value is interpreted as a
                    regular expression. If there is no regex item in the
                    dictionary then the default is False (= values is NOT regex).
                * enabled (bool):
                    True or False. The filter is only applied if 'enabled = True'
                If there are multiple conditions in the list each has to evaluate to True (AND)
            inplace (bool, optional):
                Defines if the self._df is modified (inplace) or just
                a new data frame is returned. Defaults to True.
            reset_index (bool, optional):
                Filter removes rows. If filter_index = True then the numbering
                of the index is newly calculated

        Returns:
            pd.DataFrame | None:
                A new data frame or pointer to self._df (depending on the value of 'inplace').
                None in case of an error.

        """

        if self._df is None:
            self.logger.error("Data frame is not initialized.")
            return None

        if self._df.empty:
            self.logger.error("Data frame is empty.")
            return None

        # First filtered_df is the full data frame.
        # Then it is subsequentially reduced by each condition
        # at the end it is just those rows that match all conditions.
        filtered_df = self._df if inplace else self._df.copy()

        def list_matches(row: list, values: list) -> bool:
            """Check if any item in the 'values' list is present in the given 'row' list.

            Args:
                row (list):
                    A list of items from the data frame column.
                values (list):
                    A list of values to check for in the 'row'.

            Returns:
                bool:
                    True if any item in 'values' is found in 'row', otherwise False.

            """

            return any(item in values for item in row)

        def dict_matches(row: dict, key: str, values: list) -> bool:
            """Check if the value for the dictionary 'key' is in 'values'.

            Args:
                row (dict):
                    A dictionary from the data frame column.
                key (str):
                    The key to lookup in the dictionary.
                values (list):
                    A list of values to check for in the 'row'.

            Returns:
                bool:
                    True, if the value for the dictionary key is in 'values', otherwise False.

            """

            if not row or key not in row:
                return False

            return row[key] in values

        # We traverse a list of conditions. Each condition must evaluate to True
        # otherwise the current workspace or document (i.e. the data set for these objects)
        # will be skipped.
        for condition in conditions:
            # Check if the condition is enabled. If 'enabled' is not
            # in the condition dict then we assume it is enabled.
            if not condition.get("enabled", True):
                continue
            field = condition.get("field", None)
            if not field:
                self.logger.error(
                    "Missing value for filter condition 'field' in payload!",
                )
                continue
            if "." in field:
                field, sub = field.split(".", 1)
            else:
                sub = None

            if field not in self._df.columns:
                self.logger.warning(
                    "Filter condition field -> '%s' does not exist as column in the data frame! Data frame has these columns -> %s",
                    field,
                    str(self._df.columns),
                )
                continue  # Skip filtering for columns not present in data frame

            regex = condition.get("regex", False)
            # We need the column to be of type string if we want to use regular expressions
            # so if the column is not yet a string we convert the column to string:
            if regex and filtered_df[field].dtype != "object":
                # Change type of column to string:
                filtered_df[field] = filtered_df[field].astype(str)
                filtered_df[field] = filtered_df[field].fillna("")

            value = condition.get("value", None)
            if value is None:
                # Support alternative syntax using plural.
                value = condition.get("values", None)
            if value is None:
                self.logger.error(
                    "Missing filter value(s) for filter condition field -> '%s'!",
                    field,
                )
                continue

            # if a single string is passed as value we put
            # it into an 1-item list to simplify the following code:
            if not isinstance(value, list):
                value = [value]

            # If all values in the condition are strings then we
            # want the column also to be of type string:
            if all(isinstance(v, str) for v in value):
                # Change type of column to string:
                #                filtered_df[field] = filtered_df[field].astype(str)
                #                filtered_df[field] = filtered_df[field].fillna("").astype(str)
                #                filtered_df[field] = filtered_df[field].fillna("")

                # When inplace == True, filtered_df is just a reference to self._df.
                # Using .loc[:, field] ensures that Pandas updates the column correctly in self._df.
                # When inplace == False, filtered_df is a full copy (self._df.copy() above),
                # so modifications remain in filtered_df.
                # .loc[:, field] ensures no SettingWithCopyWarning, since filtered_df is now a separate DataFrame.
                filtered_df.loc[:, field] = filtered_df[field].fillna("").astype(str)

            self.logger.info(
                "Data frame has %s row(s) and %s column(s) before filter -> %s has been applied.",
                str(filtered_df.shape[0]),
                str(filtered_df.shape[1]),
                str(condition),
            )

            # Check if the column is boolean
            if pd.api.types.is_bool_dtype(filtered_df[field]):
                # Convert string representations of booleans to actual booleans
                value = [v.lower() in ["true", "1"] if isinstance(v, str) else bool(v) for v in value]

            # Do we want to test for equalitiy or non-equality?
            # For lists equality means: value is in the list
            # For lists non-equality means: value is NOT in the list
            test_for_equal = condition.get("equal", True)

            # Check if the column contains only lists (every non-empty element in the column is a list).
            # `filtered_df[field]`: Access the column with the name specified in 'field'.
            # `.dropna()`: Drop None or NaN rows for the test.
            # `.apply(lambda x: isinstance(x, list))`: For each element in the column, check if it is a list.
            # `.all()`: Ensure that all elements in the column satisfy the condition of being a list.
            if filtered_df[field].dropna().apply(lambda x: isinstance(x, list)).all():
                if not test_for_equal:
                    filtered_df = filtered_df[~filtered_df[field].apply(list_matches, values=value)]
                else:
                    filtered_df = filtered_df[filtered_df[field].apply(list_matches, values=value)]
            # Check if the column contains only dictionaries (every non-empty element in the column is a dict).
            # `filtered_df[field]`: Access the column with the name specified in 'field'.
            # `.dropna()`: Drop None or NaN rows for the test.
            # `.apply(lambda x: isinstance(x, dict))`: For each element in the column, check if it is a dict.
            # `.all()`: Ensure that all elements in the column satisfy the condition of being a dictionary.
            elif filtered_df[field].dropna().apply(lambda x: isinstance(x, dict)).all():
                if not sub:
                    self.logger.error(
                        "Filtering on dictionary values need a key. This needs to be provided with 'field.key' syntax!",
                    )
                    continue
                if not test_for_equal:
                    filtered_df = filtered_df[~filtered_df[field].apply(dict_matches, key=sub, values=value)]
                else:
                    filtered_df = filtered_df[filtered_df[field].apply(dict_matches, key=sub, values=value)]
            # Check if the column has boolean values:
            elif pd.api.types.is_bool_dtype(filtered_df[field]):
                # For a boolean filter we can drop NA values:
                filtered_df = filtered_df.dropna(subset=[field])
                if not test_for_equal:
                    filtered_df = filtered_df[~filtered_df[field].isin(value)]
                else:
                    filtered_df = filtered_df[filtered_df[field].isin(value)]
            elif not regex:
                if pd.api.types.is_string_dtype(filtered_df[field]):
                    filtered_df[field] = filtered_df[field].str.strip()
                if not test_for_equal:
                    filtered_df = filtered_df[~filtered_df[field].isin(value)]
                else:
                    filtered_df = filtered_df[filtered_df[field].isin(value)]
            else:
                # Create a pure boolean pd.Series as a filter criterium:
                regex_condition = filtered_df[field].str.contains(
                    "|".join(value),
                    regex=True,
                    na=False,
                )
                # Apply the boolean pd.Series named 'regex_condition' as
                # a filter - either non-negated or negated (using ~):
                filtered_df = filtered_df[~regex_condition] if not test_for_equal else filtered_df[regex_condition]

            self.logger.info(
                "Data frame has %s row(s) and %s column(s) after filter -> %s has been applied.",
                str(filtered_df.shape[0]),
                str(filtered_df.shape[1]),
                str(condition),
            )
        # end for condition

        if inplace:
            self._df = filtered_df

            if reset_index:
                self._df.reset_index(inplace=True, drop=True)

        return filtered_df

    # end method definition

    def fill_na_in_column(self, column_name: str, default_value: str | int) -> None:
        """Replace NA values in a column with a defined new default value.

        Args:
            column_name (str):
                The name of the column in the data frame.
            default_value (str | int):
                The value to replace NA with.

        """

        if column_name in self._df.columns:
            self._df[column_name] = self._df[column_name].fillna(value=default_value)
        else:
            self.logger.error(
                "Cannot replace NA values as column -> '%s' does not exist in the data frame! Available columns -> %s",
                column_name,
                str(self._df.columns),
            )

    # end method definition

    def fill_forward(self, inplace: bool) -> pd.DataFrame:
        """Fill the missing cells appropriately by carrying forward the values from the previous rows where necessary.

        This has applications if a hierarchy is represented by
        nested cells e.g. in an Excel sheet.

        Args:
            inplace (bool):
                Should the modification happen inplace or not.

        Returns:
            pd.DataFrame:
                The resulting data frame.

        """

        # To convert an Excel representation of a folder structure with nested
        # columns into a format appropriate for Pandas,
        # where all cells should be filled
        df_filled = self._df.ffill(inplace=inplace)

        return df_filled

    # end method definition

    def lookup_value(
        self,
        lookup_column: str,
        lookup_value: str,
        separator: str = "|",
        single_row: bool = True,
    ) -> pd.Series | pd.DataFrame | None:
        """Lookup row(s) that includes a lookup value in the value of a given column.

        Args:
            lookup_column (str):
                The name of the column to search in.
            lookup_value (str):
                The value to search for.
            separator (str):
                The string list delimiter / separator. The pipe symbol | is the default
                as it is unlikely to appear in a normal string (other than a plain comma).
                The separator is NOT looked for in the lookup_value but in the column that
                is given by lookup_column!
            single_row (bool, optional):
                This defines if we just return the first matching row if multiple matching rows
                are found. Default is True (= single row).

        Returns:
            pd.Series | pd.DataFrame | None:
                Data frame (multiple rows) or Series (row) that matches the lookup value.
                None if no match was found.

        """

        # Use the `apply` function to filter rows where the lookup value matches a
        # whole item in the separator-divided list:
        def match_lookup_value(string_list: str | None) -> bool:
            """Check if the lookup value is in a string list.

            For this the string list is converted to a python
            list. A separator is used for the splitting.

            Args:
                string_list (str):
                    Delimiter-separated string list like "a, b, c" or "a | b | c"

            Returns:
                bool:
                    True if lookup_value is equal to one of the delimiter-separated terms.

            """

            if pd.isna(string_list):  # Handle None/NaN safely
                return False

            # Ensure that the string is a string
            string_list = str(string_list)

            return lookup_value in [item.strip() for item in string_list.split(separator)]

        # end method definition

        if self._df is None:
            return None

        df = self._df

        if lookup_column not in self._df.columns:
            self.logger.error(
                "Cannot lookup value in column -> '%s'. Column does not exist in the data frame! Data frame has these columns -> %s",
                lookup_column,
                str(self._df.columns),
            )
            return None

        # Fill NaN or None values in the lookup column with empty strings
        # df[lookup_column] = df[lookup_column].fillna("")

        # Use the `apply` function to filter rows where the lookup value is in row cell
        # of column given by lookup_column. match_lookup_value() is called with
        # the content of the individual cell contents:
        matched_rows = df[df[lookup_column].apply(match_lookup_value)]

        # If nothing was found we return None:
        if matched_rows.empty:
            return None

        # If it is OK to have multiple matches (= multiple rows = pd.DataFrame).
        # We can just return the matched_rows now which should be a pd.DataFrame:
        if not single_row:
            return matched_rows

        # Check if more than one row matches, and log a warning if so
        if len(matched_rows) > 1:
            self.logger.warning(
                "More than one match found for lookup value -> '%s' in column -> '%s'. Returning the first match.",
                lookup_value,
                lookup_column,
            )

        # Return the first matched row, if any
        return matched_rows.iloc[0]

    # end method definition

    def set_value(self, column: str, value, condition: pd.Series | None = None) -> None:  # noqa: ANN001
        """Set the value in the data frame based on a condition.

        Args:
            column (str):
                The name of the column.
            value (Any):
                The value to set for those rows that fulfill the condition.
            condition (pd.Series, optional):
                This should be a boolean Series where each element is True or False,
                representing rows in the data frame that meet a certain condition.
                If None is provided then ALL rows get the 'value' in the given
                column.

        """

        if condition is None:
            self._df[column] = value  # Set value unconditionally
        else:
            self._df.loc[condition, column] = value  # Set value based on condition

    # end method definition

    def add_column(
        self,
        new_column: str,
        data_type: str = "string",
        source_column: str = "",
        reg_exp: str = "",
        prefix: str = "",
        suffix: str = "",
        length: int | None = None,
        group_chars: int | None = None,
        group_separator: str = ".",
        group_remove_leading_zero: bool = True,
    ) -> bool:
        """Add additional column to the data frame.

        Args:
            new_column (str):
                The name of the column to add.
            data_type (str, optional):
                The data type of the new column.
            source_column (str, optional):
                The name of the source column.
            reg_exp (str, optional):
                A regular expression to apply on the content of the source column.
            prefix (str, optional):
                Prefix to add in front of the value. Defaults to "".
            suffix (str, optional):
                Suffix to add at the end of the value. Defaults to "".
            length (int | None, optional):
                Length to reduce to. Defaults to None (= unlimited).
            group_chars (int | None, optional):
                Group the resulting string in characters of group_chars. Defaults to None.
                Usable e.g. for thousand seperator "."
            group_separator (str, optional):
                Separator string for the grouping. Defaults to ".".
            group_remove_leading_zero (bool, optional):
                Remove leading zeros from the groups. Defaults to True.

        Returns:
            bool:
                True = Success, False = Failure

        """

        if self._df is None:
            return False

        # Check that the new column does not yet exist
        if new_column in self._df.columns:
            self.logger.error(
                "New column -> '%s' does already exist in data frame! Cannot add it. Data frame has these columns -> %s",
                new_column,
                str(self._df.columns),
            )
            return False

        # first we handle the very simple case to not have
        # a source column but just add an empty new column:
        if not source_column:
            self._df[new_column] = pd.Series(dtype=data_type)
            return True

        # Check if the source column exists
        if source_column not in self._df.columns:
            self.logger.error(
                "Source column -> '%s' does not exist as column in data frame! Data frame has these columns -> %s",
                source_column,
                str(self._df.columns),
            )
            return False

        # Validate the regex pattern
        try:
            re.compile(reg_exp)  # Check if the pattern is a valid regex
        except re.error:
            self.logger.error(
                "Invalid regular expression -> %s. Cannot extract data for new column -> '%s'!",
                reg_exp,
                new_column,
            )
            return False

        # Ensure the source column is of type string (convert it, if necessary)
        if self._df[source_column].dtype != "object":
            self._df[source_column] = self._df[source_column].astype(str)

        # Use str.extract to apply the regular expression to the source column
        # and then assign this modified column to the variable "extracted":
        extracted = self._df[source_column].str.extract(pat=reg_exp, expand=False)

        # Limit the result to the specified length
        if length is not None:
            extracted = extracted.str[:length]

        if group_chars is not None:

            def process_grouping(x) -> str | None:  # noqa: ANN001
                if pd.isna(x):
                    return None
                # Split into groups
                groups = [x[i : i + group_chars] for i in range(0, len(x), group_chars)]
                if group_remove_leading_zero:
                    # Remove leading zeros from each group
                    groups = [group.lstrip("0") or "0" for group in groups]
                # Join groups with separator
                return group_separator.join(groups)

            extracted = extracted.apply(process_grouping)

        # Add prefix and suffix
        if prefix or suffix:
            extracted = prefix + extracted.astype(str) + suffix

        self._df[new_column] = extracted

        return True

    # end method definition

    def convert_to_lists(self, columns: list, delimiter: str = ",") -> None:
        """Intelligently convert string values to list values, in defined data frame columns.

        The delimiter to separate values in the string value can be configured.
        The method is ignoring delimiters that are inside quotes.

        Args:
            columns (list):
                The name of the columns whose values should be converted to lists.
            delimiter (str, optional):
                Character that delimits list items. Defaults to ",".

        Returns:
            None. self._df is modified in place.

        """

        # Regex to split by the delimiter, ignoring those inside quotes or double quotes
        def split_string_ignoring_quotes(s: str, delimiter: str) -> list:
            """Split a string into a list at positions that have a delimiter character.

            Args:
                s (str): the string to split
                delimiter (str): The single character that is used for splitting.

            Returns:
                A list of splitted values.

            """

            # Escaping the delimiter in case it's a special regex character
            delimiter = re.escape(delimiter)
            # Match quoted strings and unquoted delimiters separately
            pattern = rf'(?:"[^"]*"|\'[^\']*\'|[^{delimiter}]+)'
            return re.findall(pattern, s)

        for col in columns:
            self._df[col] = self._df[col].apply(
                lambda x: (split_string_ignoring_quotes(x, delimiter) if isinstance(x, str) and delimiter in x else x),
            )

    # end method definition

    def add_column_concat(
        self,
        source_columns: list,
        new_column: str,
        concat_char: str = "",
        upper: bool = False,
        lower: bool = False,
        capitalize: bool = False,
        title: bool = False,
    ) -> None:
        """Add a column as a concatenation of the values of multiple source columns.

        Args:
            source_columns (list):
                The column names the list values are taken from.
            new_column (str):
                The name of the new column.
            concat_char (str, optional):
                Character to insert between the concatenated values. Default is "".
            upper (bool, optional):
                Convert result to uppercase if True.
            lower (bool, optional):
                Convert result to lowercase if True.
            capitalize (bool, optional):
                Capitalize the result if True.
            title (bool, optional):
                Convert result to title case if True.

        Returns:
            None. self._df is modified in place.

        """

        def concatenate(row: pd.Series) -> str:
            # Comprehension to create a list from all source column values:
            concatenated = concat_char.join(
                [str(row[col]) for col in source_columns if pd.notna(row[col])],
            )

            # Apply case transformations based on parameters
            if upper:
                concatenated = concatenated.upper()
            elif lower:
                concatenated = concatenated.lower()
            elif capitalize:
                concatenated = concatenated.capitalize()
            elif title:
                concatenated = concatenated.title()

        # end method definition

        self._df[new_column] = self._df.apply(concatenate, axis=1)

    # end method definition

    def add_column_list(self, source_columns: list, new_column: str) -> None:
        """Add a column with list objects.

        The list items are taken from a list of source columns (row by row).

        Args:
            source_columns (list):
                The column names the list values are taken from.
            new_column (str):
                The name of the new column.

        Returns:
            None. self._df is modified in place.

        """

        def create_list(row: pd.Series) -> list:
            # Comprehension to create a list from all source column values:
            return [row[col] for col in source_columns]

        self._df[new_column] = self._df.apply(create_list, axis=1)

    # end method definition

    def add_column_table(
        self,
        source_columns: list,
        new_column: str,
        delimiter: str = ",",
    ) -> None:
        """Add a column with tabular objects (list of dictionaries).

        The source columns should include lists. The resulting dictionary
        keys are the column names for the source columns.

        Example (["X", "Y"] are the source_columns, "Table" is the new_column):
        X[1] = [1, 2, 3]         # row 1
        Y[1] = ["A", "B", "C"]   # row 1
        X[2] = [4, 5, 6]         # row 2
        Y[2] = ["D", "E", "F"]   # row 2

        Table[1] = [
            {
                "X": "1"
                "Y": "A"
            },
            {
                "X": "2"
                "Y": "B"
            }
            {
                "X": "3"
                "Y": "C"
            }
        ]
        Table[2] = [
            {
                "X": "4"
                "Y": "D"
            },
            {
                "X": "5"
                "Y": "E"
            }
            {
                "X": "6"
                "Y": "F"
            }
        ]

        Args:
            source_columns (list):
                The column names the list values are taken from.
            new_column (str):
                The name of the new column.
            delimiter (str, optional):
                Character that delimits list items. Defaults to ",".

        Returns:
            None. self._df is modified in place.

        """

        # Call the convert_to_lists method to ensure the columns are converted
        self.convert_to_lists(columns=source_columns, delimiter=delimiter)

        # Sub-method to pad lists to the same length
        def pad_list(lst: list, max_len: int) -> list:
            return lst + [None] * (max_len - len(lst))

        def create_table(row: pd.Series) -> list:
            max_len = max(len(row[col]) if isinstance(row[col], list) else 1 for col in source_columns)

            # Pad lists to the maximum length, leave scalar values as they are
            for col in source_columns:
                if isinstance(row[col], list):
                    row[col] = pad_list(row[col], max_len)
                elif not pd.isna(row[col]):
                    row[col] = [
                        row[col],
                    ] * max_len  # Repeat scalar value to match the max length
                else:
                    row[col] = [None] * max_len
            # Create a list of dictionaries for each row:
            table = [{col: row[col][i] for col in source_columns} for i in range(max_len)]

            return table

        # Apply the function to create a new column with table values:
        self._df[new_column] = self._df.apply(create_table, axis=1)

__getitem__(column)

Return the column corresponding to the key from the data frame.

Parameters:

Name	Type	Description	Default
column	str	The name of the data frame column.	required

Returns:

Type	Description
Series	pd.Series: The column of the data frame with the given name.

Source code in packages/pyxecm/src/pyxecm/helper/data.py

def __getitem__(self, column: str) -> pd.Series:
    """Return the column corresponding to the key from the data frame.

    Args:
        column (str): The name of the data frame column.

    Returns:
        pd.Series: The column of the data frame with the given name.

    """

    return self._df[column]

__init__(init_data=None, logger=default_logger)

Initialize the Data object.

Parameters:

Name	Type	Description	Default
init_data	DataFrame | list	Data to initialize the data frame. Can either be another data frame (that gets copied) or a list of dictionaries. Defaults to None.	None
logger	Logger	Pass a special logging object. This is optional. If not provided, the default logger is used.	default_logger

Source code in packages/pyxecm/src/pyxecm/helper/data.py

def __init__(
    self,
    init_data: pd.DataFrame | list = None,
    logger: logging.Logger = default_logger,
) -> None:
    """Initialize the Data object.

    Args:
        init_data (pd.DataFrame | list, optional):
            Data to initialize the data frame. Can either be
            another data frame (that gets copied) or a list of dictionaries.
            Defaults to None.
        logger (logging.Logger, optional):
            Pass a special logging object. This is optional. If not provided,
            the default logger is used.

    """

    if logger != default_logger:
        self.logger = logger.getChild("data")
        for logfilter in logger.filters:
            self.logger.addFilter(logfilter)

    if init_data is not None:
        # if a data frame is passed to the constructor we
        # copy its content to the new Data object

        if isinstance(init_data, pd.DataFrame):
            self._df: pd.DataFrame = init_data.copy()
        elif isinstance(init_data, Data):
            if init_data.get_data_frame() is not None:
                self._df: pd.DataFrame = init_data.get_data_frame().copy()
        elif isinstance(init_data, list):
            self._df: pd.DataFrame = pd.DataFrame(init_data)
        elif isinstance(init_data, dict):
            # it is important to wrap the dict in a list to avoid that more than 1 row is created
            self._df: pd.DataFrame = pd.DataFrame([init_data])
        else:
            self.logger.error("Illegal initialization data for 'Data' class!")
            self._df = None
    else:
        self._df = None

__len__()

Return lenght of the embedded Pandas data frame object.

This is basically a convenience method.

Returns:

Name	Type	Description
int	int	Lenght of the data frame.

Source code in packages/pyxecm/src/pyxecm/helper/data.py

def __len__(self) -> int:
    """Return lenght of the embedded Pandas data frame object.

    This is basically a convenience method.

    Returns:
        int:
            Lenght of the data frame.

    """

    if self._df is not None:
        return len(self._df)
    return 0

__str__()

Print the Pandas data frame object.

Returns:

Name	Type	Description
str	str	String representation.

Source code in packages/pyxecm/src/pyxecm/helper/data.py

def __str__(self) -> str:
    """Print the Pandas data frame object.

    Returns:
        str:
            String representation.

    """

    # if data frame is initialized we return
    # the string representation of pd.DataFrame
    if self._df is not None:
        return str(self._df)

    return str(self)

add_column(new_column, data_type='string', source_column='', reg_exp='', prefix='', suffix='', length=None, group_chars=None, group_separator='.', group_remove_leading_zero=True)

Add additional column to the data frame.

Parameters:

Name	Type	Description	Default
new_column	str	The name of the column to add.	required
data_type	str	The data type of the new column.	'string'
source_column	str	The name of the source column.	''
reg_exp	str	A regular expression to apply on the content of the source column.	''
prefix	str	Prefix to add in front of the value. Defaults to "".	''
suffix	str	Suffix to add at the end of the value. Defaults to "".	''
length	int | None	Length to reduce to. Defaults to None (= unlimited).	None
group_chars	int | None	Group the resulting string in characters of group_chars. Defaults to None. Usable e.g. for thousand seperator "."	None
group_separator	str	Separator string for the grouping. Defaults to ".".	'.'
group_remove_leading_zero	bool	Remove leading zeros from the groups. Defaults to True.	True

Returns:

Name	Type	Description
bool	bool	True = Success, False = Failure

Source code in packages/pyxecm/src/pyxecm/helper/data.py

def add_column(
    self,
    new_column: str,
    data_type: str = "string",
    source_column: str = "",
    reg_exp: str = "",
    prefix: str = "",
    suffix: str = "",
    length: int | None = None,
    group_chars: int | None = None,
    group_separator: str = ".",
    group_remove_leading_zero: bool = True,
) -> bool:
    """Add additional column to the data frame.

    Args:
        new_column (str):
            The name of the column to add.
        data_type (str, optional):
            The data type of the new column.
        source_column (str, optional):
            The name of the source column.
        reg_exp (str, optional):
            A regular expression to apply on the content of the source column.
        prefix (str, optional):
            Prefix to add in front of the value. Defaults to "".
        suffix (str, optional):
            Suffix to add at the end of the value. Defaults to "".
        length (int | None, optional):
            Length to reduce to. Defaults to None (= unlimited).
        group_chars (int | None, optional):
            Group the resulting string in characters of group_chars. Defaults to None.
            Usable e.g. for thousand seperator "."
        group_separator (str, optional):
            Separator string for the grouping. Defaults to ".".
        group_remove_leading_zero (bool, optional):
            Remove leading zeros from the groups. Defaults to True.

    Returns:
        bool:
            True = Success, False = Failure

    """

    if self._df is None:
        return False

    # Check that the new column does not yet exist
    if new_column in self._df.columns:
        self.logger.error(
            "New column -> '%s' does already exist in data frame! Cannot add it. Data frame has these columns -> %s",
            new_column,
            str(self._df.columns),
        )
        return False

    # first we handle the very simple case to not have
    # a source column but just add an empty new column:
    if not source_column:
        self._df[new_column] = pd.Series(dtype=data_type)
        return True

    # Check if the source column exists
    if source_column not in self._df.columns:
        self.logger.error(
            "Source column -> '%s' does not exist as column in data frame! Data frame has these columns -> %s",
            source_column,
            str(self._df.columns),
        )
        return False

    # Validate the regex pattern
    try:
        re.compile(reg_exp)  # Check if the pattern is a valid regex
    except re.error:
        self.logger.error(
            "Invalid regular expression -> %s. Cannot extract data for new column -> '%s'!",
            reg_exp,
            new_column,
        )
        return False

    # Ensure the source column is of type string (convert it, if necessary)
    if self._df[source_column].dtype != "object":
        self._df[source_column] = self._df[source_column].astype(str)

    # Use str.extract to apply the regular expression to the source column
    # and then assign this modified column to the variable "extracted":
    extracted = self._df[source_column].str.extract(pat=reg_exp, expand=False)

    # Limit the result to the specified length
    if length is not None:
        extracted = extracted.str[:length]

    if group_chars is not None:

        def process_grouping(x) -> str | None:  # noqa: ANN001
            if pd.isna(x):
                return None
            # Split into groups
            groups = [x[i : i + group_chars] for i in range(0, len(x), group_chars)]
            if group_remove_leading_zero:
                # Remove leading zeros from each group
                groups = [group.lstrip("0") or "0" for group in groups]
            # Join groups with separator
            return group_separator.join(groups)

        extracted = extracted.apply(process_grouping)

    # Add prefix and suffix
    if prefix or suffix:
        extracted = prefix + extracted.astype(str) + suffix

    self._df[new_column] = extracted

    return True

add_column_concat(source_columns, new_column, concat_char='', upper=False, lower=False, capitalize=False, title=False)

Add a column as a concatenation of the values of multiple source columns.

Parameters:

Name	Type	Description	Default
source_columns	list	The column names the list values are taken from.	required
new_column	str	The name of the new column.	required
concat_char	str	Character to insert between the concatenated values. Default is "".	''
upper	bool	Convert result to uppercase if True.	False
lower	bool	Convert result to lowercase if True.	False
capitalize	bool	Capitalize the result if True.	False
title	bool	Convert result to title case if True.	False

Returns:

Type	Description
None	None. self._df is modified in place.

Source code in packages/pyxecm/src/pyxecm/helper/data.py

def add_column_concat(
    self,
    source_columns: list,
    new_column: str,
    concat_char: str = "",
    upper: bool = False,
    lower: bool = False,
    capitalize: bool = False,
    title: bool = False,
) -> None:
    """Add a column as a concatenation of the values of multiple source columns.

    Args:
        source_columns (list):
            The column names the list values are taken from.
        new_column (str):
            The name of the new column.
        concat_char (str, optional):
            Character to insert between the concatenated values. Default is "".
        upper (bool, optional):
            Convert result to uppercase if True.
        lower (bool, optional):
            Convert result to lowercase if True.
        capitalize (bool, optional):
            Capitalize the result if True.
        title (bool, optional):
            Convert result to title case if True.

    Returns:
        None. self._df is modified in place.

    """

    def concatenate(row: pd.Series) -> str:
        # Comprehension to create a list from all source column values:
        concatenated = concat_char.join(
            [str(row[col]) for col in source_columns if pd.notna(row[col])],
        )

        # Apply case transformations based on parameters
        if upper:
            concatenated = concatenated.upper()
        elif lower:
            concatenated = concatenated.lower()
        elif capitalize:
            concatenated = concatenated.capitalize()
        elif title:
            concatenated = concatenated.title()

    # end method definition

    self._df[new_column] = self._df.apply(concatenate, axis=1)

add_column_list(source_columns, new_column)

Add a column with list objects.

The list items are taken from a list of source columns (row by row).

Parameters:

Name	Type	Description	Default
source_columns	list	The column names the list values are taken from.	required
new_column	str	The name of the new column.	required

Returns:

Type	Description
None	None. self._df is modified in place.

Source code in packages/pyxecm/src/pyxecm/helper/data.py

def add_column_list(self, source_columns: list, new_column: str) -> None:
    """Add a column with list objects.

    The list items are taken from a list of source columns (row by row).

    Args:
        source_columns (list):
            The column names the list values are taken from.
        new_column (str):
            The name of the new column.

    Returns:
        None. self._df is modified in place.

    """

    def create_list(row: pd.Series) -> list:
        # Comprehension to create a list from all source column values:
        return [row[col] for col in source_columns]

    self._df[new_column] = self._df.apply(create_list, axis=1)

add_column_table(source_columns, new_column, delimiter=',')

Add a column with tabular objects (list of dictionaries).

The source columns should include lists. The resulting dictionary keys are the column names for the source columns.

Example (["X", "Y"] are the source_columns, "Table" is the new_column): X[1] = [1, 2, 3] # row 1 Y[1] = ["A", "B", "C"] # row 1 X[2] = [4, 5, 6] # row 2 Y[2] = ["D", "E", "F"] # row 2

Table[1] = [ { "X": "1" "Y": "A" }, { "X": "2" "Y": "B" } { "X": "3" "Y": "C" } ] Table[2] = [ { "X": "4" "Y": "D" }, { "X": "5" "Y": "E" } { "X": "6" "Y": "F" } ]

Parameters:

Name	Type	Description	Default
source_columns	list	The column names the list values are taken from.	required
new_column	str	The name of the new column.	required
delimiter	str	Character that delimits list items. Defaults to ",".	','

Returns:

Type	Description
None	None. self._df is modified in place.

Source code in packages/pyxecm/src/pyxecm/helper/data.py

def add_column_table(
    self,
    source_columns: list,
    new_column: str,
    delimiter: str = ",",
) -> None:
    """Add a column with tabular objects (list of dictionaries).

    The source columns should include lists. The resulting dictionary
    keys are the column names for the source columns.

    Example (["X", "Y"] are the source_columns, "Table" is the new_column):
    X[1] = [1, 2, 3]         # row 1
    Y[1] = ["A", "B", "C"]   # row 1
    X[2] = [4, 5, 6]         # row 2
    Y[2] = ["D", "E", "F"]   # row 2

    Table[1] = [
        {
            "X": "1"
            "Y": "A"
        },
        {
            "X": "2"
            "Y": "B"
        }
        {
            "X": "3"
            "Y": "C"
        }
    ]
    Table[2] = [
        {
            "X": "4"
            "Y": "D"
        },
        {
            "X": "5"
            "Y": "E"
        }
        {
            "X": "6"
            "Y": "F"
        }
    ]

    Args:
        source_columns (list):
            The column names the list values are taken from.
        new_column (str):
            The name of the new column.
        delimiter (str, optional):
            Character that delimits list items. Defaults to ",".

    Returns:
        None. self._df is modified in place.

    """

    # Call the convert_to_lists method to ensure the columns are converted
    self.convert_to_lists(columns=source_columns, delimiter=delimiter)

    # Sub-method to pad lists to the same length
    def pad_list(lst: list, max_len: int) -> list:
        return lst + [None] * (max_len - len(lst))

    def create_table(row: pd.Series) -> list:
        max_len = max(len(row[col]) if isinstance(row[col], list) else 1 for col in source_columns)

        # Pad lists to the maximum length, leave scalar values as they are
        for col in source_columns:
            if isinstance(row[col], list):
                row[col] = pad_list(row[col], max_len)
            elif not pd.isna(row[col]):
                row[col] = [
                    row[col],
                ] * max_len  # Repeat scalar value to match the max length
            else:
                row[col] = [None] * max_len
        # Create a list of dictionaries for each row:
        table = [{col: row[col][i] for col in source_columns} for i in range(max_len)]

        return table

    # Apply the function to create a new column with table values:
    self._df[new_column] = self._df.apply(create_table, axis=1)

append(add_data)

Append additional data to the data frame.

Parameters:

Name	Type	Description	Default
add_data	DataFrame | list | dict	Additional data. Can be pd.DataFrame or list of dicts (or Data).	required

Returns:

Name	Type	Description
bool	bool	True = Success, False = Error

Source code in packages/pyxecm/src/pyxecm/helper/data.py

def append(self, add_data: pd.DataFrame | list | dict) -> bool:
    """Append additional data to the data frame.

    Args:
        add_data (pd.DataFrame | list | dict):
            Additional data. Can be pd.DataFrame or list of dicts (or Data).

    Returns:
        bool:
            True = Success, False = Error

    """

    # Does the data frame has already content?
    # Then we need to concat / append. Otherwise
    # we just initialize self._df
    if self._df is not None:
        if isinstance(add_data, pd.DataFrame):
            self._df = pd.concat([self._df, add_data], ignore_index=True)
            return True
        elif isinstance(add_data, Data):
            df = add_data.get_data_frame()
            if df is not None and not df.empty:
                self._df = pd.concat([self._df, df], ignore_index=True)
            return True
        elif isinstance(add_data, list):
            if add_data:
                df = Data(add_data, logger=self.logger)
                self._df = pd.concat(
                    [self._df, df.get_data_frame()],
                    ignore_index=True,
                )
            return True
        elif isinstance(add_data, dict):
            if add_data:
                # it is important to wrap the dict in a list to avoid that more than 1 row is created
                df = Data([add_data], logger=self.logger)
                self._df = pd.concat(
                    [self._df, df.get_data_frame()],
                    ignore_index=True,
                )
            return True
        else:
            self.logger.error("Illegal data type -> '%s'", type(add_data))
            return False
    elif isinstance(add_data, pd.DataFrame):
        self._df = add_data
        return True
    elif isinstance(add_data, Data):
        self._df = add_data.get_data_frame()
        return True
    elif isinstance(add_data, list):
        self._df = pd.DataFrame(add_data)
        return True
    elif isinstance(add_data, dict):
        # it is important to wrap the dict in a list to avoid that more than 1 row is created
        self._df = pd.DataFrame([add_data])
        return True
    else:
        self.logger.error("Illegal data type -> '%s'", type(add_data))
        return False

cleanse(cleansings)

Cleanse data with regular expressions and upper/lower case conversions.

Parameters:

Name	Type	Description	Default
cleansings	dict	Dictionary with keys that equal the column names. The dictionary values are dictionaries themselves with these fields: * replacements (dict): name of a column in the data frame * upper (bool, optional, default = False): change the value to uppercase * lower (bool, optional, default = False): change the value to lowercase * capitalize (bool, optional, default = False) - first character upper case, rest lower-case * title (bool, optional, default = False) - first character of each word upper case * length (int, optional, default = 0): truncate to max length	required

Source code in packages/pyxecm/src/pyxecm/helper/data.py

def cleanse(self, cleansings: dict) -> None:
    """Cleanse data with regular expressions and upper/lower case conversions.

    Args:
        cleansings (dict):
            Dictionary with keys that equal the column names.
            The dictionary values are dictionaries themselves with
            these fields:
            * replacements (dict): name of a column in the data frame
            * upper (bool, optional, default = False): change the value to uppercase
            * lower (bool, optional, default = False): change the value to lowercase
            * capitalize (bool, optional, default = False) - first character upper case, rest lower-case
            * title (bool, optional, default = False) - first character of each word upper case
            * length (int, optional, default = 0): truncate to max length

    """

    # Iterate over each column in the cleansing dictionary
    for column, cleansing in cleansings.items():
        # Read the cleansing parameters:
        replacements = cleansing.get("replacements", {})
        upper = cleansing.get("upper", False)
        lower = cleansing.get("lower", False)
        capitalize = cleansing.get("capitalize", False)
        title = cleansing.get("title", False)
        length = cleansing.get("length", 0)

        # Handle dict columns - we expect the column name to seperate
        # main field from sub field using a dot syntax (e.g., "column.subfield")
        if "." in column:
            column, dict_key = column.split(".")
            if column not in self._df.columns:
                self.logger.error(
                    "Cannot cleanse column -> '%s'. It does not exist in the data frame! Data frame has these columns -> %s",
                    column,
                    str(self._df.columns),
                )
                continue
            # Apply cleansing to dictionary values in the main column
            self.logger.debug(
                "Cleansing for column -> '%s' has a subfield -> '%s' configured. Do cleansing for dictionary items with key -> '%s'...",
                column,
                dict_key,
                dict_key,
            )
            self._df[column] = self._df[column].apply(
                lambda x,
                dict_key=dict_key,
                replacements=replacements,
                upper=upper,
                lower=lower,
                capitalize=capitalize,
                title=title,
                length=length: self._cleanse_subfield(
                    data=x,
                    dict_key=dict_key,
                    replacements=replacements,
                    upper=upper,
                    lower=lower,
                    capitalize=capitalize,
                    title=title,
                    length=length,
                ),
            )
        # end if "." in column
        else:  # the else case handles strings and list columns
            if column not in self._df.columns:
                self.logger.error(
                    "Cannot cleanse column -> '%s'. It does not exist in the data frame! Data frame has these columns -> %s",
                    column,
                    str(self._df.columns),
                )
                continue

            # Handle string columns:
            if self.is_string_column(self._df[column]):
                # Apply cleansing operations on string column
                self.logger.debug(
                    "Column -> '%s' has string values. Do cleansing for string values...",
                    column,
                )
                self._df[column] = self._df[column].apply(
                    lambda x,
                    replacements=replacements,
                    upper=upper,
                    lower=lower,
                    capitalize=capitalize,
                    title=title,
                    length=length: (
                        self._apply_string_cleansing(
                            value=x,
                            replacements=replacements,
                            upper=upper,
                            lower=lower,
                            capitalize=capitalize,
                            title=title,
                            length=length,
                        )
                        if isinstance(x, str)
                        else x
                    ),
                )

            # Handle list columns:
            elif self.is_list_column(self._df[column]):
                # Handle list-like columns for this we iterate over each list item
                # and apply the cleansing by calling _apply_string_cleansing() for item:
                self.logger.debug(
                    "Column -> '%s' has list values. Do cleansing for each list item...",
                    column,
                )
                self._df[column] = self._df[column].apply(
                    lambda x,
                    replacements=replacements,
                    upper=upper,
                    lower=lower,
                    capitalize=capitalize,
                    title=title,
                    length=length: (
                        [
                            (
                                self._apply_string_cleansing(
                                    value=item,
                                    replacements=replacements,
                                    upper=upper,
                                    lower=lower,
                                    capitalize=capitalize,
                                    title=title,
                                    length=length,
                                )
                                if isinstance(
                                    item,
                                    str,
                                )  # we just change string list items
                                else item
                            )
                            for item in x
                        ]
                        if isinstance(x, list)
                        else x
                    ),
                )

            else:
                self.logger.error(
                    "Column -> '%s' is not a string, list, or dict-like column. Skipping cleansing...",
                    column,
                )

convert_to_lists(columns, delimiter=',')

Intelligently convert string values to list values, in defined data frame columns.

The delimiter to separate values in the string value can be configured. The method is ignoring delimiters that are inside quotes.

Parameters:

Name	Type	Description	Default
columns	list	The name of the columns whose values should be converted to lists.	required
delimiter	str	Character that delimits list items. Defaults to ",".	','

Returns:

Type	Description
None	None. self._df is modified in place.

Source code in packages/pyxecm/src/pyxecm/helper/data.py

def convert_to_lists(self, columns: list, delimiter: str = ",") -> None:
    """Intelligently convert string values to list values, in defined data frame columns.

    The delimiter to separate values in the string value can be configured.
    The method is ignoring delimiters that are inside quotes.

    Args:
        columns (list):
            The name of the columns whose values should be converted to lists.
        delimiter (str, optional):
            Character that delimits list items. Defaults to ",".

    Returns:
        None. self._df is modified in place.

    """

    # Regex to split by the delimiter, ignoring those inside quotes or double quotes
    def split_string_ignoring_quotes(s: str, delimiter: str) -> list:
        """Split a string into a list at positions that have a delimiter character.

        Args:
            s (str): the string to split
            delimiter (str): The single character that is used for splitting.

        Returns:
            A list of splitted values.

        """

        # Escaping the delimiter in case it's a special regex character
        delimiter = re.escape(delimiter)
        # Match quoted strings and unquoted delimiters separately
        pattern = rf'(?:"[^"]*"|\'[^\']*\'|[^{delimiter}]+)'
        return re.findall(pattern, s)

    for col in columns:
        self._df[col] = self._df[col].apply(
            lambda x: (split_string_ignoring_quotes(x, delimiter) if isinstance(x, str) and delimiter in x else x),
        )

deduplicate(unique_fields, inplace=True)

Remove dupclicate rows that have all fields in unique_fields in common.

Parameters:

Name	Type	Description	Default
unique_fields	list	Defines the fields for which we want a unique combination for.	required
inplace	bool	True if the deduplication happens in-place. Defaults to True.	True

Returns:

Type	Description
DataFrame	pd.DataFrame: If inplace is False than a new deduplicatd data frame is returned. Otherwise the object is modified in place and self._df is returned.

Source code in packages/pyxecm/src/pyxecm/helper/data.py

def deduplicate(self, unique_fields: list, inplace: bool = True) -> pd.DataFrame:
    """Remove dupclicate rows that have all fields in unique_fields in common.

    Args:
        unique_fields (list):
            Defines the fields for which we want a unique combination for.
        inplace (bool, optional):
            True if the deduplication happens in-place. Defaults to True.

    Returns:
        pd.DataFrame:
            If inplace is False than a new deduplicatd data frame is returned.
            Otherwise the object is modified in place and self._df is returned.

    """

    if inplace:
        self._df.drop_duplicates(subset=unique_fields, inplace=True)
        self._df.reset_index(drop=True, inplace=True)
        return self._df
    else:
        df = self._df.drop_duplicates(subset=unique_fields, inplace=False)
        df = df.reset_index(drop=True, inplace=False)
        return df

drop_columns(column_names, inplace=True)

Drop selected columns from the Pandas data frame.

Parameters:

Name	Type	Description	Default
column_names	list	The list of column names to drop.	required
inplace	bool	Whether or not the dropping should be inplace, i.e. modifying self._df. Defaults to True.	True

Returns:

Type	Description
DataFrame	pd.DataFrame: New data frame (if inplace = False) or self._df (if inplace = True)

Source code in packages/pyxecm/src/pyxecm/helper/data.py

def drop_columns(self, column_names: list, inplace: bool = True) -> pd.DataFrame:
    """Drop selected columns from the Pandas data frame.

    Args:
        column_names (list):
            The list of column names to drop.
        inplace (bool, optional):
            Whether or not the dropping should be inplace, i.e. modifying self._df.
            Defaults to True.

    Returns:
        pd.DataFrame:
            New data frame (if inplace = False) or self._df (if inplace = True)

    """

    if not all(column_name in self._df.columns for column_name in column_names):
        # Reduce the column names to those that really exist in the data frame:
        column_names = [column_name for column_name in column_names if column_name in self._df.columns]
        self.logger.info(
            "Drop columns -> %s from the data frame.",
            str(column_names),
        )

    if inplace:
        self._df.drop(column_names, axis=1, inplace=True)
        return self._df
    else:
        df = self._df.drop(column_names, axis=1, inplace=False)
        return df

explode_and_flatten(explode_fields, flatten_fields=None, make_unique=False, reset_index=False, split_string_to_list=False, separator=';,')

Explode a substructure in the Pandas data frame.

Parameters:

Name	Type	Description	Default
explode_fields	str | list	Field(s) to explode. Each field to explode should have a list structure. Exploding multiple columns at once is possible. This delivers a very different result compared to exploding one column after the other!	required
flatten_fields	list	Fields in the exploded substructure to include in the main dictionaries for easier processing.	None
make_unique	bool	If True, deduplicate the exploded data frame.	False
reset_index	(bool, False)	If True, then the index is reset, False = Index is not reset.	False
split_string_to_list	bool	If True flatten the exploded data frame.	False
separator	str	Characters used to split the string values in the given column into a list.	';,'

Returns:

Type	Description
DataFrame | None	pd.DataFrame | None: Pointer to the Pandas data frame.

Source code in packages/pyxecm/src/pyxecm/helper/data.py

def explode_and_flatten(
    self,
    explode_fields: str | list,
    flatten_fields: list | None = None,
    make_unique: bool = False,
    reset_index: bool = False,
    split_string_to_list: bool = False,
    separator: str = ";,",
) -> pd.DataFrame | None:
    """Explode a substructure in the Pandas data frame.

    Args:
        explode_fields (str | list):
            Field(s) to explode. Each field to explode should have a list structure.
            Exploding multiple columns at once is possible. This delivers
            a very different result compared to exploding one column after the other!
        flatten_fields (list):
            Fields in the exploded substructure to include
            in the main dictionaries for easier processing.
        make_unique (bool, optional):
            If True, deduplicate the exploded data frame.
        reset_index (bool, False):
            If True, then the index is reset, False = Index is not reset.
        split_string_to_list (bool, optional):
            If True flatten the exploded data frame.
        separator (str, optional):
            Characters used to split the string values in the given column into a list.

    Returns:
        pd.DataFrame | None:
            Pointer to the Pandas data frame.

    """

    def update_column(row: pd.Series, sub: str) -> str:
        """Extract the value of a sub-column from a nested dictionary within a Pandas Series.

        Args:
            row (pd.Series):
                A row from the data frame.
            sub (str):
                The sub-column name to extract.

        Returns:
            str:
                The value of the sub-column, or an empty string if not found.

        """

        if isinstance(row, dict) and sub in row:
            return row[sub]
        return ""

    # end def update_column()

    def string_to_list(value: str) -> list:
        """Convert a string to a list by splitting it using a specified separator.

        If the input is already a list, it is returned as-is. If the input is `None` or a missing value,
        an empty list is returned. Otherwise, the string is split into a list of substrings using
        the given separator. Leading and trailing spaces in the resulting substrings are removed.

        Args:
            value (str):
                The input string to be converted into a list. Can also be a list, `None`,
                or a missing value (e.g., NaN).

        Returns:
            list:
                A list of substrings if the input is a string, or an empty list if the input
                is `None` or a missing value. If the input is already a list, it is returned unchanged.

        """

        # Check if the value is already a list; if so, return it directly
        if isinstance(value, list):
            return value

        # If the value is None or a missing value (e.g., NaN), return an empty list
        if not value or pd.isna(value):
            return []

        # Use a regular expression to split the string by the separator
        # and remove leading/trailing spaces from each resulting substring
        return_list = re.split(rf"[{separator}]\s*", str(value))

        return return_list

    # end def string_to_list()

    #
    # Start of main method:
    #

    # First do a sanity check if the data frame is not yet initialized.
    if self._df is None:
        self.logger.error(
            "The data frame is not initialized or empty. Cannot explode data frame.",
        )
        return None

    # Next do a sanity check for the given explode_field. It should
    # either be a string (single column name) or a list (multiple column names):
    if isinstance(explode_fields, list):
        self.logger.info("Exploding list of columns -> %s", str(explode_fields))
    elif isinstance(explode_fields, str):
        self.logger.info("Exploding single column -> '%s'", explode_fields)
    else:
        self.logger.error(
            "Illegal explode field(s) data type -> %s. Explode field must either be a string or a list of strings.",
            type(explode_fields),
        )
        return self._df

    # Ensure explode_fields is a list for uniform processing:
    if isinstance(explode_fields, str):
        explode_fields = [explode_fields]

    # Process nested field names with '.'
    processed_fields = []
    for field in explode_fields:
        # The "." indicates that the column has dictionary values:
        if "." in field:
            main, sub = field.split(".", 1)
            if main not in self._df.columns:
                self.logger.error(
                    "The column -> '%s' does not exist in the data frame! Cannot explode it. Data frame has these columns -> %s",
                    main,
                    str(self._df.columns.tolist()),
                )
                continue

            # Use update_column to extract the dictionary key specified by the sub value:
            self.logger.info(
                "Extracting dictionary value for key -> '%s' from column -> '%s'.",
                sub,
                main,
            )
            self._df[main] = self._df[main].apply(update_column, args=(sub,))
            processed_fields.append(main)
        else:
            processed_fields.append(field)

    # Verify all processed fields exist in the data frame:
    missing_columns = [col for col in processed_fields if col not in self._df.columns]
    if missing_columns:
        self.logger.error(
            "The following columns are missing in the data frame and cannot be exploded -> %s. Data frame has these columns -> %s",
            missing_columns,
            str(self._df.columns.tolist()),
        )
        return self._df

    # Handle splitting strings into lists if required:
    if split_string_to_list:
        for field in processed_fields:
            self.logger.info(
                "Splitting strings in column -> '%s' into lists using separator -> '%s'",
                field,
                separator,
            )
            # Apply the function to convert the string values in the column (give by the name in explode_field) to lists
            # The string_to_list() sub-method above also considers the separator parameter.
            self._df[field] = self._df[field].apply(string_to_list)

    # Explode all specified columns at once.
    # explode() can either take a string field or a list of fields.
    # # It is VERY important to do the explosion of multiple columns together -
    # otherwise we get combinatorial explosion. Explosion of multiple columns 1-by-1
    # is VERY different from doing the explosion together!
    self.logger.info("Validated column(s) to explode -> %s", processed_fields)
    try:
        self._df = self._df.explode(
            column=processed_fields,
            ignore_index=reset_index,
        )
    except ValueError:
        self.logger.error(
            "Error exploding columns -> %s",
            processed_fields,
        )
        return self._df

    if flatten_fields:
        # Ensure that flatten() is called for each exploded column
        for field in processed_fields:
            self.flatten(parent_field=field, flatten_fields=flatten_fields)

        # Deduplicate rows if required
        if make_unique:
            self._df.drop_duplicates(subset=flatten_fields, inplace=True)

    # Reset index explicitly if not handled during explode
    if reset_index:
        self._df.reset_index(drop=True, inplace=True)

    return self._df

fill_forward(inplace)

Fill the missing cells appropriately by carrying forward the values from the previous rows where necessary.

This has applications if a hierarchy is represented by nested cells e.g. in an Excel sheet.

Parameters:

Name	Type	Description	Default
inplace	bool	Should the modification happen inplace or not.	required

Returns:

Type	Description
DataFrame	pd.DataFrame: The resulting data frame.

Source code in packages/pyxecm/src/pyxecm/helper/data.py

def fill_forward(self, inplace: bool) -> pd.DataFrame:
    """Fill the missing cells appropriately by carrying forward the values from the previous rows where necessary.

    This has applications if a hierarchy is represented by
    nested cells e.g. in an Excel sheet.

    Args:
        inplace (bool):
            Should the modification happen inplace or not.

    Returns:
        pd.DataFrame:
            The resulting data frame.

    """

    # To convert an Excel representation of a folder structure with nested
    # columns into a format appropriate for Pandas,
    # where all cells should be filled
    df_filled = self._df.ffill(inplace=inplace)

    return df_filled

fill_na_in_column(column_name, default_value)

Replace NA values in a column with a defined new default value.

Parameters:

Name	Type	Description	Default
column_name	str	The name of the column in the data frame.	required
default_value	str | int	The value to replace NA with.	required

Source code in packages/pyxecm/src/pyxecm/helper/data.py

def fill_na_in_column(self, column_name: str, default_value: str | int) -> None:
    """Replace NA values in a column with a defined new default value.

    Args:
        column_name (str):
            The name of the column in the data frame.
        default_value (str | int):
            The value to replace NA with.

    """

    if column_name in self._df.columns:
        self._df[column_name] = self._df[column_name].fillna(value=default_value)
    else:
        self.logger.error(
            "Cannot replace NA values as column -> '%s' does not exist in the data frame! Available columns -> %s",
            column_name,
            str(self._df.columns),
        )

filter(conditions, inplace=True, reset_index=True)

Filter the data frame based on (multiple) conditions.

Parameters:

Name	Type	Description	Default
conditions	list	Conditions are a list of dictionaries with 3 items: * field (str): The name of a column in the data frame * value (str or list): Expected value (filter criterium). If it is a list then one of the list elements must match the field value (OR) * equal (bool): Whether to test for equal or non-equal. If not specified equal is treated as True. * regex (bool): This flag controls if the value is interpreted as a regular expression. If there is no regex item in the dictionary then the default is False (= values is NOT regex). * enabled (bool): True or False. The filter is only applied if 'enabled = True' If there are multiple conditions in the list each has to evaluate to True (AND)	required
inplace	bool	Defines if the self._df is modified (inplace) or just a new data frame is returned. Defaults to True.	True
reset_index	bool	Filter removes rows. If filter_index = True then the numbering of the index is newly calculated	True

Returns:

Type	Description
DataFrame | None	pd.DataFrame | None: A new data frame or pointer to self._df (depending on the value of 'inplace'). None in case of an error.

Source code in packages/pyxecm/src/pyxecm/helper/data.py

def filter(
    self,
    conditions: list,
    inplace: bool = True,
    reset_index: bool = True,
) -> pd.DataFrame | None:
    """Filter the data frame based on (multiple) conditions.

    Args:
        conditions (list):
            Conditions are a list of dictionaries with 3 items:
            * field (str): The name of a column in the data frame
            * value (str or list):
                Expected value (filter criterium).
                If it is a list then one of the list elements must match the field value (OR)
            * equal (bool):
                Whether to test for equal or non-equal. If not specified equal is treated as True.
            * regex (bool):
                This flag controls if the value is interpreted as a
                regular expression. If there is no regex item in the
                dictionary then the default is False (= values is NOT regex).
            * enabled (bool):
                True or False. The filter is only applied if 'enabled = True'
            If there are multiple conditions in the list each has to evaluate to True (AND)
        inplace (bool, optional):
            Defines if the self._df is modified (inplace) or just
            a new data frame is returned. Defaults to True.
        reset_index (bool, optional):
            Filter removes rows. If filter_index = True then the numbering
            of the index is newly calculated

    Returns:
        pd.DataFrame | None:
            A new data frame or pointer to self._df (depending on the value of 'inplace').
            None in case of an error.

    """

    if self._df is None:
        self.logger.error("Data frame is not initialized.")
        return None

    if self._df.empty:
        self.logger.error("Data frame is empty.")
        return None

    # First filtered_df is the full data frame.
    # Then it is subsequentially reduced by each condition
    # at the end it is just those rows that match all conditions.
    filtered_df = self._df if inplace else self._df.copy()

    def list_matches(row: list, values: list) -> bool:
        """Check if any item in the 'values' list is present in the given 'row' list.

        Args:
            row (list):
                A list of items from the data frame column.
            values (list):
                A list of values to check for in the 'row'.

        Returns:
            bool:
                True if any item in 'values' is found in 'row', otherwise False.

        """

        return any(item in values for item in row)

    def dict_matches(row: dict, key: str, values: list) -> bool:
        """Check if the value for the dictionary 'key' is in 'values'.

        Args:
            row (dict):
                A dictionary from the data frame column.
            key (str):
                The key to lookup in the dictionary.
            values (list):
                A list of values to check for in the 'row'.

        Returns:
            bool:
                True, if the value for the dictionary key is in 'values', otherwise False.

        """

        if not row or key not in row:
            return False

        return row[key] in values

    # We traverse a list of conditions. Each condition must evaluate to True
    # otherwise the current workspace or document (i.e. the data set for these objects)
    # will be skipped.
    for condition in conditions:
        # Check if the condition is enabled. If 'enabled' is not
        # in the condition dict then we assume it is enabled.
        if not condition.get("enabled", True):
            continue
        field = condition.get("field", None)
        if not field:
            self.logger.error(
                "Missing value for filter condition 'field' in payload!",
            )
            continue
        if "." in field:
            field, sub = field.split(".", 1)
        else:
            sub = None

        if field not in self._df.columns:
            self.logger.warning(
                "Filter condition field -> '%s' does not exist as column in the data frame! Data frame has these columns -> %s",
                field,
                str(self._df.columns),
            )
            continue  # Skip filtering for columns not present in data frame

        regex = condition.get("regex", False)
        # We need the column to be of type string if we want to use regular expressions
        # so if the column is not yet a string we convert the column to string:
        if regex and filtered_df[field].dtype != "object":
            # Change type of column to string:
            filtered_df[field] = filtered_df[field].astype(str)
            filtered_df[field] = filtered_df[field].fillna("")

        value = condition.get("value", None)
        if value is None:
            # Support alternative syntax using plural.
            value = condition.get("values", None)
        if value is None:
            self.logger.error(
                "Missing filter value(s) for filter condition field -> '%s'!",
                field,
            )
            continue

        # if a single string is passed as value we put
        # it into an 1-item list to simplify the following code:
        if not isinstance(value, list):
            value = [value]

        # If all values in the condition are strings then we
        # want the column also to be of type string:
        if all(isinstance(v, str) for v in value):
            # Change type of column to string:
            #                filtered_df[field] = filtered_df[field].astype(str)
            #                filtered_df[field] = filtered_df[field].fillna("").astype(str)
            #                filtered_df[field] = filtered_df[field].fillna("")

            # When inplace == True, filtered_df is just a reference to self._df.
            # Using .loc[:, field] ensures that Pandas updates the column correctly in self._df.
            # When inplace == False, filtered_df is a full copy (self._df.copy() above),
            # so modifications remain in filtered_df.
            # .loc[:, field] ensures no SettingWithCopyWarning, since filtered_df is now a separate DataFrame.
            filtered_df.loc[:, field] = filtered_df[field].fillna("").astype(str)

        self.logger.info(
            "Data frame has %s row(s) and %s column(s) before filter -> %s has been applied.",
            str(filtered_df.shape[0]),
            str(filtered_df.shape[1]),
            str(condition),
        )

        # Check if the column is boolean
        if pd.api.types.is_bool_dtype(filtered_df[field]):
            # Convert string representations of booleans to actual booleans
            value = [v.lower() in ["true", "1"] if isinstance(v, str) else bool(v) for v in value]

        # Do we want to test for equalitiy or non-equality?
        # For lists equality means: value is in the list
        # For lists non-equality means: value is NOT in the list
        test_for_equal = condition.get("equal", True)

        # Check if the column contains only lists (every non-empty element in the column is a list).
        # `filtered_df[field]`: Access the column with the name specified in 'field'.
        # `.dropna()`: Drop None or NaN rows for the test.
        # `.apply(lambda x: isinstance(x, list))`: For each element in the column, check if it is a list.
        # `.all()`: Ensure that all elements in the column satisfy the condition of being a list.
        if filtered_df[field].dropna().apply(lambda x: isinstance(x, list)).all():
            if not test_for_equal:
                filtered_df = filtered_df[~filtered_df[field].apply(list_matches, values=value)]
            else:
                filtered_df = filtered_df[filtered_df[field].apply(list_matches, values=value)]
        # Check if the column contains only dictionaries (every non-empty element in the column is a dict).
        # `filtered_df[field]`: Access the column with the name specified in 'field'.
        # `.dropna()`: Drop None or NaN rows for the test.
        # `.apply(lambda x: isinstance(x, dict))`: For each element in the column, check if it is a dict.
        # `.all()`: Ensure that all elements in the column satisfy the condition of being a dictionary.
        elif filtered_df[field].dropna().apply(lambda x: isinstance(x, dict)).all():
            if not sub:
                self.logger.error(
                    "Filtering on dictionary values need a key. This needs to be provided with 'field.key' syntax!",
                )
                continue
            if not test_for_equal:
                filtered_df = filtered_df[~filtered_df[field].apply(dict_matches, key=sub, values=value)]
            else:
                filtered_df = filtered_df[filtered_df[field].apply(dict_matches, key=sub, values=value)]
        # Check if the column has boolean values:
        elif pd.api.types.is_bool_dtype(filtered_df[field]):
            # For a boolean filter we can drop NA values:
            filtered_df = filtered_df.dropna(subset=[field])
            if not test_for_equal:
                filtered_df = filtered_df[~filtered_df[field].isin(value)]
            else:
                filtered_df = filtered_df[filtered_df[field].isin(value)]
        elif not regex:
            if pd.api.types.is_string_dtype(filtered_df[field]):
                filtered_df[field] = filtered_df[field].str.strip()
            if not test_for_equal:
                filtered_df = filtered_df[~filtered_df[field].isin(value)]
            else:
                filtered_df = filtered_df[filtered_df[field].isin(value)]
        else:
            # Create a pure boolean pd.Series as a filter criterium:
            regex_condition = filtered_df[field].str.contains(
                "|".join(value),
                regex=True,
                na=False,
            )
            # Apply the boolean pd.Series named 'regex_condition' as
            # a filter - either non-negated or negated (using ~):
            filtered_df = filtered_df[~regex_condition] if not test_for_equal else filtered_df[regex_condition]

        self.logger.info(
            "Data frame has %s row(s) and %s column(s) after filter -> %s has been applied.",
            str(filtered_df.shape[0]),
            str(filtered_df.shape[1]),
            str(condition),
        )
    # end for condition

    if inplace:
        self._df = filtered_df

        if reset_index:
            self._df.reset_index(inplace=True, drop=True)

    return filtered_df

flatten(parent_field, flatten_fields, concatenator='_')

Flatten a sub-dictionary by copying selected fields to the parent dictionary.

This is e.g. useful for then de-duplicate a data frame. To flatten a data frame makes sense in situation when a column used to have a list of dictionaries and got "exploded" (see explode_and_flatten() method below). In this case the column as dictionary values that then can be flattened.

Parameters:

Name	Type	Description	Default
parent_field	str	Name prefix of the new column in the data frame. The flattened field names are added with a leading underscore.	required
flatten_fields	list	Fields in the dictionary of the source column that are copied as new columns into the data frame.	required
concatenator	str	Character or string used to concatenate the parent field with the flattened field to create a unique name.	'_'

Source code in packages/pyxecm/src/pyxecm/helper/data.py

def flatten(self, parent_field: str, flatten_fields: list, concatenator: str = "_") -> None:
    """Flatten a sub-dictionary by copying selected fields to the parent dictionary.

    This is e.g. useful for then de-duplicate a data frame.
    To flatten a data frame makes sense in situation when a column used
    to have a list of dictionaries and got "exploded" (see explode_and_flatten()
    method below). In this case the column as dictionary values that then can
    be flattened.

    Args:
        parent_field (str):
            Name prefix of the new column in the data frame. The flattened field
            names are added with a leading underscore.
        flatten_fields (list):
            Fields in the dictionary of the source column that are copied
            as new columns into the data frame.
        concatenator (str, optional):
            Character or string used to concatenate the parent field with the flattened field
            to create a unique name.

    """

    # First do a sanity check if the data frame is not yet initialized.
    if self._df is None:
        self.logger.error(
            "The data frame is not initialized or empty. Cannot flatten field(s) -> '%s' in the data frame.",
            flatten_fields,
        )
        return

    if parent_field not in self._df.columns:
        self.logger.warning(
            "The parent field -> '%s' cannot be flattened as it doesn't exist as column in the data frame!",
            parent_field,
        )
        return

    for flatten_field in flatten_fields:
        flat_field = parent_field + concatenator + flatten_field
        # The following expression generates a new column in the
        # data frame with the name of 'flat_field'.
        # In the lambda function x is a dictionary that includes the subvalues
        # and it returns the value of the given flatten field
        # (if it exists, otherwise None). So x is self._df[parent_field], i.e.
        # what the lambda function gets 'applied' on.
        self._df[flat_field] = self._df[parent_field].apply(
            lambda x, sub_field=flatten_field: (x.get(sub_field, None) if isinstance(x, dict) else None),
        )

get_columns()

Get the list of column names of the data frame.

Returns:

Type	Description
list | None	list | None: The list of column names in the data frame.

Source code in packages/pyxecm/src/pyxecm/helper/data.py

def get_columns(self) -> list | None:
    """Get the list of column names of the data frame.

    Returns:
        list | None:
            The list of column names in the data frame.

    """

    if self._df is None:
        return None

    return self._df.columns

get_data_frame()

Get the Pandas data frame object.

Returns:

Type	Description
DataFrame	pd.DataFrame: The Pandas data frame object.

Source code in packages/pyxecm/src/pyxecm/helper/data.py

def get_data_frame(self) -> pd.DataFrame:
    """Get the Pandas data frame object.

    Returns:
        pd.DataFrame: The Pandas data frame object.

    """

    return self._df

is_dict_column(column, threshold=0.5)

Safely checks if a column predominantly contains dictionary-like objects.

Parameters:

Name	Type	Description	Default
column	Series	The pandas Series (column) to check.	required
threshold	float	0.0 < threshold <= 1.0. Float representation of the percentage. Default = 0.5 (50%).	0.5

Returns:

Name	Type	Description
bool	bool	True if the column contains mostly dictionary-like objects, False otherwise.

Source code in packages/pyxecm/src/pyxecm/helper/data.py

def is_dict_column(self, column: pd.Series, threshold: float = 0.5) -> bool:
    """Safely checks if a column predominantly contains dictionary-like objects.

    Args:
        column (pd.Series):
            The pandas Series (column) to check.
        threshold (float, optional):
            0.0 < threshold <= 1.0. Float representation of the percentage.
            Default = 0.5 (50%).

    Returns:
        bool:
            True if the column contains mostly dictionary-like objects, False otherwise.

    """

    if not isinstance(column, pd.Series):
        self.logger.error(
            "Expected Pandas series, but got -> %s",
            str(type(column)),
        )
        return False
    if not 0.0 < threshold <= 1.0:
        self.logger.error(
            "Threshold must be between 0.0 and 1.0, but got -> %s",
            str(threshold),
        )
        return False

    # Drop null values (NaN or None) and check types of remaining values
    non_null_values = column.dropna()
    dict_count = non_null_values.apply(lambda x: isinstance(x, dict)).sum()

    # If more than threshold % of non-null values are dictionaries, return True.
    # Else return False.
    return dict_count / len(non_null_values) > threshold if len(non_null_values) > 0 else False

is_list_column(column, threshold=0.5)

Safely checks if a column predominantly contains list-like objects.

Parameters:

Name	Type	Description	Default
column	Series	The pandas Series (column) to check.	required
threshold	float	0.0 < threshold <= 1.0. Float representation of the percentage. Default = 0.5 (50%).	0.5

Returns:

Name	Type	Description
bool	bool	True if the column contains list-like objects, False otherwise.

Source code in packages/pyxecm/src/pyxecm/helper/data.py

def is_list_column(self, column: pd.Series, threshold: float = 0.5) -> bool:
    """Safely checks if a column predominantly contains list-like objects.

    Args:
        column (pd.Series):
            The pandas Series (column) to check.
        threshold (float, optional):
            0.0 < threshold <= 1.0. Float representation of the percentage. Default = 0.5 (50%).

    Returns:
        bool:
            True if the column contains list-like objects, False otherwise.

    """

    if not isinstance(column, pd.Series):
        self.logger.error(
            "Expected pandas series, but got -> %s",
            str(type(column)),
        )
        return False
    if not 0.0 < threshold <= 1.0:
        self.logger.error(
            "Threshold must be between 0.0 and 1.0, but got -> %s",
            str(threshold),
        )
        return False

    # Drop null values (NaN or None) and check types of remaining values
    non_null_values = column.dropna()
    list_count = non_null_values.apply(lambda x: isinstance(x, list)).sum()

    # If more than threshold % of non-null values are lists, return True.
    # Else return False.
    return list_count / len(non_null_values) > threshold if len(non_null_values) > 0 else False

is_string_column(column)

Determine if a Pandas series predominantly contains string values, ignoring NaN values.

Parameters:

Name	Type	Description	Default
column	Series	The Pandas Series to check.	required

Returns:

Name	Type	Description
bool	bool	True if all non-NaN values in the column are strings, False otherwise.

Source code in packages/pyxecm/src/pyxecm/helper/data.py

def is_string_column(self, column: pd.Series) -> bool:
    """Determine if a Pandas series predominantly contains string values, ignoring NaN values.

    Args:
        column (pd.Series):
            The Pandas Series to check.

    Returns:
        bool:
            True if all non-NaN values in the column are strings, False otherwise.

    """

    # Drop NaN values and check if remaining values are strings
    return column.dropna().map(lambda x: isinstance(x, str)).all()

keep_columns(column_names, inplace=True)

Keep only selected columns in the data frame. Drop the rest.

Parameters:

Name	Type	Description	Default
column_names	list	A list of column names to keep.	required
inplace	bool	If the keeping should be inplace, i.e. modifying self._df. Defaults to True.	True

Returns:

Type	Description
DataFrame	pd.DataFrame: New data frame (if inplace = False) or self._df (if inplace = True).

Source code in packages/pyxecm/src/pyxecm/helper/data.py

def keep_columns(self, column_names: list, inplace: bool = True) -> pd.DataFrame:
    """Keep only selected columns in the data frame. Drop the rest.

    Args:
        column_names (list):
            A list of column names to keep.
        inplace (bool, optional):
            If the keeping should be inplace, i.e. modifying self._df.
            Defaults to True.

    Returns:
        pd.DataFrame:
            New data frame (if inplace = False) or self._df (if inplace = True).

    """

    if not all(column_name in self._df.columns for column_name in column_names):
        # Reduce the column names to those that really exist in the data frame:
        column_names = [column_name for column_name in column_names if column_name in self._df.columns]
        self.logger.info(
            "Reduce columns to keep to these columns -> %s that do exist in the data frame.",
            column_names,
        )

    if inplace:
        # keep only those columns which are in column_names:
        if column_names != []:
            self._df = self._df[column_names]
        return self._df
    else:
        # keep only those columns which are in column_names:
        if column_names != []:
            df = self._df[column_names]
            return df
        return None

load_csv_data(csv_path, delimiter=',', names=None, header=0, usecols=None, encoding='utf-8')

Load CSV (Comma separated values) data into data frame.

Parameters:

Name	Type	Description	Default
csv_path	str	The path to the CSV file.	required
delimiter	str, optional, length = 1	The character used to delimit values. Default is "," (comma).	','
names	list | None	The list of column names. This is useful if file does not have a header line but just the data.	None
header	int | None	The index of the header line. Default is 0 (first line). None indicates that the file does not have a header line	0
usecols	list | None	There are three possible list values types: 1. int: These values are treated as column indices for columns to keep (first column has index 0). 2. str: The names of the columns to keep. For this to work the file needs either a header line (i.e. 'header != None') or the 'names' parameter must be specified. 3. bool: The length of the list must match the number of columns. Only columns that have a value of True are kept.	None
encoding	str	The encoding of the file. Default = "utf-8".	'utf-8'

Returns:

Name	Type	Description
bool	bool	False in case an error occured, True otherwise.

Source code in packages/pyxecm/src/pyxecm/helper/data.py

def load_csv_data(
    self,
    csv_path: str,
    delimiter: str = ",",
    names: list | None = None,
    header: int | None = 0,
    usecols: list | None = None,
    encoding: str = "utf-8",
) -> bool:
    """Load CSV (Comma separated values) data into data frame.

    Args:
        csv_path (str):
            The path to the CSV file.
        delimiter (str, optional, length = 1):
            The character used to delimit values. Default is "," (comma).
        names (list | None, optional):
            The list of column names. This is useful if file does not have a header line
            but just the data.
        header (int | None, optional):
            The index of the header line. Default is 0 (first line). None indicates
            that the file does not have a header line
        usecols (list | None, optional):
            There are three possible list values types:
            1. int:
                These values are treated as column indices for columns to keep
                (first column has index 0).
            2. str:
                The names of the columns to keep. For this to work the file needs
                either a header line (i.e. 'header != None') or the 'names'
                parameter must be specified.
            3. bool:
                The length of the list must match the number of columns. Only
                columns that have a value of True are kept.
        encoding (str, optional):
            The encoding of the file. Default = "utf-8".

    Returns:
        bool:
            False in case an error occured, True otherwise.

    """

    if csv_path.startswith("http"):
        # Download file from remote location specified by the packageUrl
        # this must be a public place without authentication:
        self.logger.debug("Download CSV file from URL -> '%s'.", csv_path)

        try:
            response = requests.get(url=csv_path, timeout=1200)
            response.raise_for_status()
        except requests.exceptions.HTTPError:
            self.logger.error("HTTP error with -> %s", csv_path)
            return False
        except requests.exceptions.ConnectionError:
            self.logger.error("Connection error with -> %s", csv_path)
            return False
        except requests.exceptions.Timeout:
            self.logger.error("Timeout error with -> %s", csv_path)
            return False
        except requests.exceptions.RequestException:
            self.logger.error("Request error with -> %s", csv_path)
            return False

        self.logger.debug(
            "Successfully downloaded CSV file -> %s; status code -> %s",
            csv_path,
            response.status_code,
        )

        # Convert bytes to a string using utf-8 and create a file-like object
        csv_file = StringIO(response.content.decode(encoding))

    elif os.path.exists(csv_path):
        self.logger.debug("Using local CSV file -> '%s'.", csv_path)
        csv_file = csv_path

    else:
        self.logger.error(
            "Missing CSV file -> '%s' you have not specified a valid path!",
            csv_path,
        )
        return False

    # Load data from CSV file or buffer
    try:
        df = pd.read_csv(
            filepath_or_buffer=csv_file,
            delimiter=delimiter,
            names=names,
            header=header,
            usecols=usecols,
            encoding=encoding,
            skipinitialspace=True,
        )
        if self._df is None:
            self._df = df
        else:
            self._df = pd.concat([self._df, df])
    except FileNotFoundError:
        self.logger.error(
            "CSV file -> '%s' not found. Please check the file path.",
            csv_path,
        )
        return False
    except PermissionError:
        self.logger.error(
            "Permission denied to access the CSV file -> '%s'.",
            csv_path,
        )
        return False
    except OSError:
        self.logger.error("An I/O error occurred!")
        return False
    except ValueError:
        self.logger.error("Invalid CSV input in file -> %s", csv_path)
        return False
    except AttributeError:
        self.logger.error("Unexpected data structure in file -> %s", csv_path)
        return False
    except TypeError:
        self.logger.error("Unexpected data type in file -> %s", csv_path)
        return False
    except KeyError:
        self.logger.error("Missing key in CSV data -> %s", csv_path)
        return False

    return True

load_directory(path_to_root)

Load directory structure into Pandas data frame.

Parameters:

Name	Type	Description	Default
path_to_root	str	Path to the root element of the directory structure.	required

Returns:

Name	Type	Description
bool	bool	True = Success, False = Failure

Source code in packages/pyxecm/src/pyxecm/helper/data.py

def load_directory(self, path_to_root: str) -> bool:
    """Load directory structure into Pandas data frame.

    Args:
        path_to_root (str):
            Path to the root element of the directory structure.

    Returns:
        bool: True = Success, False = Failure

    """

    try:
        # Check if the provided path is a directory
        if not os.path.isdir(path_to_root):
            self.logger.error(
                "The provided path -> '%s' is not a valid directory.",
                path_to_root,
            )
            return False

        # Initialize a list to hold file information
        data = []

        # Walk through the directory
        for root, _, files in os.walk(path_to_root):
            for file in files:
                file_path = os.path.join(root, file)
                file_size = os.path.getsize(file_path)
                relative_path = os.path.relpath(file_path, path_to_root)
                path_parts = relative_path.split(os.sep)

                # Create a dictionary with the path parts and file details
                entry = {"level {}".format(i): part for i, part in enumerate(path_parts[:-1], start=1)}

                entry.update(
                    {
                        "filename": path_parts[-1],
                        "size": file_size,
                        "path": path_parts[1:-1],
                        "relative_path": relative_path,
                        "download_dir": root,
                    },
                )
                data.append(entry)

        # Create data frame from list of dictionaries:
        self._df = pd.DataFrame(data)

        # Determine the maximum number of levels
        max_levels = max((len(entry) - 2 for entry in data), default=0)

        # Ensure all entries have the same number of levels:
        for entry in data:
            for i in range(1, max_levels + 1):
                entry.setdefault("level {}".format(i), "")

        # Convert to data frame again to make sure all columns are consistent:
        self._df = pd.DataFrame(data)

    except NotADirectoryError:
        self.logger.error(
            "Provided path -> '%s' is not a directory!",
            path_to_root,
        )
        return False
    except FileNotFoundError:
        self.logger.error(
            "Provided path -> '%s' does not exist in file system!",
            path_to_root,
        )
        return False
    except PermissionError:
        self.logger.error(
            "Permission error accessing path -> '%s'!",
            path_to_root,
        )
        return False

    return True

load_excel_data(xlsx_path, sheet_names=0, usecols=None, skip_rows=None, header=0, names=None, na_values=None)

Load Excel (xlsx) data into Pandas data frame.

Supports xls, xlsx, xlsm, xlsb, odf, ods and odt file extensions read from a local filesystem or URL. Supports an option to read a single sheet or a list of sheets.

Parameters:

Name	Type	Description	Default
xlsx_path	str	The path to the Excel file to load.	required
sheet_names	list | str | int	Name or Index of the sheet in the Excel workbook to load. If 'None' then all sheets will be loaded. If 0 then first sheet in workbook will be loaded (this is the Default). If string then this is interpreted as the name of the sheet to load. If a list is passed, this can be a list of index values (int) or a list of strings with the sheet names to load.	0
usecols	list | str	A list of columns to load, specified by general column names in Excel, e.g. usecols='B:D', usecols=['A', 'C', 'F']	None
skip_rows	int	List of rows to skip on top of the sheet (e.g. to not read headlines)	None
header	int | None	Excel Row (0-indexed) to use for the column labels of the parsed data frame. If file contains no header row, then you should explicitly pass header=None. Default is 0.	0
names	list	A list of column names to use. Default is None.	None
na_values	list	A list of values in the Excel that should become the Pandas NA value.	None

Returns:

Name	Type	Description
bool	bool	False in case an error occured, True otherwise.

Source code in packages/pyxecm/src/pyxecm/helper/data.py

def load_excel_data(
    self,
    xlsx_path: str,
    sheet_names: str | list | None = 0,
    usecols: str | list | None = None,
    skip_rows: int | None = None,
    header: int | None = 0,
    names: list | None = None,
    na_values: list | None = None,
) -> bool:
    """Load Excel (xlsx) data into Pandas data frame.

    Supports xls, xlsx, xlsm, xlsb, odf, ods and odt file extensions
    read from a local filesystem or URL. Supports an option to read a
    single sheet or a list of sheets.

    Args:
        xlsx_path (str):
            The path to the Excel file to load.
        sheet_names (list | str | int, optional):
            Name or Index of the sheet in the Excel workbook to load.
            If 'None' then all sheets will be loaded.
            If 0 then first sheet in workbook will be loaded (this is the Default).
            If string then this is interpreted as the name of the sheet to load.
            If a list is passed, this can be a list of index values (int) or
            a list of strings with the sheet names to load.
        usecols (list | str, optional):
            A list of columns to load, specified by general column names in Excel,
            e.g. usecols='B:D', usecols=['A', 'C', 'F']
        skip_rows (int, optional):
            List of rows to skip on top of the sheet (e.g. to not read headlines)
        header (int | None, optional):
            Excel Row (0-indexed) to use for the column labels of the parsed data frame.
            If file contains no header row, then you should explicitly pass header=None.
            Default is 0.
        names (list, optional):
            A list of column names to use. Default is None.
        na_values (list, optional):
            A list of values in the Excel that should become the Pandas NA value.

    Returns:
        bool:
            False in case an error occured, True otherwise.

    """

    if xlsx_path is not None and os.path.exists(xlsx_path):
        # Load data from Excel file
        try:
            df = pd.read_excel(
                io=xlsx_path,
                sheet_name=sheet_names,
                usecols=usecols,
                skiprows=skip_rows,
                header=header,
                names=names,
                na_values=na_values,
            )
            # If multiple sheets from an Excel workbook are loaded,
            # then read_excel() returns a dictionary. The keys are
            # the names of the sheets and the values are the data frames.
            # As this class can only handle one data frame per object,
            # We handle this case by concatenating the different sheets.
            # If you don't want this make sure your Excel workbook has only
            # one sheet or use the "sheet_name" parameter to select the one(s)
            # you want to load.
            if isinstance(df, dict):
                self.logger.info("Loading multiple Excel sheets from the workbook!")
                multi_sheet_df = pd.DataFrame()
                for sheet in df:
                    multi_sheet_df = pd.concat(
                        [multi_sheet_df, df[sheet]],
                        ignore_index=True,
                    )
                df = multi_sheet_df
            if self._df is None:
                self._df = df
            else:
                self._df = pd.concat([self._df, df], ignore_index=True)
        except FileNotFoundError:
            self.logger.error(
                "Excel file -> '%s' not found. Please check the file path.",
                xlsx_path,
            )
            return False
        except PermissionError:
            self.logger.error(
                "Missing permission to access the Excel file -> '%s'.",
                xlsx_path,
            )
            return False
        except OSError:
            self.logger.error(
                "An I/O error occurred while reading the Excel file -> '%s'",
                xlsx_path,
            )
            return False
        except ValueError:
            self.logger.error(
                "Invalid Excel input in file -> '%s'",
                xlsx_path,
            )
            return False
        except AttributeError:
            self.logger.error("Unexpected data structure in file -> %s", xlsx_path)
            return False
        except TypeError:
            self.logger.error("Unexpected data type in file -> %s", xlsx_path)
            return False
        except KeyError:
            self.logger.error("Missing key in Excel data in file -> %s", xlsx_path)
            return False

    else:
        self.logger.error(
            "Missing Excel file -> '%s'. You have not specified a valid path!",
            xlsx_path,
        )
        return False

    return True

load_json_data(json_path, convert_dates=False, index_column=None, compression=None)

Load JSON data into a Pandas data frame.

Parameters:

Name	Type	Description	Default
json_path	str	The path to the JSON file.	required
convert_dates	bool	Defines whether or not dates should be converted. The default is False = dates are NOT converted.	False
index_column	str | None	The Name of the column (i.e. JSON data field) that should become the index in the loaded data frame.	None
compression	str | None	Remove a compression: * gzip (.gz) * bz2 (.bz2) * zip (.zip) * xz (.xz) The value for compression should not include the dot. Default is None = no compression.	None

Returns:

Name	Type	Description
bool	bool	False in case an error occured, True otherwise.

Source code in packages/pyxecm/src/pyxecm/helper/data.py

def load_json_data(
    self,
    json_path: str,
    convert_dates: bool = False,
    index_column: str | None = None,
    compression: str | None = None,
) -> bool:
    """Load JSON data into a Pandas data frame.

    Args:
        json_path (str):
            The path to the JSON file.
        convert_dates (bool, optional):
            Defines whether or not dates should be converted.
            The default is False = dates are NOT converted.
        index_column (str | None, optional):
            The Name of the column (i.e. JSON data field) that should
            become the index in the loaded data frame.
        compression (str | None):
            Remove a compression:
            * gzip (.gz)
            * bz2 (.bz2)
            * zip (.zip)
            * xz (.xz)
            The value for compression should not include the dot.
            Default is None = no compression.

    Returns:
        bool: False in case an error occured, True otherwise.

    """

    if not json_path:
        self.logger.error(
            "You have not specified a JSON path!",
        )
        return False

    # If compression is enabled the file path should have
    # the matching file name extension:
    if compression:
        compression = compression.lstrip(".")  # remove a dot prefix if present
        suffix = "." + compression if compression != "gzip" else "gz"
        if not json_path.endswith(suffix):
            json_path += suffix

    if not os.path.exists(json_path):
        self.logger.error(
            "Missing JSON file - you have not specified a valid path -> '%s'.",
            json_path,
        )
        return False

    # Load data from JSON file
    try:
        df = pd.read_json(
            path_or_buf=json_path,
            convert_dates=convert_dates,
            compression=compression,
        )

        if index_column and index_column not in df.columns:
            self.logger.error(
                "Specified index column -> '%s' not found in the JSON data.",
                index_column,
            )
            return False

        if index_column:
            df = df.set_index(keys=index_column)
        if self._df is None:
            self._df = df
        else:
            self._df = pd.concat([self._df, df])
        self.logger.info(
            "After loading JSON file -> '%s', the data frame has %s row(s) and %s column(s)",
            json_path,
            self._df.shape[0],
            self._df.shape[1],
        )
    except FileNotFoundError:
        self.logger.error(
            "JSON file -> '%s' not found. Please check the file path.",
            json_path,
        )
        return False
    except PermissionError:
        self.logger.error(
            "Missing permission to access the JSON file -> '%s'.",
            json_path,
        )
        return False
    except OSError:
        self.logger.error("An I/O error occurred!")
        return False
    except json.JSONDecodeError:
        self.logger.error(
            "Unable to decode JSON file -> '%s'",
            json_path,
        )
        return False
    except ValueError:
        self.logger.error("Invalid JSON input -> %s", json_path)
        return False
    except AttributeError:
        self.logger.error("Unexpected JSON data structure in file -> %s", json_path)
        return False
    except TypeError:
        self.logger.error("Unexpected JSON data type in file -> %s", json_path)
        return False
    except KeyError:
        self.logger.error("Missing key in JSON data in file -> %s", json_path)
        return False

    return True

load_web(values, value_name, url_templates, special_values=None, special_url_templates=None, pattern='')

Traverse years and bulletin types to collect all bulletin URLs.

Parameters:

Name	Type	Description	Default
values	list	List of values to travers over	required
value_name	str	Dictionary key to construct an item in combination with a value from values	required
url_templates	list	URLs to travers per value. The URLs should contain one {} that is replace by the current value.	required
special_values	list | None	List of vales (a subset of the other values list) that we want to handle in a special way. Defaults to None.	None
special_url_templates	dict | None	URLs for the special values. Defaults to None. The dictionary keys are the special values. The dictionary values are lists of special URLs with placeholders.	None
pattern	str	Regular expression to find the proper links on the page. Defaults to r"".	''

Returns:

Name	Type	Description
bool	bool	True for success, False in case of an error.

Source code in packages/pyxecm/src/pyxecm/helper/data.py

def load_web(
    self,
    values: list,
    value_name: str,
    url_templates: list,
    special_values: list | None = None,
    special_url_templates: dict | None = None,
    pattern: str = r"",
) -> bool:
    """Traverse years and bulletin types to collect all bulletin URLs.

    Args:
        values (list):
            List of values to travers over
        value_name (str):
            Dictionary key to construct an item in combination with a value from values
        url_templates (list):
            URLs to travers per value. The URLs should contain one {} that is
            replace by the current value.
        special_values (list | None, optional):
            List of vales (a subset of the other values list)
            that we want to handle in a special way. Defaults to None.
        special_url_templates (dict | None, optional):
            URLs for the special values. Defaults to None.
            The dictionary keys are the special values. The
            dictionary values are lists of special URLs with placeholders.
        pattern (str, optional):
            Regular expression to find the proper links on the page. Defaults to r"".

    Returns:
        bool:
            True for success, False in case of an error.

    """

    result_list = []

    # We have two nested for loops below. The out traverses over all placeholder values.
    # These could be the calendar years, e.g. [2003,...,2024]
    # The inner for loop traverses over the list of specified URLs. We can have multiple for
    # each value.

    # Do we have a list of placeholder values we want to iterate over?
    if values:
        # Traverse all values in the values list:
        for value in values:
            # Do we want a special treatment for this value (e.g. the current year)
            if value in special_values:
                self.logger.debug("Processing special value -> '%s'...", value)
                if value not in special_url_templates and str(value) not in special_url_templates:
                    self.logger.error(
                        "Cannot find key -> '%s' in special URL templates dictionary -> %s! Skipping...",
                        value,
                        str(special_url_templates),
                    )
                    continue
                # If the dictionary uses string keys then we need to convert the value
                # to a string as well to avoid key errors:
                if str(value) in special_url_templates:
                    value = str(value)
                special_url_template_list = special_url_templates[value]
                for special_url_template in special_url_template_list:
                    # Now the value is inserted into the placeholder in the URL:
                    special_url = special_url_template.format(value)
                    common_data = {value_name: value} if value_name else None
                    result_list += self.load_web_links(
                        url=special_url,
                        common_data=common_data,
                        pattern=pattern,
                    )
            else:  # normal URLs
                self.logger.info("Processing value -> '%s'...", value)
                for url_template in url_templates:
                    # Now the value is inserted into the placeholder in the URL:
                    url = url_template.format(value)
                    common_data = {value_name: value} if value_name else None
                    result_list += self.load_web_links(
                        url=url,
                        common_data=common_data,
                        pattern=pattern,
                    )
    else:
        for url_template in url_templates:
            url = url_template.format(value)
            result_list += self.load_web_links(
                url=url,
                common_data=None,
                pattern=pattern,
            )

    # Add the data list to the data frame:
    self.append(result_list)

    return True

load_web_links(url, common_data=None, pattern='')

Get all linked file URLs on a given web page (url) that are following a given pattern.

Construct a list of dictionaries based on this. This method is a helper method for load_web() below.

Parameters:

Name	Type	Description	Default
url	str	The web page URL.	required
common_data	dict | None	Fields that should be added to each dictionary item. Defaults to None.	None
pattern	str	Regular Expression. Defaults to r"".	''

Returns:

Type	Description
list | None	list | None: List of links on the web page that are complying with the given regular expression.

Source code in packages/pyxecm/src/pyxecm/helper/data.py

def load_web_links(
    self,
    url: str,
    common_data: dict | None = None,
    pattern: str = r"",
) -> list | None:
    """Get all linked file URLs on a given web page (url) that are following a given pattern.

    Construct a list of dictionaries based on this. This method is a helper method for load_web() below.

    Args:
        url (str):
            The web page URL.
        common_data (dict | None, optional):
            Fields that should be added to each dictionary item. Defaults to None.
        pattern (str, optional):
            Regular Expression. Defaults to r"".

    Returns:
        list | None:
            List of links on the web page that are complying with the given regular expression.

    """

    try:
        response = requests.get(url, timeout=300)
        response.raise_for_status()
    except requests.RequestException:
        self.logger.error("Failed to retrieve page at %s", url)
        return []

    # Find all file links (hyperlinks) on the page (no file extension assumed)
    # Example filename pattern: "al022023.public.005"
    file_links = re.findall(r'href="([^"]+)"', response.text)
    if not file_links:
        self.logger.warning("No file links found on the web page -> %s", url)
        return []

    result_list = []
    base_url = url if url.endswith("/") else url + "/"

    for link in file_links:
        data = common_data.copy() if common_data else {}

        # Construct the full URL
        full_url = base_url + link.lstrip("/")

        if pattern:
            # Filter by expected naming pattern for links
            match = re.search(pattern, link)
            if not match:
                continue

            # Extract and assign groups if they exist
            # TODO(mdiefenb): these names are currently hard-coded
            # for the National Hurricane Center Dataset (NHC)
            if len(match.groups()) >= 1:
                data["Code"] = match.group(1).upper()
            if len(match.groups()) >= 2:
                data["Type"] = match.group(2)
            if len(match.groups()) >= 3:
                data["Message ID"] = match.group(3)

        data["URL"] = full_url
        data["Filename"] = link

        result_list.append(data)

    return result_list

load_xml_data(xml_path, xpath=None, xslt_path=None, encoding='utf-8')

Load XML data into a Pandas data frame.

Parameters:

Name	Type	Description	Default
xml_path	str	The path to the XML file to load.	required
xpath	str	An XPath to the elements we want to select.	None
xslt_path	str	An XSLT transformation file to convert the XML data.	None
encoding	str	The encoding of the file. Default is UTF-8.	'utf-8'

Returns:

Name	Type	Description
bool	bool	False in case an error occured, True otherwise.

Source code in packages/pyxecm/src/pyxecm/helper/data.py

def load_xml_data(
    self,
    xml_path: str,
    xpath: str | None = None,
    xslt_path: str | None = None,
    encoding: str = "utf-8",
) -> bool:
    """Load XML data into a Pandas data frame.

    Args:
        xml_path (str):
            The path to the XML file to load.
        xpath (str, optional):
            An XPath to the elements we want to select.
        xslt_path (str, optional):
            An XSLT transformation file to convert the XML data.
        encoding (str, optional):
            The encoding of the file. Default is UTF-8.

    Returns:
        bool:
            False in case an error occured, True otherwise.

    """

    if xml_path.startswith("http"):
        # Download file from remote location specified by the packageUrl
        # this must be a public place without authentication:
        self.logger.debug("Download XML file from URL -> '%s'.", xml_path)

        try:
            response = requests.get(url=xml_path, timeout=1200)
            response.raise_for_status()
        except requests.exceptions.HTTPError:
            self.logger.error("HTTP error with -> %s", xml_path)
            return False
        except requests.exceptions.ConnectionError:
            self.logger.error("Connection error with -> %s", xml_path)
            return False
        except requests.exceptions.Timeout:
            self.logger.error("Timeout error with -> %s", xml_path)
            return False
        except requests.exceptions.RequestException:
            self.logger.error("Request error with -> %s", xml_path)
            return False

        self.logger.debug(
            "Successfully downloaded XML file -> '%s'; status code -> %s",
            xml_path,
            response.status_code,
        )
        # Convert bytes to a string using utf-8 and create a file-like object
        xml_file = StringIO(response.content.decode(encoding))

    elif os.path.exists(xml_path):
        self.logger.debug("Using local XML file -> '%s'.", xml_path)
        xml_file = xml_path

    else:
        self.logger.error(
            "Missing XML file -> '%s'. You have not specified a valid path or URL!",
            xml_path,
        )
        return False

    # Load data from XML file or buffer
    try:
        df = pd.read_xml(
            path_or_buffer=xml_file,
            xpath=xpath,
            stylesheet=xslt_path,
            encoding=encoding,
        )
        # Process the loaded data as needed
        if self._df is None:
            self._df = df
        else:
            self._df = pd.concat([self._df, df])
        self.logger.info("XML file -> '%s' loaded successfully!", xml_path)
    except FileNotFoundError:
        self.logger.error("XML file -> '%s' not found.", xml_path)
        return False
    except PermissionError:
        self.logger.error(
            "Missing permission to access the XML file -> '%s'.",
            xml_path,
        )
        return False
    except OSError:
        self.logger.error("An I/O error occurred loading from -> %s", xml_path)
        return False
    except ValueError:
        self.logger.error("Invalid XML data in file -> %s", xml_path)
        return False
    except AttributeError:
        self.logger.error("Unexpected data structure in XML file -> %s", xml_path)
        return False
    except TypeError:
        self.logger.error("Unexpected data type in XML file -> %s", xml_path)
        return False
    except KeyError:
        self.logger.error("Missing key in XML file -> %s", xml_path)
        return False

    return True

load_xml_directory(path_to_root, xpath=None, xml_files=None)

Load XML files from a directory structure into Pandas data frame.

Parameters:

Name	Type	Description	Default
path_to_root	str	Path to the root element of the directory structure.	required
xpath	str	XPath to the XML elements we want to select.	None
xml_files	list | None	Names of the XML files to load from the directory.	None

Returns:

Name	Type	Description
bool	bool	True = Success, False = Failure

Source code in packages/pyxecm/src/pyxecm/helper/data.py

def load_xml_directory(
    self,
    path_to_root: str,
    xpath: str | None = None,
    xml_files: list | None = None,
) -> bool:
    """Load XML files from a directory structure into Pandas data frame.

    Args:
        path_to_root (str):
            Path to the root element of the directory structure.
        xpath (str, optional):
            XPath to the XML elements we want to select.
        xml_files (list | None, optional):
            Names of the XML files to load from the directory.

    Returns:
        bool:
            True = Success, False = Failure

    """

    # Establish a default if None is passed via the parameter:
    if not xml_files:
        xml_files = ["docovw.xml"]

    try:
        # Check if the provided path is a directory
        if not os.path.isdir(path_to_root):
            self.logger.error(
                "The provided path -> '%s' is not a valid directory.",
                path_to_root,
            )
            return False

        # Walk through the directory
        for root, _, files in os.walk(path_to_root):
            for file in files:
                file_path = os.path.join(root, file)
                file_size = os.path.getsize(file_path)
                file_name = os.path.basename(file_path)

                if file_name in xml_files:
                    self.logger.info(
                        "Load XML file -> '%s' of size -> %s from -> '%s'...",
                        file_name,
                        file_size,
                        file_path,
                    )
                    success = self.load_xml_data(file_path, xpath=xpath)
                    if success:
                        self.logger.info(
                            "Successfully loaded XML file -> '%s'.",
                            file_path,
                        )

    except NotADirectoryError:
        self.logger.error(
            "Provided path -> '%s' is not a directory",
            path_to_root,
        )
        return False
    except FileNotFoundError:
        self.logger.error(
            "Provided path -> '%s' does not exist in file system!",
            path_to_root,
        )
        return False
    except PermissionError:
        self.logger.error(
            "Missing permission to access path -> '%s'",
            path_to_root,
        )
        return False

    return True

lock()

Return the threading lock object.

Returns:

Type	Description
Lock	threading.Lock: The threading lock object.

Source code in packages/pyxecm/src/pyxecm/helper/data.py

def lock(self) -> threading.Lock:
    """Return the threading lock object.

    Returns:
        threading.Lock: The threading lock object.

    """

    return self._lock

lookup_value(lookup_column, lookup_value, separator='|', single_row=True)

Lookup row(s) that includes a lookup value in the value of a given column.

Parameters:

Name	Type	Description	Default
lookup_column	str	The name of the column to search in.	required
lookup_value	str	The value to search for.	required
separator	str	The string list delimiter / separator. The pipe symbol | is the default as it is unlikely to appear in a normal string (other than a plain comma). The separator is NOT looked for in the lookup_value but in the column that is given by lookup_column!	'|'
single_row	bool	This defines if we just return the first matching row if multiple matching rows are found. Default is True (= single row).	True

Returns:

Type	Description
Series | DataFrame | None	pd.Series | pd.DataFrame | None: Data frame (multiple rows) or Series (row) that matches the lookup value. None if no match was found.

Source code in packages/pyxecm/src/pyxecm/helper/data.py

def lookup_value(
    self,
    lookup_column: str,
    lookup_value: str,
    separator: str = "|",
    single_row: bool = True,
) -> pd.Series | pd.DataFrame | None:
    """Lookup row(s) that includes a lookup value in the value of a given column.

    Args:
        lookup_column (str):
            The name of the column to search in.
        lookup_value (str):
            The value to search for.
        separator (str):
            The string list delimiter / separator. The pipe symbol | is the default
            as it is unlikely to appear in a normal string (other than a plain comma).
            The separator is NOT looked for in the lookup_value but in the column that
            is given by lookup_column!
        single_row (bool, optional):
            This defines if we just return the first matching row if multiple matching rows
            are found. Default is True (= single row).

    Returns:
        pd.Series | pd.DataFrame | None:
            Data frame (multiple rows) or Series (row) that matches the lookup value.
            None if no match was found.

    """

    # Use the `apply` function to filter rows where the lookup value matches a
    # whole item in the separator-divided list:
    def match_lookup_value(string_list: str | None) -> bool:
        """Check if the lookup value is in a string list.

        For this the string list is converted to a python
        list. A separator is used for the splitting.

        Args:
            string_list (str):
                Delimiter-separated string list like "a, b, c" or "a | b | c"

        Returns:
            bool:
                True if lookup_value is equal to one of the delimiter-separated terms.

        """

        if pd.isna(string_list):  # Handle None/NaN safely
            return False

        # Ensure that the string is a string
        string_list = str(string_list)

        return lookup_value in [item.strip() for item in string_list.split(separator)]

    # end method definition

    if self._df is None:
        return None

    df = self._df

    if lookup_column not in self._df.columns:
        self.logger.error(
            "Cannot lookup value in column -> '%s'. Column does not exist in the data frame! Data frame has these columns -> %s",
            lookup_column,
            str(self._df.columns),
        )
        return None

    # Fill NaN or None values in the lookup column with empty strings
    # df[lookup_column] = df[lookup_column].fillna("")

    # Use the `apply` function to filter rows where the lookup value is in row cell
    # of column given by lookup_column. match_lookup_value() is called with
    # the content of the individual cell contents:
    matched_rows = df[df[lookup_column].apply(match_lookup_value)]

    # If nothing was found we return None:
    if matched_rows.empty:
        return None

    # If it is OK to have multiple matches (= multiple rows = pd.DataFrame).
    # We can just return the matched_rows now which should be a pd.DataFrame:
    if not single_row:
        return matched_rows

    # Check if more than one row matches, and log a warning if so
    if len(matched_rows) > 1:
        self.logger.warning(
            "More than one match found for lookup value -> '%s' in column -> '%s'. Returning the first match.",
            lookup_value,
            lookup_column,
        )

    # Return the first matched row, if any
    return matched_rows.iloc[0]

merge(merge_data, on=None, how='inner', left_on=None, right_on=None, left_index=False, right_index=False, suffixes=('_x', '_y'), indicator=False, validate=None)

Merge the current DataFrame (_df) with another DataFrame.

Parameters:

Name	Type	Description	Default
merge_data	DataFrame | Data	The DataFrame to merge with.	required
on	str | list[str]	Column(s) to merge on. Defaults to None.	None
how	str	Type of merge ('inner', 'outer', 'left', 'right', 'cross'). Defaults to 'inner'.	'inner'
left_on	str | list[str] | None	Column(s) from self._df to merge on. Defaults to None.	None
right_on	str | list[str] | None	Column(s) from other DataFrame to merge on. Defaults to None.	None
left_index	str | list[str]	Whether to merge on the index of self._df. Defaults to False.	False
right_index	bool	Whether to merge on the index of other. Defaults to False.	False
suffixes	tuple[str, str]	Suffixes for overlapping column names. Defaults to ('_x', '_y').	('_x', '_y')
indicator	bool	If True, adds a column showing the merge source. Defaults to False.	False
validate		If provided, checks merge integrity ('one_to_one', 'one_to_many', 'many_to_one', 'many_to_many'). Defaults to None.	None

Returns:

Type	Description
DataFrame | None	The merged DataFrame or None in case of an error.

Raises:

Type	Description
ValueError	If other is not a DataFrame.
KeyError	If required columns for merging are missing.
ValueError	If validate check fails.

Source code in packages/pyxecm/src/pyxecm/helper/data.py

def merge(
    self,
    merge_data: pd.DataFrame,
    on: str | list[str] | None = None,
    how: str = "inner",
    left_on: str | list[str] | None = None,
    right_on: str | list[str] | None = None,
    left_index: bool = False,
    right_index: bool = False,
    suffixes: tuple[str, str] = ("_x", "_y"),
    indicator: bool = False,
    validate: str | None = None,
) -> pd.DataFrame | None:
    """Merge the current DataFrame (_df) with another DataFrame.

    Args:
        merge_data (pd.DataFrame | Data):
            The DataFrame to merge with.
        on (str | list[str]):
            Column(s) to merge on. Defaults to None.
        how (str, optional):
            Type of merge ('inner', 'outer', 'left', 'right', 'cross'). Defaults to 'inner'.
        left_on (str | list[str] | None, optional):
            Column(s) from self._df to merge on. Defaults to None.
        right_on (str | list[str] | None, optional):
            Column(s) from other DataFrame to merge on. Defaults to None.
        left_index (str | list[str], optional):
             Whether to merge on the index of self._df. Defaults to False.
        right_index (bool, optional):
            Whether to merge on the index of other. Defaults to False.
        suffixes (tuple[str, str]):
            Suffixes for overlapping column names. Defaults to ('_x', '_y').
        indicator (bool, optional):
            If True, adds a column showing the merge source. Defaults to False.
        validate ():
            If provided, checks merge integrity
            ('one_to_one', 'one_to_many', 'many_to_one', 'many_to_many'). Defaults to None.

    Returns:
        The merged DataFrame or None in case of an error.

    Exceptions:
        ValueError: If `other` is not a DataFrame.
        KeyError: If required columns for merging are missing.
        ValueError: If `validate` check fails.

    """

    if self._df is None or self._df.empty:
        self._df = merge_data

    if isinstance(merge_data, Data):
        merge_data = merge_data.get_data_frame()  # Extract DataFrame from Data instance

    try:
        return self._df.merge(
            merge_data,
            how=how,
            on=on,
            left_on=left_on,
            right_on=right_on,
            left_index=left_index,
            right_index=right_index,
            suffixes=suffixes,
            indicator=indicator,
            validate=validate,
        )
    except KeyError:
        self.logger.error("Column(s) not found for merging!")
    except ValueError:
        self.logger.error("Invalid merge operation!")

    return None

partitionate(number)

Partition a data frame into equally sized partitions.

Parameters:

Name	Type	Description	Default
number	int	The number of desired partitions.	required

Returns:

Name	Type	Description
list	list	A list of created partitions.

Source code in packages/pyxecm/src/pyxecm/helper/data.py

def partitionate(self, number: int) -> list:
    """Partition a data frame into equally sized partitions.

    Args:
        number (int):
            The number of desired partitions.

    Returns:
        list:
            A list of created partitions.

    """

    # Calculate the approximate size of each partition
    size = len(self._df)

    if size >= number:
        partition_size = size // number
        remainder = size % number
    else:
        partition_size = size
        number = 1
        remainder = 0

    self.logger.info(
        "Data frame has -> %s elements. We split it into -> %s partitions with -> %s row%s and remainder -> %s...",
        str(size),
        str(number),
        str(partition_size),
        "s" if partition_size > 1 else "",
        str(remainder),
    )

    # Initialize a list to store partitions:
    partitions = []
    start_index = 0

    # Slice the data frame into equally sized partitions:
    for i in range(number):
        # Calculate the end index for this partition
        end_index = start_index + partition_size + (1 if i < remainder else 0)
        partition = self._df.iloc[start_index:end_index]
        partitions.append(partition)
        start_index = end_index

    return partitions

partitionate_by_column(column_name)

Partition a data frame based on equal values in a specified column.

Parameters:

Name	Type	Description	Default
column_name	str	The column name to partition by.	required

Returns:

Type	Description
list | None	list | None: List of partitions or None in case of an error (e.g. column name does not exist).

Source code in packages/pyxecm/src/pyxecm/helper/data.py

def partitionate_by_column(self, column_name: str) -> list | None:
    """Partition a data frame based on equal values in a specified column.

    Args:
        column_name (str):
            The column name to partition by.

    Returns:
        list | None:
            List of partitions or None in case of an error (e.g. column name does not exist).

    """

    if column_name not in self._df.columns:
        self.logger.error(
            "Cannot partitionate by column -> '%s'. Column does not exist in the data frame. Data frame has these columns -> %s",
            column_name,
            str(self._df.columns),
        )
        return None

    # Separate rows with NaN or None values in the specified column:
    nan_partitions = self._df[self._df[column_name].isna()]

    # Keep only rows where the specified column has valid (non-NaN) values:
    non_nan_df = self._df.dropna(subset=[column_name])

    # Group the non-NaN DataFrame by the specified column's values:
    grouped = non_nan_df.groupby(column_name)

    # Create a list of partitions (DataFrames) for each unique value in the column:
    partitions = [group for _, group in grouped]

    # Add each row with NaN/None as its own partition
    # iterrows() returns each row as a Series. To convert it back to a DataFrame:
    # 1. .to_frame() turns the Series into a DataFrame, but with the original column names as rows.
    # 2. .T (transpose) flips it back, turning the original row into a proper DataFrame row.
    # This ensures that even rows with NaN values are treated as DataFrame partitions.
    partitions.extend([row.to_frame().T for _, row in nan_partitions.iterrows()])

    self.logger.info(
        "Data frame has been partitioned into -> %s partitions based on the values in column -> '%s'...",
        str(len(partitions)),
        column_name,
    )

    return partitions

print_info(show_size=True, show_info=False, show_columns=False, show_first=False, show_last=False, show_sample=False, show_statistics=False, row_num=10)

Log information about the data frame.

Parameters:

Name	Type	Description	Default
show_size	bool	Show size of data frame. Defaults to True.	True
show_info	bool	Show information for data frame. Defaults to False.	False
show_columns	bool	Show columns of data frame. Defaults to False.	False
show_first	bool	Show first N items. Defaults to False. N is defined by the row_num parameter.	False
show_last	bool	Show last N items. Defaults to False. N is defined by the row_num parameter.	False
show_sample	bool	Show N sample items. Defaults to False. N is defined by the row_num parameter.	False
show_statistics	bool	Show data frame statistics. Defaults to False.	False
row_num	int	Used as the number of rows printed using show_first, show_last, show_sample. Default is 10.	10

Source code in packages/pyxecm/src/pyxecm/helper/data.py

def print_info(
    self,
    show_size: bool = True,
    show_info: bool = False,
    show_columns: bool = False,
    show_first: bool = False,
    show_last: bool = False,
    show_sample: bool = False,
    show_statistics: bool = False,
    row_num: int = 10,
) -> None:
    """Log information about the data frame.

    Args:
        show_size (bool, optional):
            Show size of data frame. Defaults to True.
        show_info (bool, optional):
            Show information for data frame. Defaults to False.
        show_columns (bool, optional):
            Show columns of data frame. Defaults to False.
        show_first (bool, optional):
            Show first N items. Defaults to False. N is defined
            by the row_num parameter.
        show_last (bool, optional):
            Show last N items. Defaults to False. N is defined
            by the row_num parameter.
        show_sample (bool, optional):
            Show N sample items. Defaults to False. N is defined
            by the row_num parameter.
        show_statistics (bool, optional):
            Show data frame statistics. Defaults to False.
        row_num (int, optional):
            Used as the number of rows printed using show_first,
            show_last, show_sample. Default is 10.

    """

    if self._df is None:
        self.logger.warning("Data frame is not initialized!")
        return

    if show_size:
        self.logger.info(
            "Data frame has %s row(s) and %s column(s)",
            self._df.shape[0],
            self._df.shape[1],
        )

    if show_info:
        # df.info() can not easily be embedded into a string
        self._df.info()

    if show_columns:
        self.logger.info("Columns:\n%s", self._df.columns)
        self.logger.info(
            "Columns with number of NaN values:\n%s",
            self._df.isna().sum(),
        )
        self.logger.info(
            "Columns with number of non-NaN values:\n%s",
            self._df.notna().sum(),
        )

    if show_first:
        # the default for head is n = 5:
        self.logger.info("First %s rows:\n%s", str(row_num), self._df.head(row_num))

    if show_last:
        # the default for tail is n = 5:
        self.logger.info("Last %s rows:\n%s", str(row_num), self._df.tail(row_num))

    if show_sample:
        # the default for sample is n = 1:
        self.logger.info(
            "%s Sample rows:\n%s",
            str(row_num),
            self._df.sample(n=row_num),
        )

    if show_statistics:
        self.logger.info(
            "Description of statistics for data frame:\n%s",
            self._df.describe(),
        )
        self.logger.info(
            "Description of statistics for data frame (transformed):\n%s",
            self._df.describe().T,
        )
        self.logger.info(
            "Description of statistics for data frame (objects):\n%s",
            self._df.describe(include="object"),
        )

rename_column(old_column_name, new_column_name)

Rename a data frame column.

Parameters:

Name	Type	Description	Default
old_column_name	str	The old name of the column.	required
new_column_name	str	The new name of the column.	required

Returns:

Name	Type	Description
bool	bool	True = Success, False = Error

Source code in packages/pyxecm/src/pyxecm/helper/data.py

def rename_column(self, old_column_name: str, new_column_name: str) -> bool:
    """Rename a data frame column.

    Args:
        old_column_name (str):
            The old name of the column.
        new_column_name (str):
            The new name of the column.

    Returns:
        bool:
            True = Success, False = Error

    """

    if self._df is None:
        return False

    if old_column_name not in self._df.columns:
        self.logger.error(
            "Cannot rename column -> '%s'. It does not exist in the data frame! Data frame has these columns -> %s",
            old_column_name,
            str(self._df.columns),
        )
        return False

    if new_column_name in self._df.columns:
        self.logger.error(
            "Cannot rename column -> '%s' to -> '%s'. New name does already exist as column in the data frame! Data frame has these columns -> %s",
            old_column_name,
            new_column_name,
            str(self._df.columns),
        )
        return False

    self._df.rename(columns={old_column_name: new_column_name}, inplace=True)

    return True

save_excel_data(excel_path, sheet_name='Pandas Export', index=False, columns=None)

Save the data frame to an Excel file, with robust error handling and logging.

Parameters:

Name	Type	Description	Default
excel_path	str	The file path to save the Excel file.	required
sheet_name	str	The sheet name where data will be saved. Default is 'Sheet1'.	'Pandas Export'
index	bool	Whether to write the row names (index). Default is False.	False
columns	list | None	A list of column names to write into the excel file.	None

Returns:

Name	Type	Description
bool	bool	True = success, False = error.

Source code in packages/pyxecm/src/pyxecm/helper/data.py

def save_excel_data(
    self,
    excel_path: str,
    sheet_name: str = "Pandas Export",
    index: bool = False,
    columns: list | None = None,
) -> bool:
    """Save the data frame to an Excel file, with robust error handling and logging.

    Args:
        excel_path (str):
            The file path to save the Excel file.
        sheet_name (str):
            The sheet name where data will be saved. Default is 'Sheet1'.
        index (bool, optional):
            Whether to write the row names (index). Default is False.
        columns (list | None, optional):
            A list of column names to write into the excel file.

    Returns:
        bool:
            True = success, False = error.

    """

    try:
        # Check if the directory exists
        directory = os.path.dirname(excel_path)
        if directory and not os.path.exists(directory):
            os.makedirs(directory)

        # Validate columns if provided
        if columns:
            existing_columns = [col for col in columns if col in self._df.columns]
            missing_columns = set(columns) - set(existing_columns)
            if missing_columns:
                self.logger.warning(
                    "The following columns do not exist in the data frame and cannot be saved to Excel -> %s",
                    ", ".join(missing_columns),
                )
            columns = existing_columns

        # Attempt to save the data frame to Excel:
        if self._df is None:
            self.logger.error(
                "Cannot write Excel file -> '%s' from empty / non-initialized data frame!", excel_path
            )
        self._df.to_excel(
            excel_path,
            sheet_name=sheet_name,
            index=index,
            columns=columns or None,  # Pass None if no columns provided
        )
        self.logger.info(
            "Data frame saved successfully to Excel file -> '%s'.",
            excel_path,
        )

    except FileNotFoundError as fnf_error:
        self.logger.error("Cannot write data frame to Excel file -> '%s'; error -> %s", excel_path, str(fnf_error))
        return False
    except PermissionError as pe:
        self.logger.error("Cannot write data frame to Excel file -> '%s'; error -> %s", excel_path, str(pe))
        return False
    except ValueError as ve:
        self.logger.error("Cannot write data frame to Excel file -> '%s'; error -> %s", excel_path, str(ve))
        return False
    except OSError as ose:
        self.logger.error("Cannot write data frame to Excel file -> '%s'; error -> %s", excel_path, str(ose))
        return False

    return True

save_json_data(json_path, orient='records', preserve_index=False, index_column='index', compression=None)

Save JSON data from data frame to file.

Parameters:

Name	Type	Description	Default
json_path	str	The path to where the JSON file should be safed.	required
orient	str	The structure of the JSON. Possible values: * "records" (this is the default) * "columns" * "index" * "table" * "split"	'records'
preserve_index	bool	Defines if the index column of the data frame should be exported as well. The default is False (index is not exported).	False
index_column	str	The Name of the column (i.e. JSON data field) that should become the index in the loaded data frame. The default is "index".	'index'
compression	str | None	Apply a compression: * gzip (.gz) * bz2 (.bz2) * zip (.zip) * xz (.xz)	None

Returns:

Name	Type	Description
bool	bool	False in case an error occured, True otherwise.

Source code in packages/pyxecm/src/pyxecm/helper/data.py

def save_json_data(
    self,
    json_path: str,
    orient: str = "records",
    preserve_index: bool = False,
    index_column: str = "index",
    compression: str | None = None,
) -> bool:
    """Save JSON data from data frame to file.

    Args:
        json_path (str): The path to where the JSON file should be safed.
        orient (str, optional):
            The structure of the JSON. Possible values:
            * "records" (this is the default)
            * "columns"
            * "index"
            * "table"
            * "split"
        preserve_index (bool, optional):
            Defines if the index column of the data frame should be exported as well.
            The default is False (index is not exported).
        index_column (str, optional):
            The Name of the column (i.e. JSON data field) that should
            become the index in the loaded data frame. The default is "index".
        compression (str | None):
            Apply a compression:
            * gzip (.gz)
            * bz2 (.bz2)
            * zip (.zip)
            * xz (.xz)

    Returns:
        bool:
            False in case an error occured, True otherwise.

    """

    if not json_path:
        self.logger.error(
            "You have not specified a JSON path!",
        )
        return False

    # If compression is enabled the file path should have
    # the matching file name extension:
    if compression:
        suffix = "." + compression if compression != "gzip" else ".gz"
        if not json_path.endswith(suffix):
            json_path += suffix

    # Save data to JSON file
    try:
        if self._df is not None:
            if not os.path.exists(os.path.dirname(json_path)):
                os.makedirs(os.path.dirname(json_path), exist_ok=True)

            # index parameter is only allowed if orient has one of the following values:
            if orient in ("columns", "index", "table", "split"):
                self._df.to_json(
                    path_or_buf=json_path,
                    index=preserve_index,
                    orient=orient,
                    indent=2,
                    compression=compression,
                    date_format="iso",
                )
            # In this case we cannot use the index parameter as this would give this error:
            # Value Error -> 'index=True' is only valid when 'orient' is 'split', 'table', 'index', or 'columns'
            # So we create a new column that preserves the original row IDs from the index. The nasme

            elif preserve_index:
                df_with_index = self._df.reset_index(
                    names=index_column,
                    inplace=False,
                )
                df_with_index.to_json(
                    path_or_buf=json_path,
                    orient=orient,
                    indent=2,
                    compression=compression,
                    date_format="iso",
                )
            else:
                self._df.to_json(
                    path_or_buf=json_path,
                    orient=orient,
                    indent=2,
                    compression=compression,
                    date_format="iso",
                )
        else:
            self.logger.warning(
                "Data frame is empty. Cannot write it to JSON file -> '%s'.",
                json_path,
            )
            return False
    except FileNotFoundError:
        self.logger.error(
            "File -> '%s' not found. Please check the file path.",
            json_path,
        )
        return False
    except PermissionError:
        self.logger.error(
            "Permission denied to access the file -> '%s'.",
            json_path,
        )
        return False
    except OSError:
        self.logger.error("An I/O error occurred accessing file -> %s", json_path)
        return False
    except ValueError:
        self.logger.error("Value error!")
        return False

    return True

set_data_frame(df)

Set the Pandas data frame object.

Parameters:

Name	Type	Description	Default
df	DataFrame	The new Pandas data frame object.	required

Source code in packages/pyxecm/src/pyxecm/helper/data.py

def set_data_frame(self, df: pd.DataFrame) -> None:
    """Set the Pandas data frame object.

    Args:
        df (pd.DataFrame): The new Pandas data frame object.

    """

    self._df = df

set_value(column, value, condition=None)

Set the value in the data frame based on a condition.

Parameters:

Name	Type	Description	Default
column	str	The name of the column.	required
value	Any	The value to set for those rows that fulfill the condition.	required
condition	Series	This should be a boolean Series where each element is True or False, representing rows in the data frame that meet a certain condition. If None is provided then ALL rows get the 'value' in the given column.	None

Source code in packages/pyxecm/src/pyxecm/helper/data.py

def set_value(self, column: str, value, condition: pd.Series | None = None) -> None:  # noqa: ANN001
    """Set the value in the data frame based on a condition.

    Args:
        column (str):
            The name of the column.
        value (Any):
            The value to set for those rows that fulfill the condition.
        condition (pd.Series, optional):
            This should be a boolean Series where each element is True or False,
            representing rows in the data frame that meet a certain condition.
            If None is provided then ALL rows get the 'value' in the given
            column.

    """

    if condition is None:
        self._df[column] = value  # Set value unconditionally
    else:
        self._df.loc[condition, column] = value  # Set value based on condition

sort(sort_fields, inplace=True)

Sort the data frame based on one or multiple fields.

Sorting can be either in place or return it as a new data frame (e.g. not modifying self._df).

Parameters:

Name	Type	Description	Default
sort_fields	list	The columns / fields to be used for sorting.	required
inplace	bool	If the sorting should be inplace, i.e. modifying self._df. Defaults to True.	True

Returns:

Type	Description
DataFrame | None	pd.DataFrame | None: New data frame (if inplace = False) or self._df (if inplace = True). None in case of an error.

Source code in packages/pyxecm/src/pyxecm/helper/data.py

def sort(self, sort_fields: list, inplace: bool = True) -> pd.DataFrame | None:
    """Sort the data frame based on one or multiple fields.

    Sorting can be either in place or return it as a new data frame
    (e.g. not modifying self._df).

    Args:
        sort_fields (list):
            The columns / fields to be used for sorting.
        inplace (bool, optional):
            If the sorting should be inplace, i.e. modifying self._df.
            Defaults to True.

    Returns:
        pd.DataFrame | None:
            New data frame (if inplace = False) or self._df (if inplace = True).
            None in case of an error.

    """

    if self._df is None:
        return None

    if not all(sort_field in self._df.columns for sort_field in sort_fields):
        self.logger.warning(
            "Not all of the given sort fields -> %s do exist in the data frame.",
            str(sort_fields),
        )
        # Reduce the sort fields to those that really exist in the data frame:
        sort_fields = [sort_field for sort_field in sort_fields if sort_field in self._df.columns]
        self.logger.warning(
            "Only these given sort fields -> %s do exist as columns in the data frame.",
            str(sort_fields),
        )

    if inplace:
        self._df.sort_values(by=sort_fields, inplace=True)
        self._df.reset_index(drop=True, inplace=True)
        return self._df
    else:
        df = self._df.sort_values(by=sort_fields, inplace=False)
        df = df.reset_index(drop=True, inplace=False)
        return df

strip(columns=None, inplace=True)

Strip leading and trailing spaces from specified columns in a data frame.

Parameters:

Name	Type	Description	Default
columns	list | None	The list of column names to strip. If None, it strips leading and trailing spaces from all string columns.	None
inplace	bool	If True, the data modification is done in place, i.e. modifying the existing data frame of the object. If False, the data frame is copied and the copy is modified and returned.	True

Returns:

Type	Description
DataFrame	pd.DataFrame: The modified data frame with stripped columns.

Source code in packages/pyxecm/src/pyxecm/helper/data.py

def strip(self, columns: list | None = None, inplace: bool = True) -> pd.DataFrame:
    """Strip leading and trailing spaces from specified columns in a data frame.

    Args:
        columns (list | None):
            The list of column names to strip. If None, it strips
            leading and trailing spaces from _all_ string columns.
        inplace (bool, optional):
            If True, the data modification is done in place, i.e.
            modifying the existing data frame of the object.
            If False, the data frame is copied and the copy is modified
            and returned.

    Returns:
        pd.DataFrame:
            The modified data frame with stripped columns.

    """

    df = self._df.copy() if not inplace else self._df

    if columns is None:
        # Strip spaces from all string columns
        df = df.apply(lambda x: x.str.strip() if x.dtype == "object" else x)
    else:
        # Strip spaces from specified columns
        for col in columns:
            if col in df.columns and df[col].dtype == "object":  # Check if the column exists and is of string type
                df[col] = df[col].str.strip()

    if inplace:
        self._df = df

    return df

XML

XML helper module.

XML

Handle XML processing, e.g. to parse and update Extended ECM transport packages.

Source code in packages/pyxecm/src/pyxecm/helper/xml.py

class XML:
    """Handle XML processing, e.g. to parse and update Extended ECM transport packages."""

    logger: logging.Logger = default_logger

    @classmethod
    def remove_xml_namespace(cls, tag: str) -> str:
        """Remove namespace from XML tag.

        Args:
            tag (str):
                The XML tag with namespace.

        Returns:
            str:
                The tag without namespace.

        """

        # In Python's ElementTree, the tag namespace
        # is put into curly braces like "{namespace}element"
        # that's why this method splits after the closing curly brace
        # and takes the last item (-1):

        return tag.split("}", 1)[-1]

    # end method definition

    @classmethod
    def xml_to_dict(cls, xml_string: str, encode: bool = False, include_attributes: bool = False) -> dict:
        """Parse XML string and return a dictionary without namespaces.

        Args:
            xml_string (str):
                The XML string to process.
            encode (bool, optional):
                True if the XML string should be encoded to UTF-8 bytes. Defaults to False.
            include_attributes (bool, optional):
                True if XML attributes should be included in the dictionary. Defaults to False.

        Returns:
            dict:
                The XML structure converted to a dictionary.

        """

        def xml_element_to_dict(element: Element) -> dict:
            """Convert XML element to dictionary.

            Args:
                element (Element):
                    The XML element.

            Returns:
                dict:
                    Dictionary representing the XML element

            """

            tag = cls.remove_xml_namespace(element.tag)
            children = list(element)
            node_dict = {}

            if element.attrib and include_attributes:
                node_dict.update({"@{}".format(k): v for k, v in element.attrib.items()})

            if children:
                child_dict = {}
                for child in children:
                    child_tag = cls.remove_xml_namespace(child.tag)
                    child_value = xml_element_to_dict(child)

                    # child_dict is {child_tag: ...}, we want the value inside
                    value = child_value[child_tag]

                    # Handle multiple occurrences of the same tag:
                    if child_tag in child_dict:
                        # If the tag already exists, ensure it becomes a list
                        if not isinstance(child_dict[child_tag], list):
                            child_dict[child_tag] = [child_dict[child_tag]]
                        child_dict[child_tag].append(value)
                    else:
                        child_dict[child_tag] = value

                # Merge children into node_dict
                node_dict.update(child_dict)

            # Add text if present and meaningful
            text = element.text.strip() if element.text and element.text.strip() else None
            if text:
                if node_dict:
                    node_dict["#text"] = text
                else:
                    return {tag: text}

            return {tag: node_dict if node_dict else text}

        if encode:
            xml_string = xml_string.encode("utf-8")
        root = etree.fromstring(xml_string)

        return xml_element_to_dict(root)

    # end method definition

    @classmethod
    def load_xml_file(
        cls,
        file_path: str,
        xpath: str,
        dir_name: str | None = None,
        logger: logging.Logger = default_logger,
    ) -> list | None:
        """Load an XML file into a Python list of dictionaries.

        Args:
            file_path (str):
                The path to XML file.
            xpath (str):
                XPath to select sub-elements.
            dir_name (str | None, optional):
                Directory name to include in each dictionary, if provided.
            logger (logging.Logger):
                The logging object used for all log messages.

        Returns:
            dict | None:
                A list of dictionaries representing the parsed XML elements,
                or None if an error occurs during file reading or parsing.

        """

        if not os.path.exists(file_path):
            logger.error("XML File -> %s does not exist!", file_path)
            return None

        try:
            tree = etree.parse(file_path)
            if not tree:
                logger.warning("Empty or invalid XML tree for file -> %s", file_path)
                return None

            # Extract elements using the XPath:
            elements = tree.xpath(xpath)
            if not elements:
                logger.warning(
                    "No elements matched XPath -> %s in file -> '%s'",
                    xpath,
                    file_path,
                )
                return None

            # Convert the selected elements to dictionaries
            results = []
            tag = xpath.split("/")[-1]
            for element in elements:
                element_dict = xmltodict.parse(etree.tostring(element))
                if tag in element_dict:
                    element_dict = element_dict[tag]
                if dir_name:
                    element_dict["directory"] = dir_name
                results.append(element_dict)

        except OSError:
            logger.error("IO Error with file -> %s", file_path)
            return None
        except etree.XMLSyntaxError:
            logger.error("XML Syntax Error in file -> %s", file_path)
            return None
        except etree.DocumentInvalid:
            logger.error("Invalid XML document -> %s", file_path)
            return None

        return results

    # end method definition

    @classmethod
    def load_xml_files_from_directory(
        cls,
        path_to_root: str,
        filenames: list | None,
        xpath: str | None = None,
        logger: logging.Logger = default_logger,
    ) -> list | None:
        """Load all XML files from a directory that matches defined file names.

        Then using the XPath to identify a set of elements and convert them
        into a Python list of dictionaries.

        Args:
            path_to_root (str):
                Path to the root element of the
                directory structure
            filenames (list):
                A list of filenames. This can also be patterns like
                "*/en/docovw.xml". If empty all filenames ending
                with ".xml" is used.
            xpath (str, optional):
                The XPath to the elements we want to select.
            logger (logging.Logger):
                The logging object used for all log messages.

        Returns:
            list:
                List of dictionaries.

        """

        if not filenames:
            filenames = ["*.xml"]

        try:
            # Check if the provided path is a directory or a zip file that can be extracted
            # into a directory:
            if not os.path.isdir(path_to_root) and not path_to_root.endswith(".zip"):
                logger.error(
                    "The provided path -> '%s' is not a valid directory or Zip file.",
                    path_to_root,
                )
                return None

            # If we have a zip file we extract it - but only if it has not been extracted before:
            if path_to_root.endswith(".zip"):
                zip_file_folder = os.path.splitext(path_to_root)[0]
                if not os.path.exists(zip_file_folder):
                    logger.info(
                        "Unzipping -> '%s' into folder -> '%s'...",
                        path_to_root,
                        zip_file_folder,
                    )
                    try:
                        with zipfile.ZipFile(path_to_root, "r") as zfile:
                            zfile.extractall(zip_file_folder)
                    except zipfile.BadZipFile:
                        logger.error(
                            "Failed to extract zip file -> '%s'",
                            path_to_root,
                        )
                        return None
                    except OSError:
                        logger.error(
                            "OS error occurred while trying to extract -> '%s'",
                            path_to_root,
                        )
                        return None
                else:
                    logger.info(
                        "Zip file is already extracted (path -> '%s' exists). Reusing extracted data...",
                        zip_file_folder,
                    )
                path_to_root = zip_file_folder

            results = []

            # Walk through the directory
            for root, _, files in os.walk(path_to_root):
                for file_data in files:
                    file_path = os.path.join(root, file_data)
                    file_size = os.path.getsize(file_path)
                    file_name = os.path.basename(file_path)
                    dir_name = os.path.dirname(file_path)

                    if any(fnmatch.fnmatch(file_path, pattern) for pattern in filenames) and file_name.endswith(".xml"):
                        logger.info(
                            "Load XML file -> '%s' of size -> %s",
                            file_path,
                            file_size,
                        )
                        elements = cls.load_xml_file(
                            file_path,
                            xpath=xpath,
                            dir_name=dir_name,
                        )
                        if elements:
                            results += elements

        except NotADirectoryError:
            logger.error(
                "The given path -> '%s' is not a directory!",
                path_to_root,
            )
            return None
        except FileNotFoundError:
            logger.error(
                "The given path -> '%s' does not exist!",
                path_to_root,
            )
            return None
        except PermissionError:
            logger.error(
                "No permission to access path -> '%s'!",
                path_to_root,
            )
            return None
        except OSError:
            logger.error("Low level OS error with file -> %s", path_to_root)
            return None

        return results

    # end method definition

    @classmethod
    def load_xml_files_from_directories(
        cls,
        directories: list[str],
        filenames: list[str] | None = None,
        xpath: str | None = None,
        logger: logging.Logger = default_logger,
    ) -> list[dict] | None:
        """Load XML files from multiple directories or zip files concurrently.

        Process them using XPath, and return a list of dictionaries containing the extracted elements.

        This method handles multiple directories or zip files, processes XML files inside them in parallel
        using threads, and extracts elements that match the specified XPath. It also supports pattern matching
        for filenames and handles errors such as missing files or permission issues.

        Args:
            directories (list[str]):
                A list of directories or zip files to process. Each item can be a path
                to a directory or a zip file that contains XML files.
            filenames (list[str] | None, optional):
                A list of filename patterns (e.g., ["*/en/docovw.xml"]) to match
                against the XML files. If None or empty, defaults to ["*.xml"].
            xpath (str | None, optional):
                An optional XPath string used to filter elements from the XML files.
            logger (logging.Logger):
                The logging object used for all log messages.

        Returns:
            list[dict] | None:
                A list of dictionaries containing the extracted XML elements. Returns None
                if any error occurs during processing.

        Raises:
            Exception: If any error occurs during processing, such as issues with directories, files, or zip extraction.

        """

        # Set default for filenames if not provided
        if not filenames:
            filenames = ["*.xml"]

        results_queue = Queue()

        def process_xml_file(file_path: str) -> None:
            """Process a single XML file.

            Args:
                file_path (str):
                    Path to the XML file.

            Results:
                Adds elements to the result_queue defined outside this sub-method.

            """

            try:
                file_size = os.path.getsize(file_path)
                file_name = os.path.basename(file_path)
                dir_name = os.path.dirname(file_path)

                if (
                    not filenames or any(fnmatch.fnmatch(file_path, pattern) for pattern in filenames)
                ) and file_name.endswith(".xml"):
                    logger.info(
                        "Load XML file -> '%s' of size -> %s",
                        file_path,
                        file_size,
                    )
                    elements = cls.load_xml_file(
                        file_path,
                        xpath=xpath,
                        dir_name=dir_name,
                    )
                    if elements:
                        results_queue.put(elements)
            except FileNotFoundError:
                logger.error("File not found -> '%s'!", file_path)
            except PermissionError:
                logger.error(
                    "Permission error with file -> '%s'!",
                    file_path,
                )
            except OSError:
                logger.error(
                    "OS error processing file -> '%s'!",
                    file_path,
                )
            except ValueError:
                logger.error(
                    "Value error processing file -> '%s'!",
                    file_path,
                )

        # end method process_xml_file

        def process_directory_or_zip(path_to_root: str) -> list | None:
            """Process all files in a directory or zip file.

            Args:
                path_to_root (str):
                    File path to the root directory or zip file.

            """

            try:
                # Handle zip files
                if path_to_root.endswith(".zip"):
                    zip_file_folder = os.path.splitext(path_to_root)[0]
                    if not os.path.exists(zip_file_folder):
                        logger.info(
                            "Unzipping -> '%s' into folder -> '%s'...",
                            path_to_root,
                            zip_file_folder,
                        )
                        try:
                            with zipfile.ZipFile(path_to_root, "r") as zfile:
                                zfile.extractall(zip_file_folder)
                        except zipfile.BadZipFile:
                            logger.error(
                                "Bad zip file -> '%s'!",
                                path_to_root,
                            )
                        except zipfile.LargeZipFile:
                            logger.error(
                                "Zip file is too large to process -> '%s'!",
                                path_to_root,
                            )
                        except PermissionError:
                            logger.error(
                                "Permission error extracting zip file -> '%s'!",
                                path_to_root,
                            )
                        except OSError:
                            logger.error(
                                "OS error occurred while extracting zip file -> '%s'!",
                                path_to_root,
                            )
                        return  # Don't proceed further if zip extraction fails

                    else:
                        logger.info(
                            "Zip file is already extracted (path -> '%s' exists). Reusing extracted data...",
                            zip_file_folder,
                        )
                    path_to_root = zip_file_folder
                # end if path_to_root.endswith(".zip")

                # Use inner threading to process files within the directory
                with ThreadPoolExecutor(
                    thread_name_prefix="ProcessXMLFile",
                ) as inner_executor:
                    for root, _, files in os.walk(path_to_root):
                        for file_data in files:
                            file_path = os.path.join(root, file_data)
                            inner_executor.submit(process_xml_file, file_path)

            except FileNotFoundError:
                logger.error(
                    "Directory or file not found -> '%s'!",
                    path_to_root,
                )
            except PermissionError:
                logger.error(
                    "Permission error with directory -> '%s'!",
                    path_to_root,
                )
            except OSError:
                logger.error(
                    "OS error processing path -> '%s'!",
                    path_to_root,
                )
            except ValueError:
                logger.error(
                    "Value error processing path -> '%s'!",
                    path_to_root,
                )

        # end method process_directory_or_zip

        try:
            # Resolve wildcards in the directories list
            expanded_directories: list[str] = []
            for directory in directories:
                if "*" in directory:
                    expanded_directory: list = glob.glob(directory)
                    logger.info(
                        "Expanding directory -> '%s' with wildcards...",
                        directory,
                    )
                    expanded_directories.extend(expanded_directory)
                else:
                    logger.info(
                        "Directory -> '%s' has no wildcards. Not expanding...",
                        directory,
                    )
                    expanded_directories.append(directory)

            # Use ThreadPoolExecutor for outer level: processing directories/zip files
            logger.info(
                "Starting %d threads for each directory or zip file...",
                len(expanded_directories),
            )
            with ThreadPoolExecutor(
                thread_name_prefix="ProcessDirOrZip",
            ) as outer_executor:
                futures = [
                    outer_executor.submit(process_directory_or_zip, directory) for directory in expanded_directories
                ]

                # Wait for all futures to complete
                for future in futures:
                    future.result()

            # Collect results from the queue
            logger.info("Collecting results from worker queue...")
            results = []
            while not results_queue.empty():
                results.extend(results_queue.get())
            logger.info("Done. Collected %d results.", len(results))

        except FileNotFoundError:
            logger.error(
                "Directory or file not found during execution!",
            )
            return None
        except PermissionError:
            logger.error("Permission error during execution!")
            return None
        except TimeoutError:
            logger.error(
                "Timeout occurred while waiting for threads!",
            )
            return None
        except BrokenPipeError:
            logger.error(
                "Broken pipe error occurred during thread communication!",
            )
            return None

        return results

    # end method definition

    @classmethod
    def get_xml_element(
        cls,
        xml_content: str,
        xpath: str,
    ) -> Element:
        """Retrieve an XML Element from a string using an XPath expression.

        Args:
            xml_content (str):
                XML file as a string
            xpath (str):
                XPath used to find the element.

        Returns:
            Element:
                The XML element.

        """

        # Parse XML content into an etree
        tree = etree.fromstring(xml_content)

        # Find the XML element specified by XPath
        element = tree.find(xpath)

        return element

    # end method definition

    @classmethod
    def modify_xml_element(
        cls,
        xml_content: str,
        xpath: str,
        new_value: str,
        logger: logging.Logger = default_logger,
    ) -> None:
        """Update the text (= content) of an XML element.

        Args:
            xml_content (str):
                The content of an XML file.
            xpath (str):
                XML Path to identify the XML element.
            new_value (str):
                The new text (content).
            logger (logging.Logger):
                The logging object used for all log messages.

        """
        element = cls.get_xml_element(xml_content=xml_content, xpath=xpath)

        if element is not None:
            # Modify the XML element with the new value
            element.text = new_value
        else:
            logger.warning("XML Element -> %s not found.", xpath)

    # end method definition

    @classmethod
    def search_setting(
        cls,
        element_text: str,
        setting_key: str,
        is_simple: bool = True,
        is_escaped: bool = False,
    ) -> str | None:
        """Search a setting in an XML element and return its value.

        The simple case covers settings like this:
        "syncCandidates":true,
        "syncCandidates":true,
        In this case the setting value is a scalar like true, false, a number or none
        the regular expression pattern searches for a setting name in "..." (quotes) followed
        by a colon (:). The value is taken from what follows the colon until the next comma (,)

        The more complex case is a string value that may itself have commas,
        so we cannot look for comma as a delimiter like in the simple case
        but we take the value for a string delimited by double quotes ("...")

        Args:
            element_text (str):
                The text to examine - typically content of an XML element.
            setting_key (str):
                The name of the setting key (before the colon).
            is_simple (bool, optional):
                True if the value is scalar (not having assocs with commas). Defaults to True.
            is_escaped (bool, optional):
                True if the quotes or escaped with ". Defaults to False.

        Returns:
            str:
                The value of the setting or None if the setting is not found.

        """

        if is_simple:
            pattern = r""{}":[^,]*".format(setting_key) if is_escaped else r'"{}":[^,]*'.format(setting_key)
        elif is_escaped:
            pattern = r""{}":".*"".format(setting_key)
        else:
            pattern = r'"{}":"([^"]*)"'.format(setting_key)

        match = re.search(pattern, element_text)
        if match:
            setting_line = match.group(0)
            setting_value = setting_line.split(":")[1]
            return setting_value
        else:
            return None

    # end method definition

    @classmethod
    def replace_setting(
        cls,
        element_text: str,
        setting_key: str,
        new_value: str,
        is_simple: bool = True,
        is_escaped: bool = False,
    ) -> str:
        """Replace the value of a defined setting with a new value.

        The simple case covers settings like this:
        "syncCandidates":true,
        "syncCandidates":true,
        In this case the setting value is a scalar like true, false, a number or none
        the regular expression pattern searches for a setting name in "..." (quotes) followed
        by a colon (:). The value is taken from what follows the colon until the next comma (,)

        The more complex case is a string value that may itself have commas,
        so we cannot look for comma as a delimiter like in the simple case
        but we take the value for a string delimited by double quotes ("...")

        Args:
            element_text (str):
                The original text of the XML element (that is to be updated).
            setting_key (str):
                The name of the setting.
            new_value (str):
                The new value of the setting.
            is_simple (bool, optional):
                True = value is a scalar like true, false, a number or none. Defaults to True.
            is_escaped (bool, optional):
                True if the value is surrrounded with ". Defaults to False.

        Returns:
            str:
                The updated element text.

        """

        if is_simple:
            pattern = r""{}":[^,]*".format(setting_key) if is_escaped else r'"{}":[^,]*'.format(setting_key)
        elif is_escaped:
            pattern = r""{}":".*"".format(setting_key)
        else:
            pattern = r'"{}":"([^"]*)"'.format(setting_key)

        new_text = re.sub(pattern, new_value, element_text)

        return new_text

    # end method definition

    @classmethod
    def replace_in_xml_files(
        cls,
        directory: str,
        search_pattern: str,
        replace_string: str,
        xpath: str = "",
        setting: str = "",
        assoc_elem: str = "",
        logger: logging.Logger = default_logger,
    ) -> bool:
        """Replace all occurrences of the search pattern with the replace string.

        This is done in all XML files in the directory and its subdirectories.

        Args:
            directory (str):
                Directory to traverse for XML files
            search_pattern (str):
                The string to search in the XML file.
                This can be empty if xpath is used!
            replace_string (str):
                The replacement string.
            xpath (str, optional):
                An XPath can be given to narrow down the replacement to an XML element.
                For now the XPath needs to be constructed in a way the it returns
                one or none element.
            setting (str, optional):
                Narrow down the replacement to the line that includes the setting with this name.
                This parameter is optional.
            assoc_elem (str, optional):
                Lookup a specific assoc element. This parameter is optional.
            logger (logging.Logger):
                The logging object used for all log messages.

        Returns:
            bool:
                True if a replacement happened, False otherwise

        """

        # Define the regular expression pattern to search for
        # search pattern can be empty if an xpath is used. So
        # be careful here:
        if search_pattern:
            pattern = re.compile(search_pattern)

        found = False

        # Traverse the directory and its subdirectories
        for subdir, _, files in os.walk(directory):
            for filename in files:
                # Check if the file is an XML file
                if filename.endswith(".xml"):
                    # Read the contents of the file
                    file_path = os.path.join(subdir, filename)

                    # if xpath is given we do an intelligent replacement
                    if xpath:
                        xml_modified = False
                        logger.debug("Replacement with xpath...")
                        logger.debug(
                            "XML path -> %s, setting -> %s, assoc element -> %s",
                            xpath,
                            setting,
                            assoc_elem,
                        )
                        tree = etree.parse(file_path)
                        if not tree:
                            logger.error(
                                "Cannot parse XML tree -> %s. Skipping...",
                                file_path,
                            )
                            continue
                        root = tree.getroot()
                        # find the matching XML elements using the given XPath:
                        elements = root.xpath(xpath)
                        if not elements:
                            logger.debug(
                                "The XML file -> %s does not have any element with the given XML path -> %s. Skipping...",
                                file_path,
                                xpath,
                            )
                            continue
                        for element in elements:
                            # as XPath returns a list
                            logger.debug(
                                "Found XML element -> %s in file -> %s using xpath -> %s",
                                element.tag,
                                filename,
                                xpath,
                            )
                            # the simple case: replace the complete text of the XML element
                            if not setting and not assoc_elem:
                                logger.debug(
                                    "Replace complete text of XML element -> %s from -> %s to -> %s",
                                    xpath,
                                    element.text,
                                    replace_string,
                                )
                                element.text = replace_string
                                xml_modified = True
                            # In this case we want to set a complete value of a setting (basically replacing a whole line)
                            elif setting and not assoc_elem:
                                logger.debug(
                                    "Replace single setting -> %s in XML element -> %s with new value -> %s",
                                    setting,
                                    xpath,
                                    replace_string,
                                )
                                setting_value = cls.search_setting(
                                    element.text,
                                    setting,
                                    is_simple=True,
                                )
                                if setting_value:
                                    logger.debug(
                                        "Found existing setting value -> %s",
                                        setting_value,
                                    )
                                    # Check if the setting value needs to be surrounded by quotes.
                                    # Only simplistic values like booleans or numeric values don't need quotes
                                    if replace_string in ("true", "false", "none") or replace_string.isnumeric():
                                        replace_setting = '"' + setting + '":' + replace_string
                                    else:
                                        replace_setting = '"' + setting + '":"' + replace_string + '"'
                                    logger.debug(
                                        "Replacement setting -> %s",
                                        replace_setting,
                                    )
                                    element.text = cls.replace_setting(
                                        element_text=element.text,
                                        setting_key=setting,
                                        new_value=replace_setting,
                                        is_simple=True,
                                    )
                                    xml_modified = True
                                else:
                                    logger.warning(
                                        "Cannot find the value for setting -> %s. Skipping...",
                                        setting,
                                    )
                                    continue
                            # in this case the text is just one assoc (no setting substructure)
                            elif not setting and assoc_elem:
                                logger.debug(
                                    "Replace single Assoc value -> %s in XML element -> %s with -> %s",
                                    assoc_elem,
                                    xpath,
                                    replace_string,
                                )
                                assoc_string: str = Assoc.extract_assoc_string(
                                    input_string=element.text,
                                )
                                logger.debug("Assoc String -> %s", assoc_string)
                                assoc_dict = Assoc.string_to_dict(
                                    assoc_string=assoc_string,
                                )
                                logger.debug("Assoc Dict -> %s", str(assoc_dict))
                                assoc_dict[assoc_elem] = replace_string  # escaped_replace_string
                                assoc_string_new: str = Assoc.dict_to_string(
                                    assoc_dict=assoc_dict,
                                )
                                logger.debug(
                                    "Replace assoc with -> %s",
                                    assoc_string_new,
                                )
                                element.text = assoc_string_new
                                element.text = element.text.replace('"', """)
                                xml_modified = True
                            # In this case we have multiple settings with their own assocs
                            elif setting and assoc_elem:
                                logger.debug(
                                    "Replace single Assoc value -> %s in setting -> %s in XML element -> %s with -> %s",
                                    assoc_elem,
                                    setting,
                                    xpath,
                                    replace_string,
                                )
                                setting_value = cls.search_setting(
                                    element.text,
                                    setting,
                                    is_simple=False,
                                )
                                if setting_value:
                                    logger.debug(
                                        "Found setting value -> %s",
                                        setting_value,
                                    )
                                    assoc_string: str = Assoc.extract_assoc_string(
                                        input_string=setting_value,
                                    )
                                    logger.debug("Assoc String -> %s", assoc_string)
                                    assoc_dict = Assoc.string_to_dict(
                                        assoc_string=assoc_string,
                                    )
                                    logger.debug("Assoc Dict -> %s", str(assoc_dict))
                                    escaped_replace_string = replace_string.replace(
                                        "'",
                                        "\\\\\u0027",
                                    )
                                    logger.debug(
                                        "Escaped replacement string -> %s",
                                        escaped_replace_string,
                                    )
                                    assoc_dict[assoc_elem] = escaped_replace_string  # escaped_replace_string
                                    assoc_string_new: str = Assoc.dict_to_string(
                                        assoc_dict=assoc_dict,
                                    )
                                    assoc_string_new = assoc_string_new.replace(
                                        "'",
                                        "\\u0027",
                                    )
                                    replace_setting = '"' + setting + '":"' + assoc_string_new + '"'
                                    logger.debug(
                                        "Replacement setting -> %s",
                                        replace_setting,
                                    )
                                    # here we need to apply a "trick". It is required
                                    # as regexp cannot handle the special unicode escapes \u0027
                                    # we require. We first insert a placeholder "PLACEHOLDER"
                                    # and let regexp find the right place for it. Then further
                                    # down we use a simple search&replace to switch the PLACEHOLDER
                                    # to the real value (replace() does not have the issues with unicode escapes)
                                    element.text = cls.replace_setting(
                                        element_text=element.text,
                                        setting_key=setting,
                                        new_value="PLACEHOLDER",
                                        is_simple=False,
                                        is_escaped=False,
                                    )
                                    element.text = element.text.replace(
                                        "PLACEHOLDER",
                                        replace_setting,
                                    )
                                    element.text = element.text.replace('"', """)
                                    xml_modified = True
                                else:
                                    logger.warning(
                                        "Cannot find the value for setting -> %s. Skipping...",
                                        setting,
                                    )
                                    continue
                        if xml_modified:
                            logger.debug(
                                "XML tree has been modified. Write updated file -> %s...",
                                file_path,
                            )

                            new_contents = etree.tostring(
                                tree,
                                pretty_print=True,
                                xml_declaration=True,
                                encoding="UTF-8",
                            )
                            # we need to undo some of the stupid things tostring() did:
                            new_contents = new_contents.replace(
                                b""",
                                b""",
                            )
                            new_contents = new_contents.replace(
                                b"'",
                                b"'",
                            )
                            new_contents = new_contents.replace(b">", b">")
                            new_contents = new_contents.replace(b"<", b"<")

                            # Replace single quotes inside double quotes strings with "'" (manual escaping)
                            # This is required as we next want to replace all double quotes with single quotes
                            pattern = b'"([^"]*)"'
                            new_contents = re.sub(
                                pattern,
                                lambda m: m.group(0).replace(b"'", b"'"),
                                new_contents,
                            )

                            # Replace single quotes in XML text elements with "'"
                            # and replace double quotes in XML text elements with """
                            # This is required as we next want to replace all double quotes with single quotes
                            # to make the XML files as similar as possible with Extended ECM's format
                            pattern = b">([^<>]+?)<"
                            replacement = lambda match: match.group(0).replace(  # noqa: E731
                                b'"',
                                b"&quot;",
                            )
                            new_contents = re.sub(pattern, replacement, new_contents)
                            replacement = lambda match: match.group(0).replace(  # noqa: E731
                                b"'",
                                b"&apos;",
                            )
                            new_contents = re.sub(pattern, replacement, new_contents)

                            # Change double quotes to single quotes across the XML file - Extended ECM has it that way:
                            new_contents = new_contents.replace(b'"', b"'")

                            # Write the updated contents to the file.
                            # We DO NOT want to use tree.write() here
                            # as it would undo the manual XML tweaks we
                            # need for Extended ECM. We also need "wb"
                            # as we have bytes and not str as a data type
                            with open(file_path, "wb") as f:
                                f.write(new_contents)

                            found = True
                    # this is not using xpath - do a simple search and replace
                    else:
                        logger.debug("Replacement without xpath...")
                        with open(file_path, encoding="UTF-8") as f:
                            contents = f.read()
                        # Replace all occurrences of the search pattern with the replace string
                        new_contents = pattern.sub(replace_string, contents)

                        # Write the updated contents to the file if there were replacements
                        if contents != new_contents:
                            logger.debug(
                                "Found search string -> %s in XML file -> %s. Write updated file...",
                                search_pattern,
                                file_path,
                            )
                            # Write the updated contents to the file
                            with open(file_path, "w", encoding="UTF-8") as f:
                                f.write(new_contents)
                            found = True

        return found

    # end method definition

    @classmethod
    def extract_from_xml_files(
        cls,
        directory: str,
        xpath: str,
        logger: logging.Logger = default_logger,
    ) -> list | None:
        """Extract the XML subtrees using an XPath in all XML files in the directory and its subdirectories.

        Args:
            directory (str):
                The directory to traverse for XML files.
            xpath (str):
                Used to determine XML elements to extract.
            logger (logging.Logger):
                The logging object used for all log messages.

        Returns:
            list | None:
                Extracted data if it is found by the XPath, None otherwise.

        """

        extracted_data_list = []

        # Traverse the directory and its subdirectories
        for subdir, _, files in os.walk(directory):
            for filename in files:
                # Check if the file is an XML file
                if filename.endswith(".xml"):
                    # Read the contents of the file
                    file_path = os.path.join(subdir, filename)

                    logger.debug("Extraction with xpath -> %s...", xpath)
                    tree = etree.parse(file_path)
                    if not tree:
                        logger.error(
                            "Cannot parse XML file -> '%s'. Skipping...",
                            file_path,
                        )
                        continue
                    root = tree.getroot()
                    # find the matching XML elements using the given XPath:
                    elements = root.xpath(xpath)
                    if not elements:
                        logger.debug(
                            "The XML file -> %s does not have any element with the given XML path -> %s. Skipping...",
                            file_path,
                            xpath,
                        )
                        continue
                    for element in elements:
                        # as XPath returns a list
                        logger.debug(
                            "Found XML element -> %s in file -> %s using xpath -> %s. Add it to result list.",
                            element.tag,
                            filename,
                            xpath,
                        )
                        extracted_content = etree.tostring(element)

                        try:
                            dict_content = xmltodict.parse(extracted_content)
                        except xmltodict.expat.ExpatError:
                            logger.error(
                                "Invalid XML syntax in file -> %s. Please check the XML file for errors.",
                                filename,
                            )
                            continue

                        extracted_data_list.append(dict_content)

        return extracted_data_list

extract_from_xml_files(directory, xpath, logger=default_logger) classmethod

Extract the XML subtrees using an XPath in all XML files in the directory and its subdirectories.

Parameters:

Name	Type	Description	Default
directory	str	The directory to traverse for XML files.	required
xpath	str	Used to determine XML elements to extract.	required
logger	Logger	The logging object used for all log messages.	default_logger

Returns:

Type	Description
list | None	list | None: Extracted data if it is found by the XPath, None otherwise.

Source code in packages/pyxecm/src/pyxecm/helper/xml.py

@classmethod
def extract_from_xml_files(
    cls,
    directory: str,
    xpath: str,
    logger: logging.Logger = default_logger,
) -> list | None:
    """Extract the XML subtrees using an XPath in all XML files in the directory and its subdirectories.

    Args:
        directory (str):
            The directory to traverse for XML files.
        xpath (str):
            Used to determine XML elements to extract.
        logger (logging.Logger):
            The logging object used for all log messages.

    Returns:
        list | None:
            Extracted data if it is found by the XPath, None otherwise.

    """

    extracted_data_list = []

    # Traverse the directory and its subdirectories
    for subdir, _, files in os.walk(directory):
        for filename in files:
            # Check if the file is an XML file
            if filename.endswith(".xml"):
                # Read the contents of the file
                file_path = os.path.join(subdir, filename)

                logger.debug("Extraction with xpath -> %s...", xpath)
                tree = etree.parse(file_path)
                if not tree:
                    logger.error(
                        "Cannot parse XML file -> '%s'. Skipping...",
                        file_path,
                    )
                    continue
                root = tree.getroot()
                # find the matching XML elements using the given XPath:
                elements = root.xpath(xpath)
                if not elements:
                    logger.debug(
                        "The XML file -> %s does not have any element with the given XML path -> %s. Skipping...",
                        file_path,
                        xpath,
                    )
                    continue
                for element in elements:
                    # as XPath returns a list
                    logger.debug(
                        "Found XML element -> %s in file -> %s using xpath -> %s. Add it to result list.",
                        element.tag,
                        filename,
                        xpath,
                    )
                    extracted_content = etree.tostring(element)

                    try:
                        dict_content = xmltodict.parse(extracted_content)
                    except xmltodict.expat.ExpatError:
                        logger.error(
                            "Invalid XML syntax in file -> %s. Please check the XML file for errors.",
                            filename,
                        )
                        continue

                    extracted_data_list.append(dict_content)

    return extracted_data_list

get_xml_element(xml_content, xpath) classmethod

Retrieve an XML Element from a string using an XPath expression.

Parameters:

Name	Type	Description	Default
xml_content	str	XML file as a string	required
xpath	str	XPath used to find the element.	required

Returns:

Name	Type	Description
Element	Element	The XML element.

Source code in packages/pyxecm/src/pyxecm/helper/xml.py

@classmethod
def get_xml_element(
    cls,
    xml_content: str,
    xpath: str,
) -> Element:
    """Retrieve an XML Element from a string using an XPath expression.

    Args:
        xml_content (str):
            XML file as a string
        xpath (str):
            XPath used to find the element.

    Returns:
        Element:
            The XML element.

    """

    # Parse XML content into an etree
    tree = etree.fromstring(xml_content)

    # Find the XML element specified by XPath
    element = tree.find(xpath)

    return element

load_xml_file(file_path, xpath, dir_name=None, logger=default_logger) classmethod

Load an XML file into a Python list of dictionaries.

Parameters:

Name	Type	Description	Default
file_path	str	The path to XML file.	required
xpath	str	XPath to select sub-elements.	required
dir_name	str | None	Directory name to include in each dictionary, if provided.	None
logger	Logger	The logging object used for all log messages.	default_logger

Returns:

Type	Description
list | None	dict | None: A list of dictionaries representing the parsed XML elements, or None if an error occurs during file reading or parsing.

Source code in packages/pyxecm/src/pyxecm/helper/xml.py

@classmethod
def load_xml_file(
    cls,
    file_path: str,
    xpath: str,
    dir_name: str | None = None,
    logger: logging.Logger = default_logger,
) -> list | None:
    """Load an XML file into a Python list of dictionaries.

    Args:
        file_path (str):
            The path to XML file.
        xpath (str):
            XPath to select sub-elements.
        dir_name (str | None, optional):
            Directory name to include in each dictionary, if provided.
        logger (logging.Logger):
            The logging object used for all log messages.

    Returns:
        dict | None:
            A list of dictionaries representing the parsed XML elements,
            or None if an error occurs during file reading or parsing.

    """

    if not os.path.exists(file_path):
        logger.error("XML File -> %s does not exist!", file_path)
        return None

    try:
        tree = etree.parse(file_path)
        if not tree:
            logger.warning("Empty or invalid XML tree for file -> %s", file_path)
            return None

        # Extract elements using the XPath:
        elements = tree.xpath(xpath)
        if not elements:
            logger.warning(
                "No elements matched XPath -> %s in file -> '%s'",
                xpath,
                file_path,
            )
            return None

        # Convert the selected elements to dictionaries
        results = []
        tag = xpath.split("/")[-1]
        for element in elements:
            element_dict = xmltodict.parse(etree.tostring(element))
            if tag in element_dict:
                element_dict = element_dict[tag]
            if dir_name:
                element_dict["directory"] = dir_name
            results.append(element_dict)

    except OSError:
        logger.error("IO Error with file -> %s", file_path)
        return None
    except etree.XMLSyntaxError:
        logger.error("XML Syntax Error in file -> %s", file_path)
        return None
    except etree.DocumentInvalid:
        logger.error("Invalid XML document -> %s", file_path)
        return None

    return results

load_xml_files_from_directories(directories, filenames=None, xpath=None, logger=default_logger) classmethod

Load XML files from multiple directories or zip files concurrently.

Process them using XPath, and return a list of dictionaries containing the extracted elements.

This method handles multiple directories or zip files, processes XML files inside them in parallel using threads, and extracts elements that match the specified XPath. It also supports pattern matching for filenames and handles errors such as missing files or permission issues.

Parameters:

Name	Type	Description	Default
directories	list[str]	A list of directories or zip files to process. Each item can be a path to a directory or a zip file that contains XML files.	required
filenames	list[str] | None	A list of filename patterns (e.g., ["/en/docovw.xml"]) to match against the XML files. If None or empty, defaults to [".xml"].	None
xpath	str | None	An optional XPath string used to filter elements from the XML files.	None
logger	Logger	The logging object used for all log messages.	default_logger

Returns:

Type	Description
list[dict] | None	list[dict] | None: A list of dictionaries containing the extracted XML elements. Returns None if any error occurs during processing.

Raises:

Type	Description
Exception	If any error occurs during processing, such as issues with directories, files, or zip extraction.

Source code in packages/pyxecm/src/pyxecm/helper/xml.py

@classmethod
def load_xml_files_from_directories(
    cls,
    directories: list[str],
    filenames: list[str] | None = None,
    xpath: str | None = None,
    logger: logging.Logger = default_logger,
) -> list[dict] | None:
    """Load XML files from multiple directories or zip files concurrently.

    Process them using XPath, and return a list of dictionaries containing the extracted elements.

    This method handles multiple directories or zip files, processes XML files inside them in parallel
    using threads, and extracts elements that match the specified XPath. It also supports pattern matching
    for filenames and handles errors such as missing files or permission issues.

    Args:
        directories (list[str]):
            A list of directories or zip files to process. Each item can be a path
            to a directory or a zip file that contains XML files.
        filenames (list[str] | None, optional):
            A list of filename patterns (e.g., ["*/en/docovw.xml"]) to match
            against the XML files. If None or empty, defaults to ["*.xml"].
        xpath (str | None, optional):
            An optional XPath string used to filter elements from the XML files.
        logger (logging.Logger):
            The logging object used for all log messages.

    Returns:
        list[dict] | None:
            A list of dictionaries containing the extracted XML elements. Returns None
            if any error occurs during processing.

    Raises:
        Exception: If any error occurs during processing, such as issues with directories, files, or zip extraction.

    """

    # Set default for filenames if not provided
    if not filenames:
        filenames = ["*.xml"]

    results_queue = Queue()

    def process_xml_file(file_path: str) -> None:
        """Process a single XML file.

        Args:
            file_path (str):
                Path to the XML file.

        Results:
            Adds elements to the result_queue defined outside this sub-method.

        """

        try:
            file_size = os.path.getsize(file_path)
            file_name = os.path.basename(file_path)
            dir_name = os.path.dirname(file_path)

            if (
                not filenames or any(fnmatch.fnmatch(file_path, pattern) for pattern in filenames)
            ) and file_name.endswith(".xml"):
                logger.info(
                    "Load XML file -> '%s' of size -> %s",
                    file_path,
                    file_size,
                )
                elements = cls.load_xml_file(
                    file_path,
                    xpath=xpath,
                    dir_name=dir_name,
                )
                if elements:
                    results_queue.put(elements)
        except FileNotFoundError:
            logger.error("File not found -> '%s'!", file_path)
        except PermissionError:
            logger.error(
                "Permission error with file -> '%s'!",
                file_path,
            )
        except OSError:
            logger.error(
                "OS error processing file -> '%s'!",
                file_path,
            )
        except ValueError:
            logger.error(
                "Value error processing file -> '%s'!",
                file_path,
            )

    # end method process_xml_file

    def process_directory_or_zip(path_to_root: str) -> list | None:
        """Process all files in a directory or zip file.

        Args:
            path_to_root (str):
                File path to the root directory or zip file.

        """

        try:
            # Handle zip files
            if path_to_root.endswith(".zip"):
                zip_file_folder = os.path.splitext(path_to_root)[0]
                if not os.path.exists(zip_file_folder):
                    logger.info(
                        "Unzipping -> '%s' into folder -> '%s'...",
                        path_to_root,
                        zip_file_folder,
                    )
                    try:
                        with zipfile.ZipFile(path_to_root, "r") as zfile:
                            zfile.extractall(zip_file_folder)
                    except zipfile.BadZipFile:
                        logger.error(
                            "Bad zip file -> '%s'!",
                            path_to_root,
                        )
                    except zipfile.LargeZipFile:
                        logger.error(
                            "Zip file is too large to process -> '%s'!",
                            path_to_root,
                        )
                    except PermissionError:
                        logger.error(
                            "Permission error extracting zip file -> '%s'!",
                            path_to_root,
                        )
                    except OSError:
                        logger.error(
                            "OS error occurred while extracting zip file -> '%s'!",
                            path_to_root,
                        )
                    return  # Don't proceed further if zip extraction fails

                else:
                    logger.info(
                        "Zip file is already extracted (path -> '%s' exists). Reusing extracted data...",
                        zip_file_folder,
                    )
                path_to_root = zip_file_folder
            # end if path_to_root.endswith(".zip")

            # Use inner threading to process files within the directory
            with ThreadPoolExecutor(
                thread_name_prefix="ProcessXMLFile",
            ) as inner_executor:
                for root, _, files in os.walk(path_to_root):
                    for file_data in files:
                        file_path = os.path.join(root, file_data)
                        inner_executor.submit(process_xml_file, file_path)

        except FileNotFoundError:
            logger.error(
                "Directory or file not found -> '%s'!",
                path_to_root,
            )
        except PermissionError:
            logger.error(
                "Permission error with directory -> '%s'!",
                path_to_root,
            )
        except OSError:
            logger.error(
                "OS error processing path -> '%s'!",
                path_to_root,
            )
        except ValueError:
            logger.error(
                "Value error processing path -> '%s'!",
                path_to_root,
            )

    # end method process_directory_or_zip

    try:
        # Resolve wildcards in the directories list
        expanded_directories: list[str] = []
        for directory in directories:
            if "*" in directory:
                expanded_directory: list = glob.glob(directory)
                logger.info(
                    "Expanding directory -> '%s' with wildcards...",
                    directory,
                )
                expanded_directories.extend(expanded_directory)
            else:
                logger.info(
                    "Directory -> '%s' has no wildcards. Not expanding...",
                    directory,
                )
                expanded_directories.append(directory)

        # Use ThreadPoolExecutor for outer level: processing directories/zip files
        logger.info(
            "Starting %d threads for each directory or zip file...",
            len(expanded_directories),
        )
        with ThreadPoolExecutor(
            thread_name_prefix="ProcessDirOrZip",
        ) as outer_executor:
            futures = [
                outer_executor.submit(process_directory_or_zip, directory) for directory in expanded_directories
            ]

            # Wait for all futures to complete
            for future in futures:
                future.result()

        # Collect results from the queue
        logger.info("Collecting results from worker queue...")
        results = []
        while not results_queue.empty():
            results.extend(results_queue.get())
        logger.info("Done. Collected %d results.", len(results))

    except FileNotFoundError:
        logger.error(
            "Directory or file not found during execution!",
        )
        return None
    except PermissionError:
        logger.error("Permission error during execution!")
        return None
    except TimeoutError:
        logger.error(
            "Timeout occurred while waiting for threads!",
        )
        return None
    except BrokenPipeError:
        logger.error(
            "Broken pipe error occurred during thread communication!",
        )
        return None

    return results

load_xml_files_from_directory(path_to_root, filenames, xpath=None, logger=default_logger) classmethod

Load all XML files from a directory that matches defined file names.

Then using the XPath to identify a set of elements and convert them into a Python list of dictionaries.

Parameters:

Name	Type	Description	Default
path_to_root	str	Path to the root element of the directory structure	required
filenames	list	A list of filenames. This can also be patterns like "*/en/docovw.xml". If empty all filenames ending with ".xml" is used.	required
xpath	str	The XPath to the elements we want to select.	None
logger	Logger	The logging object used for all log messages.	default_logger

Returns:

Name	Type	Description
list	list | None	List of dictionaries.

Source code in packages/pyxecm/src/pyxecm/helper/xml.py

@classmethod
def load_xml_files_from_directory(
    cls,
    path_to_root: str,
    filenames: list | None,
    xpath: str | None = None,
    logger: logging.Logger = default_logger,
) -> list | None:
    """Load all XML files from a directory that matches defined file names.

    Then using the XPath to identify a set of elements and convert them
    into a Python list of dictionaries.

    Args:
        path_to_root (str):
            Path to the root element of the
            directory structure
        filenames (list):
            A list of filenames. This can also be patterns like
            "*/en/docovw.xml". If empty all filenames ending
            with ".xml" is used.
        xpath (str, optional):
            The XPath to the elements we want to select.
        logger (logging.Logger):
            The logging object used for all log messages.

    Returns:
        list:
            List of dictionaries.

    """

    if not filenames:
        filenames = ["*.xml"]

    try:
        # Check if the provided path is a directory or a zip file that can be extracted
        # into a directory:
        if not os.path.isdir(path_to_root) and not path_to_root.endswith(".zip"):
            logger.error(
                "The provided path -> '%s' is not a valid directory or Zip file.",
                path_to_root,
            )
            return None

        # If we have a zip file we extract it - but only if it has not been extracted before:
        if path_to_root.endswith(".zip"):
            zip_file_folder = os.path.splitext(path_to_root)[0]
            if not os.path.exists(zip_file_folder):
                logger.info(
                    "Unzipping -> '%s' into folder -> '%s'...",
                    path_to_root,
                    zip_file_folder,
                )
                try:
                    with zipfile.ZipFile(path_to_root, "r") as zfile:
                        zfile.extractall(zip_file_folder)
                except zipfile.BadZipFile:
                    logger.error(
                        "Failed to extract zip file -> '%s'",
                        path_to_root,
                    )
                    return None
                except OSError:
                    logger.error(
                        "OS error occurred while trying to extract -> '%s'",
                        path_to_root,
                    )
                    return None
            else:
                logger.info(
                    "Zip file is already extracted (path -> '%s' exists). Reusing extracted data...",
                    zip_file_folder,
                )
            path_to_root = zip_file_folder

        results = []

        # Walk through the directory
        for root, _, files in os.walk(path_to_root):
            for file_data in files:
                file_path = os.path.join(root, file_data)
                file_size = os.path.getsize(file_path)
                file_name = os.path.basename(file_path)
                dir_name = os.path.dirname(file_path)

                if any(fnmatch.fnmatch(file_path, pattern) for pattern in filenames) and file_name.endswith(".xml"):
                    logger.info(
                        "Load XML file -> '%s' of size -> %s",
                        file_path,
                        file_size,
                    )
                    elements = cls.load_xml_file(
                        file_path,
                        xpath=xpath,
                        dir_name=dir_name,
                    )
                    if elements:
                        results += elements

    except NotADirectoryError:
        logger.error(
            "The given path -> '%s' is not a directory!",
            path_to_root,
        )
        return None
    except FileNotFoundError:
        logger.error(
            "The given path -> '%s' does not exist!",
            path_to_root,
        )
        return None
    except PermissionError:
        logger.error(
            "No permission to access path -> '%s'!",
            path_to_root,
        )
        return None
    except OSError:
        logger.error("Low level OS error with file -> %s", path_to_root)
        return None

    return results

modify_xml_element(xml_content, xpath, new_value, logger=default_logger) classmethod

Update the text (= content) of an XML element.

Parameters:

Name	Type	Description	Default
xml_content	str	The content of an XML file.	required
xpath	str	XML Path to identify the XML element.	required
new_value	str	The new text (content).	required
logger	Logger	The logging object used for all log messages.	default_logger

Source code in packages/pyxecm/src/pyxecm/helper/xml.py

@classmethod
def modify_xml_element(
    cls,
    xml_content: str,
    xpath: str,
    new_value: str,
    logger: logging.Logger = default_logger,
) -> None:
    """Update the text (= content) of an XML element.

    Args:
        xml_content (str):
            The content of an XML file.
        xpath (str):
            XML Path to identify the XML element.
        new_value (str):
            The new text (content).
        logger (logging.Logger):
            The logging object used for all log messages.

    """
    element = cls.get_xml_element(xml_content=xml_content, xpath=xpath)

    if element is not None:
        # Modify the XML element with the new value
        element.text = new_value
    else:
        logger.warning("XML Element -> %s not found.", xpath)

remove_xml_namespace(tag) classmethod

Remove namespace from XML tag.

Parameters:

Name	Type	Description	Default
tag	str	The XML tag with namespace.	required

Returns:

Name	Type	Description
str	str	The tag without namespace.

Source code in packages/pyxecm/src/pyxecm/helper/xml.py

@classmethod
def remove_xml_namespace(cls, tag: str) -> str:
    """Remove namespace from XML tag.

    Args:
        tag (str):
            The XML tag with namespace.

    Returns:
        str:
            The tag without namespace.

    """

    # In Python's ElementTree, the tag namespace
    # is put into curly braces like "{namespace}element"
    # that's why this method splits after the closing curly brace
    # and takes the last item (-1):

    return tag.split("}", 1)[-1]

replace_in_xml_files(directory, search_pattern, replace_string, xpath='', setting='', assoc_elem='', logger=default_logger) classmethod

Replace all occurrences of the search pattern with the replace string.

This is done in all XML files in the directory and its subdirectories.

Parameters:

Name	Type	Description	Default
directory	str	Directory to traverse for XML files	required
search_pattern	str	The string to search in the XML file. This can be empty if xpath is used!	required
replace_string	str	The replacement string.	required
xpath	str	An XPath can be given to narrow down the replacement to an XML element. For now the XPath needs to be constructed in a way the it returns one or none element.	''
setting	str	Narrow down the replacement to the line that includes the setting with this name. This parameter is optional.	''
assoc_elem	str	Lookup a specific assoc element. This parameter is optional.	''
logger	Logger	The logging object used for all log messages.	default_logger

Returns:

Name	Type	Description
bool	bool	True if a replacement happened, False otherwise

Source code in packages/pyxecm/src/pyxecm/helper/xml.py

@classmethod
def replace_in_xml_files(
    cls,
    directory: str,
    search_pattern: str,
    replace_string: str,
    xpath: str = "",
    setting: str = "",
    assoc_elem: str = "",
    logger: logging.Logger = default_logger,
) -> bool:
    """Replace all occurrences of the search pattern with the replace string.

    This is done in all XML files in the directory and its subdirectories.

    Args:
        directory (str):
            Directory to traverse for XML files
        search_pattern (str):
            The string to search in the XML file.
            This can be empty if xpath is used!
        replace_string (str):
            The replacement string.
        xpath (str, optional):
            An XPath can be given to narrow down the replacement to an XML element.
            For now the XPath needs to be constructed in a way the it returns
            one or none element.
        setting (str, optional):
            Narrow down the replacement to the line that includes the setting with this name.
            This parameter is optional.
        assoc_elem (str, optional):
            Lookup a specific assoc element. This parameter is optional.
        logger (logging.Logger):
            The logging object used for all log messages.

    Returns:
        bool:
            True if a replacement happened, False otherwise

    """

    # Define the regular expression pattern to search for
    # search pattern can be empty if an xpath is used. So
    # be careful here:
    if search_pattern:
        pattern = re.compile(search_pattern)

    found = False

    # Traverse the directory and its subdirectories
    for subdir, _, files in os.walk(directory):
        for filename in files:
            # Check if the file is an XML file
            if filename.endswith(".xml"):
                # Read the contents of the file
                file_path = os.path.join(subdir, filename)

                # if xpath is given we do an intelligent replacement
                if xpath:
                    xml_modified = False
                    logger.debug("Replacement with xpath...")
                    logger.debug(
                        "XML path -> %s, setting -> %s, assoc element -> %s",
                        xpath,
                        setting,
                        assoc_elem,
                    )
                    tree = etree.parse(file_path)
                    if not tree:
                        logger.error(
                            "Cannot parse XML tree -> %s. Skipping...",
                            file_path,
                        )
                        continue
                    root = tree.getroot()
                    # find the matching XML elements using the given XPath:
                    elements = root.xpath(xpath)
                    if not elements:
                        logger.debug(
                            "The XML file -> %s does not have any element with the given XML path -> %s. Skipping...",
                            file_path,
                            xpath,
                        )
                        continue
                    for element in elements:
                        # as XPath returns a list
                        logger.debug(
                            "Found XML element -> %s in file -> %s using xpath -> %s",
                            element.tag,
                            filename,
                            xpath,
                        )
                        # the simple case: replace the complete text of the XML element
                        if not setting and not assoc_elem:
                            logger.debug(
                                "Replace complete text of XML element -> %s from -> %s to -> %s",
                                xpath,
                                element.text,
                                replace_string,
                            )
                            element.text = replace_string
                            xml_modified = True
                        # In this case we want to set a complete value of a setting (basically replacing a whole line)
                        elif setting and not assoc_elem:
                            logger.debug(
                                "Replace single setting -> %s in XML element -> %s with new value -> %s",
                                setting,
                                xpath,
                                replace_string,
                            )
                            setting_value = cls.search_setting(
                                element.text,
                                setting,
                                is_simple=True,
                            )
                            if setting_value:
                                logger.debug(
                                    "Found existing setting value -> %s",
                                    setting_value,
                                )
                                # Check if the setting value needs to be surrounded by quotes.
                                # Only simplistic values like booleans or numeric values don't need quotes
                                if replace_string in ("true", "false", "none") or replace_string.isnumeric():
                                    replace_setting = '"' + setting + '":' + replace_string
                                else:
                                    replace_setting = '"' + setting + '":"' + replace_string + '"'
                                logger.debug(
                                    "Replacement setting -> %s",
                                    replace_setting,
                                )
                                element.text = cls.replace_setting(
                                    element_text=element.text,
                                    setting_key=setting,
                                    new_value=replace_setting,
                                    is_simple=True,
                                )
                                xml_modified = True
                            else:
                                logger.warning(
                                    "Cannot find the value for setting -> %s. Skipping...",
                                    setting,
                                )
                                continue
                        # in this case the text is just one assoc (no setting substructure)
                        elif not setting and assoc_elem:
                            logger.debug(
                                "Replace single Assoc value -> %s in XML element -> %s with -> %s",
                                assoc_elem,
                                xpath,
                                replace_string,
                            )
                            assoc_string: str = Assoc.extract_assoc_string(
                                input_string=element.text,
                            )
                            logger.debug("Assoc String -> %s", assoc_string)
                            assoc_dict = Assoc.string_to_dict(
                                assoc_string=assoc_string,
                            )
                            logger.debug("Assoc Dict -> %s", str(assoc_dict))
                            assoc_dict[assoc_elem] = replace_string  # escaped_replace_string
                            assoc_string_new: str = Assoc.dict_to_string(
                                assoc_dict=assoc_dict,
                            )
                            logger.debug(
                                "Replace assoc with -> %s",
                                assoc_string_new,
                            )
                            element.text = assoc_string_new
                            element.text = element.text.replace('"', """)
                            xml_modified = True
                        # In this case we have multiple settings with their own assocs
                        elif setting and assoc_elem:
                            logger.debug(
                                "Replace single Assoc value -> %s in setting -> %s in XML element -> %s with -> %s",
                                assoc_elem,
                                setting,
                                xpath,
                                replace_string,
                            )
                            setting_value = cls.search_setting(
                                element.text,
                                setting,
                                is_simple=False,
                            )
                            if setting_value:
                                logger.debug(
                                    "Found setting value -> %s",
                                    setting_value,
                                )
                                assoc_string: str = Assoc.extract_assoc_string(
                                    input_string=setting_value,
                                )
                                logger.debug("Assoc String -> %s", assoc_string)
                                assoc_dict = Assoc.string_to_dict(
                                    assoc_string=assoc_string,
                                )
                                logger.debug("Assoc Dict -> %s", str(assoc_dict))
                                escaped_replace_string = replace_string.replace(
                                    "'",
                                    "\\\\\u0027",
                                )
                                logger.debug(
                                    "Escaped replacement string -> %s",
                                    escaped_replace_string,
                                )
                                assoc_dict[assoc_elem] = escaped_replace_string  # escaped_replace_string
                                assoc_string_new: str = Assoc.dict_to_string(
                                    assoc_dict=assoc_dict,
                                )
                                assoc_string_new = assoc_string_new.replace(
                                    "'",
                                    "\\u0027",
                                )
                                replace_setting = '"' + setting + '":"' + assoc_string_new + '"'
                                logger.debug(
                                    "Replacement setting -> %s",
                                    replace_setting,
                                )
                                # here we need to apply a "trick". It is required
                                # as regexp cannot handle the special unicode escapes \u0027
                                # we require. We first insert a placeholder "PLACEHOLDER"
                                # and let regexp find the right place for it. Then further
                                # down we use a simple search&replace to switch the PLACEHOLDER
                                # to the real value (replace() does not have the issues with unicode escapes)
                                element.text = cls.replace_setting(
                                    element_text=element.text,
                                    setting_key=setting,
                                    new_value="PLACEHOLDER",
                                    is_simple=False,
                                    is_escaped=False,
                                )
                                element.text = element.text.replace(
                                    "PLACEHOLDER",
                                    replace_setting,
                                )
                                element.text = element.text.replace('"', """)
                                xml_modified = True
                            else:
                                logger.warning(
                                    "Cannot find the value for setting -> %s. Skipping...",
                                    setting,
                                )
                                continue
                    if xml_modified:
                        logger.debug(
                            "XML tree has been modified. Write updated file -> %s...",
                            file_path,
                        )

                        new_contents = etree.tostring(
                            tree,
                            pretty_print=True,
                            xml_declaration=True,
                            encoding="UTF-8",
                        )
                        # we need to undo some of the stupid things tostring() did:
                        new_contents = new_contents.replace(
                            b""",
                            b""",
                        )
                        new_contents = new_contents.replace(
                            b"'",
                            b"'",
                        )
                        new_contents = new_contents.replace(b">", b">")
                        new_contents = new_contents.replace(b"<", b"<")

                        # Replace single quotes inside double quotes strings with "'" (manual escaping)
                        # This is required as we next want to replace all double quotes with single quotes
                        pattern = b'"([^"]*)"'
                        new_contents = re.sub(
                            pattern,
                            lambda m: m.group(0).replace(b"'", b"'"),
                            new_contents,
                        )

                        # Replace single quotes in XML text elements with "'"
                        # and replace double quotes in XML text elements with """
                        # This is required as we next want to replace all double quotes with single quotes
                        # to make the XML files as similar as possible with Extended ECM's format
                        pattern = b">([^<>]+?)<"
                        replacement = lambda match: match.group(0).replace(  # noqa: E731
                            b'"',
                            b"&quot;",
                        )
                        new_contents = re.sub(pattern, replacement, new_contents)
                        replacement = lambda match: match.group(0).replace(  # noqa: E731
                            b"'",
                            b"&apos;",
                        )
                        new_contents = re.sub(pattern, replacement, new_contents)

                        # Change double quotes to single quotes across the XML file - Extended ECM has it that way:
                        new_contents = new_contents.replace(b'"', b"'")

                        # Write the updated contents to the file.
                        # We DO NOT want to use tree.write() here
                        # as it would undo the manual XML tweaks we
                        # need for Extended ECM. We also need "wb"
                        # as we have bytes and not str as a data type
                        with open(file_path, "wb") as f:
                            f.write(new_contents)

                        found = True
                # this is not using xpath - do a simple search and replace
                else:
                    logger.debug("Replacement without xpath...")
                    with open(file_path, encoding="UTF-8") as f:
                        contents = f.read()
                    # Replace all occurrences of the search pattern with the replace string
                    new_contents = pattern.sub(replace_string, contents)

                    # Write the updated contents to the file if there were replacements
                    if contents != new_contents:
                        logger.debug(
                            "Found search string -> %s in XML file -> %s. Write updated file...",
                            search_pattern,
                            file_path,
                        )
                        # Write the updated contents to the file
                        with open(file_path, "w", encoding="UTF-8") as f:
                            f.write(new_contents)
                        found = True

    return found

replace_setting(element_text, setting_key, new_value, is_simple=True, is_escaped=False) classmethod

Replace the value of a defined setting with a new value.

The simple case covers settings like this: "syncCandidates":true, "syncCandidates":true, In this case the setting value is a scalar like true, false, a number or none the regular expression pattern searches for a setting name in "..." (quotes) followed by a colon (:). The value is taken from what follows the colon until the next comma (,)

The more complex case is a string value that may itself have commas, so we cannot look for comma as a delimiter like in the simple case but we take the value for a string delimited by double quotes ("...")

Parameters:

Name	Type	Description	Default
element_text	str	The original text of the XML element (that is to be updated).	required
setting_key	str	The name of the setting.	required
new_value	str	The new value of the setting.	required
is_simple	bool	True = value is a scalar like true, false, a number or none. Defaults to True.	True
is_escaped	bool	True if the value is surrrounded with ". Defaults to False.	False

Returns:

Name	Type	Description
str	str	The updated element text.

Source code in packages/pyxecm/src/pyxecm/helper/xml.py

@classmethod
def replace_setting(
    cls,
    element_text: str,
    setting_key: str,
    new_value: str,
    is_simple: bool = True,
    is_escaped: bool = False,
) -> str:
    """Replace the value of a defined setting with a new value.

    The simple case covers settings like this:
    "syncCandidates":true,
    "syncCandidates":true,
    In this case the setting value is a scalar like true, false, a number or none
    the regular expression pattern searches for a setting name in "..." (quotes) followed
    by a colon (:). The value is taken from what follows the colon until the next comma (,)

    The more complex case is a string value that may itself have commas,
    so we cannot look for comma as a delimiter like in the simple case
    but we take the value for a string delimited by double quotes ("...")

    Args:
        element_text (str):
            The original text of the XML element (that is to be updated).
        setting_key (str):
            The name of the setting.
        new_value (str):
            The new value of the setting.
        is_simple (bool, optional):
            True = value is a scalar like true, false, a number or none. Defaults to True.
        is_escaped (bool, optional):
            True if the value is surrrounded with ". Defaults to False.

    Returns:
        str:
            The updated element text.

    """

    if is_simple:
        pattern = r""{}":[^,]*".format(setting_key) if is_escaped else r'"{}":[^,]*'.format(setting_key)
    elif is_escaped:
        pattern = r""{}":".*"".format(setting_key)
    else:
        pattern = r'"{}":"([^"]*)"'.format(setting_key)

    new_text = re.sub(pattern, new_value, element_text)

    return new_text

search_setting(element_text, setting_key, is_simple=True, is_escaped=False) classmethod

Search a setting in an XML element and return its value.

The simple case covers settings like this: "syncCandidates":true, "syncCandidates":true, In this case the setting value is a scalar like true, false, a number or none the regular expression pattern searches for a setting name in "..." (quotes) followed by a colon (:). The value is taken from what follows the colon until the next comma (,)

The more complex case is a string value that may itself have commas, so we cannot look for comma as a delimiter like in the simple case but we take the value for a string delimited by double quotes ("...")

Parameters:

Name	Type	Description	Default
element_text	str	The text to examine - typically content of an XML element.	required
setting_key	str	The name of the setting key (before the colon).	required
is_simple	bool	True if the value is scalar (not having assocs with commas). Defaults to True.	True
is_escaped	bool	True if the quotes or escaped with ". Defaults to False.	False

Returns:

Name	Type	Description
str	str | None	The value of the setting or None if the setting is not found.

Source code in packages/pyxecm/src/pyxecm/helper/xml.py

@classmethod
def search_setting(
    cls,
    element_text: str,
    setting_key: str,
    is_simple: bool = True,
    is_escaped: bool = False,
) -> str | None:
    """Search a setting in an XML element and return its value.

    The simple case covers settings like this:
    "syncCandidates":true,
    "syncCandidates":true,
    In this case the setting value is a scalar like true, false, a number or none
    the regular expression pattern searches for a setting name in "..." (quotes) followed
    by a colon (:). The value is taken from what follows the colon until the next comma (,)

    The more complex case is a string value that may itself have commas,
    so we cannot look for comma as a delimiter like in the simple case
    but we take the value for a string delimited by double quotes ("...")

    Args:
        element_text (str):
            The text to examine - typically content of an XML element.
        setting_key (str):
            The name of the setting key (before the colon).
        is_simple (bool, optional):
            True if the value is scalar (not having assocs with commas). Defaults to True.
        is_escaped (bool, optional):
            True if the quotes or escaped with ". Defaults to False.

    Returns:
        str:
            The value of the setting or None if the setting is not found.

    """

    if is_simple:
        pattern = r""{}":[^,]*".format(setting_key) if is_escaped else r'"{}":[^,]*'.format(setting_key)
    elif is_escaped:
        pattern = r""{}":".*"".format(setting_key)
    else:
        pattern = r'"{}":"([^"]*)"'.format(setting_key)

    match = re.search(pattern, element_text)
    if match:
        setting_line = match.group(0)
        setting_value = setting_line.split(":")[1]
        return setting_value
    else:
        return None

xml_to_dict(xml_string, encode=False, include_attributes=False) classmethod

Parse XML string and return a dictionary without namespaces.

Parameters:

Name	Type	Description	Default
xml_string	str	The XML string to process.	required
encode	bool	True if the XML string should be encoded to UTF-8 bytes. Defaults to False.	False
include_attributes	bool	True if XML attributes should be included in the dictionary. Defaults to False.	False

Returns:

Name	Type	Description
dict	dict	The XML structure converted to a dictionary.

Source code in packages/pyxecm/src/pyxecm/helper/xml.py

@classmethod
def xml_to_dict(cls, xml_string: str, encode: bool = False, include_attributes: bool = False) -> dict:
    """Parse XML string and return a dictionary without namespaces.

    Args:
        xml_string (str):
            The XML string to process.
        encode (bool, optional):
            True if the XML string should be encoded to UTF-8 bytes. Defaults to False.
        include_attributes (bool, optional):
            True if XML attributes should be included in the dictionary. Defaults to False.

    Returns:
        dict:
            The XML structure converted to a dictionary.

    """

    def xml_element_to_dict(element: Element) -> dict:
        """Convert XML element to dictionary.

        Args:
            element (Element):
                The XML element.

        Returns:
            dict:
                Dictionary representing the XML element

        """

        tag = cls.remove_xml_namespace(element.tag)
        children = list(element)
        node_dict = {}

        if element.attrib and include_attributes:
            node_dict.update({"@{}".format(k): v for k, v in element.attrib.items()})

        if children:
            child_dict = {}
            for child in children:
                child_tag = cls.remove_xml_namespace(child.tag)
                child_value = xml_element_to_dict(child)

                # child_dict is {child_tag: ...}, we want the value inside
                value = child_value[child_tag]

                # Handle multiple occurrences of the same tag:
                if child_tag in child_dict:
                    # If the tag already exists, ensure it becomes a list
                    if not isinstance(child_dict[child_tag], list):
                        child_dict[child_tag] = [child_dict[child_tag]]
                    child_dict[child_tag].append(value)
                else:
                    child_dict[child_tag] = value

            # Merge children into node_dict
            node_dict.update(child_dict)

        # Add text if present and meaningful
        text = element.text.strip() if element.text and element.text.strip() else None
        if text:
            if node_dict:
                node_dict["#text"] = text
            else:
                return {tag: text}

        return {tag: node_dict if node_dict else text}

    if encode:
        xml_string = xml_string.encode("utf-8")
    root = etree.fromstring(xml_string)

    return xml_element_to_dict(root)

Web

Module to implement functions to execute Web Requests.

HTTP

Class HTTP is used to issue HTTP request and test if hosts are reachable.

Source code in packages/pyxecm/src/pyxecm/helper/web.py

class HTTP:
    """Class HTTP is used to issue HTTP request and test if hosts are reachable."""

    logger: logging.Logger = default_logger

    def __init__(
        self,
        logger: logging.Logger = default_logger,
    ) -> None:
        """Initialize the HTTP object."""

        if logger != default_logger:
            self.logger = logger.getChild("http")
            for logfilter in logger.filters:
                self.logger.addFilter(logfilter)

    # end method definition

    def check_host_reachable(self, hostname: str, port: int = 80) -> bool:
        """Check if a server / web address is reachable.

        Args:
            hostname (str):
                The endpoint hostname.
            port (int):
                The endpoint port.

        Results:
            bool:
                True is reachable, False otherwise

        """

        self.logger.debug(
            "Test if host -> '%s' is reachable on port -> %s ...",
            hostname,
            str(port),
        )
        try:
            socket.getaddrinfo(hostname, port)
        except socket.gaierror as exception:
            self.logger.warning(
                "Address-related error - cannot reach host -> %s; error -> %s",
                hostname,
                exception.strerror,
            )
            return False
        except OSError as exception:
            self.logger.warning(
                "Connection error - cannot reach host -> %s; error -> %s",
                hostname,
                exception.strerror,
            )
            return False
        else:
            self.logger.debug("Host is reachable at -> %s:%s", hostname, str(port))
            return True

    # end method definition

    def http_request(
        self,
        url: str,
        method: str = "POST",
        payload: dict | None = None,
        headers: dict | None = None,
        timeout: float | None = REQUEST_TIMEOUT,
        retries: int = REQUEST_MAX_RETRIES,
        wait_time: float = REQUEST_RETRY_DELAY,
        wait_on_status: list | None = None,
        show_error: bool = True,
        stream: bool = False,
    ) -> dict | None:
        """Issues an http request to a given URL.

        Args:
            url (str):
                The URL of the request.
            method (str, optional):
                Method of the request (POST, PUT, GET, ...). Defaults to "POST".
            payload (dict, optional):
                Request payload. Defaults to None.
            headers (dict, optional):
                Request header. Defaults to None. If None then a default
                value defined in REQUEST_FORM_HEADERS is used.
            timeout (float | None, optional):
                The timeout in seconds. Defaults to REQUEST_TIMEOUT.
            retries (int, optional):
                The number of retries. If -1 then unlimited retries.
                Defaults to REQUEST_MAX_RETRIES.
            wait_time (int, optional):
                The number of seconds to wait after each try.
                Defaults to REQUEST_RETRY_DELAY.
            wait_on_status (list, optional):
                A list of status codes we want to wait on.
                If None or empty then we wait for all return codes if
                wait_time > 0.
            show_error (bool, optional):
                Whether to show an error or a warning message in case of an error.
            stream (bool, optional):
                Enable stream for response content (e.g. for downloading large files).

        Returns:
            dict | None:
                Response of call

        """

        if not headers:
            headers = REQUEST_FORM_HEADERS

        message = "Make HTTP request to URL -> '{}' using -> {} method".format(
            url,
            method,
        )
        if payload:
            message += " with payload -> {}".format(payload)
        if retries:
            message += " (max number of retries -> {}, wait time between retries -> {})".format(
                retries,
                wait_time,
            )
            try:
                retries = int(retries)
            except ValueError:
                self.logger.warning(
                    "HTTP request -> retries is not a valid integer value: %s, defaulting to 0 retries ",
                    retries,
                )
                retries = 0

        self.logger.debug(message)

        try_counter = 1

        while True:
            try:
                response = requests.request(
                    method=method,
                    url=url,
                    data=payload,
                    headers=headers,
                    timeout=timeout,
                    stream=stream,
                )
            except requests.RequestException as exc:
                response = None
                self.logger.warning(
                    "HTTP request -> %s to url -> %s failed (try %s); error -> %s",
                    method,
                    url,
                    try_counter,
                    exc,
                )

            if response is not None:
                # Do we have a successful result?
                if response.ok:
                    self.logger.debug(
                        "HTTP request -> %s to url -> %s succeeded with status -> %s!",
                        method,
                        url,
                        response.status_code,
                    )

                    if wait_on_status and response.status_code in wait_on_status:
                        self.logger.debug(
                            "%s is in wait_on_status list -> %s",
                            response.status_code,
                            wait_on_status,
                        )
                    else:
                        return response

                else:
                    message = "HTTP request -> {} to url -> {} failed; status -> {}; error -> {}".format(
                        method,
                        url,
                        response.status_code,
                        (
                            response.text
                            if response.headers.get("content-type") == "application/json"
                            else "see debug log"
                        ),
                    )
                    if show_error and retries == 0:
                        self.logger.error(message)
                    else:
                        self.logger.warning(message)
            # end if response is not None

            # Check if another retry is allowed, if not return None
            if retries == 0:
                return None

            if wait_time > 0.0:
                self.logger.warning(
                    "Sleeping %s seconds and then trying once more...",
                    str(wait_time * try_counter),
                )
                time.sleep(wait_time * try_counter)

            retries -= 1
            try_counter += 1
        # end while True:

    # end method definition

    def download_file(
        self,
        url: str,
        filename: str,
        timeout: float = REQUEST_TIMEOUT,
        retries: int | None = REQUEST_MAX_RETRIES,
        wait_time: float = REQUEST_RETRY_DELAY,
        wait_on_status: list | None = None,
        chunk_size: int = 8192,
        show_error: bool = True,
    ) -> bool:
        """Download a file from a URL.

        Args:
            url (str):
                The URL to open / load.
            filename (str):
                The filename to save the content.
            timeout (float, optional):
                The timeout in seconds.
            retries (int, optional):
                The number of retries. If -1 then unlimited retries.
            wait_time (float, optional):
                The number of seconds to wait after each try.
            wait_on_status (list, optional):
                The list of status codes we want to wait on.
                If None or empty then we wait for all return codes if
                wait_time > 0.
            chunk_size (int, optional):
                Chunk size for reading file content. Default is 8192.
            show_error (bool, optional):
                Whether or not an error show logged if download fails.
                Default is True.

        Returns:
            bool:
                True if successful, False otherwise.

        """

        # Validate the URL:
        parsed_url = urlparse(url)
        if not parsed_url.scheme or not parsed_url.netloc:
            self.logger.error("Invalid URL -> '%s' to download a file!", url)
            return False

        response = self.http_request(
            url=url,
            method="GET",
            retries=retries,
            timeout=timeout,
            wait_time=wait_time,
            wait_on_status=wait_on_status,
            show_error=show_error,
            stream=True,  # for downloads we want streaming
        )

        if not response or not response.ok:
            self.logger.error(
                "Failed to request download file -> '%s' from site -> %s%s",
                filename,
                url,
                "; error -> {}".format(response.text) if response else "",
            )
            return False

        try:
            directory = os.path.dirname(filename)
            if not os.path.exists(directory):
                self.logger.info(
                    "Download directory -> '%s' does not exist, creating it.",
                    directory,
                )
                os.makedirs(directory)
            with open(filename, "wb") as download_file:
                download_file.writelines(response.iter_content(chunk_size=chunk_size))
            self.logger.debug(
                "File downloaded successfully as -> '%s' (size -> %s).",
                filename,
                self.human_readable_size(os.path.getsize(filename)),
            )
        except (OSError, requests.exceptions.RequestException):
            self.logger.error(
                "Cannot write content to file -> '%s' in directory -> '%s'!",
                filename,
                directory,
            )
            return False
        else:
            return True

    # end method definition

    def human_readable_size(self, size_in_bytes: int) -> str:
        """Return a file size in human readable form.

        Args:
            size_in_bytes (int): The file size in bytes.

        Returns:
            str:
                The formatted size using units.

        """

        for unit in ["B", "KB", "MB", "GB", "TB"]:
            if size_in_bytes < 1024:
                return "{:.2f} {}".format(size_in_bytes, unit)
            size_in_bytes /= 1024

        # We should never get here but linter wants it:
        return "{:.2f}".format(size_in_bytes)

    # end method definition

    def extract_content(self, url: str, xpath: str) -> str | None:
        """Extract a string from a response of a HTTP request based on an XPath.

        Args:
            url (str):
                The URL to open / load.
            xpath (str):
                The XPath expression to apply to the result.

        Returns:
            str | None:
                Extracted string or None in case of an error.

        """

        # Send a GET request to the URL:
        response = self.http_request(
            url=url,
            method="GET",
        )

        # Check if request was successful
        if response and response.ok:
            # Parse the HTML content
            tree = html.fromstring(response.content)

            # Extract content using XPath
            elements = tree.xpath(xpath)

            # Get text content of all elements and join them
            content = "\n".join([elem.text_content().strip() for elem in elements])

            # Return the extracted content:
            return content

        # If request was not successful, print error message:
        self.logger.error(
            "Cannot extract content from URL -> '%s'; error code -> %s",
            url,
            response.status_code,
        )

        return None

__init__(logger=default_logger)

Initialize the HTTP object.

Source code in packages/pyxecm/src/pyxecm/helper/web.py

def __init__(
    self,
    logger: logging.Logger = default_logger,
) -> None:
    """Initialize the HTTP object."""

    if logger != default_logger:
        self.logger = logger.getChild("http")
        for logfilter in logger.filters:
            self.logger.addFilter(logfilter)

check_host_reachable(hostname, port=80)

Check if a server / web address is reachable.

Parameters:

Name	Type	Description	Default
hostname	str	The endpoint hostname.	required
port	int	The endpoint port.	80

Results

bool: True is reachable, False otherwise

Source code in packages/pyxecm/src/pyxecm/helper/web.py

def check_host_reachable(self, hostname: str, port: int = 80) -> bool:
    """Check if a server / web address is reachable.

    Args:
        hostname (str):
            The endpoint hostname.
        port (int):
            The endpoint port.

    Results:
        bool:
            True is reachable, False otherwise

    """

    self.logger.debug(
        "Test if host -> '%s' is reachable on port -> %s ...",
        hostname,
        str(port),
    )
    try:
        socket.getaddrinfo(hostname, port)
    except socket.gaierror as exception:
        self.logger.warning(
            "Address-related error - cannot reach host -> %s; error -> %s",
            hostname,
            exception.strerror,
        )
        return False
    except OSError as exception:
        self.logger.warning(
            "Connection error - cannot reach host -> %s; error -> %s",
            hostname,
            exception.strerror,
        )
        return False
    else:
        self.logger.debug("Host is reachable at -> %s:%s", hostname, str(port))
        return True

download_file(url, filename, timeout=REQUEST_TIMEOUT, retries=REQUEST_MAX_RETRIES, wait_time=REQUEST_RETRY_DELAY, wait_on_status=None, chunk_size=8192, show_error=True)

Download a file from a URL.

Parameters:

Name	Type	Description	Default
url	str	The URL to open / load.	required
filename	str	The filename to save the content.	required
timeout	float	The timeout in seconds.	REQUEST_TIMEOUT
retries	int	The number of retries. If -1 then unlimited retries.	REQUEST_MAX_RETRIES
wait_time	float	The number of seconds to wait after each try.	REQUEST_RETRY_DELAY
wait_on_status	list	The list of status codes we want to wait on. If None or empty then we wait for all return codes if wait_time > 0.	None
chunk_size	int	Chunk size for reading file content. Default is 8192.	8192
show_error	bool	Whether or not an error show logged if download fails. Default is True.	True

Returns:

Name	Type	Description
bool	bool	True if successful, False otherwise.

Source code in packages/pyxecm/src/pyxecm/helper/web.py

def download_file(
    self,
    url: str,
    filename: str,
    timeout: float = REQUEST_TIMEOUT,
    retries: int | None = REQUEST_MAX_RETRIES,
    wait_time: float = REQUEST_RETRY_DELAY,
    wait_on_status: list | None = None,
    chunk_size: int = 8192,
    show_error: bool = True,
) -> bool:
    """Download a file from a URL.

    Args:
        url (str):
            The URL to open / load.
        filename (str):
            The filename to save the content.
        timeout (float, optional):
            The timeout in seconds.
        retries (int, optional):
            The number of retries. If -1 then unlimited retries.
        wait_time (float, optional):
            The number of seconds to wait after each try.
        wait_on_status (list, optional):
            The list of status codes we want to wait on.
            If None or empty then we wait for all return codes if
            wait_time > 0.
        chunk_size (int, optional):
            Chunk size for reading file content. Default is 8192.
        show_error (bool, optional):
            Whether or not an error show logged if download fails.
            Default is True.

    Returns:
        bool:
            True if successful, False otherwise.

    """

    # Validate the URL:
    parsed_url = urlparse(url)
    if not parsed_url.scheme or not parsed_url.netloc:
        self.logger.error("Invalid URL -> '%s' to download a file!", url)
        return False

    response = self.http_request(
        url=url,
        method="GET",
        retries=retries,
        timeout=timeout,
        wait_time=wait_time,
        wait_on_status=wait_on_status,
        show_error=show_error,
        stream=True,  # for downloads we want streaming
    )

    if not response or not response.ok:
        self.logger.error(
            "Failed to request download file -> '%s' from site -> %s%s",
            filename,
            url,
            "; error -> {}".format(response.text) if response else "",
        )
        return False

    try:
        directory = os.path.dirname(filename)
        if not os.path.exists(directory):
            self.logger.info(
                "Download directory -> '%s' does not exist, creating it.",
                directory,
            )
            os.makedirs(directory)
        with open(filename, "wb") as download_file:
            download_file.writelines(response.iter_content(chunk_size=chunk_size))
        self.logger.debug(
            "File downloaded successfully as -> '%s' (size -> %s).",
            filename,
            self.human_readable_size(os.path.getsize(filename)),
        )
    except (OSError, requests.exceptions.RequestException):
        self.logger.error(
            "Cannot write content to file -> '%s' in directory -> '%s'!",
            filename,
            directory,
        )
        return False
    else:
        return True

extract_content(url, xpath)

Extract a string from a response of a HTTP request based on an XPath.

Parameters:

Name	Type	Description	Default
url	str	The URL to open / load.	required
xpath	str	The XPath expression to apply to the result.	required

Returns:

Type	Description
str | None	str | None: Extracted string or None in case of an error.

Source code in packages/pyxecm/src/pyxecm/helper/web.py

def extract_content(self, url: str, xpath: str) -> str | None:
    """Extract a string from a response of a HTTP request based on an XPath.

    Args:
        url (str):
            The URL to open / load.
        xpath (str):
            The XPath expression to apply to the result.

    Returns:
        str | None:
            Extracted string or None in case of an error.

    """

    # Send a GET request to the URL:
    response = self.http_request(
        url=url,
        method="GET",
    )

    # Check if request was successful
    if response and response.ok:
        # Parse the HTML content
        tree = html.fromstring(response.content)

        # Extract content using XPath
        elements = tree.xpath(xpath)

        # Get text content of all elements and join them
        content = "\n".join([elem.text_content().strip() for elem in elements])

        # Return the extracted content:
        return content

    # If request was not successful, print error message:
    self.logger.error(
        "Cannot extract content from URL -> '%s'; error code -> %s",
        url,
        response.status_code,
    )

    return None

http_request(url, method='POST', payload=None, headers=None, timeout=REQUEST_TIMEOUT, retries=REQUEST_MAX_RETRIES, wait_time=REQUEST_RETRY_DELAY, wait_on_status=None, show_error=True, stream=False)

Issues an http request to a given URL.

Parameters:

Name	Type	Description	Default
url	str	The URL of the request.	required
method	str	Method of the request (POST, PUT, GET, ...). Defaults to "POST".	'POST'
payload	dict	Request payload. Defaults to None.	None
headers	dict	Request header. Defaults to None. If None then a default value defined in REQUEST_FORM_HEADERS is used.	None
timeout	float | None	The timeout in seconds. Defaults to REQUEST_TIMEOUT.	REQUEST_TIMEOUT
retries	int	The number of retries. If -1 then unlimited retries. Defaults to REQUEST_MAX_RETRIES.	REQUEST_MAX_RETRIES
wait_time	int	The number of seconds to wait after each try. Defaults to REQUEST_RETRY_DELAY.	REQUEST_RETRY_DELAY
wait_on_status	list	A list of status codes we want to wait on. If None or empty then we wait for all return codes if wait_time > 0.	None
show_error	bool	Whether to show an error or a warning message in case of an error.	True
stream	bool	Enable stream for response content (e.g. for downloading large files).	False

Returns:

Type	Description
dict | None	dict | None: Response of call

Source code in packages/pyxecm/src/pyxecm/helper/web.py

def http_request(
    self,
    url: str,
    method: str = "POST",
    payload: dict | None = None,
    headers: dict | None = None,
    timeout: float | None = REQUEST_TIMEOUT,
    retries: int = REQUEST_MAX_RETRIES,
    wait_time: float = REQUEST_RETRY_DELAY,
    wait_on_status: list | None = None,
    show_error: bool = True,
    stream: bool = False,
) -> dict | None:
    """Issues an http request to a given URL.

    Args:
        url (str):
            The URL of the request.
        method (str, optional):
            Method of the request (POST, PUT, GET, ...). Defaults to "POST".
        payload (dict, optional):
            Request payload. Defaults to None.
        headers (dict, optional):
            Request header. Defaults to None. If None then a default
            value defined in REQUEST_FORM_HEADERS is used.
        timeout (float | None, optional):
            The timeout in seconds. Defaults to REQUEST_TIMEOUT.
        retries (int, optional):
            The number of retries. If -1 then unlimited retries.
            Defaults to REQUEST_MAX_RETRIES.
        wait_time (int, optional):
            The number of seconds to wait after each try.
            Defaults to REQUEST_RETRY_DELAY.
        wait_on_status (list, optional):
            A list of status codes we want to wait on.
            If None or empty then we wait for all return codes if
            wait_time > 0.
        show_error (bool, optional):
            Whether to show an error or a warning message in case of an error.
        stream (bool, optional):
            Enable stream for response content (e.g. for downloading large files).

    Returns:
        dict | None:
            Response of call

    """

    if not headers:
        headers = REQUEST_FORM_HEADERS

    message = "Make HTTP request to URL -> '{}' using -> {} method".format(
        url,
        method,
    )
    if payload:
        message += " with payload -> {}".format(payload)
    if retries:
        message += " (max number of retries -> {}, wait time between retries -> {})".format(
            retries,
            wait_time,
        )
        try:
            retries = int(retries)
        except ValueError:
            self.logger.warning(
                "HTTP request -> retries is not a valid integer value: %s, defaulting to 0 retries ",
                retries,
            )
            retries = 0

    self.logger.debug(message)

    try_counter = 1

    while True:
        try:
            response = requests.request(
                method=method,
                url=url,
                data=payload,
                headers=headers,
                timeout=timeout,
                stream=stream,
            )
        except requests.RequestException as exc:
            response = None
            self.logger.warning(
                "HTTP request -> %s to url -> %s failed (try %s); error -> %s",
                method,
                url,
                try_counter,
                exc,
            )

        if response is not None:
            # Do we have a successful result?
            if response.ok:
                self.logger.debug(
                    "HTTP request -> %s to url -> %s succeeded with status -> %s!",
                    method,
                    url,
                    response.status_code,
                )

                if wait_on_status and response.status_code in wait_on_status:
                    self.logger.debug(
                        "%s is in wait_on_status list -> %s",
                        response.status_code,
                        wait_on_status,
                    )
                else:
                    return response

            else:
                message = "HTTP request -> {} to url -> {} failed; status -> {}; error -> {}".format(
                    method,
                    url,
                    response.status_code,
                    (
                        response.text
                        if response.headers.get("content-type") == "application/json"
                        else "see debug log"
                    ),
                )
                if show_error and retries == 0:
                    self.logger.error(message)
                else:
                    self.logger.warning(message)
        # end if response is not None

        # Check if another retry is allowed, if not return None
        if retries == 0:
            return None

        if wait_time > 0.0:
            self.logger.warning(
                "Sleeping %s seconds and then trying once more...",
                str(wait_time * try_counter),
            )
            time.sleep(wait_time * try_counter)

        retries -= 1
        try_counter += 1

human_readable_size(size_in_bytes)

Return a file size in human readable form.

Parameters:

Name	Type	Description	Default
size_in_bytes	int	The file size in bytes.	required

Returns:

Name	Type	Description
str	str	The formatted size using units.

Source code in packages/pyxecm/src/pyxecm/helper/web.py

def human_readable_size(self, size_in_bytes: int) -> str:
    """Return a file size in human readable form.

    Args:
        size_in_bytes (int): The file size in bytes.

    Returns:
        str:
            The formatted size using units.

    """

    for unit in ["B", "KB", "MB", "GB", "TB"]:
        if size_in_bytes < 1024:
            return "{:.2f} {}".format(size_in_bytes, unit)
        size_in_bytes /= 1024

    # We should never get here but linter wants it:
    return "{:.2f}".format(size_in_bytes)