vmis/backend/app/services/nextcloud_client.py

"""
NextcloudClient — Nextcloud OCS API
管理 NC 使用者的查詢/建立與 quota 統計。
"""
import logging
from typing import Optional
import httpx

logger = logging.getLogger(__name__)

OCS_HEADERS = {"OCS-APIRequest": "true"}
TIMEOUT = 15.0


class NextcloudClient:
    def __init__(self, domain: str, admin_user: str = "admin", admin_password: str = ""):
        self._base = f"https://{domain}"
        self._auth = (admin_user, admin_password)

    def user_exists(self, username: str) -> bool:
        try:
            resp = httpx.get(
                f"{self._base}/ocs/v1.php/cloud/users/{username}",
                auth=self._auth,
                headers=OCS_HEADERS,
                timeout=TIMEOUT,
            )
            return resp.status_code == 200
        except Exception:
            return False

    def create_user(self, username: str, password: Optional[str], quota_gb: int = 20) -> bool:
        try:
            resp = httpx.post(
                f"{self._base}/ocs/v1.php/cloud/users",
                auth=self._auth,
                headers=OCS_HEADERS,
                data={
                    "userid": username,
                    "password": password or "",
                    "quota": f"{quota_gb}GB",
                },
                timeout=TIMEOUT,
            )
            return resp.status_code == 200
        except Exception as e:
            logger.error(f"NC create_user({username}) failed: {e}")
            return False

    def get_user_quota_used_gb(self, username: str) -> Optional[float]:
        try:
            resp = httpx.get(
                f"{self._base}/ocs/v2.php/cloud/users/{username}",
                auth=self._auth,
                headers=OCS_HEADERS,
                timeout=TIMEOUT,
            )
            if resp.status_code != 200:
                return None
            used_bytes = resp.json().get("ocs", {}).get("data", {}).get("quota", {}).get("used", 0)
            return round(used_bytes / 1073741824, 4)
        except Exception:
            return None

    def get_total_quota_used_gb(self) -> Optional[float]:
        """Sum all users' quota usage"""
        try:
            resp = httpx.get(
                f"{self._base}/ocs/v2.php/cloud/users",
                auth=self._auth,
                headers=OCS_HEADERS,
                params={"limit": 500},
                timeout=TIMEOUT,
            )
            if resp.status_code != 200:
                return None
            users = resp.json().get("ocs", {}).get("data", {}).get("users", [])
            total = 0.0
            for uid in users:
                used = self.get_user_quota_used_gb(uid)
                if used:
                    total += used
            return round(total, 4)
        except Exception:
            return None