Source code for ewoksmx.tasks.mx_ccp4.utils

import logging
import os
from datetime import datetime

logger = logging.getLogger(__name__)




class FolderManager:

    def __init__(self, is_single_run, task_name):
        self.is_single_run = is_single_run
        self.task_name = task_name

    def _get_id(self):
        """
        id is used to make the processing folder unique when single_run === False
        """

        return datetime.now().strftime("%Y_%m_%d_%H_%M_%S_%f")[:-3]



    def create_folder_if_not_exists(self, folder_path):
        """
        Create a folder at the specified path if it does not already exist.

        Args:
            folder_path (str): The path of the folder to create.

        Returns:
            None
        """
        if not os.path.exists(folder_path):
            os.makedirs(folder_path)
            logger.debug(f"Folder {folder_path} has been created")


    def _check_folder_with_raw_data_in_path(self, filepath):
        """
        Check if the given file path exists, is a folder, and contains
        'RAW_DATA' in its path.

        Args:
            filepath (str): The path to check.

        Returns:
            bool: True if the path exists, is a folder, and 'RAW_DATA' is in the path,
            False otherwise.
        """
        return (
            os.path.exists(filepath)
            and os.path.isdir(filepath)
            and "RAW_DATA" in filepath
        )



    def get_output_folder(self, raw_dataset_path):
        """
        Returns the outputfolder path where output data should be written
        It keeps the same path as the raw dataset but replacing RAW_DATA
        by PROCESSED_DATA
        """

        logger.debug(f"get_output_folder. raw_dataset_path={raw_dataset_path}")
        if raw_dataset_path is None:
            raise Exception("No dataset location provided")

        if self._check_folder_with_raw_data_in_path(raw_dataset_path[0]):
            # It replace RAW by PROCESSED and
            # appends the task name to the output folder
            output_folder = raw_dataset_path[0].replace("RAW_DATA", "PROCESSED_DATA")
            output_folder = os.path.join(output_folder, self.task_name)

            if os.path.exists(output_folder) and self.is_single_run:
                raise Exception("Folder exists and will not be overriden")
            else:
                if not self.is_single_run:
                    output_folder = os.path.join(output_folder, self._get_id())
                    if os.path.exists(output_folder):
                        raise Exception(
                            "Could not find a suitable folder based on the id"
                        )
                return output_folder

        raise Exception(
            f"Dataset file location ({raw_dataset_path}) does not exist or there is no RAW_DATA"
        )






class Security:

    def __init__(self, icat_client):
        """
        Constructor for Security.

        Args:
            icat_client: The ICAT client instance used for session management.
        """
        self.icatClient = icat_client



    def check_user_credentials(self, session_id):
        """
        Validates the user session based on the provided session ID.

        Args:
            session_id (str): The session ID to validate.

        Raises:
            Exception: If the session has expired (lifeTimeMinutes < 0).
        """
        session = self.icatClient.get_session_information(session_id)
        user_info = session.get("user", {})

        self.full_name = user_info.get("fullName", "Unknown")
        self.user_name = session.get("userName", "Unknown")
        self.lifeTimeMinutes = user_info.get("lifeTimeMinutes", 0)

        logger.debug(f"Processing launched by user_name={self.user_name}")
        logger.debug(f"Logged in. lifeTimeMinutes={self.lifeTimeMinutes}")

        if self.lifeTimeMinutes < 0:
            raise Exception("Session expired")

        return self.user_name




    def is_participant(self, session_id, investigation_id):
        investigation = self.icatClient.get_investigations_by(
            session_id=session_id, ids=investigation_id, filter="isParticipant"
        )
        print(investigation)