691 lines
25 KiB
Python
691 lines
25 KiB
Python
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass
|
|
from datetime import datetime, timezone
|
|
import re
|
|
import time
|
|
from typing import Any
|
|
|
|
import httpx
|
|
import psycopg
|
|
|
|
from ..k8s.exec import ExecError, PodExecutor
|
|
from ..k8s.pods import PodSelectionError
|
|
from ..settings import settings
|
|
from ..utils.logging import get_logger
|
|
from ..utils.passwords import random_password
|
|
from .keycloak_admin import keycloak_admin
|
|
|
|
|
|
logger = get_logger(__name__)
|
|
|
|
|
|
def _extract_attr(attrs: Any, key: str) -> str:
|
|
if not isinstance(attrs, dict):
|
|
return ""
|
|
raw = attrs.get(key)
|
|
if isinstance(raw, list):
|
|
for item in raw:
|
|
if isinstance(item, str) and item.strip():
|
|
return item.strip()
|
|
return ""
|
|
if isinstance(raw, str) and raw.strip():
|
|
return raw.strip()
|
|
return ""
|
|
|
|
|
|
def _resolve_mailu_email(username: str, user: dict[str, Any]) -> str:
|
|
attrs = user.get("attributes")
|
|
mailu_email = _extract_attr(attrs, "mailu_email")
|
|
if mailu_email:
|
|
return mailu_email
|
|
email = user.get("email")
|
|
if isinstance(email, str) and email.strip():
|
|
email = email.strip()
|
|
if email.lower().endswith(f"@{settings.mailu_domain.lower()}"):
|
|
return email
|
|
return f"{username}@{settings.mailu_domain}"
|
|
|
|
|
|
def _parse_mail_export(output: str) -> list[tuple[str, str]]:
|
|
accounts: list[tuple[str, str]] = []
|
|
account_id = ""
|
|
for line in output.splitlines():
|
|
line = line.strip()
|
|
if not line:
|
|
continue
|
|
match = re.match(r"^Account\s+(\d+):", line, flags=re.IGNORECASE)
|
|
if match:
|
|
account_id = match.group(1)
|
|
continue
|
|
match = re.match(r"^-\s*E-?mail:\s*(\S+)", line, flags=re.IGNORECASE)
|
|
if match and account_id:
|
|
accounts.append((account_id, match.group(1)))
|
|
return accounts
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class NextcloudMailSyncSummary:
|
|
processed: int
|
|
created: int
|
|
updated: int
|
|
deleted: int
|
|
skipped: int
|
|
failures: int
|
|
detail: str = ""
|
|
|
|
|
|
@dataclass
|
|
class MailSyncCounters:
|
|
processed: int = 0
|
|
created: int = 0
|
|
updated: int = 0
|
|
deleted: int = 0
|
|
skipped: int = 0
|
|
failures: int = 0
|
|
last_error: str = ""
|
|
|
|
def summary(self) -> NextcloudMailSyncSummary:
|
|
return NextcloudMailSyncSummary(
|
|
processed=self.processed,
|
|
created=self.created,
|
|
updated=self.updated,
|
|
deleted=self.deleted,
|
|
skipped=self.skipped,
|
|
failures=self.failures,
|
|
detail=self.last_error,
|
|
)
|
|
|
|
def status(self) -> str:
|
|
return "ok" if self.failures == 0 else "error"
|
|
|
|
def record_failure(self, detail: str) -> None:
|
|
self.failures += 1
|
|
if detail and not self.last_error:
|
|
self.last_error = detail
|
|
|
|
|
|
class NextcloudService:
|
|
def __init__(self) -> None:
|
|
self._executor = PodExecutor(
|
|
settings.nextcloud_namespace,
|
|
settings.nextcloud_pod_label,
|
|
settings.nextcloud_container,
|
|
)
|
|
|
|
def _exec_with_fallback(
|
|
self,
|
|
primary: list[str],
|
|
fallback: list[str],
|
|
env: dict[str, str] | None = None,
|
|
check: bool = True,
|
|
) -> ExecResult:
|
|
try:
|
|
result = self._executor.exec(
|
|
primary,
|
|
env=env,
|
|
timeout_sec=settings.nextcloud_exec_timeout_sec,
|
|
check=check,
|
|
)
|
|
except ExecError as exc:
|
|
if "runuser: may not be used by non-root users" not in str(exc):
|
|
raise
|
|
return self._executor.exec(
|
|
fallback,
|
|
env=env,
|
|
timeout_sec=settings.nextcloud_exec_timeout_sec,
|
|
check=check,
|
|
)
|
|
|
|
if not result.ok and "runuser: may not be used by non-root users" in result.stderr:
|
|
return self._executor.exec(
|
|
fallback,
|
|
env=env,
|
|
timeout_sec=settings.nextcloud_exec_timeout_sec,
|
|
check=check,
|
|
)
|
|
return result
|
|
|
|
def _occ_exec(
|
|
self,
|
|
args: list[str],
|
|
env: dict[str, str] | None = None,
|
|
check: bool = True,
|
|
) -> ExecResult:
|
|
command = ["runuser", "-u", "www-data", "--", "php", "/var/www/html/occ", *args]
|
|
fallback = ["php", "/var/www/html/occ", *args]
|
|
return self._exec_with_fallback(command, fallback, env=env, check=check)
|
|
|
|
def _occ(self, args: list[str]) -> str:
|
|
result = self._occ_exec(args, check=True)
|
|
return result.stdout
|
|
|
|
def _display_name(self, user: dict[str, Any]) -> str:
|
|
first = user.get("firstName") if isinstance(user.get("firstName"), str) else ""
|
|
last = user.get("lastName") if isinstance(user.get("lastName"), str) else ""
|
|
first = first.strip()
|
|
last = last.strip()
|
|
if first and last:
|
|
return f"{first} {last}"
|
|
return last or first
|
|
|
|
def _ensure_nextcloud_user(
|
|
self,
|
|
username: str,
|
|
mailu_email: str,
|
|
display_name: str,
|
|
) -> None:
|
|
result = self._occ_exec(["user:info", username], check=False)
|
|
if result.ok:
|
|
return
|
|
|
|
detail = f"{result.stdout}\n{result.stderr}".strip().lower()
|
|
missing_markers = ("not found", "not exist", "does not exist", "unknown user")
|
|
if detail and not any(marker in detail for marker in missing_markers):
|
|
raise RuntimeError(f"nextcloud user lookup failed: {detail[:200]}")
|
|
|
|
password = random_password(24)
|
|
env = {"OC_PASS": password}
|
|
args = ["user:add", "--password-from-env"]
|
|
name = display_name or username
|
|
if name:
|
|
args += ["--display-name", name]
|
|
if mailu_email:
|
|
args += ["--email", mailu_email]
|
|
args.append(username)
|
|
self._occ_exec(args, env=env, check=True)
|
|
|
|
def run_cron(self) -> dict[str, Any]:
|
|
if not settings.nextcloud_namespace:
|
|
raise RuntimeError("nextcloud cron not configured")
|
|
try:
|
|
self._exec_with_fallback(
|
|
["runuser", "-u", "www-data", "--", "php", "-f", "/var/www/html/cron.php"],
|
|
["php", "-f", "/var/www/html/cron.php"],
|
|
)
|
|
except (ExecError, PodSelectionError, TimeoutError) as exc:
|
|
return {"status": "error", "detail": str(exc)}
|
|
return {"status": "ok"}
|
|
|
|
def _list_mail_accounts(self, username: str) -> list[tuple[str, str]]:
|
|
output = self._occ(["mail:account:export", username])
|
|
return _parse_mail_export(output)
|
|
|
|
def _set_editor_mode_richtext(self, account_ids: list[str]) -> None:
|
|
safe_ids = [item for item in account_ids if item.isdigit()]
|
|
if not safe_ids:
|
|
return
|
|
if not settings.nextcloud_db_host or not settings.nextcloud_db_password:
|
|
logger.info(
|
|
"nextcloud editor_mode skipped",
|
|
extra={"event": "nextcloud_mail_editor_mode", "status": "skip", "reason": "missing db config"},
|
|
)
|
|
return
|
|
ids_csv = ",".join(safe_ids)
|
|
query = (
|
|
"UPDATE oc_mail_accounts SET editor_mode='richtext' "
|
|
f"WHERE id IN ({ids_csv}) AND editor_mode <> 'richtext';"
|
|
)
|
|
try:
|
|
with psycopg.connect(
|
|
host=settings.nextcloud_db_host,
|
|
port=settings.nextcloud_db_port,
|
|
dbname=settings.nextcloud_db_name,
|
|
user=settings.nextcloud_db_user,
|
|
password=settings.nextcloud_db_password,
|
|
) as conn:
|
|
with conn.cursor() as cur:
|
|
cur.execute(query)
|
|
except Exception as exc:
|
|
logger.info(
|
|
"nextcloud editor_mode update failed",
|
|
extra={"event": "nextcloud_mail_editor_mode", "status": "error", "detail": str(exc)},
|
|
)
|
|
|
|
def _set_user_mail_meta(self, user_id: str, primary_email: str, account_count: int) -> None:
|
|
synced_at = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())
|
|
attrs = {
|
|
"nextcloud_mail_primary_email": [primary_email],
|
|
"nextcloud_mail_account_count": [str(account_count)],
|
|
"nextcloud_mail_synced_at": [synced_at],
|
|
}
|
|
try:
|
|
keycloak_admin.update_user_safe(user_id, {"attributes": attrs})
|
|
except Exception:
|
|
return
|
|
|
|
def _collect_users(self, username: str | None) -> list[dict[str, Any]]:
|
|
if username is not None:
|
|
user = keycloak_admin.find_user(username)
|
|
return [user] if user else []
|
|
return list(keycloak_admin.iter_users(page_size=200, brief=False))
|
|
|
|
def _normalize_user(self, user: dict[str, Any]) -> tuple[str, str, dict[str, Any]] | None:
|
|
username_val = user.get("username") if isinstance(user.get("username"), str) else ""
|
|
username_val = username_val.strip()
|
|
if not username_val:
|
|
return None
|
|
if user.get("enabled") is False:
|
|
return None
|
|
if user.get("serviceAccountClientId") or username_val.startswith("service-account-"):
|
|
return None
|
|
|
|
user_id = user.get("id") if isinstance(user.get("id"), str) else ""
|
|
full_user = user
|
|
if user_id:
|
|
try:
|
|
full_user = keycloak_admin.get_user(user_id)
|
|
except Exception:
|
|
full_user = user
|
|
return username_val, user_id, full_user
|
|
|
|
def _list_mail_accounts_safe(
|
|
self,
|
|
username: str,
|
|
counters: MailSyncCounters,
|
|
) -> list[tuple[str, str]] | None:
|
|
try:
|
|
return self._list_mail_accounts(username)
|
|
except Exception as exc:
|
|
detail = f"mail export failed: {exc}"
|
|
counters.record_failure(detail)
|
|
logger.info(
|
|
"nextcloud mail export failed",
|
|
extra={"event": "nextcloud_mail_export", "status": "error", "detail": detail},
|
|
)
|
|
return None
|
|
|
|
def _select_primary_account(
|
|
self,
|
|
mailu_accounts: list[tuple[str, str]],
|
|
mailu_email: str,
|
|
) -> tuple[str, str]:
|
|
primary_id = ""
|
|
primary_email = ""
|
|
for account_id, account_email in mailu_accounts:
|
|
if not primary_id:
|
|
primary_id = account_id
|
|
primary_email = account_email
|
|
if account_email.lower() == mailu_email.lower():
|
|
primary_id = account_id
|
|
primary_email = account_email
|
|
break
|
|
return primary_id, primary_email
|
|
|
|
def _update_mail_account(
|
|
self,
|
|
username: str,
|
|
primary_id: str,
|
|
mailu_email: str,
|
|
app_pw: str,
|
|
) -> str | None:
|
|
try:
|
|
self._occ(
|
|
[
|
|
"mail:account:update",
|
|
"-q",
|
|
primary_id,
|
|
"--name",
|
|
username,
|
|
"--email",
|
|
mailu_email,
|
|
"--imap-host",
|
|
settings.mailu_host,
|
|
"--imap-port",
|
|
"993",
|
|
"--imap-ssl-mode",
|
|
"ssl",
|
|
"--imap-user",
|
|
mailu_email,
|
|
"--imap-password",
|
|
app_pw,
|
|
"--smtp-host",
|
|
settings.mailu_host,
|
|
"--smtp-port",
|
|
"587",
|
|
"--smtp-ssl-mode",
|
|
"tls",
|
|
"--smtp-user",
|
|
mailu_email,
|
|
"--smtp-password",
|
|
app_pw,
|
|
"--auth-method",
|
|
"password",
|
|
]
|
|
)
|
|
return None
|
|
except Exception as exc:
|
|
return str(exc)
|
|
|
|
def _create_mail_account(self, username: str, mailu_email: str, app_pw: str) -> str | None:
|
|
try:
|
|
self._occ(
|
|
[
|
|
"mail:account:create",
|
|
"-q",
|
|
username,
|
|
username,
|
|
mailu_email,
|
|
settings.mailu_host,
|
|
"993",
|
|
"ssl",
|
|
mailu_email,
|
|
app_pw,
|
|
settings.mailu_host,
|
|
"587",
|
|
"tls",
|
|
mailu_email,
|
|
app_pw,
|
|
"password",
|
|
]
|
|
)
|
|
return None
|
|
except Exception as exc:
|
|
return str(exc)
|
|
|
|
def _delete_extra_accounts(
|
|
self,
|
|
mailu_accounts: list[tuple[str, str]],
|
|
primary_id: str,
|
|
counters: MailSyncCounters,
|
|
) -> int:
|
|
deleted = 0
|
|
for account_id, _account_email in mailu_accounts:
|
|
if account_id == primary_id:
|
|
continue
|
|
try:
|
|
self._occ(["mail:account:delete", "-q", account_id])
|
|
deleted += 1
|
|
except Exception as exc:
|
|
counters.record_failure(f"mail account delete failed: {exc}")
|
|
return deleted
|
|
|
|
def _mailu_accounts(self, accounts: list[tuple[str, str]]) -> list[tuple[str, str]]:
|
|
return [
|
|
(account_id, email)
|
|
for account_id, email in accounts
|
|
if email.lower().endswith(f"@{settings.mailu_domain.lower()}")
|
|
]
|
|
|
|
def _summarize_mail_accounts(
|
|
self,
|
|
accounts: list[tuple[str, str]],
|
|
mailu_email: str,
|
|
) -> tuple[int, str, list[str]]:
|
|
mailu_accounts = self._mailu_accounts(accounts)
|
|
account_count = len(mailu_accounts)
|
|
primary_email = ""
|
|
editor_mode_ids: list[str] = []
|
|
for account_id, account_email in mailu_accounts:
|
|
editor_mode_ids.append(account_id)
|
|
if account_email.lower() == mailu_email.lower():
|
|
primary_email = account_email
|
|
break
|
|
if not primary_email:
|
|
primary_email = account_email
|
|
return account_count, primary_email, editor_mode_ids
|
|
|
|
def _mail_sync_context(
|
|
self,
|
|
user: dict[str, Any],
|
|
counters: MailSyncCounters,
|
|
) -> tuple[str, str, str, str, dict[str, Any]] | None:
|
|
normalized = self._normalize_user(user)
|
|
if not normalized:
|
|
counters.skipped += 1
|
|
return None
|
|
username, user_id, full_user = normalized
|
|
attrs = full_user.get("attributes") if isinstance(full_user.get("attributes"), dict) else {}
|
|
mailu_email = _resolve_mailu_email(username, full_user)
|
|
app_pw = _extract_attr(attrs, "mailu_app_password")
|
|
if not mailu_email or not app_pw:
|
|
counters.skipped += 1
|
|
return None
|
|
if mailu_email and not _extract_attr(attrs, "mailu_email"):
|
|
try:
|
|
keycloak_admin.set_user_attribute(username, "mailu_email", mailu_email)
|
|
except Exception:
|
|
pass
|
|
return username, user_id, mailu_email, app_pw, full_user
|
|
|
|
def _sync_mail_accounts(
|
|
self,
|
|
username: str,
|
|
mailu_email: str,
|
|
app_pw: str,
|
|
accounts: list[tuple[str, str]],
|
|
counters: MailSyncCounters,
|
|
) -> bool:
|
|
mailu_accounts = self._mailu_accounts(accounts)
|
|
if mailu_accounts:
|
|
primary_id, _primary_email = self._select_primary_account(mailu_accounts, mailu_email)
|
|
error = self._update_mail_account(username, primary_id, mailu_email, app_pw)
|
|
if error:
|
|
counters.record_failure(f"mail account update failed: {error}")
|
|
return False
|
|
counters.updated += 1
|
|
counters.deleted += self._delete_extra_accounts(mailu_accounts, primary_id, counters)
|
|
else:
|
|
error = self._create_mail_account(username, mailu_email, app_pw)
|
|
if error:
|
|
counters.record_failure(f"mail account create failed: {error}")
|
|
return False
|
|
counters.created += 1
|
|
return True
|
|
|
|
def _apply_mail_metadata(
|
|
self,
|
|
user_id: str,
|
|
mailu_email: str,
|
|
accounts: list[tuple[str, str]],
|
|
) -> None:
|
|
account_count, primary_email, editor_mode_ids = self._summarize_mail_accounts(accounts, mailu_email)
|
|
self._set_editor_mode_richtext(editor_mode_ids)
|
|
if user_id:
|
|
self._set_user_mail_meta(user_id, primary_email, account_count)
|
|
|
|
def _sync_user_mail(self, user: dict[str, Any], counters: MailSyncCounters) -> None:
|
|
context = self._mail_sync_context(user, counters)
|
|
if not context:
|
|
return
|
|
username, user_id, mailu_email, app_pw, full_user = context
|
|
|
|
try:
|
|
display_name = self._display_name(full_user)
|
|
self._ensure_nextcloud_user(username, mailu_email, display_name)
|
|
except Exception as exc:
|
|
counters.record_failure(f"nextcloud user ensure failed: {exc}")
|
|
return
|
|
|
|
accounts = self._list_mail_accounts_safe(username, counters)
|
|
if accounts is None:
|
|
return
|
|
|
|
counters.processed += 1
|
|
if not self._sync_mail_accounts(username, mailu_email, app_pw, accounts, counters):
|
|
return
|
|
|
|
accounts_after = self._list_mail_accounts_safe(username, counters)
|
|
if accounts_after is None:
|
|
return
|
|
|
|
self._apply_mail_metadata(user_id, mailu_email, accounts_after)
|
|
|
|
def sync_mail(self, username: str | None = None, wait: bool = True) -> dict[str, Any]:
|
|
if not settings.nextcloud_namespace:
|
|
raise RuntimeError("nextcloud mail sync not configured")
|
|
cleaned_username = None
|
|
if username is not None:
|
|
cleaned_username = username.strip()
|
|
if not cleaned_username:
|
|
raise RuntimeError("missing username")
|
|
if not keycloak_admin.ready():
|
|
return {"status": "error", "detail": "keycloak admin not configured"}
|
|
|
|
users = self._collect_users(cleaned_username)
|
|
if cleaned_username is not None and not users:
|
|
return {"status": "ok", "detail": "no matching user"}
|
|
|
|
counters = MailSyncCounters()
|
|
for user in users:
|
|
self._sync_user_mail(user, counters)
|
|
|
|
summary = counters.summary()
|
|
|
|
logger.info(
|
|
"nextcloud mail sync finished",
|
|
extra={
|
|
"event": "nextcloud_mail_sync",
|
|
"status": counters.status(),
|
|
"processed_count": counters.processed,
|
|
"created_count": counters.created,
|
|
"updated_count": counters.updated,
|
|
"deleted_count": counters.deleted,
|
|
"skipped_count": counters.skipped,
|
|
"failures_count": counters.failures,
|
|
"detail": summary.detail,
|
|
},
|
|
)
|
|
|
|
return {"status": counters.status(), "summary": summary, "detail": summary.detail}
|
|
|
|
def _run_shell(self, script: str, check: bool = True) -> None:
|
|
self._executor.exec(
|
|
script,
|
|
timeout_sec=settings.nextcloud_exec_timeout_sec,
|
|
check=check,
|
|
)
|
|
|
|
def _external_api(self, method: str, path: str, data: dict[str, Any] | None = None) -> dict[str, Any]:
|
|
if not settings.nextcloud_url:
|
|
raise RuntimeError("nextcloud url not configured")
|
|
if not settings.nextcloud_admin_user or not settings.nextcloud_admin_password:
|
|
raise RuntimeError("nextcloud admin credentials missing")
|
|
url = f"{settings.nextcloud_url}/ocs/v2.php/apps/external/api/v1{path}"
|
|
headers = {"OCS-APIRequest": "true"}
|
|
with httpx.Client(timeout=settings.nextcloud_exec_timeout_sec) as client:
|
|
resp = client.request(
|
|
method,
|
|
url,
|
|
headers=headers,
|
|
auth=(settings.nextcloud_admin_user, settings.nextcloud_admin_password),
|
|
data=data,
|
|
)
|
|
resp.raise_for_status()
|
|
try:
|
|
return resp.json()
|
|
except Exception:
|
|
return {}
|
|
|
|
def run_maintenance(self) -> dict[str, Any]:
|
|
if not settings.nextcloud_namespace:
|
|
raise RuntimeError("nextcloud maintenance not configured")
|
|
|
|
try:
|
|
self._run_shell(
|
|
"""
|
|
set -euo pipefail
|
|
if [ ! -d /var/www/html/lib ] && [ -d /usr/src/nextcloud/lib ]; then
|
|
if command -v rsync >/dev/null 2>&1; then
|
|
rsync -a --delete --exclude config --exclude data /usr/src/nextcloud/ /var/www/html/
|
|
else
|
|
cp -a /usr/src/nextcloud/. /var/www/html/
|
|
fi
|
|
fi
|
|
mkdir -p /var/www/html/data
|
|
chown 33:33 /var/www/html || true
|
|
chmod 775 /var/www/html || true
|
|
chown -R 33:33 /var/www/html/apps /var/www/html/custom_apps /var/www/html/data /var/www/html/config 2>/dev/null || true
|
|
""",
|
|
check=False,
|
|
)
|
|
|
|
self._occ(["config:app:set", "theming", "name", "--value", "Atlas Cloud"])
|
|
self._occ(["config:app:set", "theming", "slogan", "--value", "Unified access to Atlas services"])
|
|
theming_url = settings.nextcloud_url or "https://cloud.bstein.dev"
|
|
self._occ(["config:app:set", "theming", "url", "--value", theming_url])
|
|
self._occ(["config:app:set", "theming", "color", "--value", "#0f172a"])
|
|
self._occ(["config:app:set", "theming", "disable-user-theming", "--value", "yes"])
|
|
|
|
self._executor.exec(
|
|
["runuser", "-u", "www-data", "--", "php", "/var/www/html/occ", "app:install", "customcss"],
|
|
timeout_sec=settings.nextcloud_exec_timeout_sec,
|
|
check=False,
|
|
)
|
|
self._executor.exec(
|
|
["runuser", "-u", "www-data", "--", "php", "/var/www/html/occ", "app:enable", "customcss"],
|
|
timeout_sec=settings.nextcloud_exec_timeout_sec,
|
|
check=False,
|
|
)
|
|
|
|
mail_css = (
|
|
".mail-message-body, .mail-message-body pre, .mail-message-body code, .mail-message-body table {\n"
|
|
" font-family: \"Inter\", \"Source Sans 3\", \"Helvetica Neue\", Arial, sans-serif;\n"
|
|
" font-size: 14px;\n"
|
|
" line-height: 1.6;\n"
|
|
" color: var(--color-main-text);\n"
|
|
"}\n"
|
|
".mail-message-body pre {\n"
|
|
" background: rgba(15, 23, 42, 0.06);\n"
|
|
" padding: 12px;\n"
|
|
" border-radius: 8px;\n"
|
|
"}\n"
|
|
".mail-message-body blockquote {\n"
|
|
" border-left: 3px solid var(--color-border);\n"
|
|
" padding-left: 12px;\n"
|
|
" margin: 8px 0;\n"
|
|
" color: var(--color-text-lighter);\n"
|
|
"}\n"
|
|
".mail-message-body img {\n"
|
|
" max-width: 100%;\n"
|
|
" border-radius: 6px;\n"
|
|
"}\n"
|
|
)
|
|
self._occ(["config:app:set", "customcss", "css", "--value", mail_css])
|
|
self._occ(["config:app:set", "files", "default_quota", "--value", "250 GB"])
|
|
|
|
payload = self._external_api("GET", "?format=json")
|
|
links = payload.get("ocs", {}).get("data", []) if isinstance(payload, dict) else []
|
|
for link in links:
|
|
link_id = link.get("id") if isinstance(link, dict) else None
|
|
if link_id is not None:
|
|
self._external_api("DELETE", f"/sites/{link_id}?format=json")
|
|
|
|
sites = [
|
|
("Vaultwarden", "https://vault.bstein.dev"),
|
|
("Jellyfin", "https://stream.bstein.dev"),
|
|
("Gitea", "https://scm.bstein.dev"),
|
|
("Jenkins", "https://ci.bstein.dev"),
|
|
("Harbor", "https://registry.bstein.dev"),
|
|
("Vault", "https://secret.bstein.dev"),
|
|
("Jitsi", "https://meet.bstein.dev"),
|
|
("Grafana", "https://metrics.bstein.dev"),
|
|
("Chat LLM", "https://chat.ai.bstein.dev"),
|
|
("Vision", "https://draw.ai.bstein.dev"),
|
|
("STT/TTS", "https://talk.ai.bstein.dev"),
|
|
]
|
|
for name, url in sites:
|
|
self._external_api(
|
|
"POST",
|
|
"/sites?format=json",
|
|
data={
|
|
"name": name,
|
|
"url": url,
|
|
"lang": "",
|
|
"type": "link",
|
|
"device": "",
|
|
"icon": "",
|
|
"groups[]": "",
|
|
"redirect": "1",
|
|
},
|
|
)
|
|
except (ExecError, PodSelectionError, TimeoutError) as exc:
|
|
return {"status": "error", "detail": str(exc)}
|
|
except Exception as exc: # noqa: BLE001
|
|
return {"status": "error", "detail": str(exc)}
|
|
|
|
return {"status": "ok", "detail": "maintenance complete"}
|
|
|
|
|
|
nextcloud = NextcloudService()
|