|
| 1 | +# Licensed under the GPL-3.0 License. |
| 2 | +# Created for TagStudio: https://github.com/CyanVoxel/TagStudio |
| 3 | + |
| 4 | +from pathlib import Path |
| 5 | +import shutil |
| 6 | +import typing |
| 7 | + |
| 8 | +from PySide6.QtCore import QThreadPool |
| 9 | +from PySide6.QtGui import QDropEvent, QDragEnterEvent, QDragMoveEvent |
| 10 | +from PySide6.QtWidgets import QMessageBox |
| 11 | + |
| 12 | +from src.qt.widgets.progress import ProgressWidget |
| 13 | +from src.qt.helpers.custom_runnable import CustomRunnable |
| 14 | +from src.qt.helpers.function_iterator import FunctionIterator |
| 15 | + |
| 16 | +if typing.TYPE_CHECKING: |
| 17 | + from src.qt.ts_qt import QtDriver |
| 18 | + |
| 19 | +import logging |
| 20 | + |
| 21 | + |
| 22 | +class DropImport: |
| 23 | + def __init__(self, driver: "QtDriver"): |
| 24 | + self.driver = driver |
| 25 | + |
| 26 | + def dropEvent(self, event: QDropEvent): # noqa: N802 |
| 27 | + if ( |
| 28 | + event.source() is self.driver |
| 29 | + ): # change that if you want to drop something originating from tagstudio, for moving or so |
| 30 | + return |
| 31 | + |
| 32 | + if not event.mimeData().hasUrls(): |
| 33 | + return |
| 34 | + |
| 35 | + self.urls = event.mimeData().urls() |
| 36 | + self.import_files() |
| 37 | + |
| 38 | + def dragEnterEvent(self, event: QDragEnterEvent): # noqa: N802 |
| 39 | + if event.mimeData().hasUrls(): |
| 40 | + event.accept() |
| 41 | + else: |
| 42 | + event.ignore() |
| 43 | + |
| 44 | + def dragMoveEvent(self, event: QDragMoveEvent): # noqa: N802 |
| 45 | + if event.mimeData().hasUrls(): |
| 46 | + event.accept() |
| 47 | + else: |
| 48 | + logging.info(self.driver.selected) |
| 49 | + event.ignore() |
| 50 | + |
| 51 | + def import_files(self): |
| 52 | + self.files: list[Path] = [] |
| 53 | + self.dirs_in_root: list[Path] = [] |
| 54 | + self.duplicate_files: list[Path] = [] |
| 55 | + |
| 56 | + def displayed_text(x): |
| 57 | + text = f"Searching New Files...\n{x[0] + 1} File{'s' if x[0] + 1 != 1 else ''} Found." |
| 58 | + if x[1] == 0: |
| 59 | + return text |
| 60 | + return text + f" {x[1]} Already exist in the library folders" |
| 61 | + |
| 62 | + create_progress_bar( |
| 63 | + self.collect_files_to_import, |
| 64 | + "Searching Files", |
| 65 | + "Searching New Files...\nPreparing...", |
| 66 | + displayed_text, |
| 67 | + self.ask_user, |
| 68 | + ) |
| 69 | + |
| 70 | + def collect_files_to_import(self): |
| 71 | + for url in self.urls: |
| 72 | + if not url.isLocalFile(): |
| 73 | + continue |
| 74 | + |
| 75 | + file = Path(url.toLocalFile()) |
| 76 | + |
| 77 | + if file.is_dir(): |
| 78 | + for f in self.get_files_in_folder(file): |
| 79 | + if f.is_dir(): |
| 80 | + continue |
| 81 | + self.files.append(f) |
| 82 | + if (self.driver.lib.library_dir / self.get_relative_path(file)).exists(): |
| 83 | + self.duplicate_files.append(f) |
| 84 | + yield [len(self.files), len(self.duplicate_files)] |
| 85 | + |
| 86 | + self.dirs_in_root.append(file.parent) |
| 87 | + else: |
| 88 | + self.files.append(file) |
| 89 | + |
| 90 | + if file.parent not in self.dirs_in_root: |
| 91 | + self.dirs_in_root.append( |
| 92 | + file.parent |
| 93 | + ) # to create relative path of files not in folder |
| 94 | + |
| 95 | + if (Path(self.driver.lib.library_dir) / file.name).exists(): |
| 96 | + self.duplicate_files.append(file) |
| 97 | + |
| 98 | + yield [len(self.files), len(self.duplicate_files)] |
| 99 | + |
| 100 | + def copy_files(self): |
| 101 | + file_count = 0 |
| 102 | + duplicated_files_progress = 0 |
| 103 | + for file in self.files: |
| 104 | + if file.is_dir(): |
| 105 | + continue |
| 106 | + |
| 107 | + dest_file = self.get_relative_path(file) |
| 108 | + |
| 109 | + if file in self.duplicate_files: |
| 110 | + duplicated_files_progress += 1 |
| 111 | + if self.choice == 1: # override |
| 112 | + pass |
| 113 | + elif self.choice == 2: # rename |
| 114 | + new_name = self.get_renamed_duplicate_filename_in_lib(dest_file) |
| 115 | + dest_file = dest_file.with_name(new_name) |
| 116 | + else: # skip |
| 117 | + continue |
| 118 | + |
| 119 | + (self.driver.lib.library_dir / dest_file).parent.mkdir(parents=True, exist_ok=True) |
| 120 | + shutil.copyfile(file, self.driver.lib.library_dir / dest_file) |
| 121 | + |
| 122 | + file_count += 1 |
| 123 | + yield [file_count, duplicated_files_progress] |
| 124 | + |
| 125 | + def ask_user(self): |
| 126 | + self.choice = -1 |
| 127 | + |
| 128 | + if len(self.duplicate_files) > 0: |
| 129 | + self.choice = self.duplicates_choice() |
| 130 | + else: |
| 131 | + self.begin_transfer() |
| 132 | + |
| 133 | + def duplicate_prompt_callback(self, button): |
| 134 | + if button == self.skip_button: |
| 135 | + self.choice = 0 |
| 136 | + elif button == self.override_button: |
| 137 | + self.choice = 1 |
| 138 | + elif button == self.rename_button: |
| 139 | + self.choice = 2 |
| 140 | + else: |
| 141 | + return |
| 142 | + |
| 143 | + self.begin_transfer() |
| 144 | + |
| 145 | + def begin_transfer(self): |
| 146 | + def displayed_text(x): |
| 147 | + dupes_choice_text = ( |
| 148 | + "Skipped" if self.choice == 0 else ("Overridden" if self.choice == 1 else "Renamed") |
| 149 | + ) |
| 150 | + |
| 151 | + text = ( |
| 152 | + f"Importing New Files...\n{x[0] + 1} File{'s' if x[0] + 1 != 1 else ''} Imported." |
| 153 | + ) |
| 154 | + if x[1] == 0: |
| 155 | + return text |
| 156 | + return text + f" {x[1]} {dupes_choice_text}" |
| 157 | + |
| 158 | + create_progress_bar( |
| 159 | + self.copy_files, |
| 160 | + "Import Files", |
| 161 | + "Importing New Files...\nPreparing...", |
| 162 | + displayed_text, |
| 163 | + self.driver.add_new_files_callback, |
| 164 | + len(self.files), |
| 165 | + ) |
| 166 | + |
| 167 | + def duplicates_choice(self): |
| 168 | + display_limit: int = 5 |
| 169 | + self.msg_box = QMessageBox() |
| 170 | + self.msg_box.setWindowTitle(f"File Conflict{'s' if len(self.duplicate_files) > 1 else ''}") |
| 171 | + |
| 172 | + dupes_to_show = self.duplicate_files |
| 173 | + if len(self.duplicate_files) > display_limit: |
| 174 | + dupes_to_show = dupes_to_show[0:display_limit] |
| 175 | + |
| 176 | + self.msg_box.setText( |
| 177 | + f"The following files:\n {'\n '.join(map(lambda path: str(path), self.get_relative_paths(dupes_to_show)))} {(f'\nand {len(self.duplicate_files) - display_limit} more ') if len(self.duplicate_files) > display_limit else '\n'} have filenames that already exist in the library folder." |
| 178 | + ) |
| 179 | + self.skip_button = self.msg_box.addButton("Skip", QMessageBox.ButtonRole.YesRole) |
| 180 | + self.override_button = self.msg_box.addButton( |
| 181 | + "Override", QMessageBox.ButtonRole.DestructiveRole |
| 182 | + ) |
| 183 | + self.rename_button = self.msg_box.addButton( |
| 184 | + "Rename", QMessageBox.ButtonRole.DestructiveRole |
| 185 | + ) |
| 186 | + self.cancel_button = self.msg_box.setStandardButtons(QMessageBox.Cancel) |
| 187 | + |
| 188 | + self.msg_box.buttonClicked.connect(lambda button: self.duplicate_prompt_callback(button)) |
| 189 | + self.msg_box.open() |
| 190 | + |
| 191 | + def get_files_exists_in_library(self, path: Path) -> list[Path]: |
| 192 | + exists: list[Path] = [] |
| 193 | + if not path.is_dir(): |
| 194 | + return exists |
| 195 | + |
| 196 | + files = self.get_files_in_folder(path) |
| 197 | + for file in files: |
| 198 | + if file.is_dir(): |
| 199 | + exists += self.get_files_exists_in_library(file) |
| 200 | + elif (self.driver.lib.library_dir / self.get_relative_path(file)).exists(): |
| 201 | + exists.append(file) |
| 202 | + return exists |
| 203 | + |
| 204 | + def get_relative_paths(self, paths: list[Path]) -> list[Path]: |
| 205 | + relative_paths = [] |
| 206 | + for file in paths: |
| 207 | + relative_paths.append(self.get_relative_path(file)) |
| 208 | + return relative_paths |
| 209 | + |
| 210 | + def get_relative_path(self, path: Path) -> Path: |
| 211 | + for dir in self.dirs_in_root: |
| 212 | + if path.is_relative_to(dir): |
| 213 | + return path.relative_to(dir) |
| 214 | + return Path(path.name) |
| 215 | + |
| 216 | + def get_files_in_folder(self, path: Path) -> list[Path]: |
| 217 | + files = [] |
| 218 | + for file in path.glob("**/*"): |
| 219 | + files.append(file) |
| 220 | + return files |
| 221 | + |
| 222 | + def get_renamed_duplicate_filename_in_lib(self, filepath: Path) -> str: |
| 223 | + index = 2 |
| 224 | + o_filename = filepath.name |
| 225 | + dot_idx = o_filename.index(".") |
| 226 | + while (self.driver.lib.library_dir / filepath).exists(): |
| 227 | + filepath = filepath.with_name( |
| 228 | + o_filename[:dot_idx] + f" ({index})" + o_filename[dot_idx:] |
| 229 | + ) |
| 230 | + index += 1 |
| 231 | + return filepath.name |
| 232 | + |
| 233 | + |
| 234 | +def create_progress_bar( |
| 235 | + function, title: str, text: str, update_label_callback, done_callback, max=0 |
| 236 | +): |
| 237 | + iterator = FunctionIterator(function) |
| 238 | + pw = ProgressWidget( |
| 239 | + window_title=title, |
| 240 | + label_text=text, |
| 241 | + cancel_button_text=None, |
| 242 | + minimum=0, |
| 243 | + maximum=max, |
| 244 | + ) |
| 245 | + pw.show() |
| 246 | + iterator.value.connect(lambda x: pw.update_progress(x[0] + 1)) |
| 247 | + iterator.value.connect(lambda x: pw.update_label(update_label_callback(x))) |
| 248 | + r = CustomRunnable(lambda: iterator.run()) |
| 249 | + r.done.connect(lambda: (pw.hide(), done_callback())) # type: ignore |
| 250 | + QThreadPool.globalInstance().start(r) |
0 commit comments