Adjust version in setup.py

PFERD/ilias/__init__.py

@@ -3,7 +3,8 @@ Synchronizing files from ILIAS instances (https://www.ilias.de/).
 """
 
 from .authenticators import IliasAuthenticator, KitShibbolethAuthenticator
-from .crawler import IliasCrawler, IliasDirectoryFilter, IliasDirectoryType
+from .crawler import (IliasCrawler, IliasCrawlerEntry, IliasDirectoryFilter,
+                      IliasElementType)
 from .downloader import (IliasDownloader, IliasDownloadInfo,
                          IliasDownloadStrategy, download_everything,
                          download_modified_or_new)

PFERD/ilias/crawler.py

@@ -8,7 +8,7 @@ import logging
 import re
 from enum import Enum
 from pathlib import Path
-from typing import Any, Callable, Dict, List, Optional
+from typing import Any, Callable, Dict, List, Optional, Union
 from urllib.parse import (parse_qs, urlencode, urljoin, urlparse, urlsplit,
                           urlunsplit)
 
@@ -26,16 +26,52 @@ LOGGER = logging.getLogger(__name__)
 PRETTY = PrettyLogger(LOGGER)
 
 
-class IliasDirectoryType(Enum):
+class IliasElementType(Enum):
     """
-    The type of an ilias directory.
+    The type of an ilias element.
     """
-    FOLDER = "FOLDER"
+    REGULAR_FOLDER = "REGULAR_FOLDER"
-    VIDEO = "VIDEO"
+    VIDEO_FOLDER = "VIDEO_FOLDER"
-    EXERCISE = "EXERCISE"
+    EXERCISE_FOLDER = "EXERCISE_FOLDER"
+    REGULAR_FILE = "REGULAR_FILE"
+    VIDEO_FILE = "VIDEO_FILE"
+    FORUM = "FORUM"
+    EXTERNAL_LINK = "EXTERNAL_LINK"
 
 
-IliasDirectoryFilter = Callable[[Path, IliasDirectoryType], bool]
+IliasDirectoryFilter = Callable[[Path, IliasElementType], bool]
+
+
+class IliasCrawlerEntry:
+    # pylint: disable=too-few-public-methods
+    """
+    An ILIAS crawler entry used internally to find, catalogue and recursively crawl elements.
+    """
+
+    def __init__(
+            self,
+            path: Path,
+            url: Union[str, Callable[[], Optional[str]]],
+            entry_type: IliasElementType,
+            modification_date: Optional[datetime.datetime]
+    ):
+        self.path = path
+        if isinstance(url, str):
+            str_url = url
+            self.url: Callable[[], Optional[str]] = lambda: str_url
+        else:
+            self.url = url
+        self.entry_type = entry_type
+        self.modification_date = modification_date
+
+    def to_download_info(self) -> Optional[IliasDownloadInfo]:
+        """
+        Converts this crawler entry to an IliasDownloadInfo, if possible.
+        This method will only succeed for *File* types.
+        """
+        if self.entry_type in [IliasElementType.REGULAR_FILE, IliasElementType.VIDEO_FILE]:
+            return IliasDownloadInfo(self.path, self.url, self.modification_date)
+        return None
 
 
 class IliasCrawler:
@@ -62,12 +98,6 @@ class IliasCrawler:
         self._authenticator = authenticator
         self.dir_filter = dir_filter
 
-    def _abs_url_from_link(self, link_tag: bs4.Tag) -> str:
-        """
-        Create an absolute url from an <a> tag.
-        """
-        return urljoin(self._base_url, link_tag.get("href"))
-
     @staticmethod
     def _url_set_query_param(url: str, param: str, value: str) -> str:
         """
@@ -102,7 +132,8 @@ class IliasCrawler:
             )
 
         # And treat it as a folder
-        return self._crawl_folder(Path(""), root_url)
+        entries: List[IliasCrawlerEntry] = self._crawl_folder(Path(""), root_url)
+        return self._iterate_entries_to_download_infos(entries)
 
     def _is_course_id_valid(self, root_url: str, course_id: str) -> bool:
         response: requests.Response = self._session.get(root_url)
@@ -115,14 +146,92 @@ class IliasCrawler:
         Raises:
             FatalException: if an unrecoverable error occurs
         """
-        return self._crawl_folder(Path(""), self._base_url + "?baseClass=ilPersonalDesktopGUI")
+        entries: List[IliasCrawlerEntry] = self._crawl_folder(
+            Path(""), self._base_url + "?baseClass=ilPersonalDesktopGUI"
+        )
+        return self._iterate_entries_to_download_infos(entries)
 
-    def _switch_on_crawled_type(
+    def _iterate_entries_to_download_infos(
             self,
+            entries: List[IliasCrawlerEntry]
+    ) -> List[IliasDownloadInfo]:
+        result: List[IliasDownloadInfo] = []
+        entries_to_process: List[IliasCrawlerEntry] = entries.copy()
+        while len(entries_to_process) > 0:
+            entry = entries_to_process.pop()
+
+            if entry.entry_type == IliasElementType.EXTERNAL_LINK:
+                PRETTY.not_searching(entry.path, "external link")
+                continue
+            if entry.entry_type == IliasElementType.FORUM:
+                PRETTY.not_searching(entry.path, "forum")
+                continue
+
+            if not self.dir_filter(entry.path, entry.entry_type):
+                PRETTY.not_searching(entry.path, "user filter")
+                continue
+
+            download_info = entry.to_download_info()
+            if download_info is not None:
+                result.append(download_info)
+                continue
+
+            url = entry.url()
+
+            if url is None:
+                PRETTY.warning(f"Could not find url for {str(entry.path)!r}, skipping it")
+                continue
+
+            PRETTY.searching(entry.path)
+
+            if entry.entry_type == IliasElementType.EXERCISE_FOLDER:
+                entries_to_process += self._crawl_exercises(entry.path, url)
+                continue
+            if entry.entry_type == IliasElementType.REGULAR_FOLDER:
+                entries_to_process += self._crawl_folder(entry.path, url)
+                continue
+            if entry.entry_type == IliasElementType.VIDEO_FOLDER:
+                entries_to_process += self._crawl_video_directory(entry.path, url)
+                continue
+
+        return result
+
+    def _crawl_folder(self, folder_path: Path, url: str) -> List[IliasCrawlerEntry]:
+        """
+        Crawl all files in a folder-like element.
+        """
+        soup = self._get_page(url, {})
+
+        result: List[IliasCrawlerEntry] = []
+
+        # Fetch all links and throw them to the general interpreter
+        links: List[bs4.Tag] = soup.select("a.il_ContainerItemTitle")
+        for link in links:
+            abs_url = self._abs_url_from_link(link)
+            element_path = Path(folder_path, link.getText().strip())
+            element_type = self._find_type_from_link(element_path, link, abs_url)
+
+            if element_type == IliasElementType.REGULAR_FILE:
+                result += self._crawl_file(folder_path, link, abs_url)
+            elif element_type is not None:
+                result += [IliasCrawlerEntry(element_path, abs_url, element_type, None)]
+            else:
+                PRETTY.warning(f"Found element without a type at {str(element_path)!r}")
+
+        return result
+
+    def _abs_url_from_link(self, link_tag: bs4.Tag) -> str:
+        """
+        Create an absolute url from an <a> tag.
+        """
+        return urljoin(self._base_url, link_tag.get("href"))
+
+    @staticmethod
+    def _find_type_from_link(
             path: Path,
             link_element: bs4.Tag,
             url: str
-    ) -> List[IliasDownloadInfo]:
+    ) -> Optional[IliasElementType]:
         """
         Decides which sub crawler to use for a given top level element.
         """
@@ -131,28 +240,64 @@ class IliasCrawler:
 
         # file URLs contain "target=file"
         if "target=file_" in parsed_url.query:
-            LOGGER.debug("Interpreted as file.")
+            return IliasElementType.REGULAR_FILE
-            return self._crawl_file(path, link_element, url)
 
         # Skip forums
         if "cmd=showThreads" in parsed_url.query:
-            LOGGER.debug("Skipping forum %r", url)
+            return IliasElementType.FORUM
-            return []
 
         # Everything with a ref_id can *probably* be opened to reveal nested things
         # video groups, directories, exercises, etc
         if "ref_id=" in parsed_url.query:
-            LOGGER.debug("Processing folder-like...")
+            return IliasCrawler._find_type_from_folder_like(link_element, url)
-            return self._switch_on_folder_like(path, link_element, url)
 
         PRETTY.warning(
-            "Got unkwarning element type in switch. I am not sure what horror I found on the"
+            "Got unknown element type in switch. I am not sure what horror I found on the"
             f" ILIAS page. The element was at {str(path)!r} and it is {link_element!r})"
         )
-        return []
+        return None
 
     @staticmethod
-    def _crawl_file(path: Path, link_element: bs4.Tag, url: str) -> List[IliasDownloadInfo]:
+    def _find_type_from_folder_like(link_element: bs4.Tag, url: str) -> Optional[IliasElementType]:
+        """
+        Try crawling something that looks like a folder.
+        """
+        # pylint: disable=too-many-return-statements
+
+        # We look for the outer div of our inner link, to find information around it
+        # (mostly the icon)
+        for parent in link_element.parents:
+            if "ilContainerListItemOuter" in parent["class"]:
+                found_parent = parent
+                break
+
+        if found_parent is None:
+            PRETTY.warning(f"Could not find element icon for {url!r}")
+            return None
+
+        # Find the small descriptive icon to figure out the type
+        img_tag: Optional[bs4.Tag] = found_parent.select_one("img.ilListItemIcon")
+
+        if img_tag is None:
+            PRETTY.warning(f"Could not find image tag for {url!r}")
+            return None
+
+        if "opencast" in str(img_tag["alt"]).lower():
+            return IliasElementType.VIDEO_FOLDER
+
+        if str(img_tag["src"]).endswith("icon_exc.svg"):
+            return IliasElementType.EXERCISE_FOLDER
+
+        if str(img_tag["src"]).endswith("icon_webr.svg"):
+            return IliasElementType.EXTERNAL_LINK
+
+        if str(img_tag["src"]).endswith("frm.svg"):
+            return IliasElementType.FORUM
+
+        return IliasElementType.REGULAR_FOLDER
+
+    @staticmethod
+    def _crawl_file(path: Path, link_element: bs4.Tag, url: str) -> List[IliasCrawlerEntry]:
         """
         Crawls a file.
         """
@@ -183,80 +328,11 @@ class IliasCrawler:
         name = link_element.getText()
         full_path = Path(path, name + "." + file_type)
 
-        return [IliasDownloadInfo(full_path, url, modification_date)]
+        return [
+            IliasCrawlerEntry(full_path, url, IliasElementType.REGULAR_FILE, modification_date)
+        ]
 
-    def _switch_on_folder_like(
+    def _crawl_video_directory(self, video_dir_path: Path, url: str) -> List[IliasCrawlerEntry]:
-            self,
-            parent_path: Path,
-            link_element: bs4.Tag,
-            url: str
-    ) -> List[IliasDownloadInfo]:
-        """
-        Try crawling something that looks like a folder.
-        """
-        # pylint: disable=too-many-return-statements
-
-        element_path = Path(parent_path, link_element.getText().strip())
-
-        found_parent: Optional[bs4.Tag] = None
-
-        # We look for the outer div of our inner link, to find information around it
-        # (mostly the icon)
-        for parent in link_element.parents:
-            if "ilContainerListItemOuter" in parent["class"]:
-                found_parent = parent
-                break
-
-        if found_parent is None:
-            PRETTY.warning(f"Could not find element icon for {url!r}")
-            return []
-
-        # Find the small descriptive icon to figure out the type
-        img_tag: Optional[bs4.Tag] = found_parent.select_one("img.ilListItemIcon")
-
-        if img_tag is None:
-            PRETTY.warning(f"Could not find image tag for {url!r}")
-            return []
-
-        directory_type = IliasDirectoryType.FOLDER
-
-        if "opencast" in str(img_tag["alt"]).lower():
-            directory_type = IliasDirectoryType.VIDEO
-
-        if str(img_tag["src"]).endswith("icon_exc.svg"):
-            directory_type = IliasDirectoryType.EXERCISE
-
-        if not self.dir_filter(element_path, directory_type):
-            PRETTY.not_searching(element_path, "user filter")
-            return []
-
-        PRETTY.searching(element_path)
-
-        # A forum
-        if str(img_tag["src"]).endswith("frm.svg"):
-            LOGGER.debug("Skipping forum at %r", url)
-            PRETTY.not_searching(element_path, "forum")
-            return []
-
-        # An exercise
-        if directory_type == IliasDirectoryType.EXERCISE:
-            LOGGER.debug("Crawling exercises at %r", url)
-            return self._crawl_exercises(element_path, url)
-
-        if str(img_tag["src"]).endswith("icon_webr.svg"):
-            LOGGER.debug("Skipping external link at %r", url)
-            PRETTY.not_searching(element_path, "external link")
-            return []
-
-        # Match the opencast video plugin
-        if directory_type == IliasDirectoryType.VIDEO:
-            LOGGER.debug("Found video site: %r", url)
-            return self._crawl_video_directory(element_path, url)
-
-        # Assume it is a folder
-        return self._crawl_folder(element_path, self._abs_url_from_link(link_element))
-
-    def _crawl_video_directory(self, video_dir_path: Path, url: str) -> List[IliasDownloadInfo]:
         """
         Crawl the video overview site.
         """
@@ -272,6 +348,71 @@ class IliasCrawler:
             {"limit": 800, "cmd": "asyncGetTableGUI", "cmdMode": "asynch"}
         )
 
+        # If we find a page selected, we probably need to respect pagination
+        if self._is_paginated_video_page(video_list_soup):
+            second_stage_url = self._abs_url_from_link(content_link)
+
+            return self._crawl_paginated_video_directory(
+                video_dir_path, video_list_soup, second_stage_url
+            )
+
+        return self._crawl_video_directory_second_stage(video_dir_path, video_list_soup)
+
+    @staticmethod
+    def _is_paginated_video_page(soup: bs4.BeautifulSoup) -> bool:
+        return soup.find(id=re.compile(r"tab_page_sel.+")) is not None
+
+    def _crawl_paginated_video_directory(
+            self,
+            video_dir_path: Path,
+            paged_video_list_soup: bs4.BeautifulSoup,
+            second_stage_url: str
+    ) -> List[IliasCrawlerEntry]:
+        LOGGER.info("Found paginated video page, trying 800 elements")
+
+        # Try to find the table id. This can be used to build the query parameter indicating
+        # you want 800 elements
+
+        table_element: bs4.Tag = paged_video_list_soup.find(
+            name="table", id=re.compile(r"tbl_xoct_.+")
+        )
+        if table_element is None:
+            PRETTY.warning(
+                "Could not increase elements per page (table not found)."
+                " Some might not be crawled!"
+            )
+            return self._crawl_video_directory_second_stage(video_dir_path, paged_video_list_soup)
+
+        match = re.match(r"tbl_xoct_(.+)", table_element.attrs["id"])
+        if match is None:
+            PRETTY.warning(
+                "Could not increase elements per page (table id not found)."
+                " Some might not be crawled!"
+            )
+            return self._crawl_video_directory_second_stage(video_dir_path, paged_video_list_soup)
+        table_id = match.group(1)
+
+        extended_video_page = self._get_page(
+            second_stage_url,
+            {f"tbl_xoct_{table_id}_trows": 800, "cmd": "asyncGetTableGUI", "cmdMode": "asynch"}
+        )
+
+        if self._is_paginated_video_page(extended_video_page):
+            PRETTY.warning(
+                "800 elements do not seem to be enough (or I failed to fetch that many)."
+                " I will miss elements."
+            )
+
+        return self._crawl_video_directory_second_stage(video_dir_path, extended_video_page)
+
+    def _crawl_video_directory_second_stage(
+            self,
+            video_dir_path: Path,
+            video_list_soup: bs4.BeautifulSoup
+    ) -> List[IliasCrawlerEntry]:
+        """
+        Crawls the "second stage" video page. This page contains the actual video urls.
+        """
         direct_download_links: List[bs4.Tag] = video_list_soup.findAll(
             name="a", text=re.compile(r"\s*Download\s*")
         )
@@ -281,7 +422,7 @@ class IliasCrawler:
             name="a", text=re.compile(r"\s*Abspielen\s*")
         )
 
-        results: List[IliasDownloadInfo] = []
+        results: List[IliasCrawlerEntry] = []
 
         # We can download everything directly!
         if len(direct_download_links) == len(video_links):
@@ -298,7 +439,7 @@ class IliasCrawler:
             parent_path: Path,
             link: bs4.Tag,
             direct_download: bool
-    ) -> List[IliasDownloadInfo]:
+    ) -> List[IliasCrawlerEntry]:
         """
         Crawl a single video based on its "Abspielen" link from the video listing.
         """
@@ -316,46 +457,54 @@ class IliasCrawler:
 
         video_path: Path = Path(parent_path, title)
 
+        video_url = self._abs_url_from_link(link)
+
         # The video had a direct download button we can use instead
         if direct_download:
             LOGGER.debug("Using direct download for video %r", str(video_path))
-            return [IliasDownloadInfo(
+            return [IliasCrawlerEntry(
+                video_path, video_url, IliasElementType.VIDEO_FILE, modification_time
+            )]
+
+        return [IliasCrawlerEntry(
             video_path,
-                self._abs_url_from_link(link),
+            self._crawl_video_url_from_play_link(video_url),
+            IliasElementType.VIDEO_FILE,
             modification_time
         )]
 
+    def _crawl_video_url_from_play_link(self, play_url: str) -> Callable[[], Optional[str]]:
+        def inner() -> Optional[str]:
             # Fetch the actual video page. This is a small wrapper page initializing a javscript
             # player. Sadly we can not execute that JS. The actual video stream url is nowhere
             # on the page, but defined in a JS object inside a script tag, passed to the player
             # library.
             # We do the impossible and RegEx the stream JSON object out of the page's HTML source
-        video_page_url = self._abs_url_from_link(link)
+            video_page_soup = soupify(self._session.get(play_url))
-        video_page_soup = self._get_page(video_page_url, {})
             regex: re.Pattern = re.compile(
                 r"({\"streams\"[\s\S]+?),\s*{\"paella_config_file", re.IGNORECASE
             )
             json_match = regex.search(str(video_page_soup))
 
             if json_match is None:
-            PRETTY.warning(f"Could not find json stream info for {video_page_url!r}")
+                PRETTY.warning(f"Could not find json stream info for {play_url!r}")
-            return []
+                return None
             json_str = json_match.group(1)
 
             # parse it
             json_object = json.loads(json_str)
             # and fetch the video url!
             video_url = json_object["streams"][0]["sources"]["mp4"][0]["src"]
+            return video_url
+        return inner
 
-        return [IliasDownloadInfo(video_path, video_url, modification_time)]
+    def _crawl_exercises(self, element_path: Path, url: str) -> List[IliasCrawlerEntry]:
-
-    def _crawl_exercises(self, element_path: Path, url: str) -> List[IliasDownloadInfo]:
         """
         Crawl files offered for download in exercises.
         """
         soup = self._get_page(url, {})
 
-        results: List[IliasDownloadInfo] = []
+        results: List[IliasCrawlerEntry] = []
 
         # Each assignment is in an accordion container
         assignment_containers: List[bs4.Tag] = soup.select(".il_VAccordionInnerContainer")
@@ -382,30 +531,15 @@ class IliasCrawler:
 
                 LOGGER.debug("Found file %r at %r", file_name, url)
 
-                results.append(IliasDownloadInfo(
+                results.append(IliasCrawlerEntry(
                     Path(element_path, container_name, file_name),
                     url,
+                    IliasElementType.REGULAR_FILE,
                     None  # We do not have any timestamp
                 ))
 
         return results
 
-    def _crawl_folder(self, folder_path: Path, url: str) -> List[IliasDownloadInfo]:
-        """
-        Crawl all files in a folder-like element.
-        """
-        soup = self._get_page(url, {})
-
-        result: List[IliasDownloadInfo] = []
-
-        # Fetch all links and throw them to the general interpreter
-        links: List[bs4.Tag] = soup.select("a.il_ContainerItemTitle")
-        for link in links:
-            abs_url = self._abs_url_from_link(link)
-            result += self._switch_on_crawled_type(folder_path, link, abs_url)
-
-        return result
-
     def _get_page(self, url: str, params: Dict[str, Any]) -> bs4.BeautifulSoup:
         """
         Fetches a page from ILIAS, authenticating when needed.
@@ -416,8 +550,10 @@ class IliasCrawler:
         content_type = response.headers["content-type"]
 
         if not content_type.startswith("text/html"):
-            # TODO: Correct exception type
+            raise FatalException(
-            raise Exception(f"Invalid content type {content_type}")
+                f"Invalid content type {content_type} when crawling ilias page"
+                " {url!r} with {params!r}"
+            )
 
         soup = soupify(response)
 

PFERD/ilias/downloader.py

@@ -2,9 +2,8 @@
 
 import datetime
 import logging
-from dataclasses import dataclass
+from pathlib import Path, PurePath
-from pathlib import Path
+from typing import Callable, List, Optional, Union
-from typing import Callable, List, Optional
 
 import bs4
 import requests
@@ -24,15 +23,24 @@ class ContentTypeException(Exception):
     """Thrown when the content type of the ilias element can not be handled."""
 
 
-@dataclass
 class IliasDownloadInfo(Transformable):
     """
     This class describes a single file to be downloaded.
     """
 
-    url: str
+    def __init__(
-    modification_date: Optional[datetime.datetime]
+            self,
-    # parameters: Dict[str, Any] = field(default_factory=dict)
+            path: PurePath,
+            url: Union[str, Callable[[], Optional[str]]],
+            modifcation_date: Optional[datetime.datetime]
+    ):
+        super().__init__(path)
+        if isinstance(url, str):
+            string_url = url
+            self.url: Callable[[], Optional[str]] = lambda: string_url
+        else:
+            self.url = url
+        self.modification_date = modifcation_date
 
 
 IliasDownloadStrategy = Callable[[Organizer, IliasDownloadInfo], bool]
@@ -108,16 +116,21 @@ class IliasDownloader:
         tmp_file = self._tmp_dir.new_path()
 
         while not self._try_download(info, tmp_file):
+            LOGGER.info("Retrying download: %r", info)
             self._authenticator.authenticate(self._session)
 
         self._organizer.accept_file(tmp_file, info.path)
 
     def _try_download(self, info: IliasDownloadInfo, target: Path) -> bool:
-        with self._session.get(info.url, stream=True) as response:
+        url = info.url()
+        if url is None:
+            PRETTY.warning(f"Could not download {str(info.path)!r} as I got no URL :/")
+            return True
+
+        with self._session.get(url, stream=True) as response:
             content_type = response.headers["content-type"]
 
             if content_type.startswith("text/html"):
-                # Dangit, we're probably not logged in.
                 if self._is_logged_in(soupify(response)):
                     raise ContentTypeException("Attempting to download a web page, not a file")
 

PFERD/pferd.py

@@ -45,7 +45,6 @@ class Pferd(Location):
         self._tmp_dir = TmpDir(self.resolve(tmp_dir))
         self._test_run = test_run
 
-
     @staticmethod
     def enable_logging() -> None:
         """
@@ -54,7 +53,6 @@ class Pferd(Location):
 
         enable_logging()
 
-
     @staticmethod
     def _print_transformables(transformables: List[TF]) -> None:
         LOGGER.info("")

example_config.py

@@ -2,7 +2,7 @@ import argparse
 from pathlib import Path, PurePath
 
 from PFERD import Pferd
-from PFERD.ilias import IliasDirectoryType
+from PFERD.ilias import IliasElementType
 from PFERD.transform import (attempt, do, glob, keep, move, move_dir,
                              optionally, re_move, re_rename)
 
@@ -49,7 +49,7 @@ tf_ss_2020_pg = attempt(
 )
 
 
-def df_ss_2020_or1(path: PurePath, _type: IliasDirectoryType) -> bool:
+def df_ss_2020_or1(path: PurePath, _type: IliasElementType) -> bool:
     if glob("Tutorien/")(path):
         return True
     if glob("Tutorien/Tutorium 10, dienstags 15:45 Uhr/")(path):

setup.py

@@ -2,7 +2,7 @@ from setuptools import find_packages, setup
 
 setup(
     name="PFERD",
-    version="2.0.0",
+    version="2.1.0",
     packages=find_packages(),
     install_requires=[
         "requests>=2.21.0",