mirror of
https://github.com/Garmelon/PFERD.git
synced 2023-12-21 10:23:01 +01:00
Compare commits
50 Commits
Author | SHA1 | Date | |
---|---|---|---|
a0ae9aee27 | |||
1486a63854 | |||
733e1ae136 | |||
4ac51048c1 | |||
f2aba970fd | |||
9c4759103a | |||
316b9d7bf4 | |||
6f30adcd22 | |||
6f78fef604 | |||
f830b42a36 | |||
ef343dec7c | |||
0da2fafcd8 | |||
f4abe3197c | |||
38d4f5b4c9 | |||
9ea03bda3e | |||
07de5bea8b | |||
f0d572c110 | |||
076067e22d | |||
ebb6e63c5c | |||
0c3f35a2d2 | |||
521890ae78 | |||
3f7c73df80 | |||
43100f69d5 | |||
d73c778b0a | |||
73c3eb0984 | |||
a519cbe05d | |||
b3ad9783c4 | |||
c1ccb6c53e | |||
51a713fa04 | |||
74ea039458 | |||
aaa6a2b6a4 | |||
e32a49480b | |||
be65051f9d | |||
3387bc5f20 | |||
3f0ae729d6 | |||
8e8c1c031a | |||
55678d7fee | |||
a57ee8b96b | |||
e367da925e | |||
77a109bb7e | |||
a3e1864a26 | |||
41cbcc509c | |||
77874b432b | |||
5c4c785e60 | |||
2aed4f6d1f | |||
34152fbe54 | |||
4047fe78f3 | |||
c28347122e | |||
5b38ab8cf1 | |||
bb25d32f03 |
74
.github/workflows/package.yml
vendored
Normal file
74
.github/workflows/package.yml
vendored
Normal file
@ -0,0 +1,74 @@
|
||||
name: Package Application with Pyinstaller
|
||||
|
||||
on:
|
||||
push:
|
||||
branches:
|
||||
- "*"
|
||||
tags:
|
||||
- "v*"
|
||||
|
||||
jobs:
|
||||
build:
|
||||
|
||||
runs-on: ${{ matrix.os }}
|
||||
strategy:
|
||||
matrix:
|
||||
os: [ubuntu-latest, windows-latest, macos-latest]
|
||||
|
||||
steps:
|
||||
- uses: actions/checkout@v2
|
||||
|
||||
- uses: actions/setup-python@v2
|
||||
with:
|
||||
python-version: '3.x'
|
||||
|
||||
- name: "Install dependencies"
|
||||
run: "pip install setuptools pyinstaller rich requests beautifulsoup4 -f --upgrade"
|
||||
|
||||
- name: "Install sync_url.py"
|
||||
run: "pyinstaller sync_url.py -F"
|
||||
|
||||
- name: "Move artifact"
|
||||
run: "mv dist/sync_url* dist/sync_url-${{ matrix.os }}"
|
||||
|
||||
- uses: actions/upload-artifact@v2
|
||||
with:
|
||||
name: "Pferd Sync URL"
|
||||
path: "dist/sync_url*"
|
||||
|
||||
release:
|
||||
name: Release
|
||||
|
||||
needs: [build]
|
||||
runs-on: ubuntu-latest
|
||||
if: startsWith(github.ref, 'refs/tags/')
|
||||
|
||||
env:
|
||||
GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
|
||||
|
||||
steps:
|
||||
- name: "Checkout"
|
||||
uses: actions/checkout@v2
|
||||
|
||||
- name: "Download artifacts"
|
||||
uses: actions/download-artifact@v2
|
||||
with:
|
||||
name: "Pferd Sync URL"
|
||||
|
||||
- name: "look at folder structure"
|
||||
run: "ls -lah"
|
||||
|
||||
- name: "Rename releases"
|
||||
run: "mv sync_url-macos-latest pferd_sync_url_mac && mv sync_url-ubuntu-latest pferd_sync_url_linux && mv sync_url-windows-latest pferd_sync_url.exe"
|
||||
|
||||
- name: "Create release"
|
||||
uses: softprops/action-gh-release@v1
|
||||
|
||||
- name: "Upload release artifacts"
|
||||
uses: softprops/action-gh-release@v1
|
||||
with:
|
||||
body: "Download the correct sync_url for your platform and run it in the terminal or CMD. You might need to make it executable on Linux/Mac with `chmod +x <file>`. Also please enclose the *url you pass to the program in double quotes* or your shell might silently screw it up!"
|
||||
files: |
|
||||
pferd_sync_url_mac
|
||||
pferd_sync_url_linux
|
||||
pferd_sync_url.exe
|
4
.gitignore
vendored
4
.gitignore
vendored
@ -8,3 +8,7 @@ build/
|
||||
.env
|
||||
.vscode
|
||||
ilias_cookies.txt
|
||||
|
||||
# PyInstaller
|
||||
sync_url.spec
|
||||
dist/
|
||||
|
18
LICENSE
Normal file
18
LICENSE
Normal file
@ -0,0 +1,18 @@
|
||||
Copyright 2019-2020 Garmelon, I-Al-Istannen, danstooamerican, pavelzw
|
||||
|
||||
Permission is hereby granted, free of charge, to any person obtaining a copy of
|
||||
this software and associated documentation files (the "Software"), to deal in
|
||||
the Software without restriction, including without limitation the rights to
|
||||
use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of
|
||||
the Software, and to permit persons to whom the Software is furnished to do so,
|
||||
subject to the following conditions:
|
||||
|
||||
The above copyright notice and this permission notice shall be included in all
|
||||
copies or substantial portions of the Software.
|
||||
|
||||
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
|
||||
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS
|
||||
FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR
|
||||
COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER
|
||||
IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN
|
||||
CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
|
@ -22,7 +22,7 @@ class CookieJar:
|
||||
if cookie_file is None:
|
||||
self._cookies = LWPCookieJar()
|
||||
else:
|
||||
self._cookies = LWPCookieJar(cookie_file)
|
||||
self._cookies = LWPCookieJar(str(cookie_file.resolve()))
|
||||
|
||||
@property
|
||||
def cookies(self) -> LWPCookieJar:
|
||||
|
@ -40,9 +40,9 @@ class DownloadSummary:
|
||||
"""
|
||||
Merges ourselves with the passed summary. Modifies this object, but not the passed one.
|
||||
"""
|
||||
self._new_files += summary.new_files
|
||||
self._modified_files += summary.modified_files
|
||||
self._deleted_files += summary.deleted_files
|
||||
self._new_files = list(set(self._new_files + summary.new_files))
|
||||
self._modified_files = list(set(self._modified_files + summary.modified_files))
|
||||
self._deleted_files = list(set(self._deleted_files + summary.deleted_files))
|
||||
|
||||
def add_deleted_file(self, path: Path) -> None:
|
||||
"""
|
||||
|
@ -67,7 +67,7 @@ class KitShibbolethAuthenticator(IliasAuthenticator):
|
||||
while not self._login_successful(soup):
|
||||
# Searching the form here so that this fails before asking for
|
||||
# credentials rather than after asking.
|
||||
form = soup.find("form", {"class": "form2", "method": "post"})
|
||||
form = soup.find("form", {"class": "full content", "method": "post"})
|
||||
action = form["action"]
|
||||
|
||||
# Equivalent: Enter credentials in
|
||||
|
@ -38,6 +38,12 @@ class IliasElementType(Enum):
|
||||
FORUM = "FORUM"
|
||||
EXTERNAL_LINK = "EXTERNAL_LINK"
|
||||
|
||||
def is_folder(self) -> bool:
|
||||
"""
|
||||
Returns whether this type is some kind of folder.
|
||||
"""
|
||||
return "FOLDER" in str(self.name)
|
||||
|
||||
|
||||
IliasDirectoryFilter = Callable[[Path, IliasElementType], bool]
|
||||
|
||||
@ -110,6 +116,16 @@ class IliasCrawler:
|
||||
|
||||
return urlunsplit((scheme, netloc, path, new_query_string, fragment))
|
||||
|
||||
def recursive_crawl_url(self, url: str) -> List[IliasDownloadInfo]:
|
||||
"""
|
||||
Crawls a given url *and all reachable elements in it*.
|
||||
|
||||
Args:
|
||||
url {str} -- the *full* url to crawl
|
||||
"""
|
||||
start_entries: List[IliasCrawlerEntry] = self._crawl_folder(Path(""), url)
|
||||
return self._iterate_entries_to_download_infos(start_entries)
|
||||
|
||||
def crawl_course(self, course_id: str) -> List[IliasDownloadInfo]:
|
||||
"""
|
||||
Starts the crawl process for a course, yielding a list of elements to (potentially)
|
||||
@ -128,7 +144,7 @@ class IliasCrawler:
|
||||
|
||||
if not self._is_course_id_valid(root_url, course_id):
|
||||
raise FatalException(
|
||||
"Invalid course id? The URL the server returned did not contain my id."
|
||||
"Invalid course id? I didn't find anything looking like a course!"
|
||||
)
|
||||
|
||||
# And treat it as a folder
|
||||
@ -137,7 +153,34 @@ class IliasCrawler:
|
||||
|
||||
def _is_course_id_valid(self, root_url: str, course_id: str) -> bool:
|
||||
response: requests.Response = self._session.get(root_url)
|
||||
return course_id in response.url
|
||||
# We were redirected ==> Non-existant ID
|
||||
if course_id not in response.url:
|
||||
return False
|
||||
|
||||
link_element: bs4.Tag = self._get_page(root_url, {}).find(id="current_perma_link")
|
||||
if not link_element:
|
||||
return False
|
||||
# It wasn't a course but a category list, forum, etc.
|
||||
return "crs_" in link_element.get("value")
|
||||
|
||||
def find_course_name(self, course_id: str) -> Optional[str]:
|
||||
"""
|
||||
Returns the name of a given course. None if it is not a valid course
|
||||
or it could not be found.
|
||||
"""
|
||||
course_url = self._url_set_query_param(
|
||||
self._base_url + "/goto.php", "target", f"crs_{course_id}"
|
||||
)
|
||||
return self.find_element_name(course_url)
|
||||
|
||||
def find_element_name(self, url: str) -> Optional[str]:
|
||||
"""
|
||||
Returns the name of the element at the given URL, if it can find one.
|
||||
"""
|
||||
focus_element: bs4.Tag = self._get_page(url, {}).find(id="il_mhead_t_focus")
|
||||
if not focus_element:
|
||||
return None
|
||||
return focus_element.text
|
||||
|
||||
def crawl_personal_desktop(self) -> List[IliasDownloadInfo]:
|
||||
"""
|
||||
@ -167,7 +210,7 @@ class IliasCrawler:
|
||||
PRETTY.not_searching(entry.path, "forum")
|
||||
continue
|
||||
|
||||
if not self.dir_filter(entry.path, entry.entry_type):
|
||||
if entry.entry_type.is_folder() and not self.dir_filter(entry.path, entry.entry_type):
|
||||
PRETTY.not_searching(entry.path, "user filter")
|
||||
continue
|
||||
|
||||
@ -202,6 +245,15 @@ class IliasCrawler:
|
||||
"""
|
||||
soup = self._get_page(url, {})
|
||||
|
||||
if soup.find(id="headerimage"):
|
||||
element: bs4.Tag = soup.find(id="headerimage")
|
||||
if "opencast" in element.attrs["src"].lower():
|
||||
PRETTY.warning(f"Switched to crawling a video at {folder_path}")
|
||||
if not self.dir_filter(folder_path, IliasElementType.VIDEO_FOLDER):
|
||||
PRETTY.not_searching(folder_path, "user filter")
|
||||
return []
|
||||
return self._crawl_video_directory(folder_path, url)
|
||||
|
||||
result: List[IliasCrawlerEntry] = []
|
||||
|
||||
# Fetch all links and throw them to the general interpreter
|
||||
@ -541,10 +593,17 @@ class IliasCrawler:
|
||||
|
||||
return results
|
||||
|
||||
def _get_page(self, url: str, params: Dict[str, Any]) -> bs4.BeautifulSoup:
|
||||
def _get_page(self, url: str, params: Dict[str, Any],
|
||||
retry_count: int = 0) -> bs4.BeautifulSoup:
|
||||
"""
|
||||
Fetches a page from ILIAS, authenticating when needed.
|
||||
"""
|
||||
|
||||
if retry_count >= 4:
|
||||
raise FatalException("Could not get a proper page after 4 tries. "
|
||||
"Maybe your URL is wrong, authentication fails continuously, "
|
||||
"your ILIAS connection is spotty or ILIAS is not well.")
|
||||
|
||||
LOGGER.debug("Fetching %r", url)
|
||||
|
||||
response = self._session.get(url, params=params)
|
||||
@ -565,7 +624,7 @@ class IliasCrawler:
|
||||
|
||||
self._authenticator.authenticate(self._session)
|
||||
|
||||
return self._get_page(url, params)
|
||||
return self._get_page(url, params, retry_count + 1)
|
||||
|
||||
@staticmethod
|
||||
def _is_logged_in(soup: bs4.BeautifulSoup) -> bool:
|
||||
|
@ -2,6 +2,8 @@
|
||||
|
||||
import datetime
|
||||
import logging
|
||||
import math
|
||||
import os
|
||||
from pathlib import Path, PurePath
|
||||
from typing import Callable, List, Optional, Union
|
||||
|
||||
@ -82,9 +84,13 @@ class IliasDownloader:
|
||||
session: requests.Session,
|
||||
authenticator: IliasAuthenticator,
|
||||
strategy: IliasDownloadStrategy,
|
||||
timeout: int = 5
|
||||
):
|
||||
"""
|
||||
Create a new IliasDownloader.
|
||||
|
||||
The timeout applies to the download request only, as bwcloud uses IPv6
|
||||
and requests has a problem with that: https://github.com/psf/requests/issues/5522
|
||||
"""
|
||||
|
||||
self._tmp_dir = tmp_dir
|
||||
@ -92,6 +98,7 @@ class IliasDownloader:
|
||||
self._session = session
|
||||
self._authenticator = authenticator
|
||||
self._strategy = strategy
|
||||
self._timeout = timeout
|
||||
|
||||
def download_all(self, infos: List[IliasDownloadInfo]) -> None:
|
||||
"""
|
||||
@ -119,7 +126,15 @@ class IliasDownloader:
|
||||
LOGGER.info("Retrying download: %r", info)
|
||||
self._authenticator.authenticate(self._session)
|
||||
|
||||
self._organizer.accept_file(tmp_file, info.path)
|
||||
dst_path = self._organizer.accept_file(tmp_file, info.path)
|
||||
if dst_path and info.modification_date:
|
||||
os.utime(
|
||||
dst_path,
|
||||
times=(
|
||||
math.ceil(info.modification_date.timestamp()),
|
||||
math.ceil(info.modification_date.timestamp())
|
||||
)
|
||||
)
|
||||
|
||||
def _try_download(self, info: IliasDownloadInfo, target: Path) -> bool:
|
||||
url = info.url()
|
||||
@ -127,10 +142,11 @@ class IliasDownloader:
|
||||
PRETTY.warning(f"Could not download {str(info.path)!r} as I got no URL :/")
|
||||
return True
|
||||
|
||||
with self._session.get(url, stream=True) as response:
|
||||
with self._session.get(url, stream=True, timeout=self._timeout) as response:
|
||||
content_type = response.headers["content-type"]
|
||||
has_content_disposition = "content-disposition" in response.headers
|
||||
|
||||
if content_type.startswith("text/html"):
|
||||
if content_type.startswith("text/html") and not has_content_disposition:
|
||||
if self._is_logged_in(soupify(response)):
|
||||
raise ContentTypeException("Attempting to download a web page, not a file")
|
||||
|
||||
|
151
PFERD/ipd.py
Normal file
151
PFERD/ipd.py
Normal file
@ -0,0 +1,151 @@
|
||||
"""
|
||||
Utility functions and a scraper/downloader for the IPD pages.
|
||||
"""
|
||||
import datetime
|
||||
import logging
|
||||
import math
|
||||
import os
|
||||
from dataclasses import dataclass
|
||||
from pathlib import Path
|
||||
from typing import Callable, List, Optional
|
||||
from urllib.parse import urljoin
|
||||
|
||||
import bs4
|
||||
import requests
|
||||
|
||||
from PFERD.errors import FatalException
|
||||
from PFERD.utils import soupify
|
||||
|
||||
from .logging import PrettyLogger
|
||||
from .organizer import Organizer
|
||||
from .tmp_dir import TmpDir
|
||||
from .transform import Transformable
|
||||
from .utils import stream_to_path
|
||||
|
||||
LOGGER = logging.getLogger(__name__)
|
||||
PRETTY = PrettyLogger(LOGGER)
|
||||
|
||||
|
||||
@dataclass
|
||||
class IpdDownloadInfo(Transformable):
|
||||
"""
|
||||
Information about an ipd entry.
|
||||
"""
|
||||
url: str
|
||||
modification_date: Optional[datetime.datetime]
|
||||
|
||||
|
||||
IpdDownloadStrategy = Callable[[Organizer, IpdDownloadInfo], bool]
|
||||
|
||||
|
||||
def ipd_download_new_or_modified(organizer: Organizer, info: IpdDownloadInfo) -> bool:
|
||||
"""
|
||||
Accepts new files or files with a more recent modification date.
|
||||
"""
|
||||
resolved_file = organizer.resolve(info.path)
|
||||
if not resolved_file.exists():
|
||||
return True
|
||||
if not info.modification_date:
|
||||
PRETTY.ignored_file(info.path, "could not find modification time, file exists")
|
||||
return False
|
||||
|
||||
resolved_mod_time_seconds = resolved_file.stat().st_mtime
|
||||
|
||||
# Download if the info is newer
|
||||
if info.modification_date.timestamp() > resolved_mod_time_seconds:
|
||||
return True
|
||||
|
||||
PRETTY.ignored_file(info.path, "local file has newer or equal modification time")
|
||||
return False
|
||||
|
||||
|
||||
class IpdCrawler:
|
||||
# pylint: disable=too-few-public-methods
|
||||
"""
|
||||
A crawler for IPD pages.
|
||||
"""
|
||||
|
||||
def __init__(self, base_url: str):
|
||||
self._base_url = base_url
|
||||
|
||||
def _abs_url_from_link(self, link_tag: bs4.Tag) -> str:
|
||||
"""
|
||||
Create an absolute url from an <a> tag.
|
||||
"""
|
||||
return urljoin(self._base_url, link_tag.get("href"))
|
||||
|
||||
def crawl(self) -> List[IpdDownloadInfo]:
|
||||
"""
|
||||
Crawls the playlist given in the constructor.
|
||||
"""
|
||||
page = soupify(requests.get(self._base_url))
|
||||
|
||||
items: List[IpdDownloadInfo] = []
|
||||
|
||||
for link in page.findAll(name="a", attrs={"href": lambda x: x and x.endswith("pdf")}):
|
||||
href: str = link.attrs.get("href")
|
||||
name = href.split("/")[-1]
|
||||
|
||||
modification_date: Optional[datetime.datetime] = None
|
||||
try:
|
||||
enclosing_row: bs4.Tag = link.findParent(name="tr")
|
||||
if enclosing_row:
|
||||
date_text = enclosing_row.find(name="td").text
|
||||
modification_date = datetime.datetime.strptime(date_text, "%d.%m.%Y")
|
||||
except ValueError:
|
||||
modification_date = None
|
||||
|
||||
items.append(IpdDownloadInfo(
|
||||
Path(name),
|
||||
url=self._abs_url_from_link(link),
|
||||
modification_date=modification_date
|
||||
))
|
||||
|
||||
return items
|
||||
|
||||
|
||||
class IpdDownloader:
|
||||
"""
|
||||
A downloader for ipd files.
|
||||
"""
|
||||
|
||||
def __init__(self, tmp_dir: TmpDir, organizer: Organizer, strategy: IpdDownloadStrategy):
|
||||
self._tmp_dir = tmp_dir
|
||||
self._organizer = organizer
|
||||
self._strategy = strategy
|
||||
self._session = requests.session()
|
||||
|
||||
def download_all(self, infos: List[IpdDownloadInfo]) -> None:
|
||||
"""
|
||||
Download multiple files one after the other.
|
||||
"""
|
||||
for info in infos:
|
||||
self.download(info)
|
||||
|
||||
def download(self, info: IpdDownloadInfo) -> None:
|
||||
"""
|
||||
Download a single file.
|
||||
"""
|
||||
if not self._strategy(self._organizer, info):
|
||||
self._organizer.mark(info.path)
|
||||
return
|
||||
|
||||
with self._session.get(info.url, stream=True) as response:
|
||||
if response.status_code == 200:
|
||||
tmp_file = self._tmp_dir.new_path()
|
||||
stream_to_path(response, tmp_file, info.path.name)
|
||||
dst_path = self._organizer.accept_file(tmp_file, info.path)
|
||||
|
||||
if dst_path and info.modification_date:
|
||||
os.utime(
|
||||
dst_path,
|
||||
times=(
|
||||
math.ceil(info.modification_date.timestamp()),
|
||||
math.ceil(info.modification_date.timestamp())
|
||||
)
|
||||
)
|
||||
|
||||
elif response.status_code == 403:
|
||||
raise FatalException("Received 403. Are you not using the KIT VPN?")
|
||||
else:
|
||||
PRETTY.warning(f"Could not download file, got response {response.status_code}")
|
@ -5,9 +5,10 @@ A organizer is bound to a single directory.
|
||||
|
||||
import filecmp
|
||||
import logging
|
||||
import os
|
||||
import shutil
|
||||
from pathlib import Path, PurePath
|
||||
from typing import List, Set
|
||||
from typing import List, Optional, Set
|
||||
|
||||
from .download_summary import DownloadSummary
|
||||
from .location import Location
|
||||
@ -35,10 +36,26 @@ class Organizer(Location):
|
||||
|
||||
self.download_summary = DownloadSummary()
|
||||
|
||||
def accept_file(self, src: Path, dst: PurePath) -> None:
|
||||
"""Move a file to this organizer and mark it."""
|
||||
src_absolute = src.resolve()
|
||||
dst_absolute = self.resolve(dst)
|
||||
def accept_file(self, src: Path, dst: PurePath) -> Optional[Path]:
|
||||
"""
|
||||
Move a file to this organizer and mark it.
|
||||
|
||||
Returns the path the file was moved to, to allow the caller to adjust the metadata.
|
||||
As you might still need to adjust the metadata when the file was identical
|
||||
(e.g. update the timestamp), the path is also returned in this case.
|
||||
In all other cases (ignored, not overwritten, etc.) this method returns None.
|
||||
"""
|
||||
# Windows limits the path length to 260 for *some* historical reason
|
||||
# If you want longer paths, you will have to add the "\\?\" prefix in front of
|
||||
# your path...
|
||||
# See:
|
||||
# https://docs.microsoft.com/en-us/windows/win32/fileio/naming-a-file#maximum-path-length-limitation
|
||||
if os.name == 'nt':
|
||||
src_absolute = Path("\\\\?\\" + str(src.resolve()))
|
||||
dst_absolute = Path("\\\\?\\" + str(self.resolve(dst)))
|
||||
else:
|
||||
src_absolute = src.resolve()
|
||||
dst_absolute = self.resolve(dst)
|
||||
|
||||
if not src_absolute.exists():
|
||||
raise FileAcceptException("Source file does not exist")
|
||||
@ -52,7 +69,7 @@ class Organizer(Location):
|
||||
PRETTY.warning(f"File {str(dst_absolute)!r} was already written!")
|
||||
if not prompt_yes_no(f"Overwrite file?", default=False):
|
||||
PRETTY.ignored_file(dst_absolute, "file was written previously")
|
||||
return
|
||||
return None
|
||||
|
||||
# Destination file is directory
|
||||
if dst_absolute.exists() and dst_absolute.is_dir():
|
||||
@ -60,7 +77,7 @@ class Organizer(Location):
|
||||
shutil.rmtree(dst_absolute)
|
||||
else:
|
||||
PRETTY.warning(f"Could not add file {str(dst_absolute)!r}")
|
||||
return
|
||||
return None
|
||||
|
||||
# Destination file exists
|
||||
if dst_absolute.exists() and dst_absolute.is_file():
|
||||
@ -68,9 +85,7 @@ class Organizer(Location):
|
||||
# Bail out, nothing more to do
|
||||
PRETTY.ignored_file(dst_absolute, "same file contents")
|
||||
self.mark(dst)
|
||||
# Touch it to update the timestamp
|
||||
dst_absolute.touch()
|
||||
return
|
||||
return dst_absolute
|
||||
|
||||
self.download_summary.add_modified_file(dst_absolute)
|
||||
PRETTY.modified_file(dst_absolute)
|
||||
@ -87,6 +102,8 @@ class Organizer(Location):
|
||||
|
||||
self.mark(dst)
|
||||
|
||||
return dst_absolute
|
||||
|
||||
def mark(self, path: PurePath) -> None:
|
||||
"""Mark a file as used so it will not get cleaned up."""
|
||||
absolute_path = self.resolve(path)
|
||||
@ -107,6 +124,8 @@ class Organizer(Location):
|
||||
self._cleanup(self.path)
|
||||
|
||||
def _cleanup(self, start_dir: Path) -> None:
|
||||
if not start_dir.exists():
|
||||
return
|
||||
paths: List[Path] = list(start_dir.iterdir())
|
||||
|
||||
# Recursively clean paths
|
||||
|
143
PFERD/pferd.py
143
PFERD/pferd.py
@ -14,6 +14,8 @@ from .errors import FatalException, swallow_and_print_errors
|
||||
from .ilias import (IliasAuthenticator, IliasCrawler, IliasDirectoryFilter,
|
||||
IliasDownloader, IliasDownloadInfo, IliasDownloadStrategy,
|
||||
KitShibbolethAuthenticator, download_modified_or_new)
|
||||
from .ipd import (IpdCrawler, IpdDownloader, IpdDownloadInfo,
|
||||
IpdDownloadStrategy, ipd_download_new_or_modified)
|
||||
from .location import Location
|
||||
from .logging import PrettyLogger, enable_logging
|
||||
from .organizer import Organizer
|
||||
@ -72,7 +74,8 @@ class Pferd(Location):
|
||||
dir_filter: IliasDirectoryFilter,
|
||||
transform: Transform,
|
||||
download_strategy: IliasDownloadStrategy,
|
||||
clean: bool = True
|
||||
timeout: int,
|
||||
clean: bool = True,
|
||||
) -> Organizer:
|
||||
# pylint: disable=too-many-locals
|
||||
cookie_jar = CookieJar(to_path(cookies) if cookies else None)
|
||||
@ -81,7 +84,8 @@ class Pferd(Location):
|
||||
organizer = Organizer(self.resolve(to_path(target)))
|
||||
|
||||
crawler = IliasCrawler(base_url, session, authenticator, dir_filter)
|
||||
downloader = IliasDownloader(tmp_dir, organizer, session, authenticator, download_strategy)
|
||||
downloader = IliasDownloader(tmp_dir, organizer, session,
|
||||
authenticator, download_strategy, timeout)
|
||||
|
||||
cookie_jar.load_cookies()
|
||||
info = crawl_function(crawler)
|
||||
@ -112,6 +116,7 @@ class Pferd(Location):
|
||||
password: Optional[str] = None,
|
||||
download_strategy: IliasDownloadStrategy = download_modified_or_new,
|
||||
clean: bool = True,
|
||||
timeout: int = 5,
|
||||
) -> Organizer:
|
||||
"""
|
||||
Synchronizes a folder with the ILIAS instance of the KIT.
|
||||
@ -137,6 +142,8 @@ class Pferd(Location):
|
||||
be downloaded. Can save bandwidth and reduce the number of requests.
|
||||
(default: {download_modified_or_new})
|
||||
clean {bool} -- Whether to clean up when the method finishes.
|
||||
timeout {int} -- The download timeout for opencast videos. Sadly needed due to a
|
||||
requests bug.
|
||||
"""
|
||||
# This authenticator only works with the KIT ilias instance.
|
||||
authenticator = KitShibbolethAuthenticator(username=username, password=password)
|
||||
@ -152,6 +159,7 @@ class Pferd(Location):
|
||||
transform=transform,
|
||||
download_strategy=download_strategy,
|
||||
clean=clean,
|
||||
timeout=timeout
|
||||
)
|
||||
|
||||
self._download_summary.merge(organizer.download_summary)
|
||||
@ -175,6 +183,7 @@ class Pferd(Location):
|
||||
password: Optional[str] = None,
|
||||
download_strategy: IliasDownloadStrategy = download_modified_or_new,
|
||||
clean: bool = True,
|
||||
timeout: int = 5,
|
||||
) -> Organizer:
|
||||
"""
|
||||
Synchronizes a folder with the ILIAS instance of the KIT. This method will crawl the ILIAS
|
||||
@ -199,11 +208,14 @@ class Pferd(Location):
|
||||
be downloaded. Can save bandwidth and reduce the number of requests.
|
||||
(default: {download_modified_or_new})
|
||||
clean {bool} -- Whether to clean up when the method finishes.
|
||||
timeout {int} -- The download timeout for opencast videos. Sadly needed due to a
|
||||
requests bug.
|
||||
"""
|
||||
# This authenticator only works with the KIT ilias instance.
|
||||
authenticator = KitShibbolethAuthenticator(username=username, password=password)
|
||||
PRETTY.starting_synchronizer(target, "ILIAS", "Personal Desktop")
|
||||
return self._ilias(
|
||||
|
||||
organizer = self._ilias(
|
||||
target=target,
|
||||
base_url="https://ilias.studium.kit.edu/",
|
||||
crawl_function=lambda crawler: crawler.crawl_personal_desktop(),
|
||||
@ -213,8 +225,131 @@ class Pferd(Location):
|
||||
transform=transform,
|
||||
download_strategy=download_strategy,
|
||||
clean=clean,
|
||||
timeout=timeout
|
||||
)
|
||||
|
||||
self._download_summary.merge(organizer.download_summary)
|
||||
|
||||
return organizer
|
||||
|
||||
@swallow_and_print_errors
|
||||
def ilias_kit_folder(
|
||||
self,
|
||||
target: PathLike,
|
||||
full_url: str,
|
||||
dir_filter: IliasDirectoryFilter = lambda x, y: True,
|
||||
transform: Transform = lambda x: x,
|
||||
cookies: Optional[PathLike] = None,
|
||||
username: Optional[str] = None,
|
||||
password: Optional[str] = None,
|
||||
download_strategy: IliasDownloadStrategy = download_modified_or_new,
|
||||
clean: bool = True,
|
||||
timeout: int = 5,
|
||||
) -> Organizer:
|
||||
"""
|
||||
Synchronizes a folder with a given folder on the ILIAS instance of the KIT.
|
||||
|
||||
Arguments:
|
||||
target {Path} -- the target path to write the data to
|
||||
full_url {str} -- the full url of the folder/videos/course to crawl
|
||||
|
||||
Keyword Arguments:
|
||||
dir_filter {IliasDirectoryFilter} -- A filter for directories. Will be applied on the
|
||||
crawler level, these directories and all of their content is skipped.
|
||||
(default: {lambdax:True})
|
||||
transform {Transform} -- A transformation function for the output paths. Return None
|
||||
to ignore a file. (default: {lambdax:x})
|
||||
cookies {Optional[Path]} -- The path to store and load cookies from.
|
||||
(default: {None})
|
||||
username {Optional[str]} -- The SCC username. If none is given, it will prompt
|
||||
the user. (default: {None})
|
||||
password {Optional[str]} -- The SCC password. If none is given, it will prompt
|
||||
the user. (default: {None})
|
||||
download_strategy {DownloadStrategy} -- A function to determine which files need to
|
||||
be downloaded. Can save bandwidth and reduce the number of requests.
|
||||
(default: {download_modified_or_new})
|
||||
clean {bool} -- Whether to clean up when the method finishes.
|
||||
timeout {int} -- The download timeout for opencast videos. Sadly needed due to a
|
||||
requests bug.
|
||||
"""
|
||||
# This authenticator only works with the KIT ilias instance.
|
||||
authenticator = KitShibbolethAuthenticator(username=username, password=password)
|
||||
PRETTY.starting_synchronizer(target, "ILIAS", "An ILIAS element by url")
|
||||
|
||||
if not full_url.startswith("https://ilias.studium.kit.edu"):
|
||||
raise FatalException("Not a valid KIT ILIAS URL")
|
||||
|
||||
organizer = self._ilias(
|
||||
target=target,
|
||||
base_url="https://ilias.studium.kit.edu/",
|
||||
crawl_function=lambda crawler: crawler.recursive_crawl_url(full_url),
|
||||
authenticator=authenticator,
|
||||
cookies=cookies,
|
||||
dir_filter=dir_filter,
|
||||
transform=transform,
|
||||
download_strategy=download_strategy,
|
||||
clean=clean,
|
||||
timeout=timeout
|
||||
)
|
||||
|
||||
self._download_summary.merge(organizer.download_summary)
|
||||
|
||||
return organizer
|
||||
|
||||
@swallow_and_print_errors
|
||||
def ipd_kit(
|
||||
self,
|
||||
target: Union[PathLike, Organizer],
|
||||
url: str,
|
||||
transform: Transform = lambda x: x,
|
||||
download_strategy: IpdDownloadStrategy = ipd_download_new_or_modified,
|
||||
clean: bool = True
|
||||
) -> Organizer:
|
||||
"""
|
||||
Synchronizes a folder with a DIVA playlist.
|
||||
|
||||
Arguments:
|
||||
target {Union[PathLike, Organizer]} -- The organizer / target folder to use.
|
||||
url {str} -- the url to the page
|
||||
|
||||
Keyword Arguments:
|
||||
transform {Transform} -- A transformation function for the output paths. Return None
|
||||
to ignore a file. (default: {lambdax:x})
|
||||
download_strategy {DivaDownloadStrategy} -- A function to determine which files need to
|
||||
be downloaded. Can save bandwidth and reduce the number of requests.
|
||||
(default: {diva_download_new})
|
||||
clean {bool} -- Whether to clean up when the method finishes.
|
||||
"""
|
||||
tmp_dir = self._tmp_dir.new_subdir()
|
||||
|
||||
if target is None:
|
||||
PRETTY.starting_synchronizer("None", "IPD", url)
|
||||
raise FatalException("Got 'None' as target directory, aborting")
|
||||
|
||||
if isinstance(target, Organizer):
|
||||
organizer = target
|
||||
else:
|
||||
organizer = Organizer(self.resolve(to_path(target)))
|
||||
|
||||
PRETTY.starting_synchronizer(organizer.path, "IPD", url)
|
||||
|
||||
elements: List[IpdDownloadInfo] = IpdCrawler(url).crawl()
|
||||
transformed = apply_transform(transform, elements)
|
||||
|
||||
if self._test_run:
|
||||
self._print_transformables(transformed)
|
||||
return organizer
|
||||
|
||||
downloader = IpdDownloader(tmp_dir=tmp_dir, organizer=organizer, strategy=download_strategy)
|
||||
downloader.download_all(transformed)
|
||||
|
||||
if clean:
|
||||
organizer.cleanup()
|
||||
|
||||
self._download_summary.merge(organizer.download_summary)
|
||||
|
||||
return organizer
|
||||
|
||||
@swallow_and_print_errors
|
||||
def diva_kit(
|
||||
self,
|
||||
@ -273,4 +408,6 @@ class Pferd(Location):
|
||||
if clean:
|
||||
organizer.cleanup()
|
||||
|
||||
self._download_summary.merge(organizer.download_summary)
|
||||
|
||||
return organizer
|
||||
|
218
README.md
218
README.md
@ -2,43 +2,66 @@
|
||||
|
||||
**P**rogramm zum **F**lotten, **E**infachen **R**unterladen von **D**ateien
|
||||
|
||||
- [Quickstart with `sync_url`](#quickstart-with-sync_url)
|
||||
- [Installation](#installation)
|
||||
- [Upgrading from 2.0.0 to 2.1.0+](#upgrading-from-200-to-210)
|
||||
- [Example setup](#example-setup)
|
||||
- [Usage](#usage)
|
||||
- [General concepts](#general-concepts)
|
||||
- [Constructing transforms](#constructing-transforms)
|
||||
- [Transform creators](#transform-creators)
|
||||
- [Transform combinators](#transform-combinators)
|
||||
- [A short, but commented example](#a-short-but-commented-example)
|
||||
|
||||
## Quickstart with `sync_url`
|
||||
|
||||
The `sync_url` program allows you to just synchronize a given ILIAS URL (of a
|
||||
course, a folder, your personal desktop, etc.) without any extra configuration
|
||||
or setting up. Download the program, open ILIAS, copy the URL from the address
|
||||
bar and pass it to sync_url.
|
||||
|
||||
It bundles everything it needs in one executable and is easy to
|
||||
use, but doesn't expose all the configuration options and tweaks a full install
|
||||
does.
|
||||
|
||||
1. Download the `sync_url` binary from the [latest release](https://github.com/Garmelon/PFERD/releases/latest).
|
||||
2. Recognize that you most likely need to enclose the URL in `""` quotes to prevent your shell from interpreting `&` and other symbols
|
||||
3. Run the binary in your terminal (`./sync_url` or `sync_url.exe` in the CMD) to see the help and use it. I'd recommend using the `--cookies` option.
|
||||
If you are on **Linux/Mac**, you need to *make the file executable* using `chmod +x <file>`.
|
||||
If you are on **Mac**, you need to allow this unverified program to run (see e.g. [here](https://www.switchingtomac.com/tutorials/osx/how-to-run-unverified-apps-on-macos/))
|
||||
|
||||
## Installation
|
||||
|
||||
Ensure that you have at least Python 3.8 installed.
|
||||
|
||||
To install PFERD or update your installation to the latest version, run this
|
||||
wherever you want to install/have installed PFERD:
|
||||
wherever you want to install or have already installed PFERD:
|
||||
```
|
||||
$ pip install git+https://github.com/Garmelon/PFERD@v2.1.2
|
||||
$ pip install git+https://github.com/Garmelon/PFERD@v2.4.3
|
||||
```
|
||||
|
||||
The use of [venv](https://docs.python.org/3/library/venv.html) is recommended.
|
||||
The use of [venv] is recommended.
|
||||
|
||||
[venv]: https://docs.python.org/3/library/venv.html
|
||||
|
||||
### Upgrading from 2.0.0 to 2.1.0+
|
||||
|
||||
The `IliasDirectoryType` type was renamed to `IliasElementType` and is now far
|
||||
more detailed.
|
||||
The new values are: REGULAR_FOLDER, VIDEO_FOLDER,
|
||||
EXERCISE_FOLDER, REGULAR_FILE, VIDEO_FILE, FORUM, EXTERNAL_LINK.
|
||||
Forums and external links are skipped automatically if you use the `kit_ilias` helper.
|
||||
- The `IliasDirectoryType` type was renamed to `IliasElementType` and is now far more detailed.
|
||||
The new values are: `REGULAR_FOLDER`, `VIDEO_FOLDER`, `EXERCISE_FOLDER`, `REGULAR_FILE`, `VIDEO_FILE`, `FORUM`, `EXTERNAL_LINK`.
|
||||
- Forums and external links are skipped automatically if you use the `kit_ilias` helper.
|
||||
|
||||
## Example setup
|
||||
|
||||
In this example, `python3` refers to at least Python 3.8.
|
||||
|
||||
If you just want to get started and crawl *your entire ILIAS Desktop* instead
|
||||
of a given set of courses, please replace `example_config.py` with
|
||||
`example_config_personal_desktop.py` in all of the instructions below (`curl` call and
|
||||
`python3` run command).
|
||||
|
||||
A full example setup and initial use could look like:
|
||||
```
|
||||
$ mkdir Vorlesungen
|
||||
$ cd Vorlesungen
|
||||
$ python3 -m venv .venv
|
||||
$ .venv/bin/activate
|
||||
$ pip install git+https://github.com/Garmelon/PFERD@v2.1.2
|
||||
$ curl -O https://raw.githubusercontent.com/Garmelon/PFERD/v2.1.2/example_config.py
|
||||
$ pip install git+https://github.com/Garmelon/PFERD@v2.4.3
|
||||
$ curl -O https://raw.githubusercontent.com/Garmelon/PFERD/v2.4.3/example_config.py
|
||||
$ python3 example_config.py
|
||||
$ deactivate
|
||||
```
|
||||
@ -51,50 +74,93 @@ $ python3 example_config.py
|
||||
$ deactivate
|
||||
```
|
||||
|
||||
If you just want to get started and crawl *your entire ILIAS Desktop* instead
|
||||
of a given set of courses, please replace `example_config.py` with
|
||||
`example_config_personal_desktop.py` in all of the instructions below (`curl` call and
|
||||
`python3` run command).
|
||||
|
||||
## Usage
|
||||
|
||||
### General concepts
|
||||
|
||||
A PFERD config is a normal python file that starts multiple *synchronizers*
|
||||
which do all the heavy lifting. While you can create and wire them up manually,
|
||||
you are encouraged to use the helper methods provided in `PFERD.Pferd`.
|
||||
|
||||
The synchronizers take some input arguments specific to their service and a
|
||||
*transformer*. The transformer receives the computed path of an element in
|
||||
ILIAS and can return either an output path (so you can rename files or move
|
||||
them around as you wish) or `None` if you do not want to save the given file.
|
||||
*transform*. The transform receives the computed path of an element in ILIAS and
|
||||
can return either an output path (so you can rename files or move them around as
|
||||
you wish) or `None` if you do not want to save the given file.
|
||||
|
||||
Additionally the ILIAS synchronizer allows you to define a *crawl filter*. This
|
||||
filter also receives the computed path as the input, but is only called or
|
||||
*directoties*. If you return `True`, the directory will be crawled and
|
||||
filter also receives the computed path as the input, but is only called for
|
||||
*directories*. If you return `True`, the directory will be crawled and
|
||||
searched. If you return `False` the directory will be ignored and nothing in it
|
||||
will be passed to the transformer.
|
||||
will be passed to the transform.
|
||||
|
||||
In order to help you with writing your own transformers and filters, PFERD
|
||||
ships with a few powerful building blocks:
|
||||
### Constructing transforms
|
||||
|
||||
| Method | Description |
|
||||
|--------|-------------|
|
||||
| `glob` | Returns a transform that returns `None` if the glob does not match and the unmodified path otherwise. |
|
||||
| `predicate` | Returns a transform that returns `None` if the predicate does not match the path and the unmodified path otherwise. |
|
||||
| `move_dir(source, target)` | Returns a transform that moves all files from the `source` to the `target` dir. |
|
||||
| `move(source, target)` | Returns a transform that moves the `source` file to `target`. |
|
||||
| `rename(old, new)` | Renames a single file. |
|
||||
| `re_move(regex, sub)` | Moves all files matching the given regular expression. The different captured groups are available under their index and can be used together with normal python format methods: `re_move(r"Blatt (\d+)\.pdf", "Blätter/Blatt_{1:0>2}.pdf"),`. |
|
||||
| `re_rename(old, new)` | Same as `re_move` but operates on the path *names* instead of the full path. |
|
||||
While transforms are just normal python functions, writing them by hand can
|
||||
quickly become tedious. In order to help you with writing your own transforms
|
||||
and filters, PFERD defines a few useful transform creators and combinators in
|
||||
the `PFERD.transform` module:
|
||||
|
||||
And PFERD also offers a few combinator functions:
|
||||
#### Transform creators
|
||||
|
||||
* **`keep`**
|
||||
`keep` just returns the input path unchanged. It can be very useful as the
|
||||
last argument in an `attempt` call, to leave everything not matching a rule
|
||||
unchanged.
|
||||
* **`optionally(transformer)`**
|
||||
Wraps a given transformer and returns its result if it is not `None`.
|
||||
These methods let you create a few basic transform building blocks:
|
||||
|
||||
- **`glob(glob)`**
|
||||
Creates a transform that returns the unchanged path if the glob matches the path and `None` otherwise.
|
||||
See also [Path.match].
|
||||
Example: `glob("Übung/*.pdf")`
|
||||
- **`predicate(pred)`**
|
||||
Creates a transform that returns the unchanged path if `pred(path)` returns a truthy value.
|
||||
Returns `None` otherwise.
|
||||
Example: `predicate(lambda path: len(path.parts) == 3)`
|
||||
- **`move_dir(source, target)`**
|
||||
Creates a transform that moves all files from the `source` to the `target` directory.
|
||||
Example: `move_dir("Übung/", "Blätter/")`
|
||||
- **`move(source, target)`**
|
||||
Creates a transform that moves the `source` file to `target`.
|
||||
Example: `move("Vorlesung/VL02_Automten.pdf", "Vorlesung/VL02_Automaten.pdf")`
|
||||
- **`rename(source, target)`**
|
||||
Creates a transform that renames all files named `source` to `target`.
|
||||
This transform works on the file names, not paths, and thus works no matter where the file is located.
|
||||
Example: `rename("VL02_Automten.pdf", "VL02_Automaten.pdf")`
|
||||
- **`re_move(regex, target)`**
|
||||
Creates a transform that moves all files matching `regex` to `target`.
|
||||
The transform `str.format` on the `target` string with the contents of the capturing groups before returning it.
|
||||
The capturing groups can be accessed via their index.
|
||||
See also [Match.group].
|
||||
Example: `re_move(r"Übung/Blatt (\d+)\.pdf", "Blätter/Blatt_{1:0>2}.pdf")`
|
||||
- **`re_rename(regex, target)`**
|
||||
Creates a transform that renames all files matching `regex` to `target`.
|
||||
This transform works on the file names, not paths, and thus works no matter where the file is located.
|
||||
Example: `re_rename(r"VL(\d+)(.*)\.pdf", "Vorlesung_Nr_{1}__{2}.pdf")`
|
||||
|
||||
All movement or rename transforms above return `None` if a file doesn't match
|
||||
their movement or renaming criteria. This enables them to be used as building
|
||||
blocks to build up more complex transforms.
|
||||
|
||||
In addition, `PFERD.transform` also defines the `keep` transform which returns its input path unchanged.
|
||||
This behaviour can be very useful when creating more complex transforms.
|
||||
See below for example usage.
|
||||
|
||||
[Path.match]: https://docs.python.org/3/library/pathlib.html#pathlib.Path.match
|
||||
[Match.group]: https://docs.python.org/3/library/re.html#re.Match.group
|
||||
|
||||
#### Transform combinators
|
||||
|
||||
These methods let you combine transforms into more complex transforms:
|
||||
|
||||
- **`optionally(transform)`**
|
||||
Wraps a given transform and returns its result if it is not `None`.
|
||||
Otherwise returns the input path unchanged.
|
||||
* **`do(transformers)`**
|
||||
`do` accepts a series of transformers and applies them in the given order to
|
||||
the result of the previous one. If any transformer returns `None`, do
|
||||
short-circuits and also returns `None`. This can be used to perform multiple
|
||||
renames in a row:
|
||||
See below for example usage.
|
||||
* **`do(transforms)`**
|
||||
Accepts a series of transforms and applies them in the given order to the result of the previous one.
|
||||
If any transform returns `None`, `do` short-circuits and also returns `None`.
|
||||
This can be used to perform multiple renames in a row:
|
||||
```py
|
||||
do(
|
||||
# Move them
|
||||
@ -103,13 +169,12 @@ And PFERD also offers a few combinator functions:
|
||||
optionally(re_rename("(.*).m4v.mp4", "{1}.mp4")),
|
||||
# Remove the 'dbs' prefix (if they have any)
|
||||
optionally(re_rename("(?i)dbs-(.+)", "{1}")),
|
||||
),
|
||||
)
|
||||
```
|
||||
* **`attempt(transformers)`**
|
||||
`attempt` applies the passed transformers in the given order until it finds
|
||||
one that does not return `None`. If it does not find any, it returns `None`.
|
||||
This can be used to give a list of possible transformations and it will
|
||||
automatically pick the first one that fits:
|
||||
- **`attempt(transforms)`**
|
||||
Applies the passed transforms in the given order until it finds one that does not return `None`.
|
||||
If it does not find any, it returns `None`.
|
||||
This can be used to give a list of possible transformations and automatically pick the first one that fits:
|
||||
```py
|
||||
attempt(
|
||||
# Move all videos. If a video is passed in, this `re_move` will succeed
|
||||
@ -122,17 +187,26 @@ And PFERD also offers a few combinator functions:
|
||||
)
|
||||
```
|
||||
|
||||
All of these combinators are used in the provided example config, if you want
|
||||
to see some more true-to-live usages.
|
||||
All of these combinators are used in the provided example configs, if you want
|
||||
to see some more real-life usages.
|
||||
|
||||
### A short, but commented example
|
||||
|
||||
```py
|
||||
def filter_course(path: PurePath) -> bool:
|
||||
# Note that glob returns a Transformer
|
||||
# - a function from PurePath -> Optional[PurePath]
|
||||
# So we need to apply the result of 'glob' to our input path.
|
||||
# We need to crawl the 'Tutorien' folder as it contains the one we want.
|
||||
from pathlib import Path, PurePath
|
||||
from PFERD import Pferd
|
||||
from PFERD.ilias import IliasElementType
|
||||
from PFERD.transform import *
|
||||
|
||||
# This filter will later be used by the ILIAS crawler to decide whether it
|
||||
# should crawl a directory (or directory-like structure).
|
||||
def filter_course(path: PurePath, type: IliasElementType) -> bool:
|
||||
# Note that glob returns a Transform, which is a function from PurePath ->
|
||||
# Optional[PurePath]. Because of this, we need to apply the result of
|
||||
# 'glob' to our input path. The returned value will be truthy (a Path) if
|
||||
# the transform succeeded, or `None` if it failed.
|
||||
|
||||
# We need to crawl the 'Tutorien' folder as it contains one that we want.
|
||||
if glob("Tutorien/")(path):
|
||||
return True
|
||||
# If we found 'Tutorium 10', keep it!
|
||||
@ -145,21 +219,35 @@ def filter_course(path: PurePath) -> bool:
|
||||
# All other dirs (including subdirs of 'Tutorium 10') should be searched :)
|
||||
return True
|
||||
|
||||
enable_logging() # needed once before calling a Pferd method
|
||||
# Create a Pferd instance rooted in the same directory as the script file
|
||||
# This is not a test run, so files will be downloaded (default, can be omitted)
|
||||
|
||||
# This transform will later be used to rename a few files. It can also be used
|
||||
# to ignore some files.
|
||||
transform_course = attempt(
|
||||
# We don't care about the other tuts and would instead prefer a cleaner
|
||||
# directory structure.
|
||||
move_dir("Tutorien/Tutorium 10/", "Tutorium/"),
|
||||
# We don't want to modify any other files, so we're going to keep them
|
||||
# exactly as they are.
|
||||
keep
|
||||
)
|
||||
|
||||
# Enable and configure the text output. Needs to be called before calling any
|
||||
# other PFERD methods.
|
||||
Pferd.enable_logging()
|
||||
# Create a Pferd instance rooted in the same directory as the script file. This
|
||||
# is not a test run, so files will be downloaded (default, can be omitted).
|
||||
pferd = Pferd(Path(__file__).parent, test_run=False)
|
||||
|
||||
# Use the ilias_kit helper to synchronize an ILIAS course
|
||||
pferd.ilias_kit(
|
||||
# The folder all of the course's content should be placed in
|
||||
Path("My cool course"),
|
||||
# The directory that all of the downloaded files should be placed in
|
||||
"My_cool_course/",
|
||||
# The course ID (found in the URL when on the course page in ILIAS)
|
||||
"course id",
|
||||
# A path to a cookie jar. If you synchronize multiple ILIAS courses, setting this
|
||||
# to a common value requires you to only login once.
|
||||
# A path to a cookie jar. If you synchronize multiple ILIAS courses,
|
||||
# setting this to a common value requires you to only log in once.
|
||||
cookies=Path("ilias_cookies.txt"),
|
||||
# A transform to apply to all found paths
|
||||
# A transform can rename, move or filter out certain files
|
||||
transform=transform_course,
|
||||
# A crawl filter limits what paths the cralwer searches
|
||||
dir_filter=filter_course,
|
||||
|
2
setup.py
2
setup.py
@ -2,7 +2,7 @@ from setuptools import find_packages, setup
|
||||
|
||||
setup(
|
||||
name="PFERD",
|
||||
version="2.1.2",
|
||||
version="2.4.3",
|
||||
packages=find_packages(),
|
||||
install_requires=[
|
||||
"requests>=2.21.0",
|
||||
|
79
sync_url.py
Executable file
79
sync_url.py
Executable file
@ -0,0 +1,79 @@
|
||||
#!/usr/bin/env python
|
||||
|
||||
"""
|
||||
A simple script to download a course by name from ILIAS.
|
||||
"""
|
||||
|
||||
import argparse
|
||||
import os
|
||||
import re
|
||||
from pathlib import Path, PurePath
|
||||
from typing import Optional
|
||||
from urllib.parse import urlparse
|
||||
|
||||
from PFERD import Pferd
|
||||
from PFERD.cookie_jar import CookieJar
|
||||
from PFERD.ilias import (IliasCrawler, IliasElementType,
|
||||
KitShibbolethAuthenticator)
|
||||
from PFERD.utils import to_path
|
||||
|
||||
|
||||
def sanitize_path(path: PurePath) -> Optional[PurePath]:
|
||||
# Escape windows illegal path characters
|
||||
if os.name == 'nt':
|
||||
sanitized_parts = [re.sub(r'[<>:"/|?]', "_", x) for x in list(path.parts)]
|
||||
return PurePath(*sanitized_parts)
|
||||
return path
|
||||
|
||||
|
||||
def main() -> None:
|
||||
parser = argparse.ArgumentParser()
|
||||
parser.add_argument("--test-run", action="store_true")
|
||||
parser.add_argument('-c', '--cookies', nargs='?', default=None, help="File to store cookies in")
|
||||
parser.add_argument('--no-videos', nargs='?', default=None, help="Don't download videos")
|
||||
parser.add_argument('url', help="URL to the course page")
|
||||
parser.add_argument('folder', nargs='?', default=None, help="Folder to put stuff into")
|
||||
args = parser.parse_args()
|
||||
|
||||
url = urlparse(args.url)
|
||||
|
||||
cookie_jar = CookieJar(to_path(args.cookies) if args.cookies else None)
|
||||
session = cookie_jar.create_session()
|
||||
authenticator = KitShibbolethAuthenticator()
|
||||
crawler = IliasCrawler(url.scheme + '://' + url.netloc, session,
|
||||
authenticator, lambda x, y: True)
|
||||
|
||||
cookie_jar.load_cookies()
|
||||
|
||||
if args.folder is not None:
|
||||
folder = args.folder
|
||||
# Initialize pferd at the *parent of the passed folder*
|
||||
# This is needed so Pferd's internal protections against escaping the working directory
|
||||
# do not trigger (e.g. if somebody names a file in ILIAS '../../bad thing.txt')
|
||||
pferd = Pferd(Path(Path(__file__).parent, folder).parent, test_run=args.test_run)
|
||||
else:
|
||||
# fetch course name from ilias
|
||||
folder = crawler.find_element_name(args.url)
|
||||
cookie_jar.save_cookies()
|
||||
|
||||
# Initialize pferd at the location of the script
|
||||
pferd = Pferd(Path(__file__).parent, test_run=args.test_run)
|
||||
|
||||
def dir_filter(_: Path, element: IliasElementType) -> bool:
|
||||
if args.no_videos:
|
||||
return element not in [IliasElementType.VIDEO_FILE, IliasElementType.VIDEO_FOLDER]
|
||||
return True
|
||||
|
||||
pferd.enable_logging()
|
||||
# fetch
|
||||
pferd.ilias_kit_folder(
|
||||
target=folder,
|
||||
full_url=args.url,
|
||||
cookies=args.cookies,
|
||||
dir_filter=dir_filter,
|
||||
transform=sanitize_path
|
||||
)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
Reference in New Issue
Block a user