Skip to content

folder_handler

FolderHandler

A class that handles file and folder operations.

Source code in libs\cafex_core\src\cafex_core\handlers\folder_handler.py 
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
class FolderHandler:
    """A class that handles file and folder operations."""

    @staticmethod
    def delete_files_in_folder(folder_path):
        """Delete all files in the specified folder.

        Args:
            folder_path (str): The path of the folder containing the files to be deleted.

        Returns:
            None

        Raises:
            None
        """
        if os.path.exists(folder_path):
            for filename in os.listdir(folder_path):
                file_path = os.path.join(folder_path, filename)
                try:
                    if os.path.isfile(file_path):
                        os.remove(file_path)
                        print(f"File '{file_path}' deleted successfully.")
                    elif os.path.isdir(file_path):  # Optional: Delete subfolders
                        shutil.rmtree(file_path)
                except OSError as e:
                    print(f"Error deleting '{file_path}': {e}")

    @staticmethod
    def create_folder(root_dir, folder_name):
        """Create a folder in the specified root directory.

        Args:
            root_dir (str): The root directory where the folder will be created.
            folder_name (str): The name of the folder to be created.

        Returns:
            None

        Raises:
            None
        """
        folder_path = os.path.join(root_dir, folder_name)
        if not os.path.exists(folder_path):
            os.makedirs(folder_path)
        return folder_path

    @staticmethod
    def delete_folder(folder_path):
        """Delete the specified folder and all its contents.

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

        Returns:
            None

        Raises:
            None
        """
        if os.path.exists(folder_path):
            try:
                shutil.rmtree(folder_path)
            except Exception as e:
                print(f"Error deleting folder '{folder_path}': {e}")

    @staticmethod
    def reorganize_result_folders(conf_cwd, current_execution_uuid):
        """Reorganize result folders by moving existing results to a history
        folder, sorting them, and keeping only the 10 most recent, excluding
        the current execution.

        Args:
            conf_cwd (str): The root directory of the project.
            current_execution_uuid (str): The UUID of the current execution.

        Returns:
            None
        """
        result_dir = os.path.join(conf_cwd, "result")
        history_dir = os.path.join(result_dir, "history")

        # Create history folder if it doesn't exist
        os.makedirs(history_dir, exist_ok=True)

        # Get all subdirectories in the result folder, excluding the current execution and history
        subdirs = [
            d
            for d in os.listdir(result_dir)
            if os.path.isdir(os.path.join(result_dir, d))
            and d != "history"
            and d != current_execution_uuid
        ]

        # Rename and move subdirectories to history folder
        for subdir in subdirs:
            src_path = os.path.join(result_dir, subdir)
            creation_time = datetime.fromtimestamp(os.path.getctime(src_path))

            stop_server(Path(src_path))

            # Check if the folder name already starts with a timestamp
            if not subdir[:15].replace("_", "").isdigit():
                new_name = f"{creation_time.strftime('%Y%m%d_%H%M%S')}_{subdir}"
            else:
                new_name = subdir

            dst_path = os.path.join(history_dir, new_name)
            shutil.move(src_path, dst_path)

        # Sort history folders by creation date (newest first)
        history_subdirs = [
            d for d in os.listdir(history_dir) if os.path.isdir(os.path.join(history_dir, d))
        ]
        sorted_history_subdirs = sorted(
            history_subdirs,
            key=lambda x: os.path.getctime(os.path.join(history_dir, x)),
            reverse=True,
        )

        # Keep only the 10 most recent history folders
        for old_dir in sorted_history_subdirs[10:]:
            shutil.rmtree(os.path.join(history_dir, old_dir))

create_folder(root_dir, folder_name) staticmethod

Create a folder in the specified root directory.

Parameters:

Name Type Description Default
root_dir str

The root directory where the folder will be created.

 required
folder_name str

The name of the folder to be created.

 required

Returns:

Type Description

None
Source code in libs\cafex_core\src\cafex_core\handlers\folder_handler.py 
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
@staticmethod
def create_folder(root_dir, folder_name):
    """Create a folder in the specified root directory.

    Args:
        root_dir (str): The root directory where the folder will be created.
        folder_name (str): The name of the folder to be created.

    Returns:
        None

    Raises:
        None
    """
    folder_path = os.path.join(root_dir, folder_name)
    if not os.path.exists(folder_path):
        os.makedirs(folder_path)
    return folder_path

delete_files_in_folder(folder_path) staticmethod

Delete all files in the specified folder.

Parameters:

Name Type Description Default
folder_path str

The path of the folder containing the files to be deleted.

 required

Returns:

Type Description

None
Source code in libs\cafex_core\src\cafex_core\handlers\folder_handler.py 
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
@staticmethod
def delete_files_in_folder(folder_path):
    """Delete all files in the specified folder.

    Args:
        folder_path (str): The path of the folder containing the files to be deleted.

    Returns:
        None

    Raises:
        None
    """
    if os.path.exists(folder_path):
        for filename in os.listdir(folder_path):
            file_path = os.path.join(folder_path, filename)
            try:
                if os.path.isfile(file_path):
                    os.remove(file_path)
                    print(f"File '{file_path}' deleted successfully.")
                elif os.path.isdir(file_path):  # Optional: Delete subfolders
                    shutil.rmtree(file_path)
            except OSError as e:
                print(f"Error deleting '{file_path}': {e}")

delete_folder(folder_path) staticmethod

Delete the specified folder and all its contents.

Parameters:

Name Type Description Default
folder_path str

The path of the folder to be deleted.

 required

Returns:

Type Description

None
Source code in libs\cafex_core\src\cafex_core\handlers\folder_handler.py 
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
@staticmethod
def delete_folder(folder_path):
    """Delete the specified folder and all its contents.

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

    Returns:
        None

    Raises:
        None
    """
    if os.path.exists(folder_path):
        try:
            shutil.rmtree(folder_path)
        except Exception as e:
            print(f"Error deleting folder '{folder_path}': {e}")

reorganize_result_folders(conf_cwd, current_execution_uuid) staticmethod

Reorganize result folders by moving existing results to a history folder, sorting them, and keeping only the 10 most recent, excluding the current execution.

Parameters:

Name Type Description Default
conf_cwd str

The root directory of the project.

 required
current_execution_uuid str

The UUID of the current execution.

 required

Returns:

Type Description

None
Source code in libs\cafex_core\src\cafex_core\handlers\folder_handler.py 
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
@staticmethod
def reorganize_result_folders(conf_cwd, current_execution_uuid):
    """Reorganize result folders by moving existing results to a history
    folder, sorting them, and keeping only the 10 most recent, excluding
    the current execution.

    Args:
        conf_cwd (str): The root directory of the project.
        current_execution_uuid (str): The UUID of the current execution.

    Returns:
        None
    """
    result_dir = os.path.join(conf_cwd, "result")
    history_dir = os.path.join(result_dir, "history")

    # Create history folder if it doesn't exist
    os.makedirs(history_dir, exist_ok=True)

    # Get all subdirectories in the result folder, excluding the current execution and history
    subdirs = [
        d
        for d in os.listdir(result_dir)
        if os.path.isdir(os.path.join(result_dir, d))
        and d != "history"
        and d != current_execution_uuid
    ]

    # Rename and move subdirectories to history folder
    for subdir in subdirs:
        src_path = os.path.join(result_dir, subdir)
        creation_time = datetime.fromtimestamp(os.path.getctime(src_path))

        stop_server(Path(src_path))

        # Check if the folder name already starts with a timestamp
        if not subdir[:15].replace("_", "").isdigit():
            new_name = f"{creation_time.strftime('%Y%m%d_%H%M%S')}_{subdir}"
        else:
            new_name = subdir

        dst_path = os.path.join(history_dir, new_name)
        shutil.move(src_path, dst_path)

    # Sort history folders by creation date (newest first)
    history_subdirs = [
        d for d in os.listdir(history_dir) if os.path.isdir(os.path.join(history_dir, d))
    ]
    sorted_history_subdirs = sorted(
        history_subdirs,
        key=lambda x: os.path.getctime(os.path.join(history_dir, x)),
        reverse=True,
    )

    # Keep only the 10 most recent history folders
    for old_dir in sorted_history_subdirs[10:]:
        shutil.rmtree(os.path.join(history_dir, old_dir))