diff --git a/PFERD/organizer.py b/PFERD/organizer.py index 346df76..f63e92a 100644 --- a/PFERD/organizer.py +++ b/PFERD/organizer.py @@ -7,8 +7,9 @@ import filecmp import logging import os import shutil +from enum import Enum from pathlib import Path, PurePath -from typing import List, Optional, Set +from typing import Callable, List, Optional, Set from .download_summary import DownloadSummary from .location import Location @@ -19,6 +20,25 @@ LOGGER = logging.getLogger(__name__) PRETTY = PrettyLogger(LOGGER) +class FileConflictResolution(Enum): + """ + The reaction when confronted with a file conflict. + """ + + OVERWRITE_EXISTING = "overwrite" + KEEP_EXISTING = "keep" + DEFAULT = "default" + PROMPT = "prompt" + + +FileConflictResolver = Callable[[PurePath], FileConflictResolution] + + +def resolve_prompt_user(_path: PurePath) -> FileConflictResolution: + """Resolves conflicts by always asking the user.""" + return FileConflictResolution.PROMPT + + class FileAcceptException(Exception): """An exception while accepting a file.""" @@ -26,7 +46,7 @@ class FileAcceptException(Exception): class Organizer(Location): """A helper for managing downloaded files.""" - def __init__(self, path: Path, no_prompt: bool = False): + def __init__(self, path: Path, conflict_resolver: FileConflictResolver = resolve_prompt_user): """Create a new organizer for a given path.""" super().__init__(path) self._known_files: Set[Path] = set() @@ -36,7 +56,7 @@ class Organizer(Location): self.download_summary = DownloadSummary() - self.not_prompting = no_prompt + self.conflict_resolver = conflict_resolver def accept_file(self, src: Path, dst: PurePath) -> Optional[Path]: """ @@ -69,18 +89,14 @@ class Organizer(Location): if self._is_marked(dst): PRETTY.warning(f"File {str(dst_absolute)!r} was already written!") - default_action: bool = False - if self.not_prompting and not default_action \ - or not self.not_prompting and not prompt_yes_no(f"Overwrite file?", default=default_action): + if self._resolve_conflict(f"Overwrite file?", dst_absolute, default=False): PRETTY.ignored_file(dst_absolute, "file was written previously") return None # Destination file is directory if dst_absolute.exists() and dst_absolute.is_dir(): - default_action: bool = False - if self.not_prompting and default_action \ - or not self.not_prompting \ - and prompt_yes_no(f"Overwrite folder {dst_absolute} with file?", default=default_action): + prompt = f"Overwrite folder {dst_absolute} with file?" + if self._resolve_conflict(prompt, dst_absolute, default=False): shutil.rmtree(dst_absolute) else: PRETTY.warning(f"Could not add file {str(dst_absolute)!r}") @@ -151,8 +167,20 @@ class Organizer(Location): def _delete_file_if_confirmed(self, path: Path) -> None: prompt = f"Do you want to delete {path}" - default_action: bool = False - if self.not_prompting and default_action or \ - not self.not_prompting and prompt_yes_no(prompt, default_action): + if self._resolve_conflict(prompt, path, default=False): self.download_summary.add_deleted_file(path) path.unlink() + + def _resolve_conflict(self, prompt: str, path: Path, default: bool) -> bool: + if not self.conflict_resolver: + return prompt_yes_no(prompt, default=default) + + result = self.conflict_resolver(path) + if result == FileConflictResolution.DEFAULT: + return default + if result == FileConflictResolution.KEEP_EXISTING: + return False + if result == FileConflictResolution.OVERWRITE_EXISTING: + return True + + return prompt_yes_no(prompt, default=default) diff --git a/PFERD/pferd.py b/PFERD/pferd.py index 57b15f6..12ead8b 100644 --- a/PFERD/pferd.py +++ b/PFERD/pferd.py @@ -18,7 +18,7 @@ from .ipd import (IpdCrawler, IpdDownloader, IpdDownloadInfo, IpdDownloadStrategy, ipd_download_new_or_modified) from .location import Location from .logging import PrettyLogger, enable_logging -from .organizer import Organizer +from .organizer import FileConflictResolver, Organizer, resolve_prompt_user from .tmp_dir import TmpDir from .transform import TF, Transform, apply_transform from .utils import PathLike, to_path @@ -76,13 +76,13 @@ class Pferd(Location): download_strategy: IliasDownloadStrategy, timeout: int, clean: bool = True, - no_prompt: bool = None + file_conflict_resolver: FileConflictResolver = resolve_prompt_user ) -> Organizer: # pylint: disable=too-many-locals cookie_jar = CookieJar(to_path(cookies) if cookies else None) session = cookie_jar.create_session() tmp_dir = self._tmp_dir.new_subdir() - organizer = Organizer(self.resolve(to_path(target)), no_prompt if no_prompt is not None else False) + organizer = Organizer(self.resolve(to_path(target)), file_conflict_resolver) crawler = IliasCrawler(base_url, session, authenticator, dir_filter) downloader = IliasDownloader(tmp_dir, organizer, session, @@ -118,6 +118,7 @@ class Pferd(Location): download_strategy: IliasDownloadStrategy = download_modified_or_new, clean: bool = True, timeout: int = 5, + file_conflict_resolver: FileConflictResolver = resolve_prompt_user ) -> Organizer: """ Synchronizes a folder with the ILIAS instance of the KIT. @@ -145,6 +146,8 @@ class Pferd(Location): clean {bool} -- Whether to clean up when the method finishes. timeout {int} -- The download timeout for opencast videos. Sadly needed due to a requests bug. + file_conflict_resolver {FileConflictResolver} -- A function specifying how to deal + with overwriting or deleting files. The default always asks the user. """ # This authenticator only works with the KIT ilias instance. authenticator = KitShibbolethAuthenticator(username=username, password=password) @@ -160,7 +163,8 @@ class Pferd(Location): transform=transform, download_strategy=download_strategy, clean=clean, - timeout=timeout + timeout=timeout, + file_conflict_resolver=file_conflict_resolver ) self._download_summary.merge(organizer.download_summary) @@ -185,6 +189,7 @@ class Pferd(Location): download_strategy: IliasDownloadStrategy = download_modified_or_new, clean: bool = True, timeout: int = 5, + file_conflict_resolver: FileConflictResolver = resolve_prompt_user ) -> Organizer: """ Synchronizes a folder with the ILIAS instance of the KIT. This method will crawl the ILIAS @@ -211,6 +216,8 @@ class Pferd(Location): clean {bool} -- Whether to clean up when the method finishes. timeout {int} -- The download timeout for opencast videos. Sadly needed due to a requests bug. + file_conflict_resolver {FileConflictResolver} -- A function specifying how to deal + with overwriting or deleting files. The default always asks the user. """ # This authenticator only works with the KIT ilias instance. authenticator = KitShibbolethAuthenticator(username=username, password=password) @@ -226,7 +233,8 @@ class Pferd(Location): transform=transform, download_strategy=download_strategy, clean=clean, - timeout=timeout + timeout=timeout, + file_conflict_resolver=file_conflict_resolver ) self._download_summary.merge(organizer.download_summary) @@ -246,7 +254,7 @@ class Pferd(Location): download_strategy: IliasDownloadStrategy = download_modified_or_new, clean: bool = True, timeout: int = 5, - no_prompt: bool = None + file_conflict_resolver: FileConflictResolver = resolve_prompt_user ) -> Organizer: """ Synchronizes a folder with a given folder on the ILIAS instance of the KIT. @@ -273,6 +281,8 @@ class Pferd(Location): clean {bool} -- Whether to clean up when the method finishes. timeout {int} -- The download timeout for opencast videos. Sadly needed due to a requests bug. + file_conflict_resolver {FileConflictResolver} -- A function specifying how to deal + with overwriting or deleting files. The default always asks the user. """ # This authenticator only works with the KIT ilias instance. authenticator = KitShibbolethAuthenticator(username=username, password=password) @@ -292,7 +302,7 @@ class Pferd(Location): download_strategy=download_strategy, clean=clean, timeout=timeout, - no_prompt=no_prompt + file_conflict_resolver=file_conflict_resolver ) self._download_summary.merge(organizer.download_summary) @@ -306,7 +316,8 @@ class Pferd(Location): url: str, transform: Transform = lambda x: x, download_strategy: IpdDownloadStrategy = ipd_download_new_or_modified, - clean: bool = True + clean: bool = True, + file_conflict_resolver: FileConflictResolver = resolve_prompt_user ) -> Organizer: """ Synchronizes a folder with a DIVA playlist. @@ -322,6 +333,8 @@ class Pferd(Location): be downloaded. Can save bandwidth and reduce the number of requests. (default: {diva_download_new}) clean {bool} -- Whether to clean up when the method finishes. + file_conflict_resolver {FileConflictResolver} -- A function specifying how to deal + with overwriting or deleting files. The default always asks the user. """ tmp_dir = self._tmp_dir.new_subdir() @@ -332,7 +345,7 @@ class Pferd(Location): if isinstance(target, Organizer): organizer = target else: - organizer = Organizer(self.resolve(to_path(target))) + organizer = Organizer(self.resolve(to_path(target)), file_conflict_resolver) PRETTY.starting_synchronizer(organizer.path, "IPD", url) @@ -360,7 +373,8 @@ class Pferd(Location): playlist_location: str, transform: Transform = lambda x: x, download_strategy: DivaDownloadStrategy = diva_download_new, - clean: bool = True + clean: bool = True, + file_conflict_resolver: FileConflictResolver = resolve_prompt_user ) -> Organizer: """ Synchronizes a folder with a DIVA playlist. @@ -377,6 +391,8 @@ class Pferd(Location): be downloaded. Can save bandwidth and reduce the number of requests. (default: {diva_download_new}) clean {bool} -- Whether to clean up when the method finishes. + file_conflict_resolver {FileConflictResolver} -- A function specifying how to deal + with overwriting or deleting files. The default always asks the user. """ tmp_dir = self._tmp_dir.new_subdir() @@ -392,7 +408,7 @@ class Pferd(Location): if isinstance(target, Organizer): organizer = target else: - organizer = Organizer(self.resolve(to_path(target))) + organizer = Organizer(self.resolve(to_path(target)), file_conflict_resolver) PRETTY.starting_synchronizer(organizer.path, "DIVA", playlist_id) diff --git a/sync_url.py b/sync_url.py index 14c2c9e..e06deb6 100755 --- a/sync_url.py +++ b/sync_url.py @@ -5,24 +5,35 @@ A simple script to download a course by name from ILIAS. """ import argparse -from pathlib import Path +from pathlib import Path, PurePath from urllib.parse import urlparse from PFERD import Pferd from PFERD.cookie_jar import CookieJar from PFERD.ilias import (IliasCrawler, IliasElementType, KitShibbolethAuthenticator) +from PFERD.organizer import FileConflictResolution, resolve_prompt_user from PFERD.transform import sanitize_windows_path from PFERD.utils import to_path +def _resolve_overwrite(_path: PurePath) -> FileConflictResolution: + return FileConflictResolution.OVERWRITE_EXISTING + + +def _resolve_default(_path: PurePath) -> FileConflictResolution: + return FileConflictResolution.DEFAULT + + def main() -> None: parser = argparse.ArgumentParser() parser.add_argument("--test-run", action="store_true") parser.add_argument('-c', '--cookies', nargs='?', default=None, help="File to store cookies in") parser.add_argument('--no-videos', nargs='?', default=None, help="Don't download videos") - parser.add_argument('-p', '--passive', action="store_true", + parser.add_argument('-d', '--default', action="store_true", help="Don't prompt for confirmations and use sane defaults") + parser.add_argument('-r', '--remove', action="store_true", + help="Remove and overwrite files without prompting for confirmation") parser.add_argument('url', help="URL to the course page") parser.add_argument('folder', nargs='?', default=None, help="Folder to put stuff into") args = parser.parse_args() @@ -39,13 +50,17 @@ def main() -> None: folder = Path(args.folder) if args.folder is None: - folder = Path(crawler.find_element_name(args.url)) + element_name = crawler.find_element_name(args.url) + if not element_name: + print("Error, could not get element name. Please specify a folder yourself.") + return + folder = Path(element_name) cookie_jar.save_cookies() # files may not escape the pferd_root with relative paths # note: Path(Path.cwd, Path(folder)) == Path(folder) if it is an absolute path pferd_root = Path(Path.cwd(), Path(folder)).parent - folder = folder.name + target = folder.name pferd = Pferd(pferd_root, test_run=args.test_run) def dir_filter(_: Path, element: IliasElementType) -> bool: @@ -53,15 +68,22 @@ def main() -> None: return element not in [IliasElementType.VIDEO_FILE, IliasElementType.VIDEO_FOLDER] return True + if args.default: + file_confilict_resolver = _resolve_default + elif args.remove: + file_confilict_resolver = _resolve_overwrite + else: + file_confilict_resolver = resolve_prompt_user + pferd.enable_logging() # fetch pferd.ilias_kit_folder( - target=folder, + target=target, full_url=args.url, cookies=args.cookies, dir_filter=dir_filter, transform=sanitize_windows_path, - no_prompt=args.passive + file_conflict_resolver=file_confilict_resolver )