azcomputerguru/claudetools
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

sync: auto-sync from HOWARD-HOME at 2026-07-05 20:23:39

Browse Source 
Author: Howard Enos
Machine: HOWARD-HOME
Timestamp: 2026-07-05 20:23:39
This commit is contained in:
Howard Enos
2026-07-05 20:24:08 -07:00
parent 976bfe7957
commit a8e76ba215
10 changed files with 292 additions and 108 deletions
Download Patch File Download Diff File Expand all files Collapse all files

143
.claude/skills/owncloud/scripts/owncloud_client.py
View File

@@ -250,84 +250,115 @@ class OwncloudClient:
f"{(err or out).strip()[:300]}")
return out
def _occ_env_ok(self, env: dict, args: str,
timeout: int = EXEC_TIMEOUT) -> str:
"""occ_env variant of _occ_ok (occ_env is not wrapped by _occ_ok)."""
rc, out, err = self.occ_env(env, args, timeout=timeout)
if rc != 0:
label = args.split()[0] if args else "occ"
raise bc.BackupError(f"occ {label} failed (rc={rc}): "
f"{(err or out).strip()[:300]}")
return out
def _occ_json(self, args: str, timeout: int = EXEC_TIMEOUT) -> Any:
"""occ with --output=json; parse defensively so a non-JSON banner
(maintenance notice, PHP deprecation, update line) surfaces as a clean
BackupError rather than an uncaught JSONDecodeError traceback."""
out = self._occ_ok(args, timeout=timeout).strip()
try:
return json.loads(out or "null")
except json.JSONDecodeError:
label = args.split()[0] if args else "occ"
raise bc.BackupError(f"occ {label} returned non-JSON: {out[:300]}")
# -- user lifecycle --------------------------------------------------------
def create_user(self, uid: str, password: str,
display_name: Optional[str] = None,
email: Optional[str] = None,
groups: Optional[list[str]] = None) -> dict:
parts = [f"user:add {shlex.quote(uid)} --password-from-env -n"]
# options first, then `--`, then the positional uid so a uid that
# happens to start with '-' is never parsed by occ as an option flag.
parts = ["user:add --password-from-env -n"]
if display_name:
parts.append(f"--display-name={shlex.quote(display_name)}")
if email:
parts.append(f"--email={shlex.quote(email)}")
for g in (groups or []):
parts.append(f"-g {shlex.quote(g)}")
rc, out, err = self.occ_env({"OC_PASS": password}, " ".join(parts))
if rc != 0:
raise bc.BackupError(f"user:add {uid} failed (rc={rc}): "
f"{(err or out).strip()[:300]}")
parts.append(f"--group={shlex.quote(g)}")
parts.append(f"-- {shlex.quote(uid)}")
out = self._occ_env_ok({"OC_PASS": password}, " ".join(parts))
return {"uid": uid, "output": out.strip()}
def delete_user(self, uid: str, force: bool = False) -> str:
return self._occ_ok(f"user:delete {shlex.quote(uid)}"
+ (" -f" if force else ""), timeout=DU_TIMEOUT).strip()
return self._occ_ok(f"user:delete{' -f' if force else ''} -- "
f"{shlex.quote(uid)}", timeout=DU_TIMEOUT).strip()
def set_user_enabled(self, uid: str, enabled: bool) -> str:
verb = "user:enable" if enabled else "user:disable"
return self._occ_ok(f"{verb} {shlex.quote(uid)}").strip()
return self._occ_ok(f"{verb} -- {shlex.quote(uid)}").strip()
def reset_password(self, uid: str, password: str) -> str:
rc, out, err = self.occ_env(
return self._occ_env_ok(
{"OC_PASS": password},
f"user:resetpassword {shlex.quote(uid)} --password-from-env")
if rc != 0:
raise bc.BackupError(f"user:resetpassword {uid} failed (rc={rc}): "
f"{(err or out).strip()[:300]}")
return out.strip()
f"user:resetpassword --password-from-env -- {shlex.quote(uid)}").strip()
def modify_user(self, uid: str, key: str, value: str) -> str:
"""key is one of: displayname, email (per occ user:modify)."""
return self._occ_ok(
f"user:modify {shlex.quote(uid)} {shlex.quote(key)} {shlex.quote(value)}").strip()
f"user:modify -- {shlex.quote(uid)} {shlex.quote(key)} "
f"{shlex.quote(value)}").strip()
# -- quota (occ user:setting <uid> files quota) ----------------------------
def get_quota(self, uid: str) -> Optional[str]:
rc, out, err = self.occ(f"user:setting {shlex.quote(uid)} files quota")
return out.strip() or None if rc == 0 else None
"""Return the user's quota string, or None when it is unset (= default).
A non-existent user or a genuine occ failure raises BackupError - occ
returns rc=1 for both 'setting unset' and 'user missing', so we key off
the message ('setting does not exist' == unset)."""
rc, out, err = self.occ(f"user:setting -- {shlex.quote(uid)} files quota")
if rc == 0:
return out.strip() or None
msg = (out or err).strip()
if "setting does not exist" in msg.lower():
return None
raise bc.BackupError(f"occ user:setting quota failed (rc={rc}): {msg[:300]}")
def set_quota(self, uid: str, value: str) -> str:
return self._occ_ok(
f"user:setting {shlex.quote(uid)} files quota --value={shlex.quote(value)}").strip()
f"user:setting --value={shlex.quote(value)} "
f"-- {shlex.quote(uid)} files quota").strip()
def reset_quota(self, uid: str) -> str:
return self._occ_ok(
f"user:setting {shlex.quote(uid)} files quota --delete").strip()
f"user:setting --delete -- {shlex.quote(uid)} files quota").strip()
# -- groups ----------------------------------------------------------------
def list_groups(self) -> Any:
return json.loads(self._occ_ok("group:list --output=json") or "[]")
return self._occ_json("group:list --output=json")
def group_members(self, group: str) -> Any:
return json.loads(self._occ_ok(
f"group:list-members {shlex.quote(group)} --output=json") or "[]")
return self._occ_json(
f"group:list-members --output=json -- {shlex.quote(group)}")
def user_groups(self, uid: str) -> Any:
return json.loads(self._occ_ok(
f"user:list-groups {shlex.quote(uid)} --output=json") or "[]")
return self._occ_json(
f"user:list-groups --output=json -- {shlex.quote(uid)}")
def create_group(self, name: str) -> str:
return self._occ_ok(f"group:add {shlex.quote(name)}").strip()
return self._occ_ok(f"group:add -- {shlex.quote(name)}").strip()
def delete_group(self, name: str) -> str:
return self._occ_ok(f"group:delete {shlex.quote(name)}").strip()
return self._occ_ok(f"group:delete -- {shlex.quote(name)}").strip()
def add_group_members(self, group: str, members: list[str]) -> str:
ms = " ".join(f"-m {shlex.quote(m)}" for m in members)
return self._occ_ok(f"group:add-member {shlex.quote(group)} {ms}").strip()
# --member= binds each value even if it starts with '-'.
ms = " ".join(f"--member={shlex.quote(m)}" for m in members)
return self._occ_ok(
f"group:add-member {ms} -- {shlex.quote(group)}").strip()
def remove_group_members(self, group: str, members: list[str]) -> str:
ms = " ".join(f"-m {shlex.quote(m)}" for m in members)
return self._occ_ok(f"group:remove-member {shlex.quote(group)} {ms}").strip()
ms = " ".join(f"--member={shlex.quote(m)}" for m in members)
return self._occ_ok(
f"group:remove-member {ms} -- {shlex.quote(group)}").strip()
# -- shares (read-only, straight from the DB; occ has no share list) --------
SHARE_TYPES = {0: "user", 1: "group", 3: "public_link",
@@ -369,24 +400,35 @@ class OwncloudClient:
return self._occ_ok("maintenance:mode " + ("--on" if on else "--off")).strip()
def maintenance_status(self) -> bool:
rc, out, err = self.occ("config:system:get maintenance")
return rc == 0 and out.strip().lower() in ("true", "1", "yes")
# reuse config_get so a genuine occ failure raises (not a false 'off');
# the maintenance key is always set, so None would itself be anomalous.
val = self.config_get("maintenance")
return (val or "").strip().lower() in ("true", "1", "yes")
def list_apps(self) -> Any:
return json.loads(self._occ_ok("app:list --output=json") or "{}")
return self._occ_json("app:list --output=json")
def enable_app(self, app: str) -> str:
return self._occ_ok(f"app:enable {shlex.quote(app)}").strip()
return self._occ_ok(f"app:enable -- {shlex.quote(app)}").strip()
def disable_app(self, app: str) -> str:
return self._occ_ok(f"app:disable {shlex.quote(app)}").strip()
return self._occ_ok(f"app:disable -- {shlex.quote(app)}").strip()
def config_get(self, name: str) -> Optional[str]:
rc, out, err = self.occ(f"config:system:get {shlex.quote(name)}")
return out.strip() if rc == 0 else None
"""Return the system config value, or None when the key is genuinely
unset (occ rc=1 with empty output). A non-zero rc that carries an error
message is a real failure and raises, so callers never mistake a broken
occ read for 'value is unset'."""
rc, out, err = self.occ(f"config:system:get -- {shlex.quote(name)}")
if rc == 0:
return out.strip()
msg = (err or out).strip()
if not msg:
return None
raise bc.BackupError(f"occ config:system:get failed (rc={rc}): {msg[:300]}")
def config_list(self) -> Any:
return json.loads(self._occ_ok("config:list system --output=json") or "{}")
return self._occ_json("config:list system --output=json")
def config_set(self, name: str, value: str, vtype: str = "string") -> str:
return self._occ_ok(
@@ -396,33 +438,32 @@ class OwncloudClient:
# -- files admin -----------------------------------------------------------
def files_scan(self, uid: Optional[str] = None, all_users: bool = False,
path: Optional[str] = None) -> str:
opts = "files:scan"
if path:
opts += f" --path={shlex.quote(path)}"
if all_users:
args = "files:scan --all"
args = f"{opts} --all"
elif uid:
args = f"files:scan {shlex.quote(uid)}"
args = f"{opts} -- {shlex.quote(uid)}"
else:
raise bc.BackupError("files_scan requires a uid or all_users=True")
if path:
args += f" --path={shlex.quote(path)}"
return self._occ_ok(args, timeout=DU_TIMEOUT).strip()
def transfer_ownership(self, src: str, dst: str,
path: Optional[str] = None) -> str:
args = (f"files:transfer-ownership {shlex.quote(src)} "
f"{shlex.quote(dst)} -s")
args = "files:transfer-ownership -s"
if path:
args += f" --path={shlex.quote(path)}"
args += f" -- {shlex.quote(src)} {shlex.quote(dst)}"
return self._occ_ok(args, timeout=DU_TIMEOUT).strip()
def trashbin_cleanup(self, users: list[str]) -> str:
us = " ".join(shlex.quote(u) for u in users)
return self._occ_ok(f"trashbin:cleanup {us}".strip(),
timeout=DU_TIMEOUT).strip()
us = (" -- " + " ".join(shlex.quote(u) for u in users)) if users else ""
return self._occ_ok(f"trashbin:cleanup{us}", timeout=DU_TIMEOUT).strip()
def versions_cleanup(self, users: list[str]) -> str:
us = " ".join(shlex.quote(u) for u in users)
return self._occ_ok(f"versions:cleanup {us}".strip(),
timeout=DU_TIMEOUT).strip()
us = (" -- " + " ".join(shlex.quote(u) for u in users)) if users else ""
return self._occ_ok(f"versions:cleanup{us}", timeout=DU_TIMEOUT).strip()
def _int(v: Any) -> Optional[int]: