"""String parser. Copied from https://github.com/theorchard/python-vector-utils/blob/e42e786432aa6e1a1c279d4046c0c6d8e7109756/vector_utils/utils/parsers.py """ import hashlib import re import zlib from delivery_metadata.exceptions import UnsupportedPlaceholder def process_string( original_string: str, variables: dict[str, str | int], skip_timestamp: bool = False ) -> str: """Pattern replacement helper. Args: original_string (str): The original string variables (dict): Replacement variables skip_timestamp (bool): skip timestamp placeholder if true Returns: str: Parsed string """ md5 = False crc = False crc_pos = False if original_string[:5] == "[MD5]": md5 = True original_string = original_string[5:] elif original_string[:5] == "[CRC]": crc = True original_string = original_string[5:] elif original_string[:6] == "[CRC+]": crc_pos = True original_string = original_string[6:] breakdown = _split_string(original_string) return_string = _parse_vars(breakdown, variables, skip_timestamp) if md5: return_string = hashlib.md5(return_string.encode("utf-8")).hexdigest() elif crc: return_string = str(_get_signed_int(zlib.crc32(return_string.encode("utf-8")))) elif crc_pos: return_string = str(zlib.crc32(return_string.encode("utf-8"))) return return_string def _split_string(original_string: str) -> list[str]: """Splits string by specail characters. Args: original_string (str): The original string Returns: list: A list of the split string """ pattern = "({(?:[^{}]+(?:\"[^\"]*\"|'[^']*')?)+})" return re.split(pattern, original_string) def _parse_vars( breakdown: list[str], variables: dict[str, str | int], skip_timestamp: bool = False ) -> str: """Replace variables with actual values. Args: breakdown (list): List of string chunks variables (dict): List of values for variable replacement skip_timestamp (boo): skip timestamp placeholder if true Returns: str: Parsed string with actual values """ return_string = "" for tmp_str in breakdown: if tmp_str == "{identifier_timestamp}" and skip_timestamp: return_string = "".join([return_string, tmp_str]) continue if re.search("{.*}$", tmp_str): tmp_str = tmp_str.lstrip("{").rstrip("}") len_vars = "" if "(" in tmp_str: var_name, len_vars, *remainders = tmp_str.split("(") else: var_name = tmp_str if var_name not in variables: raise UnsupportedPlaceholder(f"{var_name} not supported") var_value = variables[var_name] if len_vars: len_vars_arr = len_vars.strip(")").split(",") if len(len_vars_arr) == 1: return_string = "".join( [ return_string, "{:0{}d}".format(int(var_value), int(len_vars_arr[0])), ] ) else: first_pos = int(len_vars_arr[0]) second_pos = int(len_vars_arr[1]) + first_pos return_string = "".join( [return_string, str(var_value)[first_pos:second_pos]] ) else: return_string = "".join([return_string, str(var_value)]) else: return_string = "".join([return_string, tmp_str]) return return_string def _get_signed_int(input_int: int) -> int: """Returns a signed integer. Args: input_int (int): Unsinged int Returns: int: Signed int """ sign_bit = 1 << 31 if input_int & sign_bit: invert = input_int ^ (2**32 - 1) return -1 * (invert + 1) return input_int