duffyduck
b88c766dd6
Bug: nextcloudhost mit https://-Präfix in CSV/GUI führte zu Doppel-Schema (https://https://cloud.foo.de) und damit zu DNS-Fehler "host=https". - models.clean_host() entfernt http(s):// und Trailing-Slash; Account ruft das im __post_init__ für plesk-/kerio-/nextcloudhost auf. - NextcloudClient wickelt requests-Connection-Fehler in NextcloudError ein, damit die Log-Ausgabe lesbar bleibt statt HTTPSConnectionPool-Stacktrace. - README: Hostname-Eingabe als tolerant dokumentiert; pleskhost-Pflicht im manual-Modus klarer (POP3-Sammler braucht es trotzdem). Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
84 lines
3.3 KiB
Python
84 lines
3.3 KiB
Python
"""Nextcloud Provisioning API (OCS) Client."""
|
||
from typing import Optional
|
||
import requests
|
||
import urllib3
|
||
|
||
|
||
class NextcloudError(Exception):
|
||
pass
|
||
|
||
|
||
class NextcloudClient:
|
||
def __init__(self, host: str, *, admin_user: str, admin_password: str,
|
||
verify: bool = True, scheme: str = "https"):
|
||
if not host:
|
||
raise NextcloudError("Nextcloud-Host ist leer")
|
||
self.base = f"{scheme}://{host}"
|
||
self.session = requests.Session()
|
||
self.session.verify = verify
|
||
if not verify:
|
||
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
|
||
self.session.auth = (admin_user, admin_password)
|
||
self.session.headers["OCS-APIRequest"] = "true"
|
||
self.session.headers["Accept"] = "application/json"
|
||
|
||
def _meta(self, r: requests.Response, op: str) -> dict:
|
||
if not r.ok and r.status_code not in (400, 401, 403, 404):
|
||
raise NextcloudError(f"Nextcloud {op} HTTP {r.status_code}: {r.text[:300]}")
|
||
try:
|
||
data = r.json()
|
||
except ValueError:
|
||
raise NextcloudError(f"Nextcloud {op}: ungültige Antwort: {r.text[:300]}")
|
||
return data.get("ocs", {})
|
||
|
||
def _request(self, method: str, url: str, op: str, **kw):
|
||
try:
|
||
return self.session.request(method, url, timeout=30, **kw)
|
||
except requests.RequestException as e:
|
||
raise NextcloudError(
|
||
f"Nextcloud {op} – Verbindung zu {url} fehlgeschlagen: {e}"
|
||
) from e
|
||
|
||
def user_exists(self, userid: str) -> bool:
|
||
r = self._request("GET", f"{self.base}/ocs/v2.php/cloud/users/{userid}",
|
||
op="user-lookup")
|
||
ocs = self._meta(r, "user-lookup")
|
||
sc = ocs.get("meta", {}).get("statuscode")
|
||
return sc in (100, 200)
|
||
|
||
def ensure_group(self, group: str) -> None:
|
||
if not group:
|
||
return
|
||
r = self._request("POST", f"{self.base}/ocs/v2.php/cloud/groups",
|
||
op="group-create", data={"groupid": group})
|
||
ocs = self._meta(r, "group-create")
|
||
sc = ocs.get("meta", {}).get("statuscode")
|
||
# 100/200 = ok, 102 = exists already
|
||
if sc not in (100, 200, 102):
|
||
raise NextcloudError(
|
||
f"Nextcloud Gruppe '{group}': {sc} {ocs.get('meta', {}).get('message')}"
|
||
)
|
||
|
||
def create_user(self, *, userid: str, password: str,
|
||
email: Optional[str] = None,
|
||
display_name: Optional[str] = None,
|
||
group: Optional[str] = None,
|
||
quota_gb: Optional[int] = None) -> None:
|
||
body = [("userid", userid), ("password", password)]
|
||
if email:
|
||
body.append(("email", email))
|
||
if display_name:
|
||
body.append(("displayName", display_name))
|
||
if group:
|
||
body.append(("groups[]", group))
|
||
body.append(("quota", f"{quota_gb} GB" if quota_gb else "none"))
|
||
|
||
r = self._request("POST", f"{self.base}/ocs/v2.php/cloud/users",
|
||
op="user-create", data=body)
|
||
ocs = self._meta(r, "user-create")
|
||
sc = ocs.get("meta", {}).get("statuscode")
|
||
if sc not in (100, 200):
|
||
raise NextcloudError(
|
||
f"Nextcloud user-create: {sc} {ocs.get('meta', {}).get('message')}"
|
||
)
|