|
1 | 1 | """WebDAV client for Nextcloud file operations.""" |
2 | 2 |
|
3 | 3 | import mimetypes |
4 | | -from typing import Tuple, Dict, Any, Optional |
| 4 | +from typing import Tuple, Dict, Any, Optional, List |
5 | 5 | import logging |
6 | 6 | from httpx import HTTPStatusError |
| 7 | +import xml.etree.ElementTree as ET |
7 | 8 |
|
8 | 9 | from .base import BaseNextcloudClient |
9 | 10 |
|
@@ -242,3 +243,174 @@ async def get_note_attachment( |
242 | 243 | f"Unexpected error fetching attachment '{filename}' for note {note_id}: {e}" |
243 | 244 | ) |
244 | 245 | raise e |
| 246 | + |
| 247 | + async def list_directory(self, path: str = "") -> List[Dict[str, Any]]: |
| 248 | + """List files and directories in the specified path via WebDAV PROPFIND.""" |
| 249 | + webdav_path = f"{self._get_webdav_base_path()}/{path.lstrip('/')}" |
| 250 | + if not webdav_path.endswith("/"): |
| 251 | + webdav_path += "/" |
| 252 | + |
| 253 | + logger.info(f"Listing directory: {webdav_path}") |
| 254 | + |
| 255 | + propfind_body = '''<?xml version="1.0"?> |
| 256 | + <d:propfind xmlns:d="DAV:"> |
| 257 | + <d:prop> |
| 258 | + <d:displayname/> |
| 259 | + <d:getcontentlength/> |
| 260 | + <d:getcontenttype/> |
| 261 | + <d:getlastmodified/> |
| 262 | + <d:resourcetype/> |
| 263 | + </d:prop> |
| 264 | + </d:propfind>''' |
| 265 | + |
| 266 | + headers = { |
| 267 | + "Depth": "1", |
| 268 | + "Content-Type": "text/xml", |
| 269 | + "OCS-APIRequest": "true" |
| 270 | + } |
| 271 | + |
| 272 | + try: |
| 273 | + response = await self._client.request( |
| 274 | + "PROPFIND", webdav_path, content=propfind_body, headers=headers |
| 275 | + ) |
| 276 | + response.raise_for_status() |
| 277 | + |
| 278 | + # Parse the XML response |
| 279 | + root = ET.fromstring(response.content) |
| 280 | + items = [] |
| 281 | + |
| 282 | + # Skip the first response (the directory itself) |
| 283 | + responses = root.findall(".//{DAV:}response")[1:] |
| 284 | + |
| 285 | + for response_elem in responses: |
| 286 | + href = response_elem.find(".//{DAV:}href") |
| 287 | + if href is None: |
| 288 | + continue |
| 289 | + |
| 290 | + # Extract file/directory name from href |
| 291 | + href_text = href.text or "" |
| 292 | + name = href_text.rstrip("/").split("/")[-1] |
| 293 | + if not name: |
| 294 | + continue |
| 295 | + |
| 296 | + # Get properties |
| 297 | + propstat = response_elem.find(".//{DAV:}propstat") |
| 298 | + if propstat is None: |
| 299 | + continue |
| 300 | + |
| 301 | + prop = propstat.find(".//{DAV:}prop") |
| 302 | + if prop is None: |
| 303 | + continue |
| 304 | + |
| 305 | + # Determine if it's a directory |
| 306 | + resourcetype = prop.find(".//{DAV:}resourcetype") |
| 307 | + is_directory = resourcetype is not None and resourcetype.find(".//{DAV:}collection") is not None |
| 308 | + |
| 309 | + # Get other properties |
| 310 | + size_elem = prop.find(".//{DAV:}getcontentlength") |
| 311 | + size = int(size_elem.text) if size_elem is not None and size_elem.text else 0 |
| 312 | + |
| 313 | + content_type_elem = prop.find(".//{DAV:}getcontenttype") |
| 314 | + content_type = content_type_elem.text if content_type_elem is not None else None |
| 315 | + |
| 316 | + modified_elem = prop.find(".//{DAV:}getlastmodified") |
| 317 | + modified = modified_elem.text if modified_elem is not None else None |
| 318 | + |
| 319 | + items.append({ |
| 320 | + "name": name, |
| 321 | + "path": f"{path.rstrip('/')}/{name}" if path else name, |
| 322 | + "is_directory": is_directory, |
| 323 | + "size": size if not is_directory else None, |
| 324 | + "content_type": content_type, |
| 325 | + "last_modified": modified |
| 326 | + }) |
| 327 | + |
| 328 | + logger.info(f"Found {len(items)} items in directory: {webdav_path}") |
| 329 | + return items |
| 330 | + |
| 331 | + except HTTPStatusError as e: |
| 332 | + logger.error(f"HTTP error listing directory '{webdav_path}': {e}") |
| 333 | + raise e |
| 334 | + except Exception as e: |
| 335 | + logger.error(f"Unexpected error listing directory '{webdav_path}': {e}") |
| 336 | + raise e |
| 337 | + |
| 338 | + async def read_file(self, path: str) -> Tuple[bytes, str]: |
| 339 | + """Read a file's content via WebDAV GET.""" |
| 340 | + webdav_path = f"{self._get_webdav_base_path()}/{path.lstrip('/')}" |
| 341 | + |
| 342 | + logger.info(f"Reading file: {webdav_path}") |
| 343 | + |
| 344 | + try: |
| 345 | + response = await self._client.get(webdav_path) |
| 346 | + response.raise_for_status() |
| 347 | + |
| 348 | + content = response.content |
| 349 | + content_type = response.headers.get("content-type", "application/octet-stream") |
| 350 | + |
| 351 | + logger.info(f"Successfully read file '{path}' ({content_type}, {len(content)} bytes)") |
| 352 | + return content, content_type |
| 353 | + |
| 354 | + except HTTPStatusError as e: |
| 355 | + logger.error(f"HTTP error reading file '{path}': {e}") |
| 356 | + raise e |
| 357 | + except Exception as e: |
| 358 | + logger.error(f"Unexpected error reading file '{path}': {e}") |
| 359 | + raise e |
| 360 | + |
| 361 | + async def write_file(self, path: str, content: bytes, content_type: Optional[str] = None) -> Dict[str, Any]: |
| 362 | + """Write content to a file via WebDAV PUT.""" |
| 363 | + webdav_path = f"{self._get_webdav_base_path()}/{path.lstrip('/')}" |
| 364 | + |
| 365 | + logger.info(f"Writing file: {webdav_path}") |
| 366 | + |
| 367 | + if not content_type: |
| 368 | + content_type, _ = mimetypes.guess_type(path) |
| 369 | + if not content_type: |
| 370 | + content_type = "application/octet-stream" |
| 371 | + |
| 372 | + headers = { |
| 373 | + "Content-Type": content_type, |
| 374 | + "OCS-APIRequest": "true" |
| 375 | + } |
| 376 | + |
| 377 | + try: |
| 378 | + response = await self._client.put(webdav_path, content=content, headers=headers) |
| 379 | + response.raise_for_status() |
| 380 | + |
| 381 | + logger.info(f"Successfully wrote file '{path}' (Status: {response.status_code})") |
| 382 | + return {"status_code": response.status_code} |
| 383 | + |
| 384 | + except HTTPStatusError as e: |
| 385 | + logger.error(f"HTTP error writing file '{path}': {e}") |
| 386 | + raise e |
| 387 | + except Exception as e: |
| 388 | + logger.error(f"Unexpected error writing file '{path}': {e}") |
| 389 | + raise e |
| 390 | + |
| 391 | + async def create_directory(self, path: str) -> Dict[str, Any]: |
| 392 | + """Create a directory via WebDAV MKCOL.""" |
| 393 | + webdav_path = f"{self._get_webdav_base_path()}/{path.lstrip('/')}" |
| 394 | + if not webdav_path.endswith("/"): |
| 395 | + webdav_path += "/" |
| 396 | + |
| 397 | + logger.info(f"Creating directory: {webdav_path}") |
| 398 | + |
| 399 | + headers = {"OCS-APIRequest": "true"} |
| 400 | + |
| 401 | + try: |
| 402 | + response = await self._client.request("MKCOL", webdav_path, headers=headers) |
| 403 | + response.raise_for_status() |
| 404 | + |
| 405 | + logger.info(f"Successfully created directory '{path}' (Status: {response.status_code})") |
| 406 | + return {"status_code": response.status_code} |
| 407 | + |
| 408 | + except HTTPStatusError as e: |
| 409 | + if e.response.status_code == 405: # Method Not Allowed - directory already exists |
| 410 | + logger.info(f"Directory '{path}' already exists") |
| 411 | + return {"status_code": 405, "message": "Directory already exists"} |
| 412 | + logger.error(f"HTTP error creating directory '{path}': {e}") |
| 413 | + raise e |
| 414 | + except Exception as e: |
| 415 | + logger.error(f"Unexpected error creating directory '{path}': {e}") |
| 416 | + raise e |
0 commit comments