Source code for mafw.tools.regexp

#  Copyright 2025–2026 European Union
#  Author: Bulgheroni Antonio (antonio.bulgheroni@ec.europa.eu)
#  SPDX-License-Identifier: EUPL-1.2
"""
Module implements some basic functions involving regular expressions.
"""

import logging
import re

from mafw.mafw_errors import UnknownProcessor

log = logging.getLogger(__name__)




def extract_protocol(url: str) -> str | None:
    """
    Extract the protocol portion from a database connection URL.

    The extract_protocol function takes a database connection URL string as input and extracts the protocol portion
    (the part before "://"). This function is useful for identifying the database type from connection strings.

    :param url: The url from which the protocol will be extracted.
    :type url: str
    :return: The protocol or None, if the extraction failed
    :rtype: str | None
    """
    pattern = r'^([a-z0-9_\-+.]+)://'
    match = re.match(pattern, url)
    if match:
        return match.group(1)
    return None





def normalize_sql_spaces(sql_string: str) -> str:
    """
    Normalize multiple consecutive spaces in SQL string to single spaces.
    Only handles spaces, preserves other whitespace characters.

    :param sql_string: The SQL string for space normalization.
    :type sql_string: str
    :return: The normalized SQL command.
    :rtype: str
    """
    return re.sub(r' +', ' ', sql_string.strip())





def parse_processor_name(processor_string: str) -> tuple[str, str | None]:
    """
    Parse a processor name string into name and replica identifier components.

    Given a string in the form 'MyProcessorName#156a', returns a tuple ('MyProcessorName', '156a').
    If the input string is 'MyProcessorName' only, then it returns ('MyProcessorName', None).
    If it gets 'MyProcessorName#', it returns ('MyProcessorName', None) but emits a warning
    informing of a possible malformed name.

    The name portion may be a bare ``Short_Name`` (a single Python identifier such as
    ``MyProcessorName``) or a dotted ``Qualified_Name`` (``package.ShortName``, for example
    ``proc_lib1.SameNameProcessor``). A qualified name combined with a replica identifier, such as
    ``proc_lib1.SameNameProcessor#156a``, yields ``('proc_lib1.SameNameProcessor', '156a')``. Each
    dot-separated segment of the name must be a valid Python identifier.

    .. versionchanged:: v2.4
        Dotted qualified names (``package.ShortName``) are now accepted. The replica suffix is split
        off first, then every dot-separated segment of the name is validated as a Python identifier,
        so the name portion is returned unchanged (dots preserved).

    :param processor_string: The processor name string to parse.
    :type processor_string: str
    :return: A tuple of (name, replica_id) where replica_id can be None.
    :rtype: tuple[str, str | None]
    :raise UnknownProcessor: if the name part is empty or contains a segment that is not a valid
        Python identifier.
    """
    # Split on the first '#' so the replica identifier is handled before validating the name. This
    # keeps the replica syntax working for both short and dotted qualified names.
    parts = processor_string.strip().split('#', 1)

    # Get the name part (always exists)
    name = parts[0]

    if len(name) == 0:
        raise UnknownProcessor('Invalid processor name (empty)')

    # Validate every dot-separated segment of the (possibly qualified) name as a Python identifier.
    # A single undotted identifier remains valid, preserving backward compatibility.
    for segment in name.split('.'):
        if not segment.isidentifier():
            raise UnknownProcessor(
                f'Invalid processor name "{name}" - segment "{segment}" is not a valid Python identifier'
            )

    # Check if there's a replica part
    if len(parts) == 1 or parts[1] == '':
        # No or empty replica id part
        if len(parts) == 2 and parts[1] == '':
            # Warn about malformed input like "Name#"
            log.warning(
                f"Malformed processor name '{processor_string}': empty replica part after '#'",
            )
        return name, None
    else:
        replica_id = parts[1]
        return name, replica_id