""" Nextcloud WebDAV client. All paths passed to these functions are relative to `settings.nextcloud_base_path`. The full WebDAV URL is: {nextcloud_url}/remote.php/dav/files/{username}/{base_path}/{relative_path} """ import xml.etree.ElementTree as ET from typing import List from urllib.parse import unquote import httpx from fastapi import HTTPException from config import settings DAV_NS = "DAV:" # Default timeout for all Nextcloud WebDAV requests (seconds) _TIMEOUT = 60.0 # Shared async client — reuses TCP connections across requests so Nextcloud # doesn't see rapid connection bursts that trigger brute-force throttling. _http_client: httpx.AsyncClient | None = None def _get_client() -> httpx.AsyncClient: global _http_client if _http_client is None or _http_client.is_closed: _http_client = httpx.AsyncClient( timeout=_TIMEOUT, follow_redirects=True, headers={"User-Agent": "BellSystems-CP/1.0"}, ) return _http_client async def close_client() -> None: """Close the shared HTTP client. Call this on application shutdown.""" global _http_client if _http_client and not _http_client.is_closed: await _http_client.aclose() _http_client = None async def keepalive_ping() -> None: """ Send a lightweight PROPFIND Depth:0 to the Nextcloud base folder to keep the TCP connection alive. Safe to call even if Nextcloud is not configured. """ if not settings.nextcloud_url: return try: url = _base_url() client = _get_client() await client.request( "PROPFIND", url, auth=_auth(), headers={"Depth": "0", "Content-Type": "application/xml"}, content=_PROPFIND_BODY, ) except Exception as e: print(f"[NEXTCLOUD KEEPALIVE] ping failed: {e}") def _dav_user() -> str: """The username used in the WebDAV URL path (may differ from the login username).""" return settings.nextcloud_dav_user or settings.nextcloud_username def _base_url() -> str: if not settings.nextcloud_url: raise HTTPException(status_code=503, detail="Nextcloud not configured") return ( f"{settings.nextcloud_url.rstrip('/')}" f"/remote.php/dav/files/{_dav_user()}" f"/{settings.nextcloud_base_path}" ) def _auth() -> tuple[str, str]: return (settings.nextcloud_username, settings.nextcloud_password) def _full_url(relative_path: str) -> str: """Build full WebDAV URL for a relative path.""" path = relative_path.strip("/") base = _base_url() return f"{base}/{path}" if path else base def _parse_propfind(xml_bytes: bytes, base_path_prefix: str) -> List[dict]: """ Parse a PROPFIND XML response. Returns list of file/folder entries, skipping the root itself. """ root = ET.fromstring(xml_bytes) results = [] # The prefix we need to strip from D:href to get the relative path back # href looks like: /remote.php/dav/files/user/BellSystems/Console/customers/abc/ dav_prefix = ( f"/remote.php/dav/files/{_dav_user()}" f"/{settings.nextcloud_base_path}/" ) for response in root.findall(f"{{{DAV_NS}}}response"): href_el = response.find(f"{{{DAV_NS}}}href") if href_el is None: continue href = unquote(href_el.text or "") # Strip DAV prefix to get relative path within base_path if href.startswith(dav_prefix): rel = href[len(dav_prefix):].rstrip("/") else: rel = href # Skip the folder itself (the root of the PROPFIND request) if rel == base_path_prefix.strip("/"): continue propstat = response.find(f"{{{DAV_NS}}}propstat") if propstat is None: continue prop = propstat.find(f"{{{DAV_NS}}}prop") if prop is None: continue # is_dir: resourcetype contains D:collection resource_type = prop.find(f"{{{DAV_NS}}}resourcetype") is_dir = resource_type is not None and resource_type.find(f"{{{DAV_NS}}}collection") is not None content_type_el = prop.find(f"{{{DAV_NS}}}getcontenttype") mime_type = content_type_el.text if content_type_el is not None else ( "inode/directory" if is_dir else "application/octet-stream" ) size_el = prop.find(f"{{{DAV_NS}}}getcontentlength") size = int(size_el.text) if size_el is not None and size_el.text else 0 modified_el = prop.find(f"{{{DAV_NS}}}getlastmodified") last_modified = modified_el.text if modified_el is not None else None filename = rel.split("/")[-1] if rel else "" results.append({ "filename": filename, "path": rel, "mime_type": mime_type, "size": size, "last_modified": last_modified, "is_dir": is_dir, }) return results async def ensure_folder(relative_path: str) -> None: """ Create a folder (and all parents) in Nextcloud via MKCOL. Includes the base_path segments so the full hierarchy is created from scratch. Silently succeeds if folders already exist. """ # Build the complete path list: base_path segments + relative_path segments base_parts = settings.nextcloud_base_path.strip("/").split("/") rel_parts = relative_path.strip("/").split("/") if relative_path.strip("/") else [] all_parts = base_parts + rel_parts dav_root = f"{settings.nextcloud_url.rstrip('/')}/remote.php/dav/files/{_dav_user()}" client = _get_client() built = "" for part in all_parts: built = f"{built}/{part}" if built else part url = f"{dav_root}/{built}" resp = await client.request("MKCOL", url, auth=_auth()) # 201 = created, 405/409 = already exists — both are fine if resp.status_code not in (201, 405, 409): raise HTTPException( status_code=502, detail=f"Failed to create Nextcloud folder '{built}': {resp.status_code}", ) async def write_info_file(customer_folder: str, customer_name: str, customer_id: str) -> None: """Write a _info.txt stub into a new customer folder for human browsability.""" content = f"Customer: {customer_name}\nID: {customer_id}\n" await upload_file( f"{customer_folder}/_info.txt", content.encode("utf-8"), "text/plain", ) _PROPFIND_BODY = b""" """ async def list_folder(relative_path: str) -> List[dict]: """ PROPFIND at depth=1 to list a folder's immediate children. relative_path is relative to nextcloud_base_path. """ url = _full_url(relative_path) client = _get_client() resp = await client.request( "PROPFIND", url, auth=_auth(), headers={"Depth": "1", "Content-Type": "application/xml"}, content=_PROPFIND_BODY, ) if resp.status_code == 404: return [] if resp.status_code not in (207, 200): raise HTTPException(status_code=502, detail=f"Nextcloud PROPFIND failed: {resp.status_code}") return _parse_propfind(resp.content, relative_path) async def list_folder_recursive(relative_path: str) -> List[dict]: """ Recursively list ALL files under a folder (any depth). Tries Depth:infinity first (single call). Falls back to manual recursion via Depth:1 if the server returns 403/400 (some servers disable infinity). Returns only file entries (is_dir=False). """ url = _full_url(relative_path) client = _get_client() resp = await client.request( "PROPFIND", url, auth=_auth(), headers={"Depth": "infinity", "Content-Type": "application/xml"}, content=_PROPFIND_BODY, ) if resp.status_code in (207, 200): all_items = _parse_propfind(resp.content, relative_path) return [item for item in all_items if not item["is_dir"]] # Depth:infinity not supported — fall back to recursive Depth:1 if resp.status_code in (403, 400, 412): return await _list_recursive_fallback(relative_path) if resp.status_code == 404: return [] raise HTTPException(status_code=502, detail=f"Nextcloud PROPFIND failed: {resp.status_code}") async def _list_recursive_fallback(relative_path: str) -> List[dict]: """Manually recurse via Depth:1 calls when Depth:infinity is blocked.""" items = await list_folder(relative_path) files = [] dirs = [] for item in items: if item["is_dir"]: dirs.append(item["path"]) else: files.append(item) for dir_path in dirs: child_files = await _list_recursive_fallback(dir_path) files.extend(child_files) return files async def upload_file(relative_path: str, content: bytes, mime_type: str) -> str: """ PUT a file to Nextcloud. Returns the relative_path on success. relative_path includes filename, e.g. "customers/abc123/media/photo.jpg" """ url = _full_url(relative_path) client = _get_client() resp = await client.put( url, auth=_auth(), content=content, headers={"Content-Type": mime_type}, ) if resp.status_code not in (200, 201, 204): raise HTTPException(status_code=502, detail=f"Nextcloud upload failed: {resp.status_code}") return relative_path async def download_file(relative_path: str) -> tuple[bytes, str]: """ GET a file from Nextcloud. Returns (bytes, mime_type). """ url = _full_url(relative_path) client = _get_client() resp = await client.get(url, auth=_auth()) if resp.status_code == 404: raise HTTPException(status_code=404, detail="File not found in Nextcloud") if resp.status_code != 200: raise HTTPException(status_code=502, detail=f"Nextcloud download failed: {resp.status_code}") mime = resp.headers.get("content-type", "application/octet-stream").split(";")[0].strip() return resp.content, mime async def delete_file(relative_path: str) -> None: """DELETE a file from Nextcloud.""" url = _full_url(relative_path) client = _get_client() resp = await client.request("DELETE", url, auth=_auth()) if resp.status_code not in (200, 204, 404): raise HTTPException(status_code=502, detail=f"Nextcloud delete failed: {resp.status_code}") async def rename_folder(old_relative_path: str, new_relative_path: str) -> None: """Rename/move a folder in Nextcloud using WebDAV MOVE.""" url = _full_url(old_relative_path) destination = _full_url(new_relative_path) client = _get_client() resp = await client.request( "MOVE", url, auth=_auth(), headers={"Destination": destination, "Overwrite": "F"}, ) if resp.status_code not in (201, 204): raise HTTPException(status_code=502, detail=f"Nextcloud rename failed: {resp.status_code}")