Source code for bhf_smarthealth.validate.metadata_validation

import logging
import re

from dataclasses import dataclass
from pathlib import Path
from typing import Optional

from bhf_smarthealth.utils import utils
from bhf_smarthealth.exceptions import MetadataValidationError, ISODateConversionError
from bhf_smarthealth.config import ENVIRONMENT
from bhf_smarthealth.models import Environments

logger = logging.getLogger(__name__)


ENVIRONMENT_ID_REGISTRY = {
    Environments.CORE: [r"^UOS\d{5}$", r"^TEST_R_[a-zA-Z0-9]{15}$"],
    Environments.ISOLATE_PROD: [r"^R_[a-zA-Z0-9]{15}$"],
    Environments.ISOLATE_DEV: [
        r"^UOS\d{5}$",
        r"^TEST_R_[a-zA-Z0-9]{15}$",
        r"^R_[a-zA-Z0-9]{15}$",
    ],
}
"""Registry of accetable regex patterns for IDs to be processed per RONIN environment"""




@dataclass
class FileMetadata:
    filename: str
    sheffield_id: Optional[str]
    file_date: Optional[str]
    validated: bool = False  # switch to show if this metadata is validated or not





def extract_metadata_from_filename(filepath: Path) -> FileMetadata:
    """Extracts Sheffield ID (sheffield_id) and date metadata from a structured filename.

    The function assumes the filename adheres to a specific format:
        <userid>_<YYYYMMDD>_<activity>.<json>

    Parameters
    ----------
    filepath : Path
        Path to file to extract metadata from

    Returns
    -------
    FileMetadata
        FileMetadata dataclass object. If metadata cannot be extracted, the sheffield_id and date will be returned as None.
    """

    basename = filepath.name

    match = re.match(r"^(\w+)_(\d{8})_(\w*)\.json$", basename)

    if not match:
        logger.debug(f"Filename does not match expected pattern: {basename}")
        # return FileMetadata dataclass in all cases
        return FileMetadata(filename=basename, sheffield_id=None, file_date=None)

    extracted_id = match.group(1)
    extracted_date = match.group(2)

    return FileMetadata(
        filename=basename,
        sheffield_id=extracted_id,
        file_date=extracted_date,
    )





def validate_metadata(
    filepath: Path,
    metadata: FileMetadata,
    dates: str | list[str],
    internal_id: Optional[str] = None,
) -> None:
    """Compare metadata extracted from a filename to its internal contents. Raised error will contain a list of errors found.

    Parameters
    ----------
    filepath : Path
        Path to file being processed
    metadata : FileMetadata
        FileMetadata object containing the filename and extracted sheffield_id and date
    dates : str | list[str]
        A list of the dates, or datetimes, inside the file being processed to be validated against the FileMetadata.date extracted from the filename

    Raises
    ------
    MetadataValidationError
        Raised if validation checks fail. Raised message contains a list of all errors found.
    """
    error_messages = set()

    if metadata.sheffield_id:
        id_pattern = "|".join(ENVIRONMENT_ID_REGISTRY[ENVIRONMENT])
        if not re.match(pattern=id_pattern, string=metadata.sheffield_id):
            error_messages.add(
                f"ID in filename ({metadata.sheffield_id}) is not suitable for processing in this RONIN environment ({ENVIRONMENT})"
            )

    # If an internal file ID is available, check it against the ID extracted from the filename
    if internal_id:
        if internal_id != metadata.sheffield_id:
            error_messages.add(
                f"ID in filename ({metadata.sheffield_id}) does not match the internal file ID ({internal_id})"
            )

    # Handle the case of only one internal file date
    if isinstance(dates, str):
        # wrap the single date in a list to normalise the input
        dates = [dates]

    # Ensure that the dates internal to the file can be converted to ISO8601 format
    internal_date_set = set()
    for date in dates:
        try:
            converted_date = utils.convert_str_to_isodate(date_str=date)
        except ISODateConversionError:
            error_messages.add(
                f"File {metadata.filename} contains dates that cannot be converted to ISO8601"
            )
        else:
            internal_date_set.add(converted_date)

    # Check if the metadata was extracted successfully
    if metadata.file_date is None:
        error_messages.add(
            f"Metadata could not be extracted from JSON file: {metadata.filename}"
        )
    else:
        # Check that the extracted date is valid
        try:
            metadata.file_date = utils.convert_str_to_isodate(metadata.file_date)
        except ISODateConversionError as e:
            error_messages.add(
                f"Extract metadata file date cannot be converted to a valid ISO8601 date.\nDate: {metadata.file_date}.\nError message: {e}"
            )

    # Check if there are any valid dates inside of the JSON file
    if len(internal_date_set) == 0:
        error_messages.add(f"No valid dates found in JSON: {metadata.filename}")
    # Check if there are multiple different date values inside of the JSON file
    elif len(internal_date_set) > 1:
        error_messages.add(
            f"Dates are inconsistent in JSON: {metadata.filename}. Number of unique dates found: {len(internal_date_set)}"
        )
    # Else check if the date extracted from the filename matches the unique internal date
    # Compares even if no valid file_date is extracted from the metadata, as useful to know if internal file structure is ok
    elif (
        len(internal_date_set) == 1 and metadata.file_date != list(internal_date_set)[0]
    ):
        error_messages.add(
            f"File date {metadata.file_date} does not match consistent internal date {list(internal_date_set)[0]}"
        )

    # Combine and raise all errors logged
    if error_messages:
        final_message = (
            f"Metadata validation failed for the following reasons:\n"
            f"{('\n').join(error_messages)}"
        )
        raise MetadataValidationError(final_message)
    else:
        logger.debug("No metadata processing errors found")





def process_metadata(
    filepath: Path | str, dates: str | list[str], internal_id: Optional[str] = None
) -> FileMetadata:
    """Orchestrator function to ensure than extract_metadata and validate_metadata are run together.

    Parameters
    ----------
    filepath : Path | str
        Path to file to be processed
    dates : str | list[str]
        A list of the dates, or datetimes, inside the file being processed to be validated against the FileMetadata.date extracted from the filename

    Returns
    -------
    FileMetadata
        FileMetadata object containing validated metadata

    Raises
    ------
    MetadataValidationError
        Raised by validate_metadata if any validation checks fail
    """
    if isinstance(filepath, str):
        filepath = Path(filepath)

    metadata = extract_metadata_from_filename(filepath)

    validate_metadata(
        filepath=filepath, dates=dates, metadata=metadata, internal_id=internal_id
    )

    # switch the validated bool to true now checks are complete. Only runs if validate_metadata is successful, else MetadataValidationError will have been raised
    metadata.validated = True

    return metadata