# Copyright (c) 2024 Jonas Heinle
"""WebDAV client implementation for listing and downloading remote files."""
import sys
import threading
import time
import urllib.parse
import uuid
from pathlib import Path, PurePosixPath
import requests
from defusedxml import ElementTree
from loguru import logger
from requests.auth import HTTPBasicAuth
REQUEST_TIMEOUT_SECONDS = 30
LOG_FILE = "logs/downloadMd_s.log"
# One loguru sink per log file: a sink per client wrote every line once per client and rotated the file N times.
_log_sinks: set[str] = set()
_log_sinks_lock = threading.Lock()
def _add_log_sink(log_file: str) -> None:
path = str(Path(log_file).absolute())
with _log_sinks_lock:
if path in _log_sinks:
return
logger.add(path, rotation="500 MB")
_log_sinks.add(path)
def _join_remote_url(*parts: str) -> str:
cleaned_parts = [part.strip("/") for part in parts if part]
return "/".join(cleaned_parts)
def _replace(part: Path, target: Path) -> None:
# Windows refuses a replace while another thread replaces the same target (PermissionError); it clears in milliseconds.
attempts = 40 if sys.platform == "win32" else 1
for attempt in range(attempts):
try:
part.replace(target)
except PermissionError:
if attempt == attempts - 1:
raise
time.sleep(0.005 * (attempt + 1))
else:
return
def _write_atomically(target: Path, response: requests.Response) -> None:
# A private file renamed into place: two downloads of one file never interleave, and a reader never sees half of it.
part = target.with_name(f".{target.name}.{uuid.uuid4().hex}.part")
try:
with part.open("xb") as f:
f.writelines(response.iter_content(chunk_size=8192))
_replace(part, target)
except BaseException:
part.unlink(missing_ok=True)
raise
[docs]
class WebDavClient:
"""A simple WebDav client for listing and downloading files and folders."""
def __init__(self, hostname: str, username: str, password: str) -> None:
"""Initialize a client with credentials and logging output folder."""
self.hostname: str = hostname
self.username: str = username
self.password: str = password
self.auth: HTTPBasicAuth = HTTPBasicAuth(
username,
password,
)
self.ensure_folder_exists(str(Path(LOG_FILE).parent))
_add_log_sink(LOG_FILE)
[docs]
def list_files(self, url: str) -> list[str]:
"""List the files directly below a WebDAV URL; OSError unless PROPFIND answers 207."""
headers: dict[str, str] = {"Content-Type": "application/xml", "Depth": "1"}
response: requests.Response = requests.request(
"PROPFIND",
url,
auth=self.auth,
headers=headers,
timeout=REQUEST_TIMEOUT_SECONDS,
)
if response.status_code != 207:
error_message: str = "Failed to list directory contents via requests: "
logger.error(
"Error message {} with code {}", error_message, response.status_code
)
error_details = f"{error_message}{response.status_code}"
raise OSError(error_details)
tree = ElementTree.fromstring(response.content)
files = []
for response in tree.findall("{DAV:}response"):
href = response.find("{DAV:}href").text
if not href.endswith("/"):
logger.debug("Found file: {} for the following url: {}", href, url)
files.append(href)
return files
[docs]
def list_folders(self, remote_base_path: str) -> list[str]:
"""List the non-hidden folders directly below a remote base path; OSError unless PROPFIND answers 207."""
headers = {"Content-Type": "application/xml", "Depth": "1"}
url: str = _join_remote_url(self.hostname, remote_base_path)
response = requests.request(
"PROPFIND",
url,
auth=self.auth,
headers=headers,
timeout=REQUEST_TIMEOUT_SECONDS,
)
if response.status_code != 207:
error_message = "Failed to list directory contents: "
logger.error("{} {}", error_message, response.status_code)
error_details = f"{error_message}{response.status_code}"
raise OSError(error_details)
tree = ElementTree.fromstring(response.content)
folders = []
for response in tree.findall("{DAV:}response"):
href = response.find("{DAV:}href").text
folder = PurePosixPath(href.rstrip("/")).name
is_folder = href.endswith("/")
is_not_remote_base_path = (
href != url + "/"
) and folder != remote_base_path.rsplit("/", maxsplit=1)[-1]
is_hidden_folder = folder.startswith(".")
if is_folder and is_not_remote_base_path and not is_hidden_folder:
logger.debug(
"Found folder: {} in the parent folder: {}",
folder,
remote_base_path,
)
folders.append(folder)
return folders
[docs]
def filter_after_global_base_path(self, path: str, remote_base_path: str) -> str:
"""Return what follows /remote_base_path/ in path, or path unchanged when it is absent."""
search_str = "/" + remote_base_path + "/"
if search_str in path:
logger.debug(
"Found folder {} for path {} and remote base path: {}",
search_str,
path,
remote_base_path,
)
url_after_removing_everything_before_and_including_remote_base_name = (
path.split(search_str, 1)[1]
)
logger.info(
"Folder structure everything after the remote_base_path is: {}",
url_after_removing_everything_before_and_including_remote_base_name,
)
return url_after_removing_everything_before_and_including_remote_base_name
logger.error("Could not find search string: {} in path: {}", search_str, path)
return path
[docs]
def ensure_folder_exists(self, path: str) -> None:
"""Ensure that the given folder exists."""
folder_path = Path(path)
if not folder_path.exists():
folder_path.mkdir(parents=True, exist_ok=True)
logger.debug("Folder created: {}", path)
else:
logger.debug("Folder already exists: {}", path)
[docs]
def get_sub_path(self, full_path: str, initial_part: str) -> str:
"""Return the part of a host-less full_path after initial_part; ValueError when it is absent."""
logger.debug("We are in the 'get_sub_path' method.")
decoded_full_path = urllib.parse.unquote(full_path)
logger.debug("The decoded full file path is: {}", decoded_full_path)
decoded_initial_part = urllib.parse.unquote(initial_part)
logger.debug("The decoded initial file path is: {}", decoded_initial_part)
# A trailing slash makes the match a whole path segment.
if not decoded_initial_part.endswith("/"):
decoded_initial_part += "/"
start_idx = decoded_full_path.find(decoded_initial_part)
if start_idx == -1:
logger.error(
"The {} string is not in the full_path={}", initial_part, full_path
)
error_details = "The full path does not contain the initial part."
raise ValueError(error_details)
# Handle the edge case where the full path is exactly the initial part
decoded_initial_part = "/" + decoded_initial_part
if full_path == decoded_initial_part.rstrip("/"):
logger.debug("The get_sub_path() method returns empty string")
return ""
if full_path.startswith(initial_part):
logger.debug(
"The get_sub_path() method returns {}", full_path[len(initial_part) :]
)
return full_path[len(initial_part) :]
sub_path_start_idx = start_idx + len(decoded_initial_part) - 1
sub_path = decoded_full_path[sub_path_start_idx:]
logger.debug("The get_sub_path() method returns {}", sub_path)
return sub_path
[docs]
def download_files(
self,
global_remote_base_path: str,
remote_base_path: str,
local_base_path: str,
) -> None:
"""Download the files directly below remote_base_path, without subfolders.
Args:
global_remote_base_path: Root folder the local folder structure is relative to.
remote_base_path: Remote folder whose files are downloaded.
local_base_path: Local folder the files land under.
"""
if not Path(local_base_path).exists():
logger.info("Dir {} will be created", local_base_path)
Path(local_base_path).mkdir(parents=True, exist_ok=True)
url = _join_remote_url(self.hostname, remote_base_path)
files_on_host = self.list_files(url)
if len(files_on_host) == 0:
logger.info("Found no files on remote_base_path: {}", remote_base_path)
for file_path in files_on_host:
logger.info("Found the file: {} on current remote_base_path", file_path)
file_name = self.filter_after_global_base_path(file_path, remote_base_path)
logger.info("The pure of filename of this file is: {}", file_name)
decoded_filename = urllib.parse.unquote(file_name)
logger.info("The decoded filename version is: {}", decoded_filename)
remote_file_url = _join_remote_url(
self.hostname,
remote_base_path,
file_name,
)
logger.info("The remote file url is: {}", remote_file_url)
sub_path = self.get_sub_path(file_path, global_remote_base_path)
if sub_path.endswith(decoded_filename):
sub_path = sub_path[: len(sub_path) - len(decoded_filename)]
if sub_path == decoded_filename:
sub_path = ""
logger.debug("The current sub path is: {}", sub_path)
local_file_path = Path(local_base_path) / sub_path / decoded_filename
logger.debug(
"The current file that is stored has the full path: {}", local_file_path
)
with requests.get(
remote_file_url,
auth=self.auth,
stream=True,
timeout=REQUEST_TIMEOUT_SECONDS,
) as response:
if response.status_code == 200:
self.ensure_folder_exists(str(local_file_path.parent))
_write_atomically(local_file_path, response)
else:
logger.debug(
"Failed to download {}: {}",
remote_file_url,
response.status_code,
)
[docs]
def download_all_files_iterative(
self,
remote_base_path: str,
local_base_path: str,
) -> None:
"""Download everything below remote_base_path into local_base_path, keeping the folder structure."""
stack: list[str] = [remote_base_path]
global_remote_base_path: str = remote_base_path
while stack:
current_remote_path: str = stack.pop()
logger.debug("Current remote path is: {}", current_remote_path)
self.download_files(
global_remote_base_path,
current_remote_path,
local_base_path,
)
folders: list[str] = self.list_folders(current_remote_path)
if len(folders) == 0:
logger.info(
"Found no subfolders for current folder: {}", current_remote_path
)
for folder in folders:
logger.info(
"Found subfolder {} for current folder: {}.",
folder,
current_remote_path,
)
relative_folder_path: str = str(
PurePosixPath(current_remote_path) / folder
)
stack.append(relative_folder_path)