Atlas
eb5b67dee2
Product layer on top of the pure engine: a wizard to connect the apps that feed the brain, an encrypted credential vault (Fernet/PBKDF2; app logins never enter brain.db), and 3 working connectors. Connector data flows through the same credential-scrub path as the engine. Atlas Corporation proprietary.
49 lines
2.2 KiB
Python
49 lines
2.2 KiB
Python
# SecondBrain — (c) 2026 Atlas Corporation. All Rights Reserved. Proprietary; see LICENSE.
|
|
"""Connector: Nextcloud / any WebDAV — index text files from a cloud folder.
|
|
For Nextcloud the base URL looks like: https://cloud.example.com/remote.php/dav/files/USERNAME/
|
|
Use a Nextcloud *app password*, not your login password."""
|
|
import os
|
|
import requests
|
|
from xml.etree import ElementTree as ET
|
|
from .base import Connector, Field, register, feed
|
|
|
|
TEXT_EXT = (".md", ".txt", ".rst", ".py", ".sh", ".json", ".csv", ".tsv", ".yml", ".yaml",
|
|
".ini", ".cfg", ".conf", ".html", ".js", ".ts", ".sql", ".xml")
|
|
|
|
@register
|
|
class WebDavConnector(Connector):
|
|
id = "webdav"
|
|
name = "Nextcloud / WebDAV files"
|
|
blurb = "Index text files from a Nextcloud (or any WebDAV) folder."
|
|
fields = [
|
|
Field("base_url", "WebDAV base URL (…/remote.php/dav/files/USER/)"),
|
|
Field("user", "Username"),
|
|
Field("password", "App password", secret=True),
|
|
Field("folder", "Folder path under base", default="/"),
|
|
]
|
|
|
|
def _list(self, cfg):
|
|
url = cfg["base_url"].rstrip("/") + "/" + cfg.get("folder", "/").strip("/")
|
|
r = requests.request("PROPFIND", url, auth=(cfg["user"], cfg["password"]),
|
|
headers={"Depth": "infinity"}, timeout=30)
|
|
r.raise_for_status()
|
|
hrefs = [e.text for e in ET.fromstring(r.content).iter("{DAV:}href")]
|
|
return [h for h in hrefs if h and h.lower().endswith(TEXT_EXT)]
|
|
|
|
def test(self, cfg):
|
|
return f"{len(self._list(cfg))} text files visible"
|
|
|
|
def sync(self, cfg):
|
|
host = cfg["base_url"].split("/remote.php")[0]
|
|
items = []
|
|
for href in self._list(cfg):
|
|
u = href if href.startswith("http") else host + href
|
|
try:
|
|
rr = requests.get(u, auth=(cfg["user"], cfg["password"]), timeout=30)
|
|
if rr.status_code != 200 or len(rr.content) > 400000:
|
|
continue
|
|
items.append({"text": rr.text, "project": "nextcloud", "path": href,
|
|
"ref": os.path.basename(href.rstrip("/")), "role": "file"})
|
|
except Exception:
|
|
continue
|
|
return feed(items, "file")
|