tfaour
b4ca795003
Build and push server image / build-and-push (push) Successful in 1m57s
File picker: an optional WebDAV browse root in Settings (User.webdav_base_url) plus a plain-PROPFIND directory listing (webdav_client.list_directory) power a "Browse..." panel on a frame's Whiteboard tab, so a file can be clicked into rather than typing its exact WebDAV URL. Manual URL entry still works unchanged either way. Force-refresh: get_or_refresh_whiteboard takes a force flag that skips the fetch throttle entirely; the Whiteboard tab's refresh button now passes it, so clicking it always re-fetches and re-renders instead of possibly just re-showing the same cached image from within the last ~20 minutes.
123 lines
5.1 KiB
Python
123 lines
5.1 KiB
Python
"""Plain authenticated WebDAV file fetch -- whiteboard frame mode's way
|
|
of pulling one specific file (a Nextcloud Whiteboard .whiteboard, or any
|
|
other WebDAV server's file, this isn't Nextcloud-specific) out of a
|
|
user's account. Deliberately just "GET this URL with Basic auth", the
|
|
same shape as calendar_feed.py's plain ICS fetch -- no discovery, no
|
|
account-wide browsing, since the caller already has (or pastes) the
|
|
exact file URL, unlike caldav_client.py's calendar-account discovery
|
|
flow which exists because a CalDAV account can hold several calendars
|
|
worth picking between.
|
|
|
|
Pure functions -- no ORM, no FastAPI Depends -- same testability
|
|
philosophy as calendar_feed.py/caldav_client.py.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from urllib.parse import unquote, urljoin, urlsplit, urlunsplit
|
|
from xml.etree import ElementTree
|
|
|
|
import httpx
|
|
|
|
HTTP_TIMEOUT_S = 15.0
|
|
FETCH_MAX_BYTES = 10 * 1024 * 1024 # a whiteboard scene is KB, not MB -- sanity cap, not an expected size
|
|
|
|
_DAV_NS = "DAV:"
|
|
_PROPFIND_BODY = (
|
|
'<?xml version="1.0" encoding="utf-8" ?>'
|
|
f'<d:propfind xmlns:d="{_DAV_NS}"><d:prop>'
|
|
"<d:displayname/><d:resourcetype/></d:prop></d:propfind>"
|
|
)
|
|
|
|
|
|
class WebDavError(Exception):
|
|
"""Fetch failed -- network, auth, a missing file, or an oversized
|
|
response. Raised loudly; callers decide what to do."""
|
|
|
|
|
|
def fetch_file(url: str, username: str, password: str) -> bytes:
|
|
"""The raw bytes of one WebDAV file, HTTP Basic auth. That's the
|
|
whole protocol surface whiteboard mode needs -- Basic auth over
|
|
plain HTTP GET is what WebDAV file access boils down to once you
|
|
already have the exact URL, no PROPFIND/discovery involved."""
|
|
try:
|
|
with httpx.stream("GET", url, auth=(username, password), timeout=HTTP_TIMEOUT_S,
|
|
follow_redirects=True) as resp:
|
|
resp.raise_for_status()
|
|
chunks = []
|
|
total = 0
|
|
for chunk in resp.iter_bytes():
|
|
total += len(chunk)
|
|
if total > FETCH_MAX_BYTES:
|
|
raise WebDavError(f"File exceeds {FETCH_MAX_BYTES} bytes")
|
|
chunks.append(chunk)
|
|
return b"".join(chunks)
|
|
except httpx.HTTPError as e:
|
|
raise WebDavError(str(e)) from e
|
|
|
|
|
|
def list_directory(url: str, username: str, password: str) -> list[dict]:
|
|
"""One level of a WebDAV directory listing -- name/url/is_dir for
|
|
each entry, folders first then alphabetical. Powers the whiteboard
|
|
file-picker (browse instead of paste the exact file URL); a plain
|
|
Depth-1 PROPFIND for displayname + resourcetype is all that needs,
|
|
same "just enough protocol, not a full WebDAV client" scope as
|
|
fetch_file above."""
|
|
try:
|
|
resp = httpx.request(
|
|
"PROPFIND", url, auth=(username, password), timeout=HTTP_TIMEOUT_S,
|
|
follow_redirects=True, headers={"Depth": "1", "Content-Type": "application/xml"},
|
|
content=_PROPFIND_BODY,
|
|
)
|
|
resp.raise_for_status()
|
|
except httpx.HTTPError as e:
|
|
raise WebDavError(str(e)) from e
|
|
|
|
try:
|
|
root = ElementTree.fromstring(resp.content)
|
|
except ElementTree.ParseError as e:
|
|
raise WebDavError(f"Server returned an unparseable directory listing: {e}") from e
|
|
|
|
self_path = urlsplit(url).path.rstrip("/")
|
|
entries = []
|
|
for response_el in root.findall(f"{{{_DAV_NS}}}response"):
|
|
href_el = response_el.find(f"{{{_DAV_NS}}}href")
|
|
if href_el is None or not href_el.text:
|
|
continue
|
|
href = href_el.text
|
|
if urlsplit(href).path.rstrip("/") == self_path:
|
|
continue # the listed directory's own entry, not a child
|
|
|
|
propstat = response_el.find(f"{{{_DAV_NS}}}propstat")
|
|
prop = propstat.find(f"{{{_DAV_NS}}}prop") if propstat is not None else None
|
|
resourcetype = prop.find(f"{{{_DAV_NS}}}resourcetype") if prop is not None else None
|
|
is_dir = resourcetype is not None and resourcetype.find(f"{{{_DAV_NS}}}collection") is not None
|
|
|
|
displayname_el = prop.find(f"{{{_DAV_NS}}}displayname") if prop is not None else None
|
|
name = (displayname_el.text or "").strip() if displayname_el is not None else ""
|
|
if not name:
|
|
name = unquote(href.rstrip("/").rsplit("/", 1)[-1])
|
|
if not name:
|
|
continue
|
|
|
|
entries.append({"name": name, "url": urljoin(url, href), "is_dir": is_dir})
|
|
|
|
entries.sort(key=lambda e: (not e["is_dir"], e["name"].casefold()))
|
|
return entries
|
|
|
|
|
|
def parent_directory_url(base_url: str, current_url: str) -> str | None:
|
|
"""One level up from `current_url`, or None if `current_url` is
|
|
already at (or above) `base_url` -- the file-picker's "Up" button
|
|
doesn't wander outside the folder the user configured as their
|
|
browse root in Settings."""
|
|
base_path = urlsplit(base_url).path.rstrip("/")
|
|
current = urlsplit(current_url)
|
|
current_path = current.path.rstrip("/")
|
|
if len(current_path) <= len(base_path):
|
|
return None
|
|
parent_path = current_path.rsplit("/", 1)[0] or "/"
|
|
if len(parent_path) < len(base_path):
|
|
parent_path = base_path
|
|
return urlunsplit((current.scheme, current.netloc, parent_path + "/", "", ""))
|