#!/usr/bin/env python3 """Synology DSM API 共用 library。""" from __future__ import annotations import json import os import ssl import urllib.parse import urllib.request import urllib.error from dataclasses import dataclass from http.cookiejar import CookieJar from pathlib import Path from typing import Any def load_dotenv(path: str = ".env") -> None: """最小 .env 讀取器：只在環境變數尚未設定時補值。""" candidates = [ Path(path), Path.cwd() / ".env", Path(__file__).resolve().parents[3] / ".env", ] seen: set[Path] = set() for p in candidates: p = p.resolve() if p in seen or not p.exists(): continue seen.add(p) for raw_line in p.read_text(encoding="utf-8").splitlines(): line = raw_line.strip() if not line or line.startswith("#") or "=" not in line: continue key, value = line.split("=", 1) key = key.strip() value = value.strip().strip('"').strip("'") if key and key not in os.environ: os.environ[key] = value @dataclass class DSMLoginResult: success: bool data: dict[str, Any] meta: dict[str, Any] class DSMApiClient: """可重用的 DSM API client。""" def __init__(self, base_url: str, timeout: int = 25, verify_ssl: bool = False) -> None: self.base_url = self.normalize_base(base_url) self.timeout = timeout self.verify_ssl = verify_ssl self.sid: str = "" self.cookie_jar = CookieJar() self._opener = urllib.request.build_opener( urllib.request.HTTPSHandler(context=self._build_ssl_context()), urllib.request.HTTPCookieProcessor(self.cookie_jar), ) @staticmethod def normalize_base(url: str) -> str: u = url.strip().rstrip("/") if not u.startswith("http://") and not u.startswith("https://"): u = "https://" + u return u def _build_ssl_context(self) -> ssl.SSLContext: ctx = ssl.create_default_context() if not self.verify_ssl: ctx.check_hostname = False ctx.verify_mode = ssl.CERT_NONE return ctx def _collect_cookies(self) -> list[dict[str, Any]]: cookies: list[dict[str, Any]] = [] for c in self.cookie_jar: cookies.append( { "name": c.name, "value": c.value, "domain": c.domain, "path": c.path, "secure": c.secure, } ) return cookies def login( self, user: str, password: str, session: str = "FileStation", fmt: str = "sid", endpoints: list[tuple[str, int]] | None = None, sessions: list[str] | None = None, ) -> DSMLoginResult: if endpoints is None: endpoints = [ ("/webapi/auth.cgi", 7), ("/webapi/auth.cgi", 6), ("/webapi/entry.cgi", 7), ("/webapi/entry.cgi", 6), ("/webapi/entry.cgi", 3), ] if sessions is None: sessions = [session, "DSM", "FileStation", "DownloadStation", "AudioStation", "VideoStation"] last_err: dict[str, Any] = {} for session_name in sessions: for path, version in endpoints: qs = { "api": "SYNO.API.Auth", "version": str(version), "method": "login", "account": user, "passwd": password, "session": session_name, "format": fmt, } url = f"{self.base_url}{path}?{urllib.parse.urlencode(qs)}" req = urllib.request.Request(url, method="GET") try: with self._opener.open(req, timeout=self.timeout) as resp: raw = resp.read().decode("utf-8", "replace") status = resp.getcode() data = json.loads(raw) if status == 200 and data.get("success") is True: self.sid = data.get("data", {}).get("sid", "") return DSMLoginResult( success=True, data=data, meta={ "base": self.base_url, "endpoint": path, "version": version, "session": session_name, "status": status, "cookies": self._collect_cookies(), }, ) last_err = { "endpoint": path, "version": version, "session": session_name, "status": status, "response": data, } except Exception as e: # noqa: BLE001 last_err = { "endpoint": path, "version": version, "session": session_name, "error": str(e), } return DSMLoginResult( success=False, data={}, meta={"base": self.base_url, "last_error": last_err}, ) def export_session(self, result: DSMLoginResult) -> dict[str, Any]: sid = result.data.get("data", {}).get("sid", "") return { "base": self.base_url, "sid": sid, "cookies": result.meta.get("cookies", []), } def api_get(self, path: str, params: dict[str, Any]) -> dict[str, Any]: """呼叫 DSM WebAPI（GET）。""" req_params = dict(params) if self.sid and "_sid" not in req_params: req_params["_sid"] = self.sid url = f"{self.base_url}{path}?{urllib.parse.urlencode(req_params)}" req = urllib.request.Request(url, method="GET") with self._opener.open(req, timeout=self.timeout) as resp: raw = resp.read().decode("utf-8", "replace") return json.loads(raw) def query_api(self, query: str = "all") -> dict[str, Any]: """查詢 DSM API 能力表（SYNO.API.Info）。""" return self.api_get( "/webapi/query.cgi", { "api": "SYNO.API.Info", "version": "1", "method": "query", "query": query, }, ) def _try_api_candidates(self, candidates: list[dict[str, Any]]) -> dict[str, Any]: last_err: dict[str, Any] = {} for c in candidates: try: data = self.api_get(c["path"], c["params"]) if data.get("success") is True: return {"success": True, "request": c, "response": data} last_err = {"request": c, "response": data} except Exception as e: # noqa: BLE001 last_err = {"request": c, "error": str(e)} return {"success": False, "error": last_err} def get_system_info(self) -> dict[str, Any]: """取得 DSM 基本 System Information。""" return self._try_api_candidates( [ {"path": "/webapi/entry.cgi", "params": {"api": "SYNO.Core.System", "version": "3", "method": "info"}}, {"path": "/webapi/entry.cgi", "params": {"api": "SYNO.Core.System", "version": "2", "method": "info"}}, {"path": "/webapi/entry.cgi", "params": {"api": "SYNO.Core.System", "version": "1", "method": "info"}}, ] ) def get_utilization(self) -> dict[str, Any]: """取得 CPU/記憶體使用率（若 NAS 支援）。""" return self._try_api_candidates( [ {"path": "/webapi/entry.cgi", "params": {"api": "SYNO.Core.System.Utilization", "version": "1", "method": "get"}}, {"path": "/webapi/entry.cgi", "params": {"api": "SYNO.Core.System.Utilization", "version": "1", "method": "load"}}, ] ) def get_volumes(self) -> dict[str, Any]: """取得 Volume 清單（若 NAS 支援）。""" return self._try_api_candidates( [ {"path": "/webapi/entry.cgi", "params": {"api": "SYNO.Core.Storage.Volume", "version": "2", "method": "list"}}, {"path": "/webapi/entry.cgi", "params": {"api": "SYNO.Core.Storage.Volume", "version": "1", "method": "list"}}, {"path": "/webapi/entry.cgi", "params": {"api": "SYNO.Storage.CGI.Volume", "version": "1", "method": "list"}}, ] ) def get_storage_hms_disk(self) -> dict[str, Any]: """取得硬碟健康 / S.M.A.R.T.（若 NAS 支援 SYNO.Storage.HMS.Disk）。""" return self._try_api_candidates( [ {"path": "/webapi/entry.cgi", "params": {"api": "SYNO.Storage.HMS.Disk", "version": "1", "method": "list"}}, {"path": "/webapi/entry.cgi", "params": {"api": "SYNO.Storage.HMS.Disk", "version": "1", "method": "get"}}, {"path": "/webapi/entry.cgi", "params": {"api": "SYNO.Storage.HMS.Disk", "version": "1", "method": "load_info"}}, ] ) def get_core_storage_disk(self) -> dict[str, Any]: """取得硬碟列表 / 介面 / 類型（若 NAS 支援 SYNO.Core.Storage.Disk）。""" return self._try_api_candidates( [ {"path": "/webapi/entry.cgi", "params": {"api": "SYNO.Core.Storage.Disk", "version": "2", "method": "list"}}, {"path": "/webapi/entry.cgi", "params": {"api": "SYNO.Core.Storage.Disk", "version": "1", "method": "list"}}, {"path": "/webapi/entry.cgi", "params": {"api": "SYNO.Core.Storage.Disk", "version": "1", "method": "get"}}, ] ) def get_core_storage_volume(self) -> dict[str, Any]: """取得 Volume 邏輯儲存資訊（若 NAS 支援 SYNO.Core.Storage.Volume）。""" return self._try_api_candidates( [ {"path": "/webapi/entry.cgi", "params": {"api": "SYNO.Core.Storage.Volume", "version": "2", "method": "list"}}, {"path": "/webapi/entry.cgi", "params": {"api": "SYNO.Core.Storage.Volume", "version": "1", "method": "list"}}, {"path": "/webapi/entry.cgi", "params": {"api": "SYNO.Core.Storage.Volume", "version": "1", "method": "get"}}, ] ) def get_storage_cgi_storage(self) -> dict[str, Any]: """取得 Storage Manager 綜合資訊（含 disks/volumes/pools）。""" return self._try_api_candidates( [ {"path": "/webapi/entry.cgi", "params": {"api": "SYNO.Storage.CGI.Storage", "version": "1", "method": "load_info"}}, {"path": "/webapi/entry.cgi", "params": {"api": "SYNO.Storage.CGI.Storage", "version": "1", "method": "load"}}, ] )