Use downloader-specific data classes

This commit is contained in:
Joscha
2020-04-20 18:06:21 +00:00
parent 55ba2f4070
commit 0926d33798
2 changed files with 67 additions and 30 deletions

View File

@ -1,7 +1,8 @@
"""Contains a downloader for ILIAS."""
from dataclasses import dataclass, field
from pathlib import Path
from typing import Any, Dict
from typing import Any, Dict, List
import bs4
import requests
@ -16,8 +17,18 @@ class ContentTypeException(Exception):
"""Thrown when the content type of the ilias element can not be handled."""
# pylint: disable=too-few-public-methods
class IliasDownloader():
@dataclass
class IliasDownloadInfo:
"""
This class describes a single file to be downloaded.
"""
path: Path
url: str
parameters: Dict[str, Any] = field(default_factory=dict)
class IliasDownloader:
"""A downloader for ILIAS."""
def __init__(self, tmp_dir: TmpDir, organizer: Organizer, authenticator: IliasAuthenticator):
@ -27,32 +38,41 @@ class IliasDownloader():
self._tmp_dir = tmp_dir
self._organizer = organizer
def download(self, url: str, target_path: Path, params: Dict[str, Any]) -> None:
"""Download a file from ILIAS.
Retries authentication until eternity, if it could not fetch the file.
def download_all(self, infos: List[IliasDownloadInfo]) -> None:
"""
Download multiple files one after the other.
"""
for info in infos:
self.download(info)
def download(self, info: IliasDownloadInfo) -> None:
"""
Download a file from ILIAS.
Retries authentication until eternity if it could not fetch the file.
"""
tmp_file = self._tmp_dir.new_file()
while not self._try_download(url, tmp_file, params):
while not self._try_download(info, tmp_file):
self._authenticator.authenticate(self._session)
self._organizer.accept_file(tmp_file, target_path)
self._organizer.accept_file(tmp_file, info.path)
def _try_download(self, url: str, target_path: Path, params: Dict[str, Any]) -> bool:
with self._session.get(url, params=params, stream=True) as response:
def _try_download(self, info: IliasDownloadInfo, target: Path) -> bool:
with self._session.get(info.url, params=info.parameters, stream=True) as response:
content_type = response.headers["content-type"]
if content_type.startswith("text/html"):
# Dangit, we're probably not logged in.
soup = soupify(response)
if self._is_logged_in(soup):
if self._is_logged_in(soupify(response)):
raise ContentTypeException("Attempting to download a web page, not a file")
return False
# Yay, we got the file :)
stream_to_path(response, target_path)
stream_to_path(response, target)
return True
@staticmethod