93 lines
2.5 KiB
Python
93 lines
2.5 KiB
Python
import httpx
|
|
import frontmatter
|
|
|
|
import pathlib
|
|
import functools
|
|
import xml.etree.ElementTree as ET
|
|
|
|
from .models import Document, DocumentClient
|
|
|
|
|
|
class NextCloudDocument(Document):
|
|
"""
|
|
NextCloud documents are either Markdown files or XML blobs
|
|
returns by the WebDAV API.
|
|
|
|
If metadata is available as frontmatter, it's parsed and made
|
|
available, otherwise the 'content' property always reflects
|
|
what's in the document (minus the frontmatter).
|
|
"""
|
|
|
|
raw: str = ""
|
|
type: str
|
|
|
|
@functools.cached_property
|
|
def _parsed_markdown(self) -> tuple[dict[str, str], str]:
|
|
try:
|
|
return frontmatter.parse(self.raw)
|
|
except Exception as e:
|
|
print(e)
|
|
return tuple()
|
|
|
|
@property
|
|
def metadata(self) -> dict[str, str]:
|
|
return self._parsed_markdown[0]
|
|
|
|
@property
|
|
def content(self) -> str:
|
|
return self._parsed_markdown[1]
|
|
|
|
|
|
class NextCloudClient(DocumentClient, arbitrary_types_allowed=True):
|
|
"""
|
|
Interface to the Nextcloud API.
|
|
"""
|
|
|
|
base_url: str
|
|
|
|
# Api credentials
|
|
user: str
|
|
password: str
|
|
|
|
@property
|
|
def _client(self) -> httpx.Client:
|
|
return httpx.Client(
|
|
auth=httpx.BasicAuth(username=self.user, password=self.password),
|
|
verify=False,
|
|
)
|
|
|
|
def get_document_tree(self, root: str) -> NextCloudDocument:
|
|
root_path = f"remote.php/dav/files/{self.user}/{root}"
|
|
request = httpx.Request("PROPFIND", f"{self.base_url}/{root_path}")
|
|
|
|
with self._client as client:
|
|
response = client.send(request)
|
|
|
|
root_properties = ET.fromstring(response.text)
|
|
|
|
root_document = NextCloudDocument(path=root_path, type="collection")
|
|
|
|
for r in root_properties:
|
|
doc_path = r.find("{DAV:}href").text
|
|
|
|
if doc_path == root_document.path:
|
|
continue
|
|
|
|
full_document_path = f"{self.base_url}/{doc_path}"
|
|
root_document.children.append(self.get_document(full_document_path))
|
|
|
|
return root_document
|
|
|
|
def get_document(self, path: str) -> NextCloudDocument:
|
|
p = pathlib.Path(path)
|
|
|
|
with self._client as client:
|
|
verb = "PROPFIND" if not p.suffix else "GET"
|
|
request = httpx.Request(verb, path)
|
|
response = client.send(request)
|
|
|
|
return NextCloudDocument(
|
|
path=path,
|
|
raw=response.text,
|
|
type="collection" if not p.suffix else "document",
|
|
)
|