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 set_user_quota(self, username: str, quota_gb: int) -> bool:
        try:
            resp = httpx.put(
                f"{self._base}/ocs/v1.php/cloud/users/{username}",
                auth=self._auth,
                headers=OCS_HEADERS,
                data={"key": "quota", "value": f"{quota_gb}GB"},
                timeout=TIMEOUT,
            )
            return resp.status_code == 200
        except Exception as e:
            logger.error(f"NC set_user_quota({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
            quota = resp.json().get("ocs", {}).get("data", {}).get("quota", {})
            if not isinstance(quota, dict):
                return 0.0
            used_bytes = quota.get("used", 0) or 0
            return round(used_bytes / 1073741824, 4)
        except Exception:
            return None

    def subscribe_calendar(
        self,
        username: str,
        cal_slug: str,
        display_name: str,
        ics_url: str,
        color: str = "#e9322d",
    ) -> bool:
        """
        為使用者建立訂閱行事曆（CalDAV MKCALENDAR with calendarserver source）。
        已存在(405)視為成功；201=新建成功。
        """
        try:
            body = (
                '<?xml version="1.0" encoding="UTF-8"?>\n'
                '<x1:mkcalendar xmlns:x1="urn:ietf:params:xml:ns:caldav">\n'
                '  <x0:set xmlns:x0="DAV:">\n'
                '    <x0:prop>\n'
                f'      <x0:displayname>{display_name}</x0:displayname>\n'
                f'      <x2:calendar-color xmlns:x2="http://apple.com/ns/ical/">{color}</x2:calendar-color>\n'
                '      <x3:source xmlns:x3="http://calendarserver.org/ns/">\n'
                f'        <x0:href>{ics_url}</x0:href>\n'
                '      </x3:source>\n'
                '    </x0:prop>\n'
                '  </x0:set>\n'
                '</x1:mkcalendar>'
            )
            url = f"{self._base}/remote.php/dav/calendars/{username}/{cal_slug}/"
            resp = httpx.request(
                "MKCALENDAR",
                url,
                auth=self._auth,
                content=body.encode("utf-8"),
                headers={"Content-Type": "application/xml; charset=utf-8"},
                timeout=TIMEOUT,
                verify=False,
            )
            # 201=建立成功, 405=已存在(Method Not Allowed on existing resource)
            return resp.status_code in (201, 405)
        except Exception as e:
            logger.error(f"NC subscribe_calendar({username}, {cal_slug}) failed: {e}")
            return False

    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