pferd/PFERD/pferd.py

414 lines
16 KiB
Python
Raw Normal View History

"""
Convenience functions for using PFERD.
"""
2020-04-23 19:55:37 +02:00
import logging
2020-04-23 11:44:13 +02:00
from pathlib import Path
from typing import Callable, List, Optional, Union
2020-04-23 11:44:13 +02:00
from .cookie_jar import CookieJar
2020-04-30 16:24:38 +02:00
from .diva import (DivaDownloader, DivaDownloadStrategy, DivaPlaylistCrawler,
diva_download_new)
from .download_summary import DownloadSummary
2020-05-09 00:00:21 +02:00
from .errors import FatalException, swallow_and_print_errors
from .ilias import (IliasAuthenticator, IliasCrawler, IliasDirectoryFilter,
IliasDownloader, IliasDownloadInfo, IliasDownloadStrategy,
KitShibbolethAuthenticator, download_modified_or_new)
2020-11-03 20:40:09 +01:00
from .ipd import (IpdCrawler, IpdDownloader, IpdDownloadInfo,
IpdDownloadStrategy, ipd_download_new_or_modified)
2020-04-23 19:38:28 +02:00
from .location import Location
2020-05-12 20:19:23 +02:00
from .logging import PrettyLogger, enable_logging
2020-04-23 11:44:13 +02:00
from .organizer import Organizer
from .tmp_dir import TmpDir
2020-04-24 20:26:28 +02:00
from .transform import TF, Transform, apply_transform
2020-04-25 19:59:58 +02:00
from .utils import PathLike, to_path
2020-04-23 11:44:13 +02:00
2020-04-23 12:49:01 +02:00
# TODO save known-good cookies as soon as possible
2020-04-23 19:55:37 +02:00
LOGGER = logging.getLogger(__name__)
PRETTY = PrettyLogger(LOGGER)
2020-04-23 12:49:01 +02:00
2020-04-23 11:44:13 +02:00
class Pferd(Location):
# pylint: disable=too-many-arguments
"""
The main entrypoint in your Pferd usage: This class combines a number of
useful shortcuts for running synchronizers in a single interface.
"""
2020-04-23 11:44:13 +02:00
2020-04-24 20:00:21 +02:00
def __init__(
self,
base_dir: Path,
tmp_dir: Path = Path(".tmp"),
test_run: bool = False
):
2020-04-23 11:44:13 +02:00
super().__init__(Path(base_dir))
self._download_summary = DownloadSummary()
2020-04-23 11:44:13 +02:00
self._tmp_dir = TmpDir(self.resolve(tmp_dir))
2020-04-24 20:00:21 +02:00
self._test_run = test_run
2020-05-12 20:19:23 +02:00
@staticmethod
def enable_logging() -> None:
"""
Enable and configure logging via the logging module.
"""
enable_logging()
2020-04-24 20:26:28 +02:00
@staticmethod
def _print_transformables(transformables: List[TF]) -> None:
2020-04-24 20:00:21 +02:00
LOGGER.info("")
LOGGER.info("Results of the test run:")
for transformable in transformables:
LOGGER.info(transformable.path)
2020-04-23 11:44:13 +02:00
def _ilias(
self,
2020-04-24 20:39:30 +02:00
target: PathLike,
2020-04-23 11:44:13 +02:00
base_url: str,
crawl_function: Callable[[IliasCrawler], List[IliasDownloadInfo]],
2020-04-23 11:44:13 +02:00
authenticator: IliasAuthenticator,
2020-04-24 20:39:30 +02:00
cookies: Optional[PathLike],
dir_filter: IliasDirectoryFilter,
2020-04-23 11:44:13 +02:00
transform: Transform,
download_strategy: IliasDownloadStrategy,
timeout: int,
clean: bool = True,
2020-04-30 16:24:38 +02:00
) -> Organizer:
# pylint: disable=too-many-locals
2020-04-24 20:39:30 +02:00
cookie_jar = CookieJar(to_path(cookies) if cookies else None)
2020-04-23 11:44:13 +02:00
session = cookie_jar.create_session()
tmp_dir = self._tmp_dir.new_subdir()
2020-04-24 20:39:30 +02:00
organizer = Organizer(self.resolve(to_path(target)))
2020-04-23 11:44:13 +02:00
crawler = IliasCrawler(base_url, session, authenticator, dir_filter)
downloader = IliasDownloader(tmp_dir, organizer, session,
authenticator, download_strategy, timeout)
2020-04-23 11:44:13 +02:00
cookie_jar.load_cookies()
info = crawl_function(crawler)
2020-04-23 11:44:13 +02:00
cookie_jar.save_cookies()
2020-04-24 20:00:21 +02:00
transformed = apply_transform(transform, info)
if self._test_run:
self._print_transformables(transformed)
2020-04-30 16:24:38 +02:00
return organizer
2020-04-24 20:00:21 +02:00
downloader.download_all(transformed)
2020-04-23 11:44:13 +02:00
cookie_jar.save_cookies()
2020-04-30 16:24:38 +02:00
if clean:
organizer.cleanup()
return organizer
2020-04-23 11:44:13 +02:00
@swallow_and_print_errors
2020-04-23 11:44:13 +02:00
def ilias_kit(
self,
2020-04-24 20:39:30 +02:00
target: PathLike,
2020-04-23 11:44:13 +02:00
course_id: str,
dir_filter: IliasDirectoryFilter = lambda x, y: True,
2020-04-23 11:44:13 +02:00
transform: Transform = lambda x: x,
2020-04-24 20:39:30 +02:00
cookies: Optional[PathLike] = None,
2020-04-23 11:44:13 +02:00
username: Optional[str] = None,
password: Optional[str] = None,
download_strategy: IliasDownloadStrategy = download_modified_or_new,
2020-04-30 16:24:38 +02:00
clean: bool = True,
timeout: int = 5,
2020-04-30 16:24:38 +02:00
) -> Organizer:
"""
Synchronizes a folder with the ILIAS instance of the KIT.
Arguments:
target {Path} -- the target path to write the data to
course_id {str} -- the id of the main course page (found in the URL after ref_id
when opening the course homepage)
Keyword Arguments:
dir_filter {IliasDirectoryFilter} -- A filter for directories. Will be applied on the
crawler level, these directories and all of their content is skipped.
(default: {lambdax:True})
transform {Transform} -- A transformation function for the output paths. Return None
to ignore a file. (default: {lambdax:x})
cookies {Optional[Path]} -- The path to store and load cookies from.
(default: {None})
username {Optional[str]} -- The SCC username. If none is given, it will prompt
the user. (default: {None})
password {Optional[str]} -- The SCC password. If none is given, it will prompt
the user. (default: {None})
download_strategy {DownloadStrategy} -- A function to determine which files need to
be downloaded. Can save bandwidth and reduce the number of requests.
(default: {download_modified_or_new})
2020-04-30 16:24:38 +02:00
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.
"""
2020-04-23 11:44:13 +02:00
# This authenticator only works with the KIT ilias instance.
authenticator = KitShibbolethAuthenticator(username=username, password=password)
2020-04-23 19:55:37 +02:00
PRETTY.starting_synchronizer(target, "ILIAS", course_id)
2020-06-25 15:41:58 +02:00
organizer = self._ilias(
2020-04-23 11:44:13 +02:00
target=target,
base_url="https://ilias.studium.kit.edu/",
crawl_function=lambda crawler: crawler.crawl_course(course_id),
authenticator=authenticator,
cookies=cookies,
dir_filter=dir_filter,
transform=transform,
download_strategy=download_strategy,
clean=clean,
timeout=timeout
)
self._download_summary.merge(organizer.download_summary)
2020-06-25 15:41:58 +02:00
return organizer
2020-06-26 15:17:44 +02:00
def print_summary(self) -> None:
2020-06-26 15:52:07 +02:00
"""
Prints the accumulated download summary.
"""
PRETTY.summary(self._download_summary)
2020-06-25 21:30:03 +02:00
@swallow_and_print_errors
def ilias_kit_personal_desktop(
self,
target: PathLike,
dir_filter: IliasDirectoryFilter = lambda x, y: True,
transform: Transform = lambda x: x,
cookies: Optional[PathLike] = None,
username: Optional[str] = None,
password: Optional[str] = None,
download_strategy: IliasDownloadStrategy = download_modified_or_new,
clean: bool = True,
timeout: int = 5,
) -> Organizer:
"""
Synchronizes a folder with the ILIAS instance of the KIT. This method will crawl the ILIAS
"personal desktop" instead of a single course.
Arguments:
target {Path} -- the target path to write the data to
Keyword Arguments:
dir_filter {IliasDirectoryFilter} -- A filter for directories. Will be applied on the
crawler level, these directories and all of their content is skipped.
(default: {lambdax:True})
transform {Transform} -- A transformation function for the output paths. Return None
to ignore a file. (default: {lambdax:x})
cookies {Optional[Path]} -- The path to store and load cookies from.
(default: {None})
username {Optional[str]} -- The SCC username. If none is given, it will prompt
the user. (default: {None})
password {Optional[str]} -- The SCC password. If none is given, it will prompt
the user. (default: {None})
download_strategy {DownloadStrategy} -- A function to determine which files need to
be downloaded. Can save bandwidth and reduce the number of requests.
(default: {download_modified_or_new})
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.
"""
# This authenticator only works with the KIT ilias instance.
authenticator = KitShibbolethAuthenticator(username=username, password=password)
PRETTY.starting_synchronizer(target, "ILIAS", "Personal Desktop")
organizer = self._ilias(
target=target,
base_url="https://ilias.studium.kit.edu/",
crawl_function=lambda crawler: crawler.crawl_personal_desktop(),
2020-04-23 11:44:13 +02:00
authenticator=authenticator,
cookies=cookies,
dir_filter=dir_filter,
2020-04-23 11:44:13 +02:00
transform=transform,
download_strategy=download_strategy,
2020-04-30 16:24:38 +02:00
clean=clean,
timeout=timeout
2020-04-23 11:44:13 +02:00
)
2020-04-30 16:24:38 +02:00
self._download_summary.merge(organizer.download_summary)
return organizer
@swallow_and_print_errors
def ilias_kit_folder(
self,
target: PathLike,
full_url: str,
dir_filter: IliasDirectoryFilter = lambda x, y: True,
transform: Transform = lambda x: x,
cookies: Optional[PathLike] = None,
username: Optional[str] = None,
password: Optional[str] = None,
download_strategy: IliasDownloadStrategy = download_modified_or_new,
clean: bool = True,
timeout: int = 5,
) -> Organizer:
"""
Synchronizes a folder with a given folder on the ILIAS instance of the KIT.
Arguments:
target {Path} -- the target path to write the data to
full_url {str} -- the full url of the folder/videos/course to crawl
Keyword Arguments:
dir_filter {IliasDirectoryFilter} -- A filter for directories. Will be applied on the
crawler level, these directories and all of their content is skipped.
(default: {lambdax:True})
transform {Transform} -- A transformation function for the output paths. Return None
to ignore a file. (default: {lambdax:x})
cookies {Optional[Path]} -- The path to store and load cookies from.
(default: {None})
username {Optional[str]} -- The SCC username. If none is given, it will prompt
the user. (default: {None})
password {Optional[str]} -- The SCC password. If none is given, it will prompt
the user. (default: {None})
download_strategy {DownloadStrategy} -- A function to determine which files need to
be downloaded. Can save bandwidth and reduce the number of requests.
(default: {download_modified_or_new})
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.
"""
# This authenticator only works with the KIT ilias instance.
authenticator = KitShibbolethAuthenticator(username=username, password=password)
PRETTY.starting_synchronizer(target, "ILIAS", "An ILIAS element by url")
if not full_url.startswith("https://ilias.studium.kit.edu"):
raise FatalException("Not a valid KIT ILIAS URL")
organizer = self._ilias(
target=target,
base_url="https://ilias.studium.kit.edu/",
crawl_function=lambda crawler: crawler.recursive_crawl_url(full_url),
authenticator=authenticator,
cookies=cookies,
dir_filter=dir_filter,
transform=transform,
download_strategy=download_strategy,
clean=clean,
timeout=timeout
)
self._download_summary.merge(organizer.download_summary)
return organizer
2020-11-03 20:40:09 +01:00
@swallow_and_print_errors
def ipd_kit(
self,
target: Union[PathLike, Organizer],
url: str,
transform: Transform = lambda x: x,
download_strategy: IpdDownloadStrategy = ipd_download_new_or_modified,
clean: bool = True
) -> Organizer:
"""
Synchronizes a folder with a DIVA playlist.
Arguments:
target {Union[PathLike, Organizer]} -- The organizer / target folder to use.
url {str} -- the url to the page
Keyword Arguments:
transform {Transform} -- A transformation function for the output paths. Return None
to ignore a file. (default: {lambdax:x})
download_strategy {DivaDownloadStrategy} -- A function to determine which files need to
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.
"""
tmp_dir = self._tmp_dir.new_subdir()
if target is None:
PRETTY.starting_synchronizer("None", "IPD", url)
raise FatalException("Got 'None' as target directory, aborting")
if isinstance(target, Organizer):
organizer = target
else:
organizer = Organizer(self.resolve(to_path(target)))
PRETTY.starting_synchronizer(organizer.path, "IPD", url)
elements: List[IpdDownloadInfo] = IpdCrawler(url).crawl()
transformed = apply_transform(transform, elements)
if self._test_run:
self._print_transformables(transformed)
return organizer
downloader = IpdDownloader(tmp_dir=tmp_dir, organizer=organizer, strategy=download_strategy)
downloader.download_all(transformed)
2020-11-04 15:06:58 +01:00
if clean:
organizer.cleanup()
self._download_summary.merge(organizer.download_summary)
2020-11-03 20:40:09 +01:00
return organizer
@swallow_and_print_errors
2020-04-30 16:24:38 +02:00
def diva_kit(
self,
target: Union[PathLike, Organizer],
playlist_location: str,
2020-04-30 16:24:38 +02:00
transform: Transform = lambda x: x,
download_strategy: DivaDownloadStrategy = diva_download_new,
clean: bool = True
) -> Organizer:
"""
Synchronizes a folder with a DIVA playlist.
Arguments:
organizer {Organizer} -- The organizer to use.
playlist_location {str} -- the playlist id or the playlist URL
in the format 'https://mediaservice.bibliothek.kit.edu/#/details/DIVA-2019-271'
2020-04-30 16:24:38 +02:00
Keyword Arguments:
transform {Transform} -- A transformation function for the output paths. Return None
to ignore a file. (default: {lambdax:x})
download_strategy {DivaDownloadStrategy} -- A function to determine which files need to
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.
"""
tmp_dir = self._tmp_dir.new_subdir()
2020-05-08 23:47:05 +02:00
if playlist_location.startswith("http"):
playlist_id = DivaPlaylistCrawler.fetch_id(playlist_link=playlist_location)
else:
playlist_id = playlist_location
2020-05-08 23:47:05 +02:00
if target is None:
PRETTY.starting_synchronizer("None", "DIVA", playlist_id)
raise FatalException("Got 'None' as target directory, aborting")
2020-04-30 16:24:38 +02:00
if isinstance(target, Organizer):
organizer = target
else:
organizer = Organizer(self.resolve(to_path(target)))
2020-05-08 23:47:05 +02:00
PRETTY.starting_synchronizer(organizer.path, "DIVA", playlist_id)
2020-04-30 16:24:38 +02:00
crawler = DivaPlaylistCrawler(playlist_id)
downloader = DivaDownloader(tmp_dir, organizer, download_strategy)
info = crawler.crawl()
transformed = apply_transform(transform, info)
if self._test_run:
self._print_transformables(transformed)
return organizer
downloader.download_all(transformed)
if clean:
organizer.cleanup()
2020-11-04 15:06:58 +01:00
self._download_summary.merge(organizer.download_summary)
2020-04-30 16:24:38 +02:00
return organizer