(self, src_dir: str, dest_dir: str, delete: bool = False, sync: bool = False, replace: bool = False, size_limit=0, thread_num=16, ignore_symlinks=False)
| 24 | progress_bar = None |
| 25 | |
| 26 | def __init__(self, src_dir: str, dest_dir: str, delete: bool = False, sync: bool = False, replace: bool = False, size_limit=0, thread_num=16, ignore_symlinks=False): |
| 27 | self.replace = replace |
| 28 | self.delete = delete |
| 29 | self.sync = sync |
| 30 | self.size_limit = size_limit |
| 31 | self.thread_num = thread_num |
| 32 | self.ignore_symlinks = ignore_symlinks |
| 33 | self.src_dir = Path(src_dir).absolute() |
| 34 | self.size = 0 |
| 35 | if not self.src_dir.exists(): |
| 36 | raise ValueError( |
| 37 | 'Error: source directory {} does not exist.'.format(self.src_dir)) |
| 38 | self.dest_dir = Path(dest_dir).absolute() |
| 39 | if self.src_dir == self.dest_dir: |
| 40 | raise ValueError("Error: same source and destination directory.") |
| 41 | self.dest_dir.mkdir(exist_ok=True) |
| 42 | file_list = [] |
| 43 | folders = [self.src_dir] |
| 44 | ignore_count = 0 |
| 45 | i, n = 0, 1 |
| 46 | while i < n: |
| 47 | folder: Path = folders[i] |
| 48 | for path in folder.iterdir(): |
| 49 | if path.is_symlink(): |
| 50 | if not ignore_symlinks: |
| 51 | file_list.append(path) |
| 52 | elif path.is_dir(): |
| 53 | folders.append(path) |
| 54 | (self.dest_dir / path.relative_to(self.src_dir)).mkdir(exist_ok=True) |
| 55 | n += 1 |
| 56 | elif path.is_file(): |
| 57 | if self.size_limit and path.stat().st_size > self.size_limit: |
| 58 | ignore_count += 1 |
| 59 | else: |
| 60 | self.size += path.stat().st_size |
| 61 | file_list.append(path) |
| 62 | i += 1 |
| 63 | if ignore_count: |
| 64 | print( |
| 65 | f"Ignoring {ignore_count} file(s), larger than {sizeof_fmt(size_limit)}") |
| 66 | print(f"{len(file_list)} file(s) to copy from {self.src_dir} to {self.dest_dir} with a size of {sizeof_fmt(self.size)}") |
| 67 | if sync: |
| 68 | dir_delete_count = 0 |
| 69 | delete_count = 0 |
| 70 | exist_count = 0 |
| 71 | print("Syncing files...") |
| 72 | file_set = {file.relative_to(self.src_dir) for file in file_list} |
| 73 | folder_set = {path.relative_to(self.src_dir) for path in folders} |
| 74 | dest_folders = [self.dest_dir] |
| 75 | while dest_folders: |
| 76 | folder: Path = dest_folders.pop() |
| 77 | for path in folder.iterdir(): |
| 78 | if path.is_file() or path.is_symlink(): |
| 79 | rel_path = path.relative_to(self.dest_dir) |
| 80 | if rel_path not in file_set: |
| 81 | path.unlink() |
| 82 | delete_count += 1 |
| 83 | elif not replace: |
nothing calls this directly
no test coverage detected