Timmy
321456d4e6
- Add synology_api.py: DSM API client library - Add synology_api_login.py: CLI tool for login and system info - Add SKILL.md: Skill definition with usage examples - Add CLAUDE.md: Developer documentation - Add references/: Example configs and usage guide Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
284 lines
11 KiB
Python
284 lines
11 KiB
Python
#!/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"}},
|
||
]
|
||
)
|