Coverage for kataglyphis_webdavclient/webdavclient.py: 94%
170 statements
« prev ^ index » next coverage.py v7.16.2, created at 2026-10-20 21:53 +0000
1# Copyright (c) 2024 Jonas Heinle
3"""WebDAV client implementation for listing and downloading remote files."""
5import sys
6import threading
7import time
8import urllib.parse
9import uuid
10from pathlib import Path, PurePosixPath
12import requests
13from defusedxml import ElementTree
14from loguru import logger
15from requests.auth import HTTPBasicAuth
18REQUEST_TIMEOUT_SECONDS = 30
19LOG_FILE = "logs/downloadMd_s.log"
21# One loguru sink per log file: a sink per client wrote every line once per client and rotated the file N times.
22_log_sinks: set[str] = set()
23_log_sinks_lock = threading.Lock()
26def _add_log_sink(log_file: str) -> None:
27 path = str(Path(log_file).absolute())
28 with _log_sinks_lock:
29 if path in _log_sinks:
30 return
31 logger.add(path, rotation="500 MB")
32 _log_sinks.add(path)
35def _join_remote_url(*parts: str) -> str:
36 cleaned_parts = [part.strip("/") for part in parts if part]
37 return "/".join(cleaned_parts)
40def _replace(part: Path, target: Path) -> None:
41 # Windows refuses a replace while another thread replaces the same target (PermissionError); it clears in milliseconds.
42 attempts = 40 if sys.platform == "win32" else 1
43 for attempt in range(attempts):
44 try:
45 part.replace(target)
46 except PermissionError:
47 if attempt == attempts - 1:
48 raise
49 time.sleep(0.005 * (attempt + 1))
50 else:
51 return
54def _write_atomically(target: Path, response: requests.Response) -> None:
55 # A private file renamed into place: two downloads of one file never interleave, and a reader never sees half of it.
56 part = target.with_name(f".{target.name}.{uuid.uuid4().hex}.part")
57 try:
58 with part.open("xb") as f:
59 f.writelines(response.iter_content(chunk_size=8192))
60 _replace(part, target)
61 except BaseException:
62 part.unlink(missing_ok=True)
63 raise
66class WebDavClient:
67 """A simple WebDav client for listing and downloading files and folders."""
69 def __init__(self, hostname: str, username: str, password: str) -> None:
70 """Initialize a client with credentials and logging output folder."""
71 self.hostname: str = hostname
72 self.username: str = username
73 self.password: str = password
74 self.auth: HTTPBasicAuth = HTTPBasicAuth(
75 username,
76 password,
77 )
79 self.ensure_folder_exists(str(Path(LOG_FILE).parent))
80 _add_log_sink(LOG_FILE)
82 def list_files(self, url: str) -> list[str]:
83 """List the files directly below a WebDAV URL; OSError unless PROPFIND answers 207."""
84 headers: dict[str, str] = {"Content-Type": "application/xml", "Depth": "1"}
85 response: requests.Response = requests.request(
86 "PROPFIND",
87 url,
88 auth=self.auth,
89 headers=headers,
90 timeout=REQUEST_TIMEOUT_SECONDS,
91 )
92 if response.status_code != 207:
93 error_message: str = "Failed to list directory contents via requests: "
94 logger.error(
95 "Error message {} with code {}", error_message, response.status_code
96 )
97 error_details = f"{error_message}{response.status_code}"
98 raise OSError(error_details)
100 tree = ElementTree.fromstring(response.content)
101 files = []
102 for response in tree.findall("{DAV:}response"):
103 href = response.find("{DAV:}href").text
104 if not href.endswith("/"):
105 logger.debug("Found file: {} for the following url: {}", href, url)
106 files.append(href)
107 return files
109 def list_folders(self, remote_base_path: str) -> list[str]:
110 """List the non-hidden folders directly below a remote base path; OSError unless PROPFIND answers 207."""
111 headers = {"Content-Type": "application/xml", "Depth": "1"}
112 url: str = _join_remote_url(self.hostname, remote_base_path)
113 response = requests.request(
114 "PROPFIND",
115 url,
116 auth=self.auth,
117 headers=headers,
118 timeout=REQUEST_TIMEOUT_SECONDS,
119 )
120 if response.status_code != 207:
121 error_message = "Failed to list directory contents: "
122 logger.error("{} {}", error_message, response.status_code)
123 error_details = f"{error_message}{response.status_code}"
124 raise OSError(error_details)
126 tree = ElementTree.fromstring(response.content)
127 folders = []
128 for response in tree.findall("{DAV:}response"):
129 href = response.find("{DAV:}href").text
130 folder = PurePosixPath(href.rstrip("/")).name
131 is_folder = href.endswith("/")
132 is_not_remote_base_path = (
133 href != url + "/"
134 ) and folder != remote_base_path.rsplit("/", maxsplit=1)[-1]
135 is_hidden_folder = folder.startswith(".")
136 if is_folder and is_not_remote_base_path and not is_hidden_folder:
137 logger.debug(
138 "Found folder: {} in the parent folder: {}",
139 folder,
140 remote_base_path,
141 )
142 folders.append(folder)
143 return folders
145 def filter_after_global_base_path(self, path: str, remote_base_path: str) -> str:
146 """Return what follows /remote_base_path/ in path, or path unchanged when it is absent."""
147 search_str = "/" + remote_base_path + "/"
148 if search_str in path:
149 logger.debug(
150 "Found folder {} for path {} and remote base path: {}",
151 search_str,
152 path,
153 remote_base_path,
154 )
155 url_after_removing_everything_before_and_including_remote_base_name = (
156 path.split(search_str, 1)[1]
157 )
158 logger.info(
159 "Folder structure everything after the remote_base_path is: {}",
160 url_after_removing_everything_before_and_including_remote_base_name,
161 )
162 return url_after_removing_everything_before_and_including_remote_base_name
163 logger.error("Could not find search string: {} in path: {}", search_str, path)
164 return path
166 def ensure_folder_exists(self, path: str) -> None:
167 """Ensure that the given folder exists."""
168 folder_path = Path(path)
169 if not folder_path.exists():
170 folder_path.mkdir(parents=True, exist_ok=True)
171 logger.debug("Folder created: {}", path)
172 else:
173 logger.debug("Folder already exists: {}", path)
175 def get_sub_path(self, full_path: str, initial_part: str) -> str:
176 """Return the part of a host-less full_path after initial_part; ValueError when it is absent."""
177 logger.debug("We are in the 'get_sub_path' method.")
178 decoded_full_path = urllib.parse.unquote(full_path)
179 logger.debug("The decoded full file path is: {}", decoded_full_path)
180 decoded_initial_part = urllib.parse.unquote(initial_part)
181 logger.debug("The decoded initial file path is: {}", decoded_initial_part)
182 # A trailing slash makes the match a whole path segment.
183 if not decoded_initial_part.endswith("/"):
184 decoded_initial_part += "/"
186 start_idx = decoded_full_path.find(decoded_initial_part)
188 if start_idx == -1:
189 logger.error(
190 "The {} string is not in the full_path={}", initial_part, full_path
191 )
192 error_details = "The full path does not contain the initial part."
193 raise ValueError(error_details)
195 # Handle the edge case where the full path is exactly the initial part
196 decoded_initial_part = "/" + decoded_initial_part
197 if full_path == decoded_initial_part.rstrip("/"):
198 logger.debug("The get_sub_path() method returns empty string")
199 return ""
201 if full_path.startswith(initial_part):
202 logger.debug(
203 "The get_sub_path() method returns {}", full_path[len(initial_part) :]
204 )
205 return full_path[len(initial_part) :]
206 sub_path_start_idx = start_idx + len(decoded_initial_part) - 1
208 sub_path = decoded_full_path[sub_path_start_idx:]
210 logger.debug("The get_sub_path() method returns {}", sub_path)
212 return sub_path
214 def download_files(
215 self,
216 global_remote_base_path: str,
217 remote_base_path: str,
218 local_base_path: str,
219 ) -> None:
220 """Download the files directly below remote_base_path, without subfolders.
222 Args:
223 global_remote_base_path: Root folder the local folder structure is relative to.
224 remote_base_path: Remote folder whose files are downloaded.
225 local_base_path: Local folder the files land under.
226 """
227 if not Path(local_base_path).exists():
228 logger.info("Dir {} will be created", local_base_path)
229 Path(local_base_path).mkdir(parents=True, exist_ok=True)
230 url = _join_remote_url(self.hostname, remote_base_path)
231 files_on_host = self.list_files(url)
233 if len(files_on_host) == 0:
234 logger.info("Found no files on remote_base_path: {}", remote_base_path)
236 for file_path in files_on_host:
237 logger.info("Found the file: {} on current remote_base_path", file_path)
238 file_name = self.filter_after_global_base_path(file_path, remote_base_path)
239 logger.info("The pure of filename of this file is: {}", file_name)
240 decoded_filename = urllib.parse.unquote(file_name)
241 logger.info("The decoded filename version is: {}", decoded_filename)
242 remote_file_url = _join_remote_url(
243 self.hostname,
244 remote_base_path,
245 file_name,
246 )
247 logger.info("The remote file url is: {}", remote_file_url)
248 sub_path = self.get_sub_path(file_path, global_remote_base_path)
250 if sub_path.endswith(decoded_filename):
251 sub_path = sub_path[: len(sub_path) - len(decoded_filename)]
253 if sub_path == decoded_filename:
254 sub_path = ""
256 logger.debug("The current sub path is: {}", sub_path)
257 local_file_path = Path(local_base_path) / sub_path / decoded_filename
258 logger.debug(
259 "The current file that is stored has the full path: {}", local_file_path
260 )
261 with requests.get(
262 remote_file_url,
263 auth=self.auth,
264 stream=True,
265 timeout=REQUEST_TIMEOUT_SECONDS,
266 ) as response:
267 if response.status_code == 200:
268 self.ensure_folder_exists(str(local_file_path.parent))
269 _write_atomically(local_file_path, response)
270 else:
271 logger.debug(
272 "Failed to download {}: {}",
273 remote_file_url,
274 response.status_code,
275 )
277 def download_all_files_iterative(
278 self,
279 remote_base_path: str,
280 local_base_path: str,
281 ) -> None:
282 """Download everything below remote_base_path into local_base_path, keeping the folder structure."""
283 stack: list[str] = [remote_base_path]
285 global_remote_base_path: str = remote_base_path
287 while stack:
288 current_remote_path: str = stack.pop()
289 logger.debug("Current remote path is: {}", current_remote_path)
291 self.download_files(
292 global_remote_base_path,
293 current_remote_path,
294 local_base_path,
295 )
297 folders: list[str] = self.list_folders(current_remote_path)
298 if len(folders) == 0:
299 logger.info(
300 "Found no subfolders for current folder: {}", current_remote_path
301 )
302 for folder in folders:
303 logger.info(
304 "Found subfolder {} for current folder: {}.",
305 folder,
306 current_remote_path,
307 )
308 relative_folder_path: str = str(
309 PurePosixPath(current_remote_path) / folder
310 )
311 stack.append(relative_folder_path)