Compare commits

..

3 commits

Author SHA1 Message Date
I-Al-Istannen
7291382430 Bump version to 3.8.0 2025-04-15 11:32:22 +02:00
I-Al-Istannen
1a430ad5d1 Update minimum Python version to 3.11 2025-04-15 11:31:39 +02:00
I-Al-Istannen
f6bdeb6b9d Support ILIAS 9 2025-04-15 11:19:53 +02:00
49 changed files with 1110 additions and 2280 deletions

View file

@ -1 +0,0 @@
2cf0e060ed126537dd993896b6aa793e2a6b9e80

View file

@ -18,13 +18,19 @@ jobs:
steps: steps:
- uses: actions/checkout@v4 - uses: actions/checkout@v4
- name: Install uv - uses: actions/setup-python@v5
uses: astral-sh/setup-uv@v7
with: with:
python-version: ${{ matrix.python }} python-version: ${{ matrix.python }}
- name: Set up project - name: Set up project
run: uv sync if: matrix.os != 'windows-latest'
run: ./scripts/setup
- name: Set up project on windows
if: matrix.os == 'windows-latest'
# For some reason, `pip install --upgrade pip` doesn't work on
# 'windows-latest'. The installed pip version works fine however.
run: ./scripts/setup --no-pip
- name: Run checks - name: Run checks
run: | run: |

View file

@ -22,55 +22,14 @@ ambiguous situations.
## Unreleased ## Unreleased
## Added ## 3.8.0 - 2025-04-15
- Store the description when using the `internet-shortcut` link format
- Support for basic auth with the kit-ipd crawler
## Fixed
- Event loop errors on Windows with Python 3.14
- Sanitize `/` in headings in kit-ipd crawler
- Crawl info tab again
## 3.8.3 - 2025-07-01
## Added
- Support for link collections.
In "fancy" mode, a single HTML file with multiple links is generated.
In all other modes, PFERD creates a folder for the collection and a new file
for every link inside.
## Fixed
- Crawling of exercises with instructions
- Don't download unavailable elements.
Elements that are unavailable (for example, because their availability is
time restricted) will not download the HTML for the info page anymore.
- `base_url` argument for `ilias-web` crawler causing crashes
## 3.8.2 - 2025-04-29
## Changed
- Explicitly mention that wikis are not supported at the moment and ignore them
## Fixed
- Ilias-native login
- Exercise crawling
## 3.8.1 - 2025-04-17
## Fixed
- Description html files now specify at UTF-8 encoding
- Images in descriptions now always have a white background
## 3.8.0 - 2025-04-16
### Added ### Added
- Support for ILIAS 9 - Support for ILIAS 9
### Changed ### Changed
- Added prettier CSS to forum threads - Added prettier CSS to forum threads
- Downloaded forum threads now link to the forum instead of the ILIAS thread
- Increase minimum supported Python version to 3.11 - Increase minimum supported Python version to 3.11
- Do not crawl nested courses (courses linked in other courses)
## Fixed ## Fixed
- File links in report on Windows - File links in report on Windows

View file

@ -153,7 +153,6 @@ requests is likely a good idea.
- `link_regex`: A regex that is matched against the `href` part of links. If it - `link_regex`: A regex that is matched against the `href` part of links. If it
matches, the given link is downloaded as a file. This is used to extract matches, the given link is downloaded as a file. This is used to extract
files from KIT-IPD pages. (Default: `^.*?[^/]+\.(pdf|zip|c|cpp|java)$`) files from KIT-IPD pages. (Default: `^.*?[^/]+\.(pdf|zip|c|cpp|java)$`)
- `auth`: Name of auth section to use for basic authentication. (Optional)
### The `ilias-web` crawler ### The `ilias-web` crawler
@ -165,14 +164,12 @@ out of the box for the corresponding universities:
[ilias-dl]: https://github.com/V3lop5/ilias-downloader/blob/main/configs "ilias-downloader configs" [ilias-dl]: https://github.com/V3lop5/ilias-downloader/blob/main/configs "ilias-downloader configs"
| University | `base_url` | `login_type` | `client_id` | | University | `base_url` | `login_type` | `client_id` |
|-----------------|-----------------------------------------|--------------|---------------| |---------------|-----------------------------------------|--------------|---------------|
| FH Aachen | https://www.ili.fh-aachen.de | local | elearning | | FH Aachen | https://www.ili.fh-aachen.de | local | elearning |
| HHU Düsseldorf | https://ilias.hhu.de | local | UniRZ |
| Uni Köln | https://www.ilias.uni-koeln.de/ilias | local | uk | | Uni Köln | https://www.ilias.uni-koeln.de/ilias | local | uk |
| Uni Konstanz | https://ilias.uni-konstanz.de | local | ILIASKONSTANZ | | Uni Konstanz | https://ilias.uni-konstanz.de | local | ILIASKONSTANZ |
| Uni Stuttgart | https://ilias3.uni-stuttgart.de | local | Uni_Stuttgart | | Uni Stuttgart | https://ilias3.uni-stuttgart.de | local | Uni_Stuttgart |
| Uni Tübingen | https://ovidius.uni-tuebingen.de/ilias3 | shibboleth | | | Uni Tübingen | https://ovidius.uni-tuebingen.de/ilias3 | shibboleth | |
| KIT ILIAS Pilot | https://pilot.ilias.studium.kit.edu | shibboleth | pilot |
If your university isn't listed, try navigating to your instance's login page. If your university isn't listed, try navigating to your instance's login page.
Assuming no custom login service is used, the URL will look something like this: Assuming no custom login service is used, the URL will look something like this:

21
DEV.md
View file

@ -9,25 +9,30 @@ particular [this][ppug-1] and [this][ppug-2] guide).
## Setting up a dev environment ## Setting up a dev environment
The use of [venv][venv] and [uv][uv] is recommended. To initially set up a The use of [venv][venv] is recommended. To initially set up a development
development environment, run these commands in the same directory as this file: environment, run these commands in the same directory as this file:
``` ```
$ uv sync $ python -m venv .venv
$ . .venv/bin/activate $ . .venv/bin/activate
$ ./scripts/setup
``` ```
This install all required dependencies and tools. It also installs PFERD as The setup script installs a few required dependencies and tools. It also
*editable*, which means that you can just run `pferd` as if it was installed installs PFERD via `pip install --editable .`, which means that you can just run
normally. Since PFERD was installed with `--editable`, there is no need to `pferd` as if it was installed normally. Since PFERD was installed with
re-run `uv sync` when the source code is changed. `--editable`, there is no need to re-run `pip install` when the source code is
changed.
If you get any errors because pip can't update itself, try running
`./scripts/setup --no-pip` instead of `./scripts/setup`.
For more details, see [this part of the Python Tutorial][venv-tut] and For more details, see [this part of the Python Tutorial][venv-tut] and
[this section on "development mode"][ppug-dev]. [this section on "development mode"][ppug-dev].
[venv]: <https://docs.python.org/3/library/venv.html> "venv - Creation of virtual environments" [venv]: <https://docs.python.org/3/library/venv.html> "venv - Creation of virtual environments"
[venv-tut]: <https://docs.python.org/3/tutorial/venv.html> "12. Virtual Environments and Packages" [venv-tut]: <https://docs.python.org/3/tutorial/venv.html> "12. Virtual Environments and Packages"
[uv]: <https://docs.astral.sh/uv/> "uv - An extremely fast Python package and project manager" [ppug-dev]: <https://packaging.python.org/guides/distributing-packages-using-setuptools/#working-in-development-mode> "Working in “development mode”"
## Checking and formatting the code ## Checking and formatting the code

View file

@ -133,8 +133,7 @@ def main() -> None:
# https://bugs.python.org/issue39232 # https://bugs.python.org/issue39232
# https://github.com/encode/httpx/issues/914#issuecomment-780023632 # https://github.com/encode/httpx/issues/914#issuecomment-780023632
# TODO Fix this properly # TODO Fix this properly
loop = asyncio.new_event_loop() loop = asyncio.get_event_loop()
asyncio.set_event_loop(loop)
loop.run_until_complete(pferd.run(args.debug_transforms)) loop.run_until_complete(pferd.run(args.debug_transforms))
loop.run_until_complete(asyncio.sleep(1)) loop.run_until_complete(asyncio.sleep(1))
loop.close() loop.close()

View file

@ -1,5 +1,5 @@
from collections.abc import Callable
from configparser import SectionProxy from configparser import SectionProxy
from typing import Callable, Dict
from ..config import Config from ..config import Config
from .authenticator import Authenticator, AuthError, AuthLoadError, AuthSection # noqa: F401 from .authenticator import Authenticator, AuthError, AuthLoadError, AuthSection # noqa: F401
@ -9,19 +9,21 @@ from .pass_ import PassAuthenticator, PassAuthSection
from .simple import SimpleAuthenticator, SimpleAuthSection from .simple import SimpleAuthenticator, SimpleAuthSection
from .tfa import TfaAuthenticator from .tfa import TfaAuthenticator
AuthConstructor = Callable[ AuthConstructor = Callable[[
[
str, # Name (without the "auth:" prefix) str, # Name (without the "auth:" prefix)
SectionProxy, # Authenticator's section of global config SectionProxy, # Authenticator's section of global config
Config, # Global config Config, # Global config
], ], Authenticator]
Authenticator,
]
AUTHENTICATORS: dict[str, AuthConstructor] = { AUTHENTICATORS: Dict[str, AuthConstructor] = {
"credential-file": lambda n, s, c: CredentialFileAuthenticator(n, CredentialFileAuthSection(s), c), "credential-file": lambda n, s, c:
"keyring": lambda n, s, c: KeyringAuthenticator(n, KeyringAuthSection(s)), CredentialFileAuthenticator(n, CredentialFileAuthSection(s), c),
"pass": lambda n, s, c: PassAuthenticator(n, PassAuthSection(s)), "keyring": lambda n, s, c:
"simple": lambda n, s, c: SimpleAuthenticator(n, SimpleAuthSection(s)), KeyringAuthenticator(n, KeyringAuthSection(s)),
"tfa": lambda n, s, c: TfaAuthenticator(n), "pass": lambda n, s, c:
PassAuthenticator(n, PassAuthSection(s)),
"simple": lambda n, s, c:
SimpleAuthenticator(n, SimpleAuthSection(s)),
"tfa": lambda n, s, c:
TfaAuthenticator(n),
} }

View file

@ -1,4 +1,5 @@
from abc import ABC, abstractmethod from abc import ABC, abstractmethod
from typing import Tuple
from ..config import Section from ..config import Section
@ -34,7 +35,7 @@ class Authenticator(ABC):
self.name = name self.name = name
@abstractmethod @abstractmethod
async def credentials(self) -> tuple[str, str]: async def credentials(self) -> Tuple[str, str]:
pass pass
async def username(self) -> str: async def username(self) -> str:

View file

@ -1,4 +1,5 @@
from pathlib import Path from pathlib import Path
from typing import Tuple
from ..config import Config from ..config import Config
from ..utils import fmt_real_path from ..utils import fmt_real_path
@ -22,9 +23,7 @@ class CredentialFileAuthenticator(Authenticator):
with open(path, encoding="utf-8") as f: with open(path, encoding="utf-8") as f:
lines = list(f) lines = list(f)
except UnicodeDecodeError: except UnicodeDecodeError:
raise AuthLoadError( raise AuthLoadError(f"Credential file at {fmt_real_path(path)} is not encoded using UTF-8")
f"Credential file at {fmt_real_path(path)} is not encoded using UTF-8"
) from None
except OSError as e: except OSError as e:
raise AuthLoadError(f"No credential file at {fmt_real_path(path)}") from e raise AuthLoadError(f"No credential file at {fmt_real_path(path)}") from e
@ -43,5 +42,5 @@ class CredentialFileAuthenticator(Authenticator):
self._username = uline[9:] self._username = uline[9:]
self._password = pline[9:] self._password = pline[9:]
async def credentials(self) -> tuple[str, str]: async def credentials(self) -> Tuple[str, str]:
return self._username, self._password return self._username, self._password

View file

@ -1,4 +1,4 @@
from typing import Optional from typing import Optional, Tuple, cast
import keyring import keyring
@ -13,10 +13,11 @@ class KeyringAuthSection(AuthSection):
return self.s.get("username") return self.s.get("username")
def keyring_name(self) -> str: def keyring_name(self) -> str:
return self.s.get("keyring_name", fallback=NAME) return cast(str, self.s.get("keyring_name", fallback=NAME))
class KeyringAuthenticator(Authenticator): class KeyringAuthenticator(Authenticator):
def __init__(self, name: str, section: KeyringAuthSection) -> None: def __init__(self, name: str, section: KeyringAuthSection) -> None:
super().__init__(name) super().__init__(name)
@ -27,7 +28,7 @@ class KeyringAuthenticator(Authenticator):
self._password_invalidated = False self._password_invalidated = False
self._username_fixed = section.username() is not None self._username_fixed = section.username() is not None
async def credentials(self) -> tuple[str, str]: async def credentials(self) -> Tuple[str, str]:
# Request the username # Request the username
if self._username is None: if self._username is None:
async with log.exclusive_output(): async with log.exclusive_output():

View file

@ -1,5 +1,6 @@
import re import re
import subprocess import subprocess
from typing import List, Tuple
from ..logging import log from ..logging import log
from .authenticator import Authenticator, AuthError, AuthSection from .authenticator import Authenticator, AuthError, AuthSection
@ -11,11 +12,11 @@ class PassAuthSection(AuthSection):
self.missing_value("passname") self.missing_value("passname")
return value return value
def username_prefixes(self) -> list[str]: def username_prefixes(self) -> List[str]:
value = self.s.get("username_prefixes", "login,username,user") value = self.s.get("username_prefixes", "login,username,user")
return [prefix.lower() for prefix in value.split(",")] return [prefix.lower() for prefix in value.split(",")]
def password_prefixes(self) -> list[str]: def password_prefixes(self) -> List[str]:
value = self.s.get("password_prefixes", "password,pass,secret") value = self.s.get("password_prefixes", "password,pass,secret")
return [prefix.lower() for prefix in value.split(",")] return [prefix.lower() for prefix in value.split(",")]
@ -30,14 +31,14 @@ class PassAuthenticator(Authenticator):
self._username_prefixes = section.username_prefixes() self._username_prefixes = section.username_prefixes()
self._password_prefixes = section.password_prefixes() self._password_prefixes = section.password_prefixes()
async def credentials(self) -> tuple[str, str]: async def credentials(self) -> Tuple[str, str]:
log.explain_topic("Obtaining credentials from pass") log.explain_topic("Obtaining credentials from pass")
try: try:
log.explain(f"Calling 'pass show {self._passname}'") log.explain(f"Calling 'pass show {self._passname}'")
result = subprocess.check_output(["pass", "show", self._passname], text=True) result = subprocess.check_output(["pass", "show", self._passname], text=True)
except subprocess.CalledProcessError as e: except subprocess.CalledProcessError as e:
raise AuthError(f"Failed to get password info from {self._passname}: {e}") from e raise AuthError(f"Failed to get password info from {self._passname}: {e}")
prefixed = {} prefixed = {}
unprefixed = [] unprefixed = []

View file

@ -1,4 +1,4 @@
from typing import Optional from typing import Optional, Tuple
from ..logging import log from ..logging import log
from ..utils import agetpass, ainput from ..utils import agetpass, ainput
@ -23,7 +23,7 @@ class SimpleAuthenticator(Authenticator):
self._username_fixed = self.username is not None self._username_fixed = self.username is not None
self._password_fixed = self.password is not None self._password_fixed = self.password is not None
async def credentials(self) -> tuple[str, str]: async def credentials(self) -> Tuple[str, str]:
if self._username is not None and self._password is not None: if self._username is not None and self._password is not None:
return self._username, self._password return self._username, self._password

View file

@ -1,3 +1,5 @@
from typing import Tuple
from ..logging import log from ..logging import log
from ..utils import ainput from ..utils import ainput
from .authenticator import Authenticator, AuthError from .authenticator import Authenticator, AuthError
@ -15,7 +17,7 @@ class TfaAuthenticator(Authenticator):
code = await ainput("TFA code: ") code = await ainput("TFA code: ")
return code return code
async def credentials(self) -> tuple[str, str]: async def credentials(self) -> Tuple[str, str]:
raise AuthError("TFA authenticator does not support usernames") raise AuthError("TFA authenticator does not support usernames")
def invalidate_username(self) -> None: def invalidate_username(self) -> None:

View file

@ -21,20 +21,23 @@ GROUP.add_argument(
"--base-url", "--base-url",
type=str, type=str,
metavar="BASE_URL", metavar="BASE_URL",
help="The base url of the ilias instance", help="The base url of the ilias instance"
) )
GROUP.add_argument( GROUP.add_argument(
"--client-id", "--client-id",
type=str, type=str,
metavar="CLIENT_ID", metavar="CLIENT_ID",
help="The client id of the ilias instance", help="The client id of the ilias instance"
) )
configure_common_group_args(GROUP) configure_common_group_args(GROUP)
def load(args: argparse.Namespace, parser: configparser.ConfigParser) -> None: def load(
args: argparse.Namespace,
parser: configparser.ConfigParser,
) -> None:
log.explain(f"Creating config for command '{COMMAND_NAME}'") log.explain(f"Creating config for command '{COMMAND_NAME}'")
parser["crawl:ilias"] = {} parser["crawl:ilias"] = {}
@ -42,8 +45,8 @@ def load(args: argparse.Namespace, parser: configparser.ConfigParser) -> None:
load_crawler(args, section) load_crawler(args, section)
section["type"] = COMMAND_NAME section["type"] = COMMAND_NAME
if args.base_url is not None: if args.ilias_url is not None:
section["base_url"] = args.base_url section["base_url"] = args.ilias_url
if args.client_id is not None: if args.client_id is not None:
section["client_id"] = args.client_id section["client_id"] = args.client_id

View file

@ -18,24 +18,19 @@ GROUP.add_argument(
"--link-regex", "--link-regex",
type=str, type=str,
metavar="REGEX", metavar="REGEX",
help="href-matching regex to identify downloadable files", help="href-matching regex to identify downloadable files"
)
GROUP.add_argument(
"--basic-auth",
action="store_true",
help="enable basic authentication",
) )
GROUP.add_argument( GROUP.add_argument(
"target", "target",
type=str, type=str,
metavar="TARGET", metavar="TARGET",
help="url to crawl", help="url to crawl"
) )
GROUP.add_argument( GROUP.add_argument(
"output", "output",
type=Path, type=Path,
metavar="OUTPUT", metavar="OUTPUT",
help="output directory", help="output directory"
) )
@ -55,11 +50,5 @@ def load(
if args.link_regex: if args.link_regex:
section["link_regex"] = str(args.link_regex) section["link_regex"] = str(args.link_regex)
if args.basic_auth:
section["auth"] = "auth:kit-ipd"
parser["auth:kit-ipd"] = {}
auth_section = parser["auth:kit-ipd"]
auth_section["type"] = "simple"
SUBPARSER.set_defaults(command=load) SUBPARSER.set_defaults(command=load)

View file

@ -18,31 +18,31 @@ GROUP.add_argument(
"target", "target",
type=Path, type=Path,
metavar="TARGET", metavar="TARGET",
help="directory to crawl", help="directory to crawl"
) )
GROUP.add_argument( GROUP.add_argument(
"output", "output",
type=Path, type=Path,
metavar="OUTPUT", metavar="OUTPUT",
help="output directory", help="output directory"
) )
GROUP.add_argument( GROUP.add_argument(
"--crawl-delay", "--crawl-delay",
type=float, type=float,
metavar="SECONDS", metavar="SECONDS",
help="artificial delay to simulate for crawl requests", help="artificial delay to simulate for crawl requests"
) )
GROUP.add_argument( GROUP.add_argument(
"--download-delay", "--download-delay",
type=float, type=float,
metavar="SECONDS", metavar="SECONDS",
help="artificial delay to simulate for download requests", help="artificial delay to simulate for download requests"
) )
GROUP.add_argument( GROUP.add_argument(
"--download-speed", "--download-speed",
type=int, type=int,
metavar="BYTES_PER_SECOND", metavar="BYTES_PER_SECOND",
help="download speed to simulate", help="download speed to simulate"
) )

View file

@ -12,60 +12,58 @@ def configure_common_group_args(group: argparse._ArgumentGroup) -> None:
"target", "target",
type=str, type=str,
metavar="TARGET", metavar="TARGET",
help="course id, 'desktop', or ILIAS URL to crawl", help="course id, 'desktop', or ILIAS URL to crawl"
) )
group.add_argument( group.add_argument(
"output", "output",
type=Path, type=Path,
metavar="OUTPUT", metavar="OUTPUT",
help="output directory", help="output directory"
) )
group.add_argument( group.add_argument(
"--username", "--username", "-u",
"-u",
type=str, type=str,
metavar="USERNAME", metavar="USERNAME",
help="user name for authentication", help="user name for authentication"
) )
group.add_argument( group.add_argument(
"--keyring", "--keyring",
action=BooleanOptionalAction, action=BooleanOptionalAction,
help="use the system keyring to store and retrieve passwords", help="use the system keyring to store and retrieve passwords"
) )
group.add_argument( group.add_argument(
"--credential-file", "--credential-file",
type=Path, type=Path,
metavar="PATH", metavar="PATH",
help="read username and password from a credential file", help="read username and password from a credential file"
) )
group.add_argument( group.add_argument(
"--links", "--links",
type=show_value_error(Links.from_string), type=show_value_error(Links.from_string),
metavar="OPTION", metavar="OPTION",
help="how to represent external links", help="how to represent external links"
) )
group.add_argument( group.add_argument(
"--link-redirect-delay", "--link-redirect-delay",
type=int, type=int,
metavar="SECONDS", metavar="SECONDS",
help="time before 'fancy' links redirect to to their target (-1 to disable)", help="time before 'fancy' links redirect to to their target (-1 to disable)"
) )
group.add_argument( group.add_argument(
"--videos", "--videos",
action=BooleanOptionalAction, action=BooleanOptionalAction,
help="crawl and download videos", help="crawl and download videos"
) )
group.add_argument( group.add_argument(
"--forums", "--forums",
action=BooleanOptionalAction, action=BooleanOptionalAction,
help="crawl and download forum posts", help="crawl and download forum posts"
) )
group.add_argument( group.add_argument(
"--http-timeout", "--http-timeout", "-t",
"-t",
type=float, type=float,
metavar="SECONDS", metavar="SECONDS",
help="timeout for all HTTP requests", help="timeout for all HTTP requests"
) )

View file

@ -1,9 +1,8 @@
import argparse import argparse
import configparser import configparser
from argparse import ArgumentTypeError from argparse import ArgumentTypeError
from collections.abc import Callable, Sequence
from pathlib import Path from pathlib import Path
from typing import Any, Optional from typing import Any, Callable, List, Optional, Sequence, Union
from ..output_dir import OnConflict, Redownload from ..output_dir import OnConflict, Redownload
from ..version import NAME, VERSION from ..version import NAME, VERSION
@ -17,7 +16,7 @@ class ParserLoadError(Exception):
class BooleanOptionalAction(argparse.Action): class BooleanOptionalAction(argparse.Action):
def __init__( def __init__(
self, self,
option_strings: list[str], option_strings: List[str],
dest: Any, dest: Any,
default: Any = None, default: Any = None,
type: Any = None, type: Any = None,
@ -52,7 +51,7 @@ class BooleanOptionalAction(argparse.Action):
self, self,
parser: argparse.ArgumentParser, parser: argparse.ArgumentParser,
namespace: argparse.Namespace, namespace: argparse.Namespace,
values: str | Sequence[Any] | None, values: Union[str, Sequence[Any], None],
option_string: Optional[str] = None, option_string: Optional[str] = None,
) -> None: ) -> None:
if option_string and option_string in self.option_strings: if option_string and option_string in self.option_strings:
@ -68,13 +67,11 @@ def show_value_error(inner: Callable[[str], Any]) -> Callable[[str], Any]:
Some validation functions (like the from_string in our enums) raise a ValueError. Some validation functions (like the from_string in our enums) raise a ValueError.
Argparse only pretty-prints ArgumentTypeErrors though, so we need to wrap our ValueErrors. Argparse only pretty-prints ArgumentTypeErrors though, so we need to wrap our ValueErrors.
""" """
def wrapper(input: str) -> Any: def wrapper(input: str) -> Any:
try: try:
return inner(input) return inner(input)
except ValueError as e: except ValueError as e:
raise ArgumentTypeError(e) from e raise ArgumentTypeError(e)
return wrapper return wrapper
@ -84,51 +81,46 @@ CRAWLER_PARSER_GROUP = CRAWLER_PARSER.add_argument_group(
description="arguments common to all crawlers", description="arguments common to all crawlers",
) )
CRAWLER_PARSER_GROUP.add_argument( CRAWLER_PARSER_GROUP.add_argument(
"--redownload", "--redownload", "-r",
"-r",
type=show_value_error(Redownload.from_string), type=show_value_error(Redownload.from_string),
metavar="OPTION", metavar="OPTION",
help="when to download a file that's already present locally", help="when to download a file that's already present locally"
) )
CRAWLER_PARSER_GROUP.add_argument( CRAWLER_PARSER_GROUP.add_argument(
"--on-conflict", "--on-conflict",
type=show_value_error(OnConflict.from_string), type=show_value_error(OnConflict.from_string),
metavar="OPTION", metavar="OPTION",
help="what to do when local and remote files or directories differ", help="what to do when local and remote files or directories differ"
) )
CRAWLER_PARSER_GROUP.add_argument( CRAWLER_PARSER_GROUP.add_argument(
"--transform", "--transform", "-T",
"-T",
action="append", action="append",
type=str, type=str,
metavar="RULE", metavar="RULE",
help="add a single transformation rule. Can be specified multiple times", help="add a single transformation rule. Can be specified multiple times"
) )
CRAWLER_PARSER_GROUP.add_argument( CRAWLER_PARSER_GROUP.add_argument(
"--tasks", "--tasks", "-n",
"-n",
type=int, type=int,
metavar="N", metavar="N",
help="maximum number of concurrent tasks (crawling, downloading)", help="maximum number of concurrent tasks (crawling, downloading)"
) )
CRAWLER_PARSER_GROUP.add_argument( CRAWLER_PARSER_GROUP.add_argument(
"--downloads", "--downloads", "-N",
"-N",
type=int, type=int,
metavar="N", metavar="N",
help="maximum number of tasks that may download data at the same time", help="maximum number of tasks that may download data at the same time"
) )
CRAWLER_PARSER_GROUP.add_argument( CRAWLER_PARSER_GROUP.add_argument(
"--task-delay", "--task-delay", "-d",
"-d",
type=float, type=float,
metavar="SECONDS", metavar="SECONDS",
help="time the crawler should wait between subsequent tasks", help="time the crawler should wait between subsequent tasks"
) )
CRAWLER_PARSER_GROUP.add_argument( CRAWLER_PARSER_GROUP.add_argument(
"--windows-paths", "--windows-paths",
action=BooleanOptionalAction, action=BooleanOptionalAction,
help="whether to repair invalid paths on windows", help="whether to repair invalid paths on windows"
) )
@ -160,73 +152,73 @@ PARSER.add_argument(
version=f"{NAME} {VERSION} (https://github.com/Garmelon/PFERD)", version=f"{NAME} {VERSION} (https://github.com/Garmelon/PFERD)",
) )
PARSER.add_argument( PARSER.add_argument(
"--config", "--config", "-c",
"-c",
type=Path, type=Path,
metavar="PATH", metavar="PATH",
help="custom config file", help="custom config file"
) )
PARSER.add_argument( PARSER.add_argument(
"--dump-config", "--dump-config",
action="store_true", action="store_true",
help="dump current configuration to the default config path and exit", help="dump current configuration to the default config path and exit"
) )
PARSER.add_argument( PARSER.add_argument(
"--dump-config-to", "--dump-config-to",
metavar="PATH", metavar="PATH",
help="dump current configuration to a file and exit. Use '-' as path to print to stdout instead", help="dump current configuration to a file and exit."
" Use '-' as path to print to stdout instead"
) )
PARSER.add_argument( PARSER.add_argument(
"--debug-transforms", "--debug-transforms",
action="store_true", action="store_true",
help="apply transform rules to files of previous run", help="apply transform rules to files of previous run"
) )
PARSER.add_argument( PARSER.add_argument(
"--crawler", "--crawler", "-C",
"-C",
action="append", action="append",
type=str, type=str,
metavar="NAME", metavar="NAME",
help="only execute a single crawler. Can be specified multiple times to execute multiple crawlers", help="only execute a single crawler."
" Can be specified multiple times to execute multiple crawlers"
) )
PARSER.add_argument( PARSER.add_argument(
"--skip", "--skip", "-S",
"-S",
action="append", action="append",
type=str, type=str,
metavar="NAME", metavar="NAME",
help="don't execute this particular crawler. Can be specified multiple times to skip multiple crawlers", help="don't execute this particular crawler."
" Can be specified multiple times to skip multiple crawlers"
) )
PARSER.add_argument( PARSER.add_argument(
"--working-dir", "--working-dir",
type=Path, type=Path,
metavar="PATH", metavar="PATH",
help="custom working directory", help="custom working directory"
) )
PARSER.add_argument( PARSER.add_argument(
"--explain", "--explain",
action=BooleanOptionalAction, action=BooleanOptionalAction,
help="log and explain in detail what PFERD is doing", help="log and explain in detail what PFERD is doing"
) )
PARSER.add_argument( PARSER.add_argument(
"--status", "--status",
action=BooleanOptionalAction, action=BooleanOptionalAction,
help="print status updates while PFERD is crawling", help="print status updates while PFERD is crawling"
) )
PARSER.add_argument( PARSER.add_argument(
"--report", "--report",
action=BooleanOptionalAction, action=BooleanOptionalAction,
help="print a report of all local changes before exiting", help="print a report of all local changes before exiting"
) )
PARSER.add_argument( PARSER.add_argument(
"--share-cookies", "--share-cookies",
action=BooleanOptionalAction, action=BooleanOptionalAction,
help="whether crawlers should share cookies where applicable", help="whether crawlers should share cookies where applicable"
) )
PARSER.add_argument( PARSER.add_argument(
"--show-not-deleted", "--show-not-deleted",
action=BooleanOptionalAction, action=BooleanOptionalAction,
help="print messages in status and report when PFERD did not delete a local only file", help="print messages in status and report when PFERD did not delete a local only file"
) )

View file

@ -3,7 +3,7 @@ import os
import sys import sys
from configparser import ConfigParser, SectionProxy from configparser import ConfigParser, SectionProxy
from pathlib import Path from pathlib import Path
from typing import Any, NoReturn, Optional from typing import Any, List, NoReturn, Optional, Tuple
from rich.markup import escape from rich.markup import escape
@ -126,13 +126,13 @@ class Config:
with open(path, encoding="utf-8") as f: with open(path, encoding="utf-8") as f:
parser.read_file(f, source=str(path)) parser.read_file(f, source=str(path))
except FileNotFoundError: except FileNotFoundError:
raise ConfigLoadError(path, "File does not exist") from None raise ConfigLoadError(path, "File does not exist")
except IsADirectoryError: except IsADirectoryError:
raise ConfigLoadError(path, "That's a directory, not a file") from None raise ConfigLoadError(path, "That's a directory, not a file")
except PermissionError: except PermissionError:
raise ConfigLoadError(path, "Insufficient permissions") from None raise ConfigLoadError(path, "Insufficient permissions")
except UnicodeDecodeError: except UnicodeDecodeError:
raise ConfigLoadError(path, "File is not encoded using UTF-8") from None raise ConfigLoadError(path, "File is not encoded using UTF-8")
def dump(self, path: Optional[Path] = None) -> None: def dump(self, path: Optional[Path] = None) -> None:
""" """
@ -150,8 +150,8 @@ class Config:
try: try:
path.parent.mkdir(parents=True, exist_ok=True) path.parent.mkdir(parents=True, exist_ok=True)
except PermissionError as e: except PermissionError:
raise ConfigDumpError(path, "Could not create parent directory") from e raise ConfigDumpError(path, "Could not create parent directory")
try: try:
# Ensuring we don't accidentally overwrite any existing files by # Ensuring we don't accidentally overwrite any existing files by
@ -167,16 +167,16 @@ class Config:
with open(path, "w", encoding="utf-8") as f: with open(path, "w", encoding="utf-8") as f:
self._parser.write(f) self._parser.write(f)
else: else:
raise ConfigDumpError(path, "File already exists") from None raise ConfigDumpError(path, "File already exists")
except IsADirectoryError: except IsADirectoryError:
raise ConfigDumpError(path, "That's a directory, not a file") from None raise ConfigDumpError(path, "That's a directory, not a file")
except PermissionError as e: except PermissionError:
raise ConfigDumpError(path, "Insufficient permissions") from e raise ConfigDumpError(path, "Insufficient permissions")
def dump_to_stdout(self) -> None: def dump_to_stdout(self) -> None:
self._parser.write(sys.stdout) self._parser.write(sys.stdout)
def crawl_sections(self) -> list[tuple[str, SectionProxy]]: def crawl_sections(self) -> List[Tuple[str, SectionProxy]]:
result = [] result = []
for name, proxy in self._parser.items(): for name, proxy in self._parser.items():
if name.startswith("crawl:"): if name.startswith("crawl:"):
@ -184,7 +184,7 @@ class Config:
return result return result
def auth_sections(self) -> list[tuple[str, SectionProxy]]: def auth_sections(self) -> List[Tuple[str, SectionProxy]]:
result = [] result = []
for name, proxy in self._parser.items(): for name, proxy in self._parser.items():
if name.startswith("auth:"): if name.startswith("auth:"):

View file

@ -1,5 +1,5 @@
from collections.abc import Callable
from configparser import SectionProxy from configparser import SectionProxy
from typing import Callable, Dict
from ..auth import Authenticator from ..auth import Authenticator
from ..config import Config from ..config import Config
@ -8,19 +8,20 @@ from .ilias import IliasWebCrawler, IliasWebCrawlerSection, KitIliasWebCrawler,
from .kit_ipd_crawler import KitIpdCrawler, KitIpdCrawlerSection from .kit_ipd_crawler import KitIpdCrawler, KitIpdCrawlerSection
from .local_crawler import LocalCrawler, LocalCrawlerSection from .local_crawler import LocalCrawler, LocalCrawlerSection
CrawlerConstructor = Callable[ CrawlerConstructor = Callable[[
[
str, # Name (without the "crawl:" prefix) str, # Name (without the "crawl:" prefix)
SectionProxy, # Crawler's section of global config SectionProxy, # Crawler's section of global config
Config, # Global config Config, # Global config
dict[str, Authenticator], # Loaded authenticators by name Dict[str, Authenticator], # Loaded authenticators by name
], ], Crawler]
Crawler,
]
CRAWLERS: dict[str, CrawlerConstructor] = { CRAWLERS: Dict[str, CrawlerConstructor] = {
"local": lambda n, s, c, a: LocalCrawler(n, LocalCrawlerSection(s), c), "local": lambda n, s, c, a:
"ilias-web": lambda n, s, c, a: IliasWebCrawler(n, IliasWebCrawlerSection(s), c, a), LocalCrawler(n, LocalCrawlerSection(s), c),
"kit-ilias-web": lambda n, s, c, a: KitIliasWebCrawler(n, KitIliasWebCrawlerSection(s), c, a), "ilias-web": lambda n, s, c, a:
"kit-ipd": lambda n, s, c, a: KitIpdCrawler(n, KitIpdCrawlerSection(s), c, a), IliasWebCrawler(n, IliasWebCrawlerSection(s), c, a),
"kit-ilias-web": lambda n, s, c, a:
KitIliasWebCrawler(n, KitIliasWebCrawlerSection(s), c, a),
"kit-ipd": lambda n, s, c, a:
KitIpdCrawler(n, KitIpdCrawlerSection(s), c),
} }

View file

@ -1,10 +1,10 @@
import asyncio import asyncio
import os import os
from abc import ABC, abstractmethod from abc import ABC, abstractmethod
from collections.abc import Awaitable, Callable, Coroutine, Sequence from collections.abc import Awaitable, Coroutine
from datetime import datetime from datetime import datetime
from pathlib import Path, PurePath from pathlib import Path, PurePath
from typing import Any, Optional, TypeVar from typing import Any, Callable, Dict, List, Optional, Sequence, Set, Tuple, TypeVar
from ..auth import Authenticator from ..auth import Authenticator
from ..config import Config, Section from ..config import Config, Section
@ -116,7 +116,7 @@ class CrawlToken(ReusableAsyncContextManager[ProgressBar]):
return bar return bar
class DownloadToken(ReusableAsyncContextManager[tuple[ProgressBar, FileSink]]): class DownloadToken(ReusableAsyncContextManager[Tuple[ProgressBar, FileSink]]):
def __init__(self, limiter: Limiter, fs_token: FileSinkToken, path: PurePath): def __init__(self, limiter: Limiter, fs_token: FileSinkToken, path: PurePath):
super().__init__() super().__init__()
@ -128,13 +128,12 @@ class DownloadToken(ReusableAsyncContextManager[tuple[ProgressBar, FileSink]]):
def path(self) -> PurePath: def path(self) -> PurePath:
return self._path return self._path
async def _on_aenter(self) -> tuple[ProgressBar, FileSink]: async def _on_aenter(self) -> Tuple[ProgressBar, FileSink]:
await self._stack.enter_async_context(self._limiter.limit_download()) await self._stack.enter_async_context(self._limiter.limit_download())
sink = await self._stack.enter_async_context(self._fs_token) sink = await self._stack.enter_async_context(self._fs_token)
# The "Downloaded ..." message is printed in the output dir, not here # The "Downloaded ..." message is printed in the output dir, not here
bar = self._stack.enter_context( bar = self._stack.enter_context(log.download_bar("[bold bright_cyan]", "Downloading",
log.download_bar("[bold bright_cyan]", "Downloading", fmt_path(self._path)) fmt_path(self._path)))
)
return bar, sink return bar, sink
@ -205,7 +204,7 @@ class CrawlerSection(Section):
on_windows = os.name == "nt" on_windows = os.name == "nt"
return self.s.getboolean("windows_paths", fallback=on_windows) return self.s.getboolean("windows_paths", fallback=on_windows)
def auth(self, authenticators: dict[str, Authenticator]) -> Authenticator: def auth(self, authenticators: Dict[str, Authenticator]) -> Authenticator:
value = self.s.get("auth") value = self.s.get("auth")
if value is None: if value is None:
self.missing_value("auth") self.missing_value("auth")
@ -262,7 +261,7 @@ class Crawler(ABC):
return self._output_dir return self._output_dir
@staticmethod @staticmethod
async def gather(awaitables: Sequence[Awaitable[Any]]) -> list[Any]: async def gather(awaitables: Sequence[Awaitable[Any]]) -> List[Any]:
""" """
Similar to asyncio.gather. However, in the case of an exception, all Similar to asyncio.gather. However, in the case of an exception, all
still running tasks are cancelled and the exception is rethrown. still running tasks are cancelled and the exception is rethrown.
@ -309,7 +308,11 @@ class Crawler(ABC):
return False return False
should_download = self._output_dir.should_try_download( should_download = self._output_dir.should_try_download(
path, etag_differs=etag_differs, mtime=mtime, redownload=redownload, on_conflict=on_conflict path,
etag_differs=etag_differs,
mtime=mtime,
redownload=redownload,
on_conflict=on_conflict
) )
if should_download: if should_download:
log.explain("Answer: Yes") log.explain("Answer: Yes")
@ -343,7 +346,7 @@ class Crawler(ABC):
etag_differs=etag_differs, etag_differs=etag_differs,
mtime=mtime, mtime=mtime,
redownload=redownload, redownload=redownload,
on_conflict=on_conflict, on_conflict=on_conflict
) )
if fs_token is None: if fs_token is None:
log.explain("Answer: No") log.explain("Answer: No")
@ -394,7 +397,7 @@ class Crawler(ABC):
log.warn("Couldn't find or load old report") log.warn("Couldn't find or load old report")
return return
seen: set[PurePath] = set() seen: Set[PurePath] = set()
for known in sorted(self.prev_report.found_paths): for known in sorted(self.prev_report.found_paths):
looking_at = list(reversed(known.parents)) + [known] looking_at = list(reversed(known.parents)) + [known]
for path in looking_at: for path in looking_at:

View file

@ -3,7 +3,7 @@ import http.cookies
import ssl import ssl
from datetime import datetime from datetime import datetime
from pathlib import Path, PurePath from pathlib import Path, PurePath
from typing import Any, Optional from typing import Any, Dict, List, Optional, Tuple, cast
import aiohttp import aiohttp
import certifi import certifi
@ -13,7 +13,7 @@ from bs4 import Tag
from ..auth import Authenticator from ..auth import Authenticator
from ..config import Config from ..config import Config
from ..logging import log from ..logging import log
from ..utils import fmt_real_path, sanitize_path_name from ..utils import fmt_real_path
from ..version import NAME, VERSION from ..version import NAME, VERSION
from .crawler import Crawler, CrawlerSection from .crawler import Crawler, CrawlerSection
@ -43,7 +43,7 @@ class HttpCrawler(Crawler):
self._http_timeout = section.http_timeout() self._http_timeout = section.http_timeout()
self._cookie_jar_path = self._output_dir.resolve(self.COOKIE_FILE) self._cookie_jar_path = self._output_dir.resolve(self.COOKIE_FILE)
self._shared_cookie_jar_paths: Optional[list[Path]] = None self._shared_cookie_jar_paths: Optional[List[Path]] = None
self._shared_auth = shared_auth self._shared_auth = shared_auth
self._output_dir.register_reserved(self.COOKIE_FILE) self._output_dir.register_reserved(self.COOKIE_FILE)
@ -98,7 +98,7 @@ class HttpCrawler(Crawler):
""" """
raise RuntimeError("_authenticate() was called but crawler doesn't provide an implementation") raise RuntimeError("_authenticate() was called but crawler doesn't provide an implementation")
def share_cookies(self, shared: dict[Authenticator, list[Path]]) -> None: def share_cookies(self, shared: Dict[Authenticator, List[Path]]) -> None:
if not self._shared_auth: if not self._shared_auth:
return return
@ -187,12 +187,12 @@ class HttpCrawler(Crawler):
if level == 0 or (level == 1 and drop_h1): if level == 0 or (level == 1 and drop_h1):
return PurePath() return PurePath()
level_heading = tag.find_previous(name=f"h{level}") level_heading = cast(Optional[Tag], tag.find_previous(name=f"h{level}"))
if level_heading is None: if level_heading is None:
return find_associated_headings(tag, level - 1) return find_associated_headings(tag, level - 1)
folder_name = sanitize_path_name(level_heading.get_text().strip()) folder_name = level_heading.get_text().strip()
return find_associated_headings(level_heading, level - 1) / folder_name return find_associated_headings(level_heading, level - 1) / folder_name
# start at level <h3> because paragraph-level headings are usually too granular for folder names # start at level <h3> because paragraph-level headings are usually too granular for folder names
@ -219,7 +219,7 @@ class HttpCrawler(Crawler):
etags[str(path)] = etag etags[str(path)] = etag
self._output_dir.report.add_custom_value(ETAGS_CUSTOM_REPORT_VALUE_KEY, etags) self._output_dir.report.add_custom_value(ETAGS_CUSTOM_REPORT_VALUE_KEY, etags)
async def _request_resource_version(self, resource_url: str) -> tuple[Optional[str], Optional[datetime]]: async def _request_resource_version(self, resource_url: str) -> Tuple[Optional[str], Optional[datetime]]:
""" """
Requests the ETag and Last-Modified headers of a resource via a HEAD request. Requests the ETag and Last-Modified headers of a resource via a HEAD request.
If no entity tag / modification date can be obtained, the according value will be None. If no entity tag / modification date can be obtained, the according value will be None.
@ -268,7 +268,7 @@ class HttpCrawler(Crawler):
# Without this aiohttp will mangle the redirect header from Shibboleth, invalidating the # Without this aiohttp will mangle the redirect header from Shibboleth, invalidating the
# passed signature. Shibboleth will not accept the broken signature and authentication will # passed signature. Shibboleth will not accept the broken signature and authentication will
# fail. # fail.
requote_redirect_url=False, requote_redirect_url=False
) as session: ) as session:
self.session = session self.session = session
try: try:

View file

@ -1,9 +1,5 @@
from .kit_ilias_web_crawler import ( from .kit_ilias_web_crawler import (IliasWebCrawler, IliasWebCrawlerSection, KitIliasWebCrawler,
IliasWebCrawler, KitIliasWebCrawlerSection)
IliasWebCrawlerSection,
KitIliasWebCrawler,
KitIliasWebCrawlerSection,
)
__all__ = [ __all__ = [
"IliasWebCrawler", "IliasWebCrawler",

View file

@ -1,6 +1,5 @@
import asyncio import asyncio
from collections.abc import Callable from typing import Any, Callable, Optional
from typing import Any, Optional
import aiohttp import aiohttp
@ -16,9 +15,9 @@ def _iorepeat(attempts: int, name: str, failure_is_error: bool = False) -> Calla
try: try:
return await f(*args, **kwargs) return await f(*args, **kwargs)
except aiohttp.ContentTypeError: # invalid content type except aiohttp.ContentTypeError: # invalid content type
raise CrawlWarning("ILIAS returned an invalid content type") from None raise CrawlWarning("ILIAS returned an invalid content type")
except aiohttp.TooManyRedirects: except aiohttp.TooManyRedirects:
raise CrawlWarning("Got stuck in a redirect loop") from None raise CrawlWarning("Got stuck in a redirect loop")
except aiohttp.ClientPayloadError as e: # encoding or not enough bytes except aiohttp.ClientPayloadError as e: # encoding or not enough bytes
last_exception = e last_exception = e
except aiohttp.ClientConnectionError as e: # e.g. timeout, disconnect, resolve failed, etc. except aiohttp.ClientConnectionError as e: # e.g. timeout, disconnect, resolve failed, etc.

View file

@ -1,5 +1,3 @@
import dataclasses
import re
from enum import Enum from enum import Enum
from typing import Optional, cast from typing import Optional, cast
@ -14,9 +12,7 @@ _link_template_fancy = """
<head> <head>
<meta charset="UTF-8"> <meta charset="UTF-8">
<title>ILIAS - Link: {{name}}</title> <title>ILIAS - Link: {{name}}</title>
<!-- REPEAT REMOVE START -->
<meta http-equiv = "refresh" content = "{{redirect_delay}}; url = {{link}}" /> <meta http-equiv = "refresh" content = "{{redirect_delay}}; url = {{link}}" />
<!-- REPEAT REMOVE END -->
</head> </head>
<style> <style>
@ -27,8 +23,6 @@ _link_template_fancy = """
display: flex; display: flex;
align-items: center; align-items: center;
justify-content: center; justify-content: center;
flex-direction: column;
gap: 4px;
} }
body { body {
padding: 0; padding: 0;
@ -37,15 +31,10 @@ _link_template_fancy = """
font-family: "Open Sans", Verdana, Arial, Helvetica, sans-serif; font-family: "Open Sans", Verdana, Arial, Helvetica, sans-serif;
height: 100vh; height: 100vh;
} }
.column {
min-width: 500px;
max-width: 90vw;
display: flex;
flex-direction: column;
row-gap: 5px;
}
.row { .row {
background-color: white; background-color: white;
min-width: 500px;
max-width: 90vw;
display: flex; display: flex;
padding: 1em; padding: 1em;
} }
@ -86,8 +75,6 @@ _link_template_fancy = """
} }
</style> </style>
<body class="center-flex"> <body class="center-flex">
<div class="column">
<!-- REPEAT START -->
<div class="row"> <div class="row">
<div class="logo center-flex"> <div class="logo center-flex">
<svg xmlns="http://www.w3.org/2000/svg" width="24" height="24" viewBox="0 0 24 24"> <svg xmlns="http://www.w3.org/2000/svg" width="24" height="24" viewBox="0 0 24 24">
@ -102,8 +89,6 @@ _link_template_fancy = """
</div> </div>
<div class="menu-button center-flex"> </div> <div class="menu-button center-flex"> </div>
</div> </div>
<!-- REPEAT END -->
</div>
</body> </body>
</html> </html>
""".strip() # noqa: E501 line too long """.strip() # noqa: E501 line too long
@ -111,7 +96,6 @@ _link_template_fancy = """
_link_template_internet_shortcut = """ _link_template_internet_shortcut = """
[InternetShortcut] [InternetShortcut]
URL={{link}} URL={{link}}
Desc={{description}}
""".strip() """.strip()
_learning_module_template = """ _learning_module_template = """
@ -254,29 +238,21 @@ def learning_module_template(body: bs4.Tag, name: str, prev: Optional[str], next
) )
if bot_nav := body.select_one(".ilc_page_bnav_BottomNavigation"): if bot_nav := body.select_one(".ilc_page_bnav_BottomNavigation"):
bot_nav.replace_with( bot_nav.replace_with(soupify(nav_template.replace(
soupify(nav_template.replace("{{left}}", left).replace("{{right}}", right).encode()) "{{left}}", left).replace("{{right}}", right).encode())
) )
body_str = body.prettify() body_str = cast(str, body.prettify())
return _learning_module_template.replace("{{body}}", body_str).replace("{{name}}", name) return _learning_module_template.replace("{{body}}", body_str).replace("{{name}}", name)
def forum_thread_template(name: str, url: str, heading: bs4.Tag, content: bs4.Tag) -> str: def forum_thread_template(name: str, url: str, heading: bs4.Tag, content: bs4.Tag) -> str:
if title := heading.find(name="b"): if title := cast(Optional[bs4.Tag], heading.find(name="b")):
title.wrap(bs4.Tag(name="a", attrs={"href": url})) title.wrap(bs4.Tag(name="a", attrs={"href": url}))
return ( return _forum_thread_template \
_forum_thread_template.replace("{{name}}", name) .replace("{{name}}", name) \
.replace("{{heading}}", heading.prettify()) .replace("{{heading}}", cast(str, heading.prettify())) \
.replace("{{content}}", content.prettify()) .replace("{{content}}", cast(str, content.prettify()))
)
@dataclasses.dataclass
class LinkData:
name: str
url: str
description: str
class Links(Enum): class Links(Enum):
@ -296,9 +272,6 @@ class Links(Enum):
return None return None
raise ValueError("Missing switch case") raise ValueError("Missing switch case")
def collection_as_one(self) -> bool:
return self == Links.FANCY
def extension(self) -> Optional[str]: def extension(self) -> Optional[str]:
if self == Links.FANCY: if self == Links.FANCY:
return ".html" return ".html"
@ -310,47 +283,10 @@ class Links(Enum):
return None return None
raise ValueError("Missing switch case") raise ValueError("Missing switch case")
def interpolate(self, redirect_delay: int, collection_name: str, links: list[LinkData]) -> str:
template = self.template()
if template is None:
raise ValueError("Cannot interpolate ignored links")
if len(links) == 1:
link = links[0]
content = template
content = content.replace("{{link}}", link.url)
content = content.replace("{{name}}", link.name)
content = content.replace("{{description}}", link.description)
content = content.replace("{{redirect_delay}}", str(redirect_delay))
return content
if self == Links.PLAINTEXT or self == Links.INTERNET_SHORTCUT:
return "\n".join(f"{link.url}" for link in links)
# All others get coerced to fancy
content = cast(str, Links.FANCY.template())
repeated_content = cast(
re.Match[str], re.search(r"<!-- REPEAT START -->([\s\S]+)<!-- REPEAT END -->", content)
).group(1)
parts = []
for link in links:
instance = repeated_content
instance = instance.replace("{{link}}", link.url)
instance = instance.replace("{{name}}", link.name)
instance = instance.replace("{{description}}", link.description)
instance = instance.replace("{{redirect_delay}}", str(redirect_delay))
parts.append(instance)
content = content.replace(repeated_content, "\n".join(parts))
content = content.replace("{{name}}", collection_name)
content = re.sub(r"<!-- REPEAT REMOVE START -->[\s\S]+<!-- REPEAT REMOVE END -->", "", content)
return content
@staticmethod @staticmethod
def from_string(string: str) -> "Links": def from_string(string: str) -> "Links":
try: try:
return Links(string) return Links(string)
except ValueError: except ValueError:
options = [f"'{option.value}'" for option in Links] raise ValueError("must be one of 'ignore', 'plaintext',"
raise ValueError(f"must be one of {', '.join(options)}") from None " 'html', 'internet-shortcut'")

View file

@ -39,10 +39,6 @@ _STYLE_TAG_CONTENT = """
margin: 0.5rem 0; margin: 0.5rem 0;
} }
img {
background-color: white;
}
body { body {
padding: 1em; padding: 1em;
grid-template-columns: 1fr min(60rem, 90%) 1fr; grid-template-columns: 1fr min(60rem, 90%) 1fr;
@ -60,11 +56,12 @@ _ARTICLE_WORTHY_CLASSES = [
def insert_base_markup(soup: BeautifulSoup) -> BeautifulSoup: def insert_base_markup(soup: BeautifulSoup) -> BeautifulSoup:
head = soup.new_tag("head") head = soup.new_tag("head")
soup.insert(0, head) soup.insert(0, head)
# Force UTF-8 encoding
head.append(soup.new_tag("meta", charset="utf-8"))
simplecss_link: Tag = soup.new_tag("link")
# <link rel="stylesheet" href="https://cdn.simplecss.org/simple.css"> # <link rel="stylesheet" href="https://cdn.simplecss.org/simple.css">
head.append(soup.new_tag("link", rel="stylesheet", href="https://cdn.simplecss.org/simple.css")) simplecss_link["rel"] = "stylesheet"
simplecss_link["href"] = "https://cdn.simplecss.org/simple.css"
head.append(simplecss_link)
# Basic style tags for compat # Basic style tags for compat
style: Tag = soup.new_tag("style") style: Tag = soup.new_tag("style")

View file

@ -4,7 +4,7 @@ import os
import re import re
from collections.abc import Awaitable, Coroutine from collections.abc import Awaitable, Coroutine
from pathlib import PurePath from pathlib import PurePath
from typing import Any, Literal, Optional, cast from typing import Any, Dict, List, Literal, Optional, Set, Union, cast
from urllib.parse import urljoin from urllib.parse import urljoin
import aiohttp import aiohttp
@ -15,24 +15,17 @@ from ...auth import Authenticator
from ...config import Config from ...config import Config
from ...logging import ProgressBar, log from ...logging import ProgressBar, log
from ...output_dir import FileSink, Redownload from ...output_dir import FileSink, Redownload
from ...utils import fmt_path, sanitize_path_name, soupify, url_set_query_param from ...utils import fmt_path, soupify, url_set_query_param
from ..crawler import CrawlError, CrawlToken, CrawlWarning, DownloadToken, anoncritical from ..crawler import CrawlError, CrawlToken, CrawlWarning, DownloadToken, anoncritical
from ..http_crawler import HttpCrawler, HttpCrawlerSection from ..http_crawler import HttpCrawler, HttpCrawlerSection
from .async_helper import _iorepeat from .async_helper import _iorepeat
from .file_templates import LinkData, Links, forum_thread_template, learning_module_template from .file_templates import Links, forum_thread_template, learning_module_template
from .ilias_html_cleaner import clean, insert_base_markup from .ilias_html_cleaner import clean, insert_base_markup
from .kit_ilias_html import ( from .kit_ilias_html import (IliasElementType, IliasForumThread, IliasLearningModulePage, IliasPage,
IliasElementType, IliasPageElement, IliasSoup, _sanitize_path_name, parse_ilias_forum_export)
IliasForumThread,
IliasLearningModulePage,
IliasPage,
IliasPageElement,
IliasSoup,
parse_ilias_forum_export,
)
from .shibboleth_login import ShibbolethLogin from .shibboleth_login import ShibbolethLogin
TargetType = str | int TargetType = Union[str, int]
class LoginTypeLocal: class LoginTypeLocal:
@ -48,7 +41,7 @@ class IliasWebCrawlerSection(HttpCrawlerSection):
return base_url return base_url
def login(self) -> Literal["shibboleth"] | LoginTypeLocal: def login(self) -> Union[Literal["shibboleth"], LoginTypeLocal]:
login_type = self.s.get("login_type") login_type = self.s.get("login_type")
if not login_type: if not login_type:
self.missing_value("login_type") self.missing_value("login_type")
@ -62,7 +55,9 @@ class IliasWebCrawlerSection(HttpCrawlerSection):
self.invalid_value("login_type", login_type, "Should be <shibboleth | local>") self.invalid_value("login_type", login_type, "Should be <shibboleth | local>")
def tfa_auth(self, authenticators: dict[str, Authenticator]) -> Optional[Authenticator]: def tfa_auth(
self, authenticators: Dict[str, Authenticator]
) -> Optional[Authenticator]:
value: Optional[str] = self.s.get("tfa_auth") value: Optional[str] = self.s.get("tfa_auth")
if value is None: if value is None:
return None return None
@ -109,10 +104,9 @@ class IliasWebCrawlerSection(HttpCrawlerSection):
return self.s.getboolean("forums", fallback=False) return self.s.getboolean("forums", fallback=False)
_DIRECTORY_PAGES: set[IliasElementType] = { _DIRECTORY_PAGES: Set[IliasElementType] = {
IliasElementType.EXERCISE, IliasElementType.EXERCISE,
IliasElementType.EXERCISE_FILES, IliasElementType.EXERCISE_FILES,
IliasElementType.EXERCISE_OVERVIEW,
IliasElementType.FOLDER, IliasElementType.FOLDER,
IliasElementType.INFO_TAB, IliasElementType.INFO_TAB,
IliasElementType.MEDIACAST_VIDEO_FOLDER, IliasElementType.MEDIACAST_VIDEO_FOLDER,
@ -121,7 +115,7 @@ _DIRECTORY_PAGES: set[IliasElementType] = {
IliasElementType.OPENCAST_VIDEO_FOLDER_MAYBE_PAGINATED, IliasElementType.OPENCAST_VIDEO_FOLDER_MAYBE_PAGINATED,
} }
_VIDEO_ELEMENTS: set[IliasElementType] = { _VIDEO_ELEMENTS: Set[IliasElementType] = {
IliasElementType.MEDIACAST_VIDEO, IliasElementType.MEDIACAST_VIDEO,
IliasElementType.MEDIACAST_VIDEO_FOLDER, IliasElementType.MEDIACAST_VIDEO_FOLDER,
IliasElementType.OPENCAST_VIDEO, IliasElementType.OPENCAST_VIDEO,
@ -171,19 +165,17 @@ class IliasWebCrawler(HttpCrawler):
name: str, name: str,
section: IliasWebCrawlerSection, section: IliasWebCrawlerSection,
config: Config, config: Config,
authenticators: dict[str, Authenticator], authenticators: Dict[str, Authenticator]
): ):
# Setting a main authenticator for cookie sharing # Setting a main authenticator for cookie sharing
auth = section.auth(authenticators) auth = section.auth(authenticators)
super().__init__(name, section, config, shared_auth=auth) super().__init__(name, section, config, shared_auth=auth)
if section.tasks() > 1: if section.tasks() > 1:
log.warn( log.warn("""
"""
Please avoid using too many parallel requests as these are the KIT ILIAS Please avoid using too many parallel requests as these are the KIT ILIAS
instance's greatest bottleneck. instance's greatest bottleneck.
""".strip() """.strip())
)
self._auth = auth self._auth = auth
self._base_url = section.base_url() self._base_url = section.base_url()
@ -200,7 +192,7 @@ instance's greatest bottleneck.
self._links = section.links() self._links = section.links()
self._videos = section.videos() self._videos = section.videos()
self._forums = section.forums() self._forums = section.forums()
self._visited_urls: dict[str, PurePath] = dict() self._visited_urls: Dict[str, PurePath] = dict()
async def _run(self) -> None: async def _run(self) -> None:
if isinstance(self._target, int): if isinstance(self._target, int):
@ -217,23 +209,18 @@ instance's greatest bottleneck.
# Start crawling at the given course # Start crawling at the given course
root_url = url_set_query_param( root_url = url_set_query_param(
urljoin(self._base_url + "/", "goto.php"), urljoin(self._base_url + "/", "goto.php"),
"target", "target", f"crs_{course_id}",
f"crs_{course_id}",
) )
await self._crawl_url(root_url, expected_id=course_id) await self._crawl_url(root_url, expected_id=course_id)
async def _crawl_desktop(self) -> None: async def _crawl_desktop(self) -> None:
await self._crawl_url( await self._crawl_url(
urljoin(self._base_url, "/ilias.php?baseClass=ilDashboardGUI&cmd=show"), crawl_nested_courses=True urljoin(self._base_url, "/ilias.php?baseClass=ilDashboardGUI&cmd=show")
) )
async def _crawl_url( async def _crawl_url(self, url: str, expected_id: Optional[int] = None) -> None:
self, url: str, expected_id: Optional[int] = None, crawl_nested_courses: bool = False if awaitable := await self._handle_ilias_page(url, None, PurePath("."), expected_id):
) -> None:
if awaitable := await self._handle_ilias_page(
url, None, PurePath("."), expected_id, crawl_nested_courses
):
await awaitable await awaitable
async def _handle_ilias_page( async def _handle_ilias_page(
@ -242,7 +229,6 @@ instance's greatest bottleneck.
current_element: Optional[IliasPageElement], current_element: Optional[IliasPageElement],
path: PurePath, path: PurePath,
expected_course_id: Optional[int] = None, expected_course_id: Optional[int] = None,
crawl_nested_courses: bool = False,
) -> Optional[Coroutine[Any, Any, None]]: ) -> Optional[Coroutine[Any, Any, None]]:
maybe_cl = await self.crawl(path) maybe_cl = await self.crawl(path)
if not maybe_cl: if not maybe_cl:
@ -250,9 +236,7 @@ instance's greatest bottleneck.
if current_element: if current_element:
self._ensure_not_seen(current_element, path) self._ensure_not_seen(current_element, path)
return self._crawl_ilias_page( return self._crawl_ilias_page(url, current_element, maybe_cl, expected_course_id)
url, current_element, maybe_cl, expected_course_id, crawl_nested_courses
)
@anoncritical @anoncritical
async def _crawl_ilias_page( async def _crawl_ilias_page(
@ -261,11 +245,10 @@ instance's greatest bottleneck.
current_element: Optional[IliasPageElement], current_element: Optional[IliasPageElement],
cl: CrawlToken, cl: CrawlToken,
expected_course_id: Optional[int] = None, expected_course_id: Optional[int] = None,
crawl_nested_courses: bool = False,
) -> None: ) -> None:
elements: list[IliasPageElement] = [] elements: List[IliasPageElement] = []
# A list as variable redefinitions are not propagated to outer scopes # A list as variable redefinitions are not propagated to outer scopes
description: list[BeautifulSoup] = [] description: List[BeautifulSoup] = []
@_iorepeat(3, "crawling folder") @_iorepeat(3, "crawling folder")
async def gather_elements() -> None: async def gather_elements() -> None:
@ -297,8 +280,6 @@ instance's greatest bottleneck.
page = cast(IliasPage, page) page = cast(IliasPage, page)
elements.extend(page.get_child_elements()) elements.extend(page.get_child_elements())
if current_element is None and (info_tab := page.get_info_tab()):
elements.append(info_tab)
if description_string := page.get_description(): if description_string := page.get_description():
description.append(description_string) description.append(description_string)
@ -310,9 +291,9 @@ instance's greatest bottleneck.
elements.sort(key=lambda e: e.id()) elements.sort(key=lambda e: e.id())
tasks: list[Awaitable[None]] = [] tasks: List[Awaitable[None]] = []
for element in elements: for element in elements:
if handle := await self._handle_ilias_element(cl.path, element, crawl_nested_courses): if handle := await self._handle_ilias_element(cl.path, element):
tasks.append(asyncio.create_task(handle)) tasks.append(asyncio.create_task(handle))
# And execute them # And execute them
@ -325,28 +306,22 @@ instance's greatest bottleneck.
# works correctly. # works correctly.
@anoncritical @anoncritical
async def _handle_ilias_element( async def _handle_ilias_element(
self, parent_path: PurePath, element: IliasPageElement, crawl_nested_courses: bool = False self,
parent_path: PurePath,
element: IliasPageElement,
) -> Optional[Coroutine[Any, Any, None]]: ) -> Optional[Coroutine[Any, Any, None]]:
# element.name might contain `/` if the crawler created nested elements, # element.name might contain `/` if the crawler created nested elements,
# so we can not sanitize it here. We trust in the output dir to thwart worst-case # so we can not sanitize it here. We trust in the output dir to thwart worst-case
# directory escape attacks. # directory escape attacks.
element_path = PurePath(parent_path, element.name) element_path = PurePath(parent_path, element.name)
# This is symptomatic of no access to the element, for example, because if element.type in _VIDEO_ELEMENTS:
# of time availability restrictions. if not self._videos:
if "cmdClass=ilInfoScreenGUI" in element.url and "cmd=showSummary" in element.url:
log.explain(
"Skipping element as url points to info screen, "
"this should only happen with not-yet-released elements"
)
return None
if element.type in _VIDEO_ELEMENTS and not self._videos:
log.status( log.status(
"[bold bright_black]", "[bold bright_black]",
"Ignored", "Ignored",
fmt_path(element_path), fmt_path(element_path),
"[bright_black](enable with option 'videos')", "[bright_black](enable with option 'videos')"
) )
return None return None
@ -358,7 +333,7 @@ instance's greatest bottleneck.
"[bold bright_black]", "[bold bright_black]",
"Ignored", "Ignored",
fmt_path(element_path), fmt_path(element_path),
"[bright_black](enable with option 'forums')", "[bright_black](enable with option 'forums')"
) )
return None return None
return await self._handle_forum(element, element_path) return await self._handle_forum(element, element_path)
@ -367,7 +342,7 @@ instance's greatest bottleneck.
"[bold bright_black]", "[bold bright_black]",
"Ignored", "Ignored",
fmt_path(element_path), fmt_path(element_path),
"[bright_black](tests contain no relevant data)", "[bright_black](tests contain no relevant data)"
) )
return None return None
elif element.type == IliasElementType.SURVEY: elif element.type == IliasElementType.SURVEY:
@ -375,7 +350,7 @@ instance's greatest bottleneck.
"[bold bright_black]", "[bold bright_black]",
"Ignored", "Ignored",
fmt_path(element_path), fmt_path(element_path),
"[bright_black](surveys contain no relevant data)", "[bright_black](surveys contain no relevant data)"
) )
return None return None
elif element.type == IliasElementType.SCORM_LEARNING_MODULE: elif element.type == IliasElementType.SCORM_LEARNING_MODULE:
@ -383,7 +358,7 @@ instance's greatest bottleneck.
"[bold bright_black]", "[bold bright_black]",
"Ignored", "Ignored",
fmt_path(element_path), fmt_path(element_path),
"[bright_black](scorm learning modules are not supported)", "[bright_black](scorm learning modules are not supported)"
) )
return None return None
elif element.type == IliasElementType.LITERATURE_LIST: elif element.type == IliasElementType.LITERATURE_LIST:
@ -391,7 +366,7 @@ instance's greatest bottleneck.
"[bold bright_black]", "[bold bright_black]",
"Ignored", "Ignored",
fmt_path(element_path), fmt_path(element_path),
"[bright_black](literature lists are not currently supported)", "[bright_black](literature lists are not currently supported)"
) )
return None return None
elif element.type == IliasElementType.LEARNING_MODULE_HTML: elif element.type == IliasElementType.LEARNING_MODULE_HTML:
@ -399,7 +374,7 @@ instance's greatest bottleneck.
"[bold bright_black]", "[bold bright_black]",
"Ignored", "Ignored",
fmt_path(element_path), fmt_path(element_path),
"[bright_black](HTML learning modules are not supported)", "[bright_black](HTML learning modules are not supported)"
) )
return None return None
elif element.type == IliasElementType.BLOG: elif element.type == IliasElementType.BLOG:
@ -407,7 +382,7 @@ instance's greatest bottleneck.
"[bold bright_black]", "[bold bright_black]",
"Ignored", "Ignored",
fmt_path(element_path), fmt_path(element_path),
"[bright_black](blogs are not currently supported)", "[bright_black](blogs are not currently supported)"
) )
return None return None
elif element.type == IliasElementType.DCL_RECORD_LIST: elif element.type == IliasElementType.DCL_RECORD_LIST:
@ -415,7 +390,7 @@ instance's greatest bottleneck.
"[bold bright_black]", "[bold bright_black]",
"Ignored", "Ignored",
fmt_path(element_path), fmt_path(element_path),
"[bright_black](dcl record lists are not currently supported)", "[bright_black](dcl record lists are not currently supported)"
) )
return None return None
elif element.type == IliasElementType.MEDIA_POOL: elif element.type == IliasElementType.MEDIA_POOL:
@ -423,33 +398,21 @@ instance's greatest bottleneck.
"[bold bright_black]", "[bold bright_black]",
"Ignored", "Ignored",
fmt_path(element_path), fmt_path(element_path),
"[bright_black](media pools are not currently supported)", "[bright_black](media pools are not currently supported)"
) )
return None return None
elif element.type == IliasElementType.COURSE: elif element.type == IliasElementType.COURSE:
if crawl_nested_courses:
return await self._handle_ilias_page(element.url, element, element_path)
log.status( log.status(
"[bold bright_black]", "[bold bright_black]",
"Ignored", "Ignored",
fmt_path(element_path), fmt_path(element_path),
"[bright_black](not descending into linked course)", "[bright_black](not descending into linked course, download it separately)"
)
return None
elif element.type == IliasElementType.WIKI:
log.status(
"[bold bright_black]",
"Ignored",
fmt_path(element_path),
"[bright_black](wikis are not currently supported)",
) )
return None return None
elif element.type == IliasElementType.LEARNING_MODULE: elif element.type == IliasElementType.LEARNING_MODULE:
return await self._handle_learning_module(element, element_path) return await self._handle_learning_module(element, element_path)
elif element.type == IliasElementType.LINK: elif element.type == IliasElementType.LINK:
return await self._handle_link(element, element_path) return await self._handle_link(element, element_path)
elif element.type == IliasElementType.LINK_COLLECTION:
return await self._handle_link(element, element_path)
elif element.type == IliasElementType.BOOKING: elif element.type == IliasElementType.BOOKING:
return await self._handle_booking(element, element_path) return await self._handle_booking(element, element_path)
elif element.type == IliasElementType.OPENCAST_VIDEO: elif element.type == IliasElementType.OPENCAST_VIDEO:
@ -475,94 +438,45 @@ instance's greatest bottleneck.
log.explain_topic(f"Decision: Crawl Link {fmt_path(element_path)}") log.explain_topic(f"Decision: Crawl Link {fmt_path(element_path)}")
log.explain(f"Links type is {self._links}") log.explain(f"Links type is {self._links}")
export_url = url_set_query_param(element.url, "cmd", "exportHTML") link_template_maybe = self._links.template()
resolved = await self._resolve_link_target(export_url) link_extension = self._links.extension()
if resolved == "none": if not link_template_maybe or not link_extension:
links = [LinkData(element.name, "", element.description or "")]
else:
links = self._parse_link_content(element, cast(BeautifulSoup, resolved))
maybe_extension = self._links.extension()
if not maybe_extension:
log.explain("Answer: No") log.explain("Answer: No")
return None return None
else: else:
log.explain("Answer: Yes") log.explain("Answer: Yes")
element_path = element_path.with_name(element_path.name + link_extension)
if len(links) <= 1 or self._links.collection_as_one():
element_path = element_path.with_name(element_path.name + maybe_extension)
maybe_dl = await self.download(element_path, mtime=element.mtime) maybe_dl = await self.download(element_path, mtime=element.mtime)
if not maybe_dl: if not maybe_dl:
return None return None
return self._download_link(self._links, element.name, links, maybe_dl)
maybe_cl = await self.crawl(element_path) return self._download_link(element, link_template_maybe, maybe_dl)
if not maybe_cl:
return None
# Required for download_all closure
cl = maybe_cl
extension = maybe_extension
async def download_all() -> None:
for link in links:
path = cl.path / (sanitize_path_name(link.name) + extension)
if dl := await self.download(path, mtime=element.mtime):
await self._download_link(self._links, element.name, [link], dl)
return download_all()
@anoncritical @anoncritical
@_iorepeat(3, "resolving link") @_iorepeat(3, "resolving link")
async def _download_link( async def _download_link(self, element: IliasPageElement, link_template: str, dl: DownloadToken) -> None:
self, link_renderer: Links, collection_name: str, links: list[LinkData], dl: DownloadToken
) -> None:
async with dl as (bar, sink): async with dl as (bar, sink):
rendered = link_renderer.interpolate(self._link_file_redirect_delay, collection_name, links) export_url = element.url.replace("cmd=calldirectlink", "cmd=exportHTML")
sink.file.write(rendered.encode("utf-8")) real_url = await self._resolve_link_target(export_url)
self._write_link_content(link_template, real_url, element.name, element.description, sink)
def _write_link_content(
self,
link_template: str,
url: str,
name: str,
description: Optional[str],
sink: FileSink,
) -> None:
content = link_template
content = content.replace("{{link}}", url)
content = content.replace("{{name}}", name)
content = content.replace("{{description}}", str(description))
content = content.replace("{{redirect_delay}}", str(self._link_file_redirect_delay))
sink.file.write(content.encode("utf-8"))
sink.done() sink.done()
async def _resolve_link_target(self, export_url: str) -> BeautifulSoup | Literal["none"]:
async def impl() -> Optional[BeautifulSoup | Literal["none"]]:
async with self.session.get(export_url, allow_redirects=False) as resp:
# No redirect means we were authenticated
if hdrs.LOCATION not in resp.headers:
return soupify(await resp.read()) # .select_one("a").get("href").strip() # type: ignore
# We are either unauthenticated or the link is not active
new_url = resp.headers[hdrs.LOCATION].lower()
if "baseclass=illinkresourcehandlergui" in new_url and "cmd=infoscreen" in new_url:
return "none"
return None
auth_id = await self._current_auth_id()
target = await impl()
if target is not None:
return target
await self.authenticate(auth_id)
target = await impl()
if target is not None:
return target
raise CrawlError("resolve_link_target failed even after authenticating")
@staticmethod
def _parse_link_content(element: IliasPageElement, content: BeautifulSoup) -> list[LinkData]:
links = list(content.select("a"))
if len(links) == 1:
url = str(links[0].get("href")).strip()
return [LinkData(name=element.name, description=element.description or "", url=url)]
results = []
for link in links:
url = str(link.get("href")).strip()
name = link.get_text(strip=True)
description = cast(Tag, link.find_next_sibling("dd")).get_text(strip=True)
results.append(LinkData(name=name, description=description, url=url.strip()))
return results
async def _handle_booking( async def _handle_booking(
self, self,
element: IliasPageElement, element: IliasPageElement,
@ -586,7 +500,7 @@ instance's greatest bottleneck.
self._ensure_not_seen(element, element_path) self._ensure_not_seen(element, element_path)
return self._download_booking(element, maybe_dl) return self._download_booking(element, link_template_maybe, maybe_dl)
@anoncritical @anoncritical
@_iorepeat(1, "downloading description") @_iorepeat(1, "downloading description")
@ -599,7 +513,7 @@ instance's greatest bottleneck.
async with dl as (_bar, sink): async with dl as (_bar, sink):
description = clean(insert_base_markup(description)) description = clean(insert_base_markup(description))
description_tag = await self.internalize_images(description) description_tag = await self.internalize_images(description)
sink.file.write(description_tag.prettify().encode("utf-8")) sink.file.write(cast(str, description_tag.prettify()).encode("utf-8"))
sink.done() sink.done()
@anoncritical @anoncritical
@ -607,13 +521,36 @@ instance's greatest bottleneck.
async def _download_booking( async def _download_booking(
self, self,
element: IliasPageElement, element: IliasPageElement,
link_template: str,
dl: DownloadToken, dl: DownloadToken,
) -> None: ) -> None:
async with dl as (bar, sink): async with dl as (bar, sink):
links = [LinkData(name=element.name, description=element.description or "", url=element.url)] self._write_link_content(link_template, element.url, element.name, element.description, sink)
rendered = self._links.interpolate(self._link_file_redirect_delay, element.name, links)
sink.file.write(rendered.encode("utf-8")) async def _resolve_link_target(self, export_url: str) -> str:
sink.done() async def impl() -> Optional[str]:
async with self.session.get(export_url, allow_redirects=False) as resp:
# No redirect means we were authenticated
if hdrs.LOCATION not in resp.headers:
return soupify(await resp.read()).select_one("a").get("href").strip() # type: ignore
# We are either unauthenticated or the link is not active
new_url = resp.headers[hdrs.LOCATION].lower()
if "baseclass=illinkresourcehandlergui" in new_url and "cmd=infoscreen" in new_url:
return ""
return None
auth_id = await self._current_auth_id()
target = await impl()
if target is not None:
return target
await self.authenticate(auth_id)
target = await impl()
if target is not None:
return target
raise CrawlError("resolve_link_target failed even after authenticating")
async def _handle_opencast_video( async def _handle_opencast_video(
self, self,
@ -624,7 +561,7 @@ instance's greatest bottleneck.
if self.prev_report: if self.prev_report:
self.report.add_custom_value( self.report.add_custom_value(
_get_video_cache_key(element), _get_video_cache_key(element),
self.prev_report.get_custom_value(_get_video_cache_key(element)), self.prev_report.get_custom_value(_get_video_cache_key(element))
) )
# A video might contain other videos, so let's "crawl" the video first # A video might contain other videos, so let's "crawl" the video first
@ -658,7 +595,7 @@ instance's greatest bottleneck.
def _previous_contained_opencast_videos( def _previous_contained_opencast_videos(
self, element: IliasPageElement, element_path: PurePath self, element: IliasPageElement, element_path: PurePath
) -> list[PurePath]: ) -> List[PurePath]:
if not self.prev_report: if not self.prev_report:
return [] return []
custom_value = self.prev_report.get_custom_value(_get_video_cache_key(element)) custom_value = self.prev_report.get_custom_value(_get_video_cache_key(element))
@ -696,7 +633,7 @@ instance's greatest bottleneck.
def add_to_report(paths: list[str]) -> None: def add_to_report(paths: list[str]) -> None:
self.report.add_custom_value( self.report.add_custom_value(
_get_video_cache_key(element), _get_video_cache_key(element),
{"known_paths": paths, "own_path": str(self._transformer.transform(dl.path))}, {"known_paths": paths, "own_path": str(self._transformer.transform(dl.path))}
) )
async with dl as (bar, sink): async with dl as (bar, sink):
@ -714,7 +651,7 @@ instance's greatest bottleneck.
add_to_report([str(self._transformer.transform(dl.path))]) add_to_report([str(self._transformer.transform(dl.path))])
return return
contained_video_paths: list[str] = [] contained_video_paths: List[str] = []
for stream_element in stream_elements: for stream_element in stream_elements:
video_path = dl.path.parent / stream_element.name video_path = dl.path.parent / stream_element.name
@ -750,7 +687,11 @@ instance's greatest bottleneck.
await self._stream_from_url(element, sink, bar, is_video) await self._stream_from_url(element, sink, bar, is_video)
async def _stream_from_url( async def _stream_from_url(
self, element: IliasPageElement, sink: FileSink, bar: ProgressBar, is_video: bool self,
element: IliasPageElement,
sink: FileSink,
bar: ProgressBar,
is_video: bool
) -> None: ) -> None:
url = element.url url = element.url
@ -818,23 +759,70 @@ instance's greatest bottleneck.
@_iorepeat(3, "crawling forum") @_iorepeat(3, "crawling forum")
@anoncritical @anoncritical
async def _crawl_forum(self, element: IliasPageElement, cl: CrawlToken) -> None: async def _crawl_forum(self, element: IliasPageElement, cl: CrawlToken) -> None:
elements: List[IliasForumThread] = []
async with cl: async with cl:
inner = IliasPage(await self._get_page(element.url), element) next_stage_url = element.url
export_url = inner.get_forum_export_url() page = None
if not export_url:
log.warn("Could not extract forum export url") while next_stage_url:
log.explain_topic(f"Parsing HTML page for {fmt_path(cl.path)}")
log.explain(f"URL: {next_stage_url}")
soup = await self._get_page(next_stage_url)
page = IliasPage(soup, element)
if next := page.get_next_stage_element():
next_stage_url = next.url
else:
break
forum_threads: list[tuple[IliasPageElement, bool]] = []
for entry in cast(IliasPage, page).get_forum_entries():
path = cl.path / (_sanitize_path_name(entry.name) + ".html")
forum_threads.append((entry, self.should_try_download(path, mtime=entry.mtime)))
# Sort the ids. The forum download will *preserve* this ordering
forum_threads.sort(key=lambda elem: elem[0].id())
if not forum_threads:
log.explain("Forum had no threads")
return return
export = await self._post( download_data = cast(IliasPage, page).get_download_forum_data(
export_url, [thread.id() for thread, download in forum_threads if download]
{"format": "html", "cmd[createExportFile]": ""}, )
if not download_data:
raise CrawlWarning("Failed to extract forum data")
if not download_data.empty:
html = await self._post_authenticated(download_data.url, download_data.form_data)
elements = parse_ilias_forum_export(soupify(html))
else:
elements = []
# Verify that ILIAS does not change the order, as we depend on it later. Otherwise, we could not call
# download in the correct order, potentially messing up duplication handling.
expected_element_titles = [thread.name for thread, download in forum_threads if download]
actual_element_titles = [_sanitize_path_name(thread.name) for thread in elements]
if expected_element_titles != actual_element_titles:
raise CrawlWarning(
f"Forum thread order mismatch: {expected_element_titles} != {actual_element_titles}"
) )
elements = parse_ilias_forum_export(soupify(export)) tasks: List[Awaitable[None]] = []
for thread, download in forum_threads:
tasks: list[Awaitable[None]] = [] if download:
for thread in elements: # This only works because ILIAS keeps the order in the export
tasks.append(asyncio.create_task(self._download_forum_thread(cl.path, thread, element.url))) elem = elements.pop(0)
tasks.append(asyncio.create_task(self._download_forum_thread(cl.path, elem, thread)))
else:
# We only downloaded the threads we "should_try_download"ed. This can be an
# over-approximation and all will be fine.
# If we selected too few, e.g. because there was a duplicate title and the mtime of the
# original is newer than the update of the duplicate.
# This causes stale data locally, but I consider this problem acceptable right now.
tasks.append(asyncio.create_task(self._download_forum_thread(cl.path, thread, thread)))
# And execute them # And execute them
await self.gather(tasks) await self.gather(tasks)
@ -842,16 +830,22 @@ instance's greatest bottleneck.
@anoncritical @anoncritical
@_iorepeat(3, "saving forum thread") @_iorepeat(3, "saving forum thread")
async def _download_forum_thread( async def _download_forum_thread(
self, parent_path: PurePath, thread: IliasForumThread | IliasPageElement, forum_url: str self,
parent_path: PurePath,
thread: Union[IliasForumThread, IliasPageElement],
element: IliasPageElement
) -> None: ) -> None:
path = parent_path / (sanitize_path_name(thread.name) + ".html") path = parent_path / (_sanitize_path_name(thread.name) + ".html")
maybe_dl = await self.download(path, mtime=thread.mtime) maybe_dl = await self.download(path, mtime=thread.mtime)
if not maybe_dl or not isinstance(thread, IliasForumThread): if not maybe_dl or not isinstance(thread, IliasForumThread):
return return
async with maybe_dl as (bar, sink): async with maybe_dl as (bar, sink):
rendered = forum_thread_template( rendered = forum_thread_template(
thread.name, forum_url, thread.name_tag, await self.internalize_images(thread.content_tag) thread.name,
element.url,
thread.name_tag,
await self.internalize_images(thread.content_tag)
) )
sink.file.write(rendered.encode("utf-8")) sink.file.write(rendered.encode("utf-8"))
sink.done() sink.done()
@ -871,7 +865,7 @@ instance's greatest bottleneck.
@_iorepeat(3, "crawling learning module") @_iorepeat(3, "crawling learning module")
@anoncritical @anoncritical
async def _crawl_learning_module(self, element: IliasPageElement, cl: CrawlToken) -> None: async def _crawl_learning_module(self, element: IliasPageElement, cl: CrawlToken) -> None:
elements: list[IliasLearningModulePage] = [] elements: List[IliasLearningModulePage] = []
async with cl: async with cl:
log.explain_topic(f"Parsing initial HTML page for {fmt_path(cl.path)}") log.explain_topic(f"Parsing initial HTML page for {fmt_path(cl.path)}")
@ -879,25 +873,25 @@ instance's greatest bottleneck.
soup = await self._get_page(element.url) soup = await self._get_page(element.url)
page = IliasPage(soup, element) page = IliasPage(soup, element)
if next := page.get_learning_module_data(): if next := page.get_learning_module_data():
elements.extend( elements.extend(await self._crawl_learning_module_direction(
await self._crawl_learning_module_direction(cl.path, next.previous_url, "left", element) cl.path, next.previous_url, "left", element
) ))
elements.append(next) elements.append(next)
elements.extend( elements.extend(await self._crawl_learning_module_direction(
await self._crawl_learning_module_direction(cl.path, next.next_url, "right", element) cl.path, next.next_url, "right", element
) ))
# Reflect their natural ordering in the file names # Reflect their natural ordering in the file names
for index, lm_element in enumerate(elements): for index, lm_element in enumerate(elements):
lm_element.title = f"{index:02}_{lm_element.title}" lm_element.title = f"{index:02}_{lm_element.title}"
tasks: list[Awaitable[None]] = [] tasks: List[Awaitable[None]] = []
for index, elem in enumerate(elements): for index, elem in enumerate(elements):
prev_url = elements[index - 1].title if index > 0 else None prev_url = elements[index - 1].title if index > 0 else None
next_url = elements[index + 1].title if index < len(elements) - 1 else None next_url = elements[index + 1].title if index < len(elements) - 1 else None
tasks.append( tasks.append(asyncio.create_task(
asyncio.create_task(self._download_learning_module_page(cl.path, elem, prev_url, next_url)) self._download_learning_module_page(cl.path, elem, prev_url, next_url)
) ))
# And execute them # And execute them
await self.gather(tasks) await self.gather(tasks)
@ -906,10 +900,10 @@ instance's greatest bottleneck.
self, self,
path: PurePath, path: PurePath,
start_url: Optional[str], start_url: Optional[str],
dir: Literal["left"] | Literal["right"], dir: Union[Literal["left"], Literal["right"]],
parent_element: IliasPageElement, parent_element: IliasPageElement
) -> list[IliasLearningModulePage]: ) -> List[IliasLearningModulePage]:
elements: list[IliasLearningModulePage] = [] elements: List[IliasLearningModulePage] = []
if not start_url: if not start_url:
return elements return elements
@ -923,7 +917,10 @@ instance's greatest bottleneck.
page = IliasPage(soup, parent_element) page = IliasPage(soup, parent_element)
if next := page.get_learning_module_data(): if next := page.get_learning_module_data():
elements.append(next) elements.append(next)
next_element_url = next.previous_url if dir == "left" else next.next_url if dir == "left":
next_element_url = next.previous_url
else:
next_element_url = next.next_url
counter += 1 counter += 1
return elements return elements
@ -935,9 +932,9 @@ instance's greatest bottleneck.
parent_path: PurePath, parent_path: PurePath,
element: IliasLearningModulePage, element: IliasLearningModulePage,
prev: Optional[str], prev: Optional[str],
next: Optional[str], next: Optional[str]
) -> None: ) -> None:
path = parent_path / (sanitize_path_name(element.title) + ".html") path = parent_path / (_sanitize_path_name(element.title) + ".html")
maybe_dl = await self.download(path) maybe_dl = await self.download(path)
if not maybe_dl: if not maybe_dl:
return return
@ -946,11 +943,17 @@ instance's greatest bottleneck.
return return
if prev: if prev:
prev_p = self._transformer.transform(parent_path / (sanitize_path_name(prev) + ".html")) prev_p = self._transformer.transform(parent_path / (_sanitize_path_name(prev) + ".html"))
prev = os.path.relpath(prev_p, my_path.parent) if prev_p else None if prev_p:
prev = cast(str, os.path.relpath(prev_p, my_path.parent))
else:
prev = None
if next: if next:
next_p = self._transformer.transform(parent_path / (sanitize_path_name(next) + ".html")) next_p = self._transformer.transform(parent_path / (_sanitize_path_name(next) + ".html"))
next = os.path.relpath(next_p, my_path.parent) if next_p else None if next_p:
next = cast(str, os.path.relpath(next_p, my_path.parent))
else:
next = None
async with maybe_dl as (bar, sink): async with maybe_dl as (bar, sink):
content = element.content content = element.content
@ -964,7 +967,10 @@ instance's greatest bottleneck.
""" """
log.explain_topic("Internalizing images") log.explain_topic("Internalizing images")
for elem in tag.find_all(recursive=True): for elem in tag.find_all(recursive=True):
if elem.name == "img" and (src := elem.attrs.get("src", None)): if not isinstance(elem, Tag):
continue
if elem.name == "img":
if src := elem.attrs.get("src", None):
url = urljoin(self._base_url, cast(str, src)) url = urljoin(self._base_url, cast(str, src))
if not url.startswith(self._base_url): if not url.startswith(self._base_url):
continue continue
@ -1015,15 +1021,29 @@ instance's greatest bottleneck.
) )
return soup return soup
async def _post(self, url: str, data: dict[str, str | list[str]]) -> bytes: async def _post_authenticated(
self,
url: str,
data: dict[str, Union[str, List[str]]]
) -> bytes:
auth_id = await self._current_auth_id()
form_data = aiohttp.FormData() form_data = aiohttp.FormData()
for key, val in data.items(): for key, val in data.items():
form_data.add_field(key, val) form_data.add_field(key, val)
async with self.session.post(url, data=form_data()) as request: async with self.session.post(url, data=form_data(), allow_redirects=False) as request:
if request.status == 200: if request.status == 200:
return await request.read() return await request.read()
raise CrawlError(f"post failed with status {request.status}")
# We weren't authenticated, so try to do that
await self.authenticate(auth_id)
# Retry once after authenticating. If this fails, we will die.
async with self.session.post(url, data=data, allow_redirects=False) as request:
if request.status == 200:
return await request.read()
raise CrawlError("post_authenticated failed even after authenticating")
async def _get_authenticated(self, url: str) -> bytes: async def _get_authenticated(self, url: str) -> bytes:
auth_id = await self._current_auth_id() auth_id = await self._current_auth_id()
@ -1053,7 +1073,7 @@ instance's greatest bottleneck.
async with self.session.get(urljoin(self._base_url, "/login.php"), params=params) as request: async with self.session.get(urljoin(self._base_url, "/login.php"), params=params) as request:
login_page = soupify(await request.read()) login_page = soupify(await request.read())
login_form = login_page.find("form", attrs={"name": "login_form"}) login_form = cast(Optional[Tag], login_page.find("form", attrs={"name": "formlogin"}))
if login_form is None: if login_form is None:
raise CrawlError("Could not find the login form! Specified client id might be invalid.") raise CrawlError("Could not find the login form! Specified client id might be invalid.")
@ -1063,12 +1083,14 @@ instance's greatest bottleneck.
username, password = await self._auth.credentials() username, password = await self._auth.credentials()
login_form_data = aiohttp.FormData() login_data = {
login_form_data.add_field("login_form/input_3/input_4", username) "username": username,
login_form_data.add_field("login_form/input_3/input_5", password) "password": password,
"cmd[doStandardAuthentication]": "Login",
}
# do the actual login # do the actual login
async with self.session.post(urljoin(self._base_url, login_url), data=login_form_data) as request: async with self.session.post(urljoin(self._base_url, login_url), data=login_data) as request:
soup = IliasSoup(soupify(await request.read()), str(request.url)) soup = IliasSoup(soupify(await request.read()), str(request.url))
if not IliasPage.is_logged_in(soup): if not IliasPage.is_logged_in(soup):
self._auth.invalidate_credentials() self._auth.invalidate_credentials()

File diff suppressed because it is too large Load diff

View file

@ -1,4 +1,4 @@
from typing import Literal from typing import Dict, Literal
from ...auth import Authenticator from ...auth import Authenticator
from ...config import Config from ...config import Config
@ -26,7 +26,7 @@ class KitIliasWebCrawler(IliasWebCrawler):
name: str, name: str,
section: KitIliasWebCrawlerSection, section: KitIliasWebCrawlerSection,
config: Config, config: Config,
authenticators: dict[str, Authenticator], authenticators: Dict[str, Authenticator],
): ):
super().__init__(name, section, config, authenticators) super().__init__(name, section, config, authenticators)

View file

@ -38,7 +38,9 @@ class ShibbolethLogin:
async with sess.get(url) as response: async with sess.get(url) as response:
shib_url = response.url shib_url = response.url
if str(shib_url).startswith(self._ilias_url): if str(shib_url).startswith(self._ilias_url):
log.explain("ILIAS recognized our shib token and logged us in in the background, returning") log.explain(
"ILIAS recognized our shib token and logged us in in the background, returning"
)
return return
soup: BeautifulSoup = soupify(await response.read()) soup: BeautifulSoup = soupify(await response.read())
@ -79,7 +81,7 @@ class ShibbolethLogin:
# (or clicking "Continue" if you have JS disabled) # (or clicking "Continue" if you have JS disabled)
relay_state = cast(Tag, soup.find("input", {"name": "RelayState"})) relay_state = cast(Tag, soup.find("input", {"name": "RelayState"}))
saml_response = cast(Tag, soup.find("input", {"name": "SAMLResponse"})) saml_response = cast(Tag, soup.find("input", {"name": "SAMLResponse"}))
url = cast(str, cast(Tag, soup.find("form", {"method": "post"}))["action"]) url = form = soup.find("form", {"method": "post"})["action"] # type: ignore
data = { # using the info obtained in the while loop above data = { # using the info obtained in the while loop above
"RelayState": cast(str, relay_state["value"]), "RelayState": cast(str, relay_state["value"]),
"SAMLResponse": cast(str, saml_response["value"]), "SAMLResponse": cast(str, saml_response["value"]),

View file

@ -1,21 +1,17 @@
import os import os
import re import re
from collections.abc import Awaitable, Generator, Iterable
from dataclasses import dataclass from dataclasses import dataclass
from datetime import datetime from datetime import datetime
from pathlib import PurePath from pathlib import PurePath
from re import Pattern from typing import Any, Awaitable, Generator, Iterable, List, Optional, Pattern, Tuple, Union, cast
from typing import Any, Optional, Union, cast
from urllib.parse import urljoin from urllib.parse import urljoin
import aiohttp
from bs4 import BeautifulSoup, Tag from bs4 import BeautifulSoup, Tag
from ..auth import Authenticator
from ..config import Config from ..config import Config
from ..logging import ProgressBar, log from ..logging import ProgressBar, log
from ..output_dir import FileSink from ..output_dir import FileSink
from ..utils import sanitize_path_name, soupify from ..utils import soupify
from .crawler import CrawlError from .crawler import CrawlError
from .http_crawler import HttpCrawler, HttpCrawlerSection from .http_crawler import HttpCrawler, HttpCrawlerSection
@ -35,15 +31,6 @@ class KitIpdCrawlerSection(HttpCrawlerSection):
regex = self.s.get("link_regex", r"^.*?[^/]+\.(pdf|zip|c|cpp|java)$") regex = self.s.get("link_regex", r"^.*?[^/]+\.(pdf|zip|c|cpp|java)$")
return re.compile(regex) return re.compile(regex)
def basic_auth(self, authenticators: dict[str, Authenticator]) -> Optional[Authenticator]:
value: Optional[str] = self.s.get("auth")
if value is None:
return None
auth = authenticators.get(value)
if auth is None:
self.invalid_value("auth", value, "No such auth section exists")
return auth
@dataclass @dataclass
class KitIpdFile: class KitIpdFile:
@ -57,7 +44,7 @@ class KitIpdFile:
@dataclass @dataclass
class KitIpdFolder: class KitIpdFolder:
name: str name: str
entries: list[Union[KitIpdFile, "KitIpdFolder"]] entries: List[Union[KitIpdFile, "KitIpdFolder"]]
def explain(self) -> None: def explain(self) -> None:
log.explain_topic(f"Folder {self.name!r}") log.explain_topic(f"Folder {self.name!r}")
@ -66,29 +53,23 @@ class KitIpdFolder:
class KitIpdCrawler(HttpCrawler): class KitIpdCrawler(HttpCrawler):
def __init__( def __init__(
self, self,
name: str, name: str,
section: KitIpdCrawlerSection, section: KitIpdCrawlerSection,
config: Config, config: Config,
authenticators: dict[str, Authenticator],
): ):
super().__init__(name, section, config) super().__init__(name, section, config)
self._url = section.target() self._url = section.target()
self._file_regex = section.link_regex() self._file_regex = section.link_regex()
self._authenticator = section.basic_auth(authenticators)
self._basic_auth: Optional[aiohttp.BasicAuth] = None
async def _run(self) -> None: async def _run(self) -> None:
if self._authenticator:
username, password = await self._authenticator.credentials()
self._basic_auth = aiohttp.BasicAuth(username, password)
maybe_cl = await self.crawl(PurePath(".")) maybe_cl = await self.crawl(PurePath("."))
if not maybe_cl: if not maybe_cl:
return return
tasks: list[Awaitable[None]] = [] tasks: List[Awaitable[None]] = []
async with maybe_cl: async with maybe_cl:
for item in await self._fetch_items(): for item in await self._fetch_items():
@ -106,7 +87,7 @@ class KitIpdCrawler(HttpCrawler):
await self.gather(tasks) await self.gather(tasks)
async def _crawl_folder(self, parent: PurePath, folder: KitIpdFolder) -> None: async def _crawl_folder(self, parent: PurePath, folder: KitIpdFolder) -> None:
path = parent / sanitize_path_name(folder.name) path = parent / folder.name
if not await self.crawl(path): if not await self.crawl(path):
return return
@ -123,9 +104,13 @@ class KitIpdCrawler(HttpCrawler):
await self.gather(tasks) await self.gather(tasks)
async def _download_file( async def _download_file(
self, parent: PurePath, file: KitIpdFile, etag: Optional[str], mtime: Optional[datetime] self,
parent: PurePath,
file: KitIpdFile,
etag: Optional[str],
mtime: Optional[datetime]
) -> None: ) -> None:
element_path = parent / sanitize_path_name(file.name) element_path = parent / file.name
prev_etag = self._get_previous_etag_from_report(element_path) prev_etag = self._get_previous_etag_from_report(element_path)
etag_differs = None if prev_etag is None else prev_etag != etag etag_differs = None if prev_etag is None else prev_etag != etag
@ -140,9 +125,9 @@ class KitIpdCrawler(HttpCrawler):
async with maybe_dl as (bar, sink): async with maybe_dl as (bar, sink):
await self._stream_from_url(file.url, element_path, sink, bar) await self._stream_from_url(file.url, element_path, sink, bar)
async def _fetch_items(self) -> Iterable[KitIpdFile | KitIpdFolder]: async def _fetch_items(self) -> Iterable[Union[KitIpdFile, KitIpdFolder]]:
page, url = await self.get_page() page, url = await self.get_page()
elements: list[Tag] = self._find_file_links(page) elements: List[Tag] = self._find_file_links(page)
# do not add unnecessary nesting for a single <h1> heading # do not add unnecessary nesting for a single <h1> heading
drop_h1: bool = len(page.find_all(name="h1")) <= 1 drop_h1: bool = len(page.find_all(name="h1")) <= 1
@ -171,21 +156,16 @@ class KitIpdCrawler(HttpCrawler):
name = os.path.basename(url) name = os.path.basename(url)
return KitIpdFile(name, url) return KitIpdFile(name, url)
def _find_file_links(self, tag: Tag | BeautifulSoup) -> list[Tag]: def _find_file_links(self, tag: Union[Tag, BeautifulSoup]) -> list[Tag]:
return cast(list[Tag], tag.find_all(name="a", attrs={"href": self._file_regex})) return cast(list[Tag], tag.find_all(name="a", attrs={"href": self._file_regex}))
def _abs_url_from_link(self, url: str, link_tag: Tag) -> str: def _abs_url_from_link(self, url: str, link_tag: Tag) -> str:
return urljoin(url, cast(str, link_tag.get("href"))) return urljoin(url, cast(str, link_tag.get("href")))
async def _stream_from_url(self, url: str, path: PurePath, sink: FileSink, bar: ProgressBar) -> None: async def _stream_from_url(self, url: str, path: PurePath, sink: FileSink, bar: ProgressBar) -> None:
async with self.session.get(url, allow_redirects=False, auth=self._basic_auth) as resp: async with self.session.get(url, allow_redirects=False) as resp:
if resp.status == 403: if resp.status == 403:
raise CrawlError("Received a 403. Are you within the KIT network/VPN?") raise CrawlError("Received a 403. Are you within the KIT network/VPN?")
if resp.status == 401:
raise CrawlError("Received a 401. Do you maybe need credentials?")
if resp.status >= 400:
raise CrawlError(f"Received HTTP {resp.status} when trying to download {url!r}")
if resp.content_length: if resp.content_length:
bar.set_total(resp.content_length) bar.set_total(resp.content_length)
@ -197,8 +177,8 @@ class KitIpdCrawler(HttpCrawler):
self._add_etag_to_report(path, resp.headers.get("ETag")) self._add_etag_to_report(path, resp.headers.get("ETag"))
async def get_page(self) -> tuple[BeautifulSoup, str]: async def get_page(self) -> Tuple[BeautifulSoup, str]:
async with self.session.get(self._url, auth=self._basic_auth) as request: async with self.session.get(self._url) as request:
# The web page for Algorithmen für Routenplanung contains some # The web page for Algorithmen für Routenplanung contains some
# weird comments that beautifulsoup doesn't parse correctly. This # weird comments that beautifulsoup doesn't parse correctly. This
# hack enables those pages to be crawled, and should hopefully not # hack enables those pages to be crawled, and should hopefully not

View file

@ -18,19 +18,22 @@ class LocalCrawlerSection(CrawlerSection):
def crawl_delay(self) -> float: def crawl_delay(self) -> float:
value = self.s.getfloat("crawl_delay", fallback=0.0) value = self.s.getfloat("crawl_delay", fallback=0.0)
if value < 0: if value < 0:
self.invalid_value("crawl_delay", value, "Must not be negative") self.invalid_value("crawl_delay", value,
"Must not be negative")
return value return value
def download_delay(self) -> float: def download_delay(self) -> float:
value = self.s.getfloat("download_delay", fallback=0.0) value = self.s.getfloat("download_delay", fallback=0.0)
if value < 0: if value < 0:
self.invalid_value("download_delay", value, "Must not be negative") self.invalid_value("download_delay", value,
"Must not be negative")
return value return value
def download_speed(self) -> Optional[int]: def download_speed(self) -> Optional[int]:
value = self.s.getint("download_speed") value = self.s.getint("download_speed")
if value is not None and value <= 0: if value is not None and value <= 0:
self.invalid_value("download_speed", value, "Must be greater than 0") self.invalid_value("download_speed", value,
"Must be greater than 0")
return value return value
@ -71,12 +74,10 @@ class LocalCrawler(Crawler):
tasks = [] tasks = []
async with cl: async with cl:
await asyncio.sleep( await asyncio.sleep(random.uniform(
random.uniform(
0.5 * self._crawl_delay, 0.5 * self._crawl_delay,
self._crawl_delay, self._crawl_delay,
) ))
)
for child in path.iterdir(): for child in path.iterdir():
pure_child = cl.path / child.name pure_child = cl.path / child.name
@ -92,12 +93,10 @@ class LocalCrawler(Crawler):
return return
async with dl as (bar, sink): async with dl as (bar, sink):
await asyncio.sleep( await asyncio.sleep(random.uniform(
random.uniform(
0.5 * self._download_delay, 0.5 * self._download_delay,
self._download_delay, self._download_delay,
) ))
)
bar.set_total(stat.st_size) bar.set_total(stat.st_size)

View file

@ -1,5 +1,5 @@
from collections.abc import Iterator
from pathlib import PurePath from pathlib import PurePath
from typing import Iterator, Set
from .logging import log from .logging import log
from .utils import fmt_path from .utils import fmt_path
@ -16,34 +16,15 @@ def name_variants(path: PurePath) -> Iterator[PurePath]:
class Deduplicator: class Deduplicator:
FORBIDDEN_CHARS = '<>:"/\\|?*' + "".join([chr(i) for i in range(0, 32)]) FORBIDDEN_CHARS = '<>:"/\\|?*' + "".join([chr(i) for i in range(0, 32)])
FORBIDDEN_NAMES = { FORBIDDEN_NAMES = {
"CON", "CON", "PRN", "AUX", "NUL",
"PRN", "COM1", "COM2", "COM3", "COM4", "COM5", "COM6", "COM7", "COM8", "COM9",
"AUX", "LPT1", "LPT2", "LPT3", "LPT4", "LPT5", "LPT6", "LPT7", "LPT8", "LPT9",
"NUL",
"COM1",
"COM2",
"COM3",
"COM4",
"COM5",
"COM6",
"COM7",
"COM8",
"COM9",
"LPT1",
"LPT2",
"LPT3",
"LPT4",
"LPT5",
"LPT6",
"LPT7",
"LPT8",
"LPT9",
} }
def __init__(self, windows_paths: bool) -> None: def __init__(self, windows_paths: bool) -> None:
self._windows_paths = windows_paths self._windows_paths = windows_paths
self._known: set[PurePath] = set() self._known: Set[PurePath] = set()
def _add(self, path: PurePath) -> None: def _add(self, path: PurePath) -> None:
self._known.add(path) self._known.add(path)

View file

@ -1,9 +1,8 @@
import asyncio import asyncio
import time import time
from collections.abc import AsyncIterator
from contextlib import asynccontextmanager from contextlib import asynccontextmanager
from dataclasses import dataclass from dataclasses import dataclass
from typing import Optional from typing import AsyncIterator, Optional
@dataclass @dataclass
@ -13,7 +12,12 @@ class Slot:
class Limiter: class Limiter:
def __init__(self, task_limit: int, download_limit: int, task_delay: float): def __init__(
self,
task_limit: int,
download_limit: int,
task_delay: float
):
if task_limit <= 0: if task_limit <= 0:
raise ValueError("task limit must be at least 1") raise ValueError("task limit must be at least 1")
if download_limit <= 0: if download_limit <= 0:

View file

@ -1,23 +1,15 @@
import asyncio import asyncio
import sys import sys
import traceback import traceback
from collections.abc import AsyncIterator, Iterator
from contextlib import AbstractContextManager, asynccontextmanager, contextmanager from contextlib import AbstractContextManager, asynccontextmanager, contextmanager
from typing import Any, Optional from typing import AsyncIterator, Iterator, List, Optional
from rich.console import Console, Group from rich.console import Console, Group
from rich.live import Live from rich.live import Live
from rich.markup import escape from rich.markup import escape
from rich.panel import Panel from rich.panel import Panel
from rich.progress import ( from rich.progress import (BarColumn, DownloadColumn, Progress, TaskID, TextColumn, TimeRemainingColumn,
BarColumn, TransferSpeedColumn)
DownloadColumn,
Progress,
TaskID,
TextColumn,
TimeRemainingColumn,
TransferSpeedColumn,
)
from rich.table import Column from rich.table import Column
@ -61,7 +53,7 @@ class Log:
self._showing_progress = False self._showing_progress = False
self._progress_suspended = False self._progress_suspended = False
self._lock = asyncio.Lock() self._lock = asyncio.Lock()
self._lines: list[str] = [] self._lines: List[str] = []
# Whether different parts of the output are enabled or disabled # Whether different parts of the output are enabled or disabled
self.output_explain = False self.output_explain = False
@ -122,7 +114,7 @@ class Log:
for line in self._lines: for line in self._lines:
self.print(line) self.print(line)
def print(self, text: Any) -> None: def print(self, text: str) -> None:
""" """
Print a normal message. Allows markup. Print a normal message. Allows markup.
""" """
@ -184,14 +176,10 @@ class Log:
# Our print function doesn't take types other than strings, but the # Our print function doesn't take types other than strings, but the
# underlying rich.print function does. This call is a special case # underlying rich.print function does. This call is a special case
# anyways, and we're calling it internally, so this should be fine. # anyways, and we're calling it internally, so this should be fine.
self.print( self.print(Panel.fit("""
Panel.fit(
"""
Please copy your program output and send it to the PFERD maintainers, either Please copy your program output and send it to the PFERD maintainers, either
directly or as a GitHub issue: https://github.com/Garmelon/PFERD/issues/new directly or as a GitHub issue: https://github.com/Garmelon/PFERD/issues/new
""".strip() """.strip())) # type: ignore
)
)
def explain_topic(self, text: str) -> None: def explain_topic(self, text: str) -> None:
""" """

View file

@ -4,13 +4,12 @@ import os
import random import random
import shutil import shutil
import string import string
from collections.abc import Iterator from contextlib import contextmanager
from contextlib import contextmanager, suppress
from dataclasses import dataclass from dataclasses import dataclass
from datetime import datetime from datetime import datetime
from enum import Enum from enum import Enum
from pathlib import Path, PurePath from pathlib import Path, PurePath
from typing import BinaryIO, Optional from typing import BinaryIO, Iterator, Optional, Tuple
from .logging import log from .logging import log
from .report import Report, ReportLoadError from .report import Report, ReportLoadError
@ -36,7 +35,8 @@ class Redownload(Enum):
try: try:
return Redownload(string) return Redownload(string)
except ValueError: except ValueError:
raise ValueError("must be one of 'never', 'never-smart', 'always', 'always-smart'") from None raise ValueError("must be one of 'never', 'never-smart',"
" 'always', 'always-smart'")
class OnConflict(Enum): class OnConflict(Enum):
@ -51,10 +51,8 @@ class OnConflict(Enum):
try: try:
return OnConflict(string) return OnConflict(string)
except ValueError: except ValueError:
raise ValueError( raise ValueError("must be one of 'prompt', 'local-first',"
"must be one of 'prompt', 'local-first'," " 'remote-first', 'no-delete', 'no-delete-prompt-overwrite'")
" 'remote-first', 'no-delete', 'no-delete-prompt-overwrite'"
) from None
@dataclass @dataclass
@ -120,8 +118,7 @@ class FileSinkToken(ReusableAsyncContextManager[FileSink]):
sink = FileSink(file) sink = FileSink(file)
async def after_download() -> None: async def after_download() -> None:
await self._output_dir._after_download( await self._output_dir._after_download(DownloadInfo(
DownloadInfo(
self._remote_path, self._remote_path,
self._path, self._path,
self._local_path, self._local_path,
@ -129,8 +126,7 @@ class FileSinkToken(ReusableAsyncContextManager[FileSink]):
self._heuristics, self._heuristics,
self._on_conflict, self._on_conflict,
sink.is_done(), sink.is_done(),
) ))
)
self._stack.push_async_callback(after_download) self._stack.push_async_callback(after_download)
self._stack.enter_context(file) self._stack.enter_context(file)
@ -178,8 +174,8 @@ class OutputDirectory:
try: try:
self._root.mkdir(parents=True, exist_ok=True) self._root.mkdir(parents=True, exist_ok=True)
except OSError as e: except OSError:
raise OutputDirError("Failed to create base directory") from e raise OutputDirError("Failed to create base directory")
def register_reserved(self, path: PurePath) -> None: def register_reserved(self, path: PurePath) -> None:
self._report.mark_reserved(path) self._report.mark_reserved(path)
@ -359,7 +355,7 @@ class OutputDirectory:
async def _create_tmp_file( async def _create_tmp_file(
self, self,
local_path: Path, local_path: Path,
) -> tuple[Path, BinaryIO]: ) -> Tuple[Path, BinaryIO]:
""" """
May raise an OutputDirError. May raise an OutputDirError.
""" """
@ -510,8 +506,10 @@ class OutputDirectory:
await self._cleanup(child, pure_child) await self._cleanup(child, pure_child)
if delete_self: if delete_self:
with suppress(OSError): try:
path.rmdir() path.rmdir()
except OSError:
pass
async def _cleanup_file(self, path: Path, pure: PurePath) -> None: async def _cleanup_file(self, path: Path, pure: PurePath) -> None:
if self._report.is_marked(pure): if self._report.is_marked(pure):

View file

@ -1,5 +1,5 @@
from pathlib import Path, PurePath from pathlib import Path, PurePath
from typing import Optional from typing import Dict, List, Optional
from rich.markup import escape from rich.markup import escape
@ -15,7 +15,7 @@ class PferdLoadError(Exception):
class Pferd: class Pferd:
def __init__(self, config: Config, cli_crawlers: Optional[list[str]], cli_skips: Optional[list[str]]): def __init__(self, config: Config, cli_crawlers: Optional[List[str]], cli_skips: Optional[List[str]]):
""" """
May throw PferdLoadError. May throw PferdLoadError.
""" """
@ -23,10 +23,10 @@ class Pferd:
self._config = config self._config = config
self._crawlers_to_run = self._find_crawlers_to_run(config, cli_crawlers, cli_skips) self._crawlers_to_run = self._find_crawlers_to_run(config, cli_crawlers, cli_skips)
self._authenticators: dict[str, Authenticator] = {} self._authenticators: Dict[str, Authenticator] = {}
self._crawlers: dict[str, Crawler] = {} self._crawlers: Dict[str, Crawler] = {}
def _find_config_crawlers(self, config: Config) -> list[str]: def _find_config_crawlers(self, config: Config) -> List[str]:
crawl_sections = [] crawl_sections = []
for name, section in config.crawl_sections(): for name, section in config.crawl_sections():
@ -37,7 +37,7 @@ class Pferd:
return crawl_sections return crawl_sections
def _find_cli_crawlers(self, config: Config, cli_crawlers: list[str]) -> list[str]: def _find_cli_crawlers(self, config: Config, cli_crawlers: List[str]) -> List[str]:
if len(cli_crawlers) != len(set(cli_crawlers)): if len(cli_crawlers) != len(set(cli_crawlers)):
raise PferdLoadError("Some crawlers were selected multiple times") raise PferdLoadError("Some crawlers were selected multiple times")
@ -68,12 +68,12 @@ class Pferd:
def _find_crawlers_to_run( def _find_crawlers_to_run(
self, self,
config: Config, config: Config,
cli_crawlers: Optional[list[str]], cli_crawlers: Optional[List[str]],
cli_skips: Optional[list[str]], cli_skips: Optional[List[str]],
) -> list[str]: ) -> List[str]:
log.explain_topic("Deciding which crawlers to run") log.explain_topic("Deciding which crawlers to run")
crawlers: list[str] crawlers: List[str]
if cli_crawlers is None: if cli_crawlers is None:
log.explain("No crawlers specified on CLI") log.explain("No crawlers specified on CLI")
log.explain("Running crawlers specified in config") log.explain("Running crawlers specified in config")
@ -104,7 +104,7 @@ class Pferd:
def _load_crawlers(self) -> None: def _load_crawlers(self) -> None:
# Cookie sharing # Cookie sharing
kit_ilias_web_paths: dict[Authenticator, list[Path]] = {} kit_ilias_web_paths: Dict[Authenticator, List[Path]] = {}
for name, section in self._config.crawl_sections(): for name, section in self._config.crawl_sections():
log.print(f"[bold bright_cyan]Loading[/] {escape(name)}") log.print(f"[bold bright_cyan]Loading[/] {escape(name)}")
@ -117,7 +117,8 @@ class Pferd:
crawler = crawler_constructor(name, section, self._config, self._authenticators) crawler = crawler_constructor(name, section, self._config, self._authenticators)
self._crawlers[name] = crawler self._crawlers[name] = crawler
if self._config.default_section.share_cookies() and isinstance(crawler, KitIliasWebCrawler): if self._config.default_section.share_cookies():
if isinstance(crawler, KitIliasWebCrawler):
crawler.share_cookies(kit_ilias_web_paths) crawler.share_cookies(kit_ilias_web_paths)
def debug_transforms(self) -> None: def debug_transforms(self) -> None:
@ -160,10 +161,9 @@ class Pferd:
def print_report(self) -> None: def print_report(self) -> None:
for name in self._crawlers_to_run: for name in self._crawlers_to_run:
crawlerOpt = self._crawlers.get(name) crawler = self._crawlers.get(name)
if crawlerOpt is None: if crawler is None:
continue # Crawler failed to load continue # Crawler failed to load
crawler = crawlerOpt
log.report("") log.report("")
log.report(f"[bold bright_cyan]Report[/] for {escape(name)}") log.report(f"[bold bright_cyan]Report[/] for {escape(name)}")

View file

@ -1,6 +1,6 @@
import json import json
from pathlib import Path, PurePath from pathlib import Path, PurePath
from typing import Any, Optional from typing import Any, Dict, List, Optional, Set
class ReportLoadError(Exception): class ReportLoadError(Exception):
@ -42,32 +42,32 @@ class Report:
def __init__(self) -> None: def __init__(self) -> None:
# Paths found by the crawler, untransformed # Paths found by the crawler, untransformed
self.found_paths: set[PurePath] = set() self.found_paths: Set[PurePath] = set()
# Files reserved for metadata files (e. g. the report file or cookies) # Files reserved for metadata files (e. g. the report file or cookies)
# that can't be overwritten by user transforms and won't be cleaned up # that can't be overwritten by user transforms and won't be cleaned up
# at the end. # at the end.
self.reserved_files: set[PurePath] = set() self.reserved_files: Set[PurePath] = set()
# Files found by the crawler, transformed. Only includes files that # Files found by the crawler, transformed. Only includes files that
# were downloaded (or a download was attempted) # were downloaded (or a download was attempted)
self.known_files: set[PurePath] = set() self.known_files: Set[PurePath] = set()
self.added_files: set[PurePath] = set() self.added_files: Set[PurePath] = set()
self.changed_files: set[PurePath] = set() self.changed_files: Set[PurePath] = set()
self.deleted_files: set[PurePath] = set() self.deleted_files: Set[PurePath] = set()
# Files that should have been deleted by the cleanup but weren't # Files that should have been deleted by the cleanup but weren't
self.not_deleted_files: set[PurePath] = set() self.not_deleted_files: Set[PurePath] = set()
# Custom crawler-specific data # Custom crawler-specific data
self.custom: dict[str, Any] = dict() self.custom: Dict[str, Any] = dict()
# Encountered errors and warnings # Encountered errors and warnings
self.encountered_warnings: list[str] = [] self.encountered_warnings: List[str] = []
self.encountered_errors: list[str] = [] self.encountered_errors: List[str] = []
@staticmethod @staticmethod
def _get_list_of_strs(data: dict[str, Any], key: str) -> list[str]: def _get_list_of_strs(data: Dict[str, Any], key: str) -> List[str]:
result: Any = data.get(key, []) result: Any = data.get(key, [])
if not isinstance(result, list): if not isinstance(result, list):
@ -80,8 +80,8 @@ class Report:
return result return result
@staticmethod @staticmethod
def _get_str_dictionary(data: dict[str, Any], key: str) -> dict[str, Any]: def _get_str_dictionary(data: Dict[str, Any], key: str) -> Dict[str, Any]:
result: dict[str, Any] = data.get(key, {}) result: Dict[str, Any] = data.get(key, {})
if not isinstance(result, dict): if not isinstance(result, dict):
raise ReportLoadError(f"Incorrect format: {key!r} is not a dictionary") raise ReportLoadError(f"Incorrect format: {key!r} is not a dictionary")
@ -170,7 +170,7 @@ class Report:
self.known_files.add(path) self.known_files.add(path)
@property @property
def marked(self) -> set[PurePath]: def marked(self) -> Set[PurePath]:
return self.known_files | self.reserved_files return self.known_files | self.reserved_files
def is_marked(self, path: PurePath) -> bool: def is_marked(self, path: PurePath) -> bool:

View file

@ -1,12 +1,10 @@
import ast import ast
import contextlib
import re import re
from abc import ABC, abstractmethod from abc import ABC, abstractmethod
from collections.abc import Callable, Sequence
from dataclasses import dataclass from dataclasses import dataclass
from enum import Enum from enum import Enum
from pathlib import PurePath from pathlib import PurePath
from typing import Optional, TypeVar from typing import Callable, Dict, List, Optional, Sequence, TypeVar, Union
from .logging import log from .logging import log
from .utils import fmt_path, str_path from .utils import fmt_path, str_path
@ -25,7 +23,7 @@ class Empty:
pass pass
RightSide = str | Ignore | Empty RightSide = Union[str, Ignore, Empty]
@dataclass @dataclass
@ -37,7 +35,7 @@ class Ignored:
pass pass
TransformResult = Transformed | Ignored | None TransformResult = Optional[Union[Transformed, Ignored]]
@dataclass @dataclass
@ -49,7 +47,7 @@ class Rule:
right: RightSide right: RightSide
right_index: int right_index: int
def right_result(self, path: PurePath) -> str | Transformed | Ignored: def right_result(self, path: PurePath) -> Union[str, Transformed, Ignored]:
if isinstance(self.right, str): if isinstance(self.right, str):
return self.right return self.right
elif isinstance(self.right, Ignore): elif isinstance(self.right, Ignore):
@ -95,20 +93,24 @@ class ExactReTf(Transformation):
# since elements of "match.groups()" can be None, mypy is wrong. # since elements of "match.groups()" can be None, mypy is wrong.
groups: Sequence[Optional[str]] = [match[0]] + list(match.groups()) groups: Sequence[Optional[str]] = [match[0]] + list(match.groups())
locals_dir: dict[str, str | int | float] = {} locals_dir: Dict[str, Union[str, int, float]] = {}
for i, group in enumerate(groups): for i, group in enumerate(groups):
if group is None: if group is None:
continue continue
locals_dir[f"g{i}"] = group locals_dir[f"g{i}"] = group
with contextlib.suppress(ValueError): try:
locals_dir[f"i{i}"] = int(group) locals_dir[f"i{i}"] = int(group)
except ValueError:
pass
with contextlib.suppress(ValueError): try:
locals_dir[f"f{i}"] = float(group) locals_dir[f"f{i}"] = float(group)
except ValueError:
pass
named_groups: dict[str, str] = match.groupdict() named_groups: Dict[str, str] = match.groupdict()
for name, capture in named_groups.items(): for name, capture in named_groups.items():
locals_dir[name] = capture locals_dir[name] = capture
@ -206,7 +208,7 @@ class Line:
@property @property
def rest(self) -> str: def rest(self) -> str:
return self.line[self.index :] return self.line[self.index:]
def peek(self, amount: int = 1) -> str: def peek(self, amount: int = 1) -> str:
return self.rest[:amount] return self.rest[:amount]
@ -226,7 +228,7 @@ class Line:
self.expect(string) self.expect(string)
return value return value
def one_of(self, parsers: list[Callable[[], T]], description: str) -> T: def one_of(self, parsers: List[Callable[[], T]], description: str) -> T:
for parser in parsers: for parser in parsers:
index = self.index index = self.index
try: try:
@ -313,7 +315,7 @@ def parse_left(line: Line) -> str:
return parse_str(line) return parse_str(line)
def parse_right(line: Line) -> str | Ignore: def parse_right(line: Line) -> Union[str, Ignore]:
c = line.peek() c = line.peek()
if c in QUOTATION_MARKS: if c in QUOTATION_MARKS:
return parse_quoted_str(line) return parse_quoted_str(line)
@ -325,27 +327,21 @@ def parse_right(line: Line) -> str | Ignore:
def parse_arrow_name(line: Line) -> str: def parse_arrow_name(line: Line) -> str:
return line.one_of( return line.one_of([
[
lambda: line.expect("exact-re"), lambda: line.expect("exact-re"),
lambda: line.expect("exact"), lambda: line.expect("exact"),
lambda: line.expect("name-re"), lambda: line.expect("name-re"),
lambda: line.expect("name"), lambda: line.expect("name"),
lambda: line.expect("re"), lambda: line.expect("re"),
lambda: line.expect(""), lambda: line.expect(""),
], ], "Expected arrow name")
"Expected arrow name",
)
def parse_arrow_head(line: Line) -> ArrowHead: def parse_arrow_head(line: Line) -> ArrowHead:
return line.one_of( return line.one_of([
[
lambda: line.expect_with(">>", ArrowHead.SEQUENCE), lambda: line.expect_with(">>", ArrowHead.SEQUENCE),
lambda: line.expect_with(">", ArrowHead.NORMAL), lambda: line.expect_with(">", ArrowHead.NORMAL),
], ], "Expected arrow head")
"Expected arrow head",
)
def parse_eol(line: Line) -> None: def parse_eol(line: Line) -> None:
@ -417,12 +413,12 @@ class Transformer:
def transform(self, path: PurePath) -> Optional[PurePath]: def transform(self, path: PurePath) -> Optional[PurePath]:
for i, (line, tf) in enumerate(self._tfs): for i, (line, tf) in enumerate(self._tfs):
log.explain(f"Testing rule {i + 1}: {line}") log.explain(f"Testing rule {i+1}: {line}")
try: try:
result = tf.transform(path) result = tf.transform(path)
except Exception as e: except Exception as e:
log.warn(f"Error while testing rule {i + 1}: {line}") log.warn(f"Error while testing rule {i+1}: {line}")
log.warn_contd(str(e)) log.warn_contd(str(e))
continue continue

View file

@ -3,11 +3,10 @@ import getpass
import sys import sys
import threading import threading
from abc import ABC, abstractmethod from abc import ABC, abstractmethod
from collections.abc import Callable
from contextlib import AsyncExitStack from contextlib import AsyncExitStack
from pathlib import Path, PurePath from pathlib import Path, PurePath
from types import TracebackType from types import TracebackType
from typing import Any, Generic, Optional, TypeVar from typing import Any, Callable, Dict, Generic, Optional, Type, TypeVar
from urllib.parse import parse_qs, urlencode, urlsplit, urlunsplit from urllib.parse import parse_qs, urlencode, urlsplit, urlunsplit
import bs4 import bs4
@ -80,7 +79,7 @@ def url_set_query_param(url: str, param: str, value: str) -> str:
return urlunsplit((scheme, netloc, path, new_query_string, fragment)) return urlunsplit((scheme, netloc, path, new_query_string, fragment))
def url_set_query_params(url: str, params: dict[str, str]) -> str: def url_set_query_params(url: str, params: Dict[str, str]) -> str:
""" """
Sets multiple query parameters in an url, overwriting existing ones. Sets multiple query parameters in an url, overwriting existing ones.
""" """
@ -106,10 +105,6 @@ def fmt_real_path(path: Path) -> str:
return repr(str(path.absolute())) return repr(str(path.absolute()))
def sanitize_path_name(name: str) -> str:
return name.replace("/", "-").replace("\\", "-").strip()
class ReusableAsyncContextManager(ABC, Generic[T]): class ReusableAsyncContextManager(ABC, Generic[T]):
def __init__(self) -> None: def __init__(self) -> None:
self._active = False self._active = False
@ -129,15 +124,15 @@ class ReusableAsyncContextManager(ABC, Generic[T]):
# See https://stackoverflow.com/a/13075071 # See https://stackoverflow.com/a/13075071
try: try:
result: T = await self._on_aenter() result: T = await self._on_aenter()
return result except: # noqa: E722 do not use bare 'except'
except:
if not await self.__aexit__(*sys.exc_info()): if not await self.__aexit__(*sys.exc_info()):
raise raise
raise
return result
async def __aexit__( async def __aexit__(
self, self,
exc_type: Optional[type[BaseException]], exc_type: Optional[Type[BaseException]],
exc_value: Optional[BaseException], exc_value: Optional[BaseException],
traceback: Optional[TracebackType], traceback: Optional[TracebackType],
) -> Optional[bool]: ) -> Optional[bool]:

View file

@ -1,2 +1,2 @@
NAME = "PFERD" NAME = "PFERD"
VERSION = "3.8.3" VERSION = "3.8.0"

8
flake.lock generated
View file

@ -2,16 +2,16 @@
"nodes": { "nodes": {
"nixpkgs": { "nixpkgs": {
"locked": { "locked": {
"lastModified": 1760725957, "lastModified": 1744440957,
"narHash": "sha256-tdoIhL/NlER290HfSjOkgi4jfmjeqmqrzgnmiMtGepE=", "narHash": "sha256-FHlSkNqFmPxPJvy+6fNLaNeWnF1lZSgqVCl/eWaJRc4=",
"owner": "NixOS", "owner": "NixOS",
"repo": "nixpkgs", "repo": "nixpkgs",
"rev": "81b927b14b7b3988334d5282ef9cba802e193fe1", "rev": "26d499fc9f1d567283d5d56fcf367edd815dba1d",
"type": "github" "type": "github"
}, },
"original": { "original": {
"owner": "NixOS", "owner": "NixOS",
"ref": "nixos-25.05", "ref": "nixos-24.11",
"repo": "nixpkgs", "repo": "nixpkgs",
"type": "github" "type": "github"
} }

View file

@ -2,7 +2,7 @@
description = "Tool for downloading course-related files from ILIAS"; description = "Tool for downloading course-related files from ILIAS";
inputs = { inputs = {
nixpkgs.url = "github:NixOS/nixpkgs/nixos-25.05"; nixpkgs.url = "github:NixOS/nixpkgs/nixos-24.11";
}; };
outputs = { self, nixpkgs }: outputs = { self, nixpkgs }:

View file

@ -20,33 +20,23 @@ pferd = "PFERD.__main__:main"
[tool.setuptools.dynamic] [tool.setuptools.dynamic]
version = {attr = "PFERD.version.VERSION"} version = {attr = "PFERD.version.VERSION"}
[tool.ruff] [tool.flake8]
line-length = 110 max-line-length = 110
[tool.ruff.lint] [tool.isort]
select = [ line_length = 110
# pycodestyle
"E",
# Pyflakes
"F",
# pyupgrade
"UP",
# flake8-bugbear
"B",
# flake8-simplify
"SIM",
# isort
"I",
]
ignore = [
"UP045",
"SIM114",
"B023"
]
[dependency-groups] [tool.autopep8]
dev = [ max_line_length = 110
"pyinstaller>=6.16.0", in-place = true
"pyright>=1.1.406", recursive = true
"ruff>=0.14.1",
] [tool.mypy]
disallow_any_generics = true
disallow_untyped_defs = true
disallow_incomplete_defs = true
no_implicit_optional = true
warn_unused_ignores = true
warn_unreachable = true
show_error_context = true
ignore_missing_imports = true

View file

@ -2,4 +2,4 @@
set -e set -e
uv run pyinstaller --onefile pferd.py pyinstaller --onefile pferd.py

View file

@ -2,5 +2,5 @@
set -e set -e
uv run pyright . mypy .
uv run ruff check flake8 PFERD

View file

@ -2,4 +2,5 @@
set -e set -e
uv run ruff format autopep8 .
isort .

1056
uv.lock generated

File diff suppressed because it is too large Load diff