Additional tasks

README.md

@@ -1 +1,46 @@
-# disk-defragmentation
+# Проект "Дефрагментатор"
+
+## Авторы
+* Овчинников Кирилл
+* Зуев Кирилл
+
+## Описание
+Дефрагментатор - программа, которая устраняет 
+фрагментацию файлов на дисках для оптимизации процесса записи и чтения с диска.
+
+## Требования
+* Python 3.8 и выше
+* Для работы дефрагментатора не нужно сторонних библиотек, однако для тестирования понадобятся
+сторонние библиотеки:
+```pip install -r requirements.txt```
+
+## Запуск
+Дефрагментатор поддерживает три команды:
+1. check - выводит на экран все файлы, находящиеся на диске, а также фрагментированные файлы\
+Пример использование:\
+```python main.py check <путь до образа диска> ```
+2. defragment - выполняет дефрагментацию диска\
+Пример использования:\
+```python main.py defragment <путь до образа диска> ```
+3. fragment - выполняет фрагментацию определённого файла, если его размер позволяет это сделать\
+Пример использования:\
+```python main.py fragment <путь до образа диска> <путь до файла, который будет фрагментирован>```
+
+## Тесты
+Код покрыт тестами, процент покрытия - 83.\
+Тесты находятся в директории ```tests```
+```
+Name                               Stmts   Miss  Cover   Missing
+----------------------------------------------------------------
+defragmenter\__init__.py               0      0   100%
+defragmenter\bpb.py                   13      0   100%
+defragmenter\cluster.py               10      0   100%
+defragmenter\cluster_manager.py       52     19    63%   47-59, 71-74, 86-91
+defragmenter\defragmenter.py          63     19    70%   73-88, 94-98
+defragmenter\directory_parser.py     140     22    84%   39-41, 106, 136, 191-210
+defragmenter\fat_attributes.py         9      0   100%
+defragmenter\fat_reader.py            54      3    94%   38, 52-53
+defragmenter\fragmenter.py            34      0   100%
+----------------------------------------------------------------
+TOTAL                                375     63    83%
+```

defragmenter/cluster_manager.py

@@ -0,0 +1,91 @@
+import struct
+from pathlib import Path
+
+from defragmenter.cluster import Cluster
+from defragmenter.directory_parser import DirectoryParser
+from defragmenter.fat_reader import FatReader
+
+FAT_ENTRY_MASK = 0x0FFFFFFF
+FAT_FREE_MASK = 0x00000000
+
+class ClusterManager:
+    """
+    Базовый класс для управления кластерами в файловой системе FAT32.
+    """
+    def __init__(self, image_path: Path, fat_reader: FatReader, directory_parser: DirectoryParser) -> None:
+        self._image_path = image_path
+        self._fat_reader = fat_reader
+        self._directory_parser = directory_parser
+        self._bpb = fat_reader.bpb
+        self._free_clusters: list[int] = self._find_free_clusters()
+
+    def find_fragmented_files(self, all_files: list[dict]) -> list[dict]:
+        fragmented_files = []
+        for file_entry in all_files:
+            cluster_chain = self._fat_reader.get_cluster_chain(file_entry["starting_cluster"])
+            if self._is_fragmented(cluster_chain):
+                fragmented_files.append({
+                    "path": file_entry["path"],
+                    "cluster_chain": [cluster.index for cluster in cluster_chain]
+                })
+
+        return fragmented_files
+
+    def _is_fragmented(self, cluster_chain: list[Cluster]) -> bool:
+        """
+        Проверяет, является ли кластерная цепочка фрагментированной.
+        """
+        for cluster_index in range(len(cluster_chain) - 1):
+            if cluster_chain[cluster_index].next_index != (cluster_chain[cluster_index].index + 1):
+                return True
+        return False
+
+    def _write_fat(self) -> None:
+        """
+        Записывает обновлённую FAT таблицу обратно в образ диска.
+        """
+        with open(self._image_path, 'r+b') as f:
+            fat_start = self._bpb.reserved_sec_cnt * self._bpb.byts_per_sec
+            fat_size = self._bpb.fat_size_32 * self._bpb.byts_per_sec
+            fat_data = bytearray()
+
+            for cluster in self._fat_reader.clusters:
+                fat_entry = cluster.next_index & FAT_ENTRY_MASK
+                fat_data += struct.pack("<I", fat_entry)
+
+            f.seek(fat_start)
+            f.write(fat_data[:fat_size])
+
+        print("FAT таблицы обновлены.")
+
+    def _find_free_clusters(self):
+        """
+        Находит все свободные кластеры.
+        """
+        return [cluster.index for cluster in self._fat_reader.clusters if cluster.next_index == FAT_FREE_MASK]
+
+    def _copy_cluster_data(self, old_cluster_index: int, new_cluster_index: int) -> None:
+        """
+        Копирует данные из одного кластера в другой
+        """
+        with open(self._image_path, 'r+b') as f:
+            old_data = self._fat_reader.read_cluster_data(self._fat_reader.clusters[old_cluster_index])
+            f.seek(self._fat_reader.get_cluster_offset(new_cluster_index))
+            f.write(old_data)
+
+    def _update_directory_entry(self, file_entry: dict, new_start_cluster_index: int) -> None:
+        """
+        Обновляет поле starting_cluster для файла в каталоге.
+        """
+        self._directory_parser.update_starting_cluster(file_entry['path'], new_start_cluster_index)
+
+    def _update_fat(self, old_clusters_indices: list[int], new_clusters_indices: list[int]) -> None:
+        """
+        Обновляет FAT таблицу: освобождает старые кластеры и связывает новые кластеры.
+        """
+        for cluster in old_clusters_indices:
+            self._fat_reader.clusters[cluster].next_index = FAT_FREE_MASK
+
+        for i in range(len(new_clusters_indices) - 1):
+            self._fat_reader.clusters[new_clusters_indices[i]].next_index = new_clusters_indices[i + 1]
+        self._fat_reader.clusters[new_clusters_indices[-1]].next_index = FAT_ENTRY_MASK

defragmenter.py → defragmenter/defragmenter.py

@@ -1,26 +1,20 @@
-import struct
 from pathlib import Path
 
-from cluster import Cluster
-from directory_parser import DirectoryParser
-from fat_reader import FatReader
+from defragmenter.cluster_manager import ClusterManager
+from defragmenter.directory_parser import DirectoryParser
+from defragmenter.fat_reader import FatReader
 
-FAT_END_MASK = 0x0FFFFFF8
 FAT_ENTRY_MASK = 0x0FFFFFFF
 FAT_FREE_MASK = 0x00000000
 
 ClusterIndexList = list[int]
 
-class Defragmenter:
+class Defragmenter(ClusterManager):
     """
     Класс для дефрагментации файловой системы FAT32.
     """
     def __init__(self, image_path: Path, fat_reader: FatReader, directory_parser: DirectoryParser) -> None:
-        self._image_path = image_path
-        self._fat_reader = fat_reader
-        self._directory_parser = directory_parser
-        self._bpb = fat_reader.bpb
-        self._free_clusters: list[int] = self._find_free_clusters()
+        super().__init__(image_path, fat_reader, directory_parser)
 
     def defragment(self) -> None:
         """
@@ -49,21 +43,6 @@ def defragment(self) -> None:
         self._write_fat()
         print("Дефрагментация завершена успешно.")
 
-    def _is_fragmented(self, cluster_chain: list[Cluster]) -> bool:
-        """
-        Проверяет, является ли кластерная цепочка фрагментированной.
-        """
-        for cluster_index in range(len(cluster_chain) - 1):
-            if cluster_chain[cluster_index].next_index != (cluster_chain[cluster_index].index + 1):
-                return True
-        return False
-
-    def _find_free_clusters(self):
-        """
-        Находит все свободные кластеры.
-        """
-        return [cluster.index for cluster in self._fat_reader.clusters if cluster.next_index == FAT_FREE_MASK]
-
     def _find_free_blocks(self) -> list[ClusterIndexList]:
         """
         Находит все непрерывные блоки свободных кластеров.
@@ -116,50 +95,4 @@ def _allocate_clusters(self, clusters_count: int) -> list[int]:
             self._free_clusters.remove(cluster)
         return new_clusters
 
-    def _copy_cluster_data(self, old_cluster_index: int, new_cluster_index: int) -> None:
-        """
-        Копирует данные из одного кластера в другой
-        """
-        with open(self._image_path, 'r+b') as f:
-            old_data = self._fat_reader.read_cluster_data(self._fat_reader.clusters[old_cluster_index])
-            f.seek(self._fat_reader.get_cluster_offset(new_cluster_index))
-            f.write(old_data)
-
-    def _update_fat(self, old_clusters_indices: list[int], new_clusters_indices: list[int]) -> None:
-        """
-        Обновляет FAT таблицу: освобождает старые кластеры и связывает новые кластеры.
-        """
-        for cluster in old_clusters_indices:
-            self._fat_reader.clusters[cluster].next_index = FAT_FREE_MASK
-
-        for i in range(len(new_clusters_indices)):
-            current_cluster = new_clusters_indices[i]
-            if i < len(new_clusters_indices) - 1:
-                self._fat_reader.clusters[current_cluster].next_index = new_clusters_indices[i + 1]
-            else:
-                self._fat_reader.clusters[current_cluster].next_index = FAT_ENTRY_MASK
-        print(f"FAT таблица обновлена для новых кластеров: {new_clusters_indices}")
-
-    def _update_directory_entry(self, file_entry: dict, new_start_cluster_index: int) -> None:
-        """
-        Обновляет поле starting_cluster для файла в каталоге.
-        """
-        self._directory_parser.update_starting_cluster(file_entry['path'], new_start_cluster_index)
-
-    def _write_fat(self) -> None:
-        """
-        Записывает обновлённую FAT таблицу обратно в образ диска.
-        """
-        with open(self._image_path, 'r+b') as f:
-            fat_start = self._bpb.reserved_sec_cnt * self._bpb.byts_per_sec
-            fat_size = self._bpb.fat_size_32 * self._bpb.byts_per_sec
-            fat_data = bytearray()
-
-            for cluster in self._fat_reader.clusters:
-                fat_entry = cluster.next_index & FAT_ENTRY_MASK
-                fat_data += struct.pack("<I", fat_entry)
-
-            f.seek(fat_start)
-            f.write(fat_data[:fat_size])
 
-        print("FAT таблицы обновлены.")

directory_parser.py → defragmenter/directory_parser.py

@@ -1,8 +1,8 @@
 import struct
 from typing import Any
 
-from fat_attributes import FatAttributes
-from fat_reader import FatReader
+from defragmenter.fat_attributes import FatAttributes
+from defragmenter.fat_reader import FatReader
 
 ENTRY_SIZE = 32
 EMPTY_ENTRY_MARK = 0x00

fat_reader.py → defragmenter/fat_reader.py

@@ -1,8 +1,8 @@
 import struct
 from pathlib import Path
 
-from bpb import BPB
-from cluster import Cluster
+from defragmenter.bpb import BPB
+from defragmenter.cluster import Cluster
 
 FAT_ENTRY_SIZE = 4
 FAT_ENTRY_MASK = 0x0FFFFFFF

defragmenter/fragmenter.py

@@ -0,0 +1,50 @@
+import random
+from pathlib import Path
+
+from defragmenter.fat_reader import FatReader
+from defragmenter.directory_parser import DirectoryParser
+from defragmenter.cluster_manager import ClusterManager
+
+FAT_ENTRY_MASK = 0x0FFFFFFF
+FAT_FREE_MASK = 0x00000000
+
+
+class Fragmenter(ClusterManager):
+    """
+    Класс для фрагментации файлов в файловой системе FAT32.
+    """
+
+    def __init__(self, image_path: Path, fat_reader: FatReader, directory_parser: DirectoryParser) -> None:
+        super().__init__(image_path, fat_reader, directory_parser)
+
+    def fragment_file(self, file_path: Path) -> None:
+        """
+        Фрагментирует указанный файл, разбивая его на несмежные кластеры.
+        """
+        all_files = self._directory_parser.get_all_files(self._bpb.root_clus)
+        target_file = next((f for f in all_files if Path(f["path"]) == file_path), None)
+        if not target_file:
+            raise FileNotFoundError(f"Файл {file_path} не найден")
+
+        cluster_chain = self._fat_reader.get_cluster_chain(target_file["starting_cluster"])
+        cluster_indices = [cluster.index for cluster in cluster_chain]
+
+        if len(cluster_indices) < 2:
+            raise ValueError(f"Файл '{file_path}' слишком мал для фрагментации.")
+
+        print(f"Фрагментируем файл '{file_path}'")
+
+        new_clusters = []
+        for old_cluster_index in cluster_indices:
+            if not self._free_clusters:
+                print("Нет свободных кластеров для фрагментации.")
+                break
+            index = random.randrange(len(self._free_clusters))
+            new_cluster = self._free_clusters.pop(index)
+            self._copy_cluster_data(old_cluster_index, new_cluster)
+            new_clusters.append(new_cluster)
+
+        self._update_directory_entry(target_file, new_clusters[0])
+        self._update_fat(cluster_indices, new_clusters)
+        self._write_fat()
+        print(f"Файл '{file_path}' успешно фрагментирован.")

main.py

@@ -2,22 +2,59 @@
 import argparse
 from pathlib import Path
 
-from bpb import BPB
-from directory_parser import DirectoryParser
-from fat_reader import FatReader
-from defragmenter import Defragmenter
+from defragmenter.bpb import BPB
+from defragmenter.directory_parser import DirectoryParser
+from defragmenter.fat_reader import FatReader
+from defragmenter.defragmenter import Defragmenter
+from defragmenter.fragmenter import Fragmenter
+from defragmenter.cluster_manager import ClusterManager
 
 arg_parser = argparse.ArgumentParser()
-arg_parser.add_argument("image_path", type=str)
+subparsers = arg_parser.add_subparsers(dest='command', help='Доступные команды', required=True)
+
+defrag_parser = subparsers.add_parser('defragment')
+defrag_parser.add_argument("image_path", type=str, help="Путь к образу файловой системы FAT32")
+
+frag_parser = subparsers.add_parser('fragment')
+frag_parser.add_argument("image_path", type=str, help="Путь к образу файловой системы FAT32")
+frag_parser.add_argument("file_path", type=str, help="Путь к файлу для фрагментации")
+
+check_parser = subparsers.add_parser('check', help="Просмотреть фрагментированные файлы")
+check_parser.add_argument("image_path", type=str, help="Путь к образу файловой системы FAT32")
+
 
 if __name__ == "__main__":
     args = arg_parser.parse_args()
+    command = args.command
+
     image_path = Path(args.image_path)
-    final_image_path = image_path.with_name(f"{image_path.name}_defragmented")
+    final_image_path = image_path.with_name(f"{image_path.name}_{command}ed")
+
+    if command == "check":
+        final_image_path = image_path
+    else:
+        shutil.copyfile(image_path, final_image_path)
 
-    shutil.copyfile(image_path, final_image_path)
     bpb = BPB(final_image_path)
     fat_reader = FatReader(final_image_path, bpb)
     parser = DirectoryParser(fat_reader)
-    defragmenter = Defragmenter(image_path, fat_reader, parser)
-    defragmenter.defragment()
+
+    if command == "check":
+        all_files = parser.get_all_files(bpb.root_clus)
+        fragmented_files = ClusterManager(image_path, fat_reader, parser).find_fragmented_files(all_files)
+        print("\nВсе файлы:")
+        for file in all_files:
+            print(f"path: {file['path']}, starting_cluster: {file['starting_cluster']}, size: {file['size']}")
+
+        print("\nФрагментированные файлы:")
+        for file in fragmented_files:
+            print(f"path: {file['path']}, cluster_chain: {file['cluster_chain']}")
+
+    elif command == "defragment":
+        defragmenter = Defragmenter(final_image_path, fat_reader, parser)
+        defragmenter.defragment()
+
+    elif command == "fragment":
+        file_path = Path(args.file_path)
+        fragmenter = Fragmenter(final_image_path, fat_reader, parser)
+        fragmenter.fragment_file(Path(args.file_path))

pytest.ini

@@ -0,0 +1,4 @@
+[pytest]
+minversion = 6.0
+addopts = --cov=defragmenter --cov-report=term-missing
+testpaths = tests