Initial commit: Synology DSM API skill

- 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>
This commit is contained in:
2026-03-26 17:55:43 +08:00
commit 321456d4e6
8 changed files with 1046 additions and 0 deletions

19
.gitignore vendored Normal file
View File

@@ -0,0 +1,19 @@
# Python
__pycache__/
*.py[cod]
*$py.class
*.so
.Python
*.egg-info/
dist/
build/
# IDE
.vscode/
.idea/
*.swp
*.swo
# OS
.DS_Store
Thumbs.db

91
CLAUDE.md Normal file
View File

@@ -0,0 +1,91 @@
# CLAUDE.md
This file provides guidance to Claude Code (claude.ai/code) when working with code in this repository.
## Overview
This is a Synology DSM API client library and CLI tool. It provides a Python interface to interact with Synology NAS via the WebAPI, supporting login, session reuse, and system information retrieval.
## Architecture
### Core Components
- **`scripts/synology_api.py`** - The `DSMApiClient` library class. Handles authentication, session management, and API calls. Uses only stdlib (urllib, ssl, json).
- **`scripts/synology_api_login.py`** - CLI wrapper that uses the library. Supports profile-based configuration, SID session caching, and system info retrieval with human-readable or JSON output.
### Session Management
The client caches SID sessions to avoid repeated logins:
1. On run, checks `tmp/synology_session.json` for a valid SID
2. If SID is still valid (probed via `get_system_info`), reuses it
3. Otherwise, performs full login and saves new session
### Configuration Methods
Two ways to configure credentials:
1. **Environment variables** (`.env` in workspace root):
```
DS_URL=https://syno.example.com/
DS_USER=username
DS_PASSWORD=password
```
2. **Profiles** (`references/hosts.json`):
```json
{
"profile-name": {
"url": "https://nas.local:5001",
"user": "username",
"password_env": "DS_PASSWORD_PROFILE"
}
}
```
Use with `--profile profile-name`
## Common Commands
### Basic login (with session caching)
```bash
python3 scripts/synology_api_login.py
```
### Get system info with human-readable summary
```bash
python3 scripts/synology/api_login.py --system-info --summary
```
### Get system info as compact JSON (for scripting)
```bash
python3 scripts/synology_api_login.py --system-info --json-compact
```
### Using a profile from hosts.json
```bash
python3 scripts/synology_api_login.py --profile home-nas --system-info --summary
```
### Direct credentials (overrides env vars)
```bash
python3 scripts/synology_api_login.py --url "https://192.168.1.10:5001" --user admin --password "xxx" --system-info
```
## Output Files
- `tmp/synology_api_login_result.json` - Login result metadata
- `tmp/synology_session.json` - Cached SID session
- `tmp/synology_system_info.json` - Full API responses bundle
## API Fallback Pattern
DSM versions vary in API support. The client uses `_try_api_candidates()` which tries multiple endpoint/version combinations and returns the first successful response. When adding new API methods, follow this pattern for compatibility.
## Compact JSON Schema
The `--json-compact` output normalizes across DSM versions and includes:
- `model`, `serial`, `ip` (prefers eth0)
- `dsm_version`, `cpu_usage_pct`, `cpu_clock_mhz`
- `memory_total_mb`, `memory_usage_pct`, `temperature_c`
- `volumes` (list), `disk_size_usage` (bytes + GiB), `disk_health`
- `time`, `uptime`

94
SKILL.md Normal file
View File

@@ -0,0 +1,94 @@
---
name: synology
description: 透過 Synology DSM WebAPI 進行登入、重用 SID session、抓取 NAS 系統資訊與輸出精簡 JSON 摘要。當使用者要查 Synology NAS 狀態、DSM 版本、CPU/記憶體用量、溫度、Volume 容量、硬碟健康、網路 IP或要把 Synology 查詢流程做成可重複執行腳本時使用。
---
# Synology
使用這個 skill 操作 Synology DSM API不走瀏覽器。
## 快速開始
先確認環境變數或 profile 可用。可參考:
- `references/env.example`
- `references/hosts.json`
- `references/usage.md`
建議做法:
- 多台 Synology`references/hosts.json``url``user``password_env`
- 密碼放 workspace 根目錄 `.env`
- 腳本會自動讀 `.env`,不用每次手動 `export`
常用指令:
```bash
python3 skills/synology/scripts/synology_api_login.py --system-info --summary
```
若要輸出精簡 JSON
```bash
python3 skills/synology/scripts/synology_api_login.py --system-info --json-compact
```
若有多台 Synology建議改用 profile
```bash
python3 skills/synology/scripts/synology_api_login.py --profile home-nas --system-info --summary
```
搭配 workspace 根目錄 `.env`
```env
DS_PASSWORD_HOME=你的密碼
DS_PASSWORD_OFFICE=你的密碼
```
## 任務
### 1. 登入 DSM
```bash
python3 skills/synology/scripts/synology_api_login.py
```
重點:
- 會先嘗試沿用已快取的 SID
- SID 失效才重新登入
- 預設吃 `DS_URL``DS_USER``DS_PASSWORD`
- 若指定 `--profile`,則改讀 `references/hosts.json`
### 2. 查系統資訊
```bash
python3 skills/synology/scripts/synology_api_login.py --system-info --summary
```
回覆時優先整理:
- NAS 型號
- DSM 版本
- IP
- CPU 使用率
- 記憶體使用率
- 溫度
- Volume 容量
- 硬碟健康
### 3. 輸出給其他腳本吃的 JSON
```bash
python3 skills/synology/scripts/synology_api_login.py --system-info --json-compact
```
適合:
- 後續 shell / Python 串接
- 定期監控
- 給其他 skill 或腳本使用
## 注意事項
- 若 DSM 使用自簽章憑證,這套腳本預設可接受;如需強制驗證憑證,再調整 client。
-`tmp/` 內已有 session cache腳本可能直接沿用舊 SID。
- 若查不到資料,先確認 `DS_URL``DS_USER``DS_PASSWORD` 是否正確。
- 若只是要看怎麼設定 `.env`,先讀 `references/env.example`
- 若只是要看用法範例,先讀 `references/usage.md`

14
references/env.example Normal file
View File

@@ -0,0 +1,14 @@
# 放在 workspace 根目錄 .env 後Synology skill 會自動讀取
# 對應 references/hosts.json 內的 password_env
DS_PASSWORD_HOME=請改成你的密碼
DS_PASSWORD_OFFICE=請改成你的密碼
# 若你只操作單台,也可沿用這組預設欄位
DS_URL=https://syno.lotimmy.com/
DS_USER=你的帳號
DS_PASSWORD=請改成你的密碼
# 可選:輸出路徑
DS_LOGIN_RESULT=tmp/synology_api_login_result.json
DS_SESSION_FILE=tmp/synology_session.json
DS_SYSINFO_FILE=tmp/synology_system_info.json

12
references/hosts.json Normal file
View File

@@ -0,0 +1,12 @@
{
"home-nas": {
"url": "https://192.168.42.20:5001",
"user": "timmy",
"password_env": "DS_PASSWORD_HOME"
},
"office-nas": {
"url": "https://syno.office.local:5001",
"user": "timmy",
"password_env": "DS_PASSWORD_OFFICE"
}
}

102
references/usage.md Normal file
View File

@@ -0,0 +1,102 @@
# Synology API 工具使用說明
這個目錄放的是 Synology DSM API 相關腳本:
- `synology_api.py`:共用 API client library
- `synology_api_login.py`:登入、復用 SID、抓系統資訊
---
## 1) 先設定環境變數(建議)
可放在 workspace 根目錄 `.env`
```env
DS_URL=https://syno.lotimmy.com/
DS_USER=你的帳號
DS_PASSWORD=你的密碼
# 可選:輸出檔案路徑
DS_LOGIN_RESULT=tmp/synology_api_login_result.json
DS_SESSION_FILE=tmp/synology_session.json
DS_SYSINFO_FILE=tmp/synology_system_info.json
```
---
## 2) 基本用法
### 只登入(會先嘗試沿用舊 SID失效才重登
```bash
python3 synology_api_login.py
```
### 登入後抓系統資訊
```bash
python3 synology_api_login.py --system-info
```
### 顯示人類可讀摘要
```bash
python3 synology_api_login.py --system-info --summary
```
### 輸出精簡 JSON適合餵腳本
```bash
python3 synology_api_login.py --system-info --json-compact
```
---
## 3) 常用參數
- `--url`DSM URL預設取 `DS_URL`
- `--user`DSM 帳號(預設取 `DS_USER`
- `--password`DSM 密碼(預設取 `DS_PASSWORD`
- `--system-info`:抓系統資訊 bundle
- `--summary`:印出摘要(需搭配 `--system-info`
- `--json-compact`:印出精簡 JSON需搭配 `--system-info`
範例:
```bash
python3 synology_api_login.py \
--url "https://192.168.42.20:5001" \
--user "admin" \
--password "******" \
--system-info --json-compact
```
---
## 4) 主要輸出檔
- `tmp/synology_api_login_result.json`:登入結果(含是否沿用 cached SID
- `tmp/synology_session.json`SID session cache
- `tmp/synology_system_info.json`:完整系統資訊 API 回傳
---
## 5) 目前精簡 JSON 會包含
- `model`
- `serial`
- `ip`(優先 `eth0`
- `dsm_version`
- `cpu_usage_pct`, `cpu_clock_mhz`
- `memory_total_mb`, `memory_usage_pct`
- `temperature_c`
- `volumes`
- `disk_size_usage`(同時有 `*_bytes``*_gib`
- `disk_health`vendor 已去除多餘空白)
- `time`, `uptime`
---
## 6) 執行位置
本 README 的指令都以「目前目錄在 `scripts/synology/`」為前提。

283
scripts/synology_api.py Normal file
View File

@@ -0,0 +1,283 @@
#!/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 WebAPIGET"""
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"}},
]
)

View File

@@ -0,0 +1,431 @@
#!/usr/bin/env python3
"""
Synology DSM API 自動登入 CLI使用 synology_api.py
預設:
DS_URL=https://syno.lotimmy.com/
DS_USER, DS_PASSWORD 從環境變數或 .env 讀取
用法:
python3 scripts/synology_api_login.py
python3 scripts/synology_api_login.py --system-info
python3 scripts/synology_api_login.py --system-info --summary
python3 scripts/synology_api_login.py --system-info --json-compact
輸出:
- tmp/synology_api_login_result.json (回應摘要)
- tmp/synology_session.json (sid 與 cookie)
- tmp/synology_system_info.json (System Information--system-info 時)
"""
from __future__ import annotations
import argparse
import json
import os
import sys
from pathlib import Path
from typing import Any, Callable
HOSTS_FILE = Path(__file__).resolve().parents[1] / "references" / "hosts.json"
from synology_api import DSMApiClient, load_dotenv
load_dotenv()
DEFAULT_URL = os.getenv("DS_URL", "https://syno.lotimmy.com/")
DEFAULT_USER = os.getenv("DS_USER", "")
DEFAULT_PASSWORD = os.getenv("DS_PASSWORD", "")
OUT_RESULT = os.getenv("DS_LOGIN_RESULT", "tmp/synology_api_login_result.json")
OUT_SESSION = os.getenv("DS_SESSION_FILE", "tmp/synology_session.json")
OUT_SYSINFO = os.getenv("DS_SYSINFO_FILE", "tmp/synology_system_info.json")
def load_hosts() -> dict[str, Any]:
if not HOSTS_FILE.exists():
return {}
try:
return json.loads(HOSTS_FILE.read_text(encoding="utf-8"))
except Exception: # noqa: BLE001
return {}
def resolve_profile(profile_name: str) -> tuple[str, str, str]:
hosts = load_hosts()
profile = hosts.get(profile_name)
if not isinstance(profile, dict):
raise SystemExit(f"[x] 找不到 profile{profile_name}(檔案:{HOSTS_FILE}")
url = str(profile.get("url", "")).strip()
user = str(profile.get("user", "")).strip()
password = str(profile.get("password", "")).strip()
password_env = str(profile.get("password_env", "")).strip()
if not password and password_env:
password = os.getenv(password_env, "").strip()
if not url or not user or not password:
raise SystemExit(f"[x] profile {profile_name} 缺少 url/user/password或 password_env 未設定)")
return url, user, password
def _find_key(obj: Any, keys: tuple[str, ...]) -> Any:
if isinstance(obj, dict):
for k, v in obj.items():
lk = k.lower()
if any(x in lk for x in keys):
return v
for v in obj.values():
hit = _find_key(v, keys)
if hit is not None:
return hit
elif isinstance(obj, list):
for item in obj:
hit = _find_key(item, keys)
if hit is not None:
return hit
return None
def _to_int(value: Any) -> int | None:
try:
if value is None:
return None
if isinstance(value, bool):
return None
if isinstance(value, (int, float)):
return int(value)
v = str(value).strip()
if not v:
return None
return int(float(v))
except (TypeError, ValueError):
return None
def _to_gib(value: int | None) -> float | None:
if value is None:
return None
return round(value / (1024 ** 3), 2)
def _build_compact_summary(bundle: dict[str, Any]) -> dict[str, Any]:
sysinfo = bundle.get("system_info", {})
util = bundle.get("utilization", {})
vols = bundle.get("volumes", {})
hms_disk = bundle.get("storage_hms_disk", {})
storage_cgi = bundle.get("storage_cgi_storage", {})
network_ethernet = bundle.get("network_ethernet", {})
sdata = sysinfo.get("response", {}).get("data", {})
udata = util.get("response", {}).get("data", {})
vdata = vols.get("response", {}).get("data", {})
hdata = hms_disk.get("response", {}).get("data", {})
scdata = storage_cgi.get("response", {}).get("data", {})
ndata = network_ethernet.get("response", {}).get("data", [])
if not ndata and isinstance(network_ethernet.get("data"), list):
ndata = network_ethernet.get("data", [])
cpu_obj = udata.get("cpu", {}) if isinstance(udata, dict) else {}
cpu_usage_pct = None
if isinstance(cpu_obj, dict):
cpu_usage_pct = cpu_obj.get("user_load", 0) + cpu_obj.get("system_load", 0) + cpu_obj.get("other_load", 0)
mem_obj = udata.get("memory", {}) if isinstance(udata, dict) else {}
mem_usage_pct = mem_obj.get("real_usage") if isinstance(mem_obj, dict) else None
volume_obj = vdata.get("volumes", vdata.get("volume", vdata))
if not volume_obj:
volume_obj = scdata.get("volumes", []) if isinstance(scdata, dict) else []
if not volume_obj:
volume_obj = udata.get("space", {}).get("volume", []) if isinstance(udata, dict) else []
volumes: list[str] = []
volume_usage: dict[str, Any] = {}
disk_size_usage: dict[str, Any] = {}
if isinstance(volume_obj, list) and volume_obj:
for v in volume_obj:
if not isinstance(v, dict):
continue
name = (
v.get("display_name")
or v.get("name")
or v.get("volume_path")
or v.get("path")
or (f"volume{v.get('num_id')}" if v.get('num_id') is not None else None)
or v.get("device")
or v.get("id")
)
if not name:
continue
volumes.append(name)
usage = v.get("utilization")
if usage is not None:
volume_usage[name] = {
"utilization_pct": usage,
"read_byte": v.get("read_byte"),
"write_byte": v.get("write_byte"),
}
size_obj = v.get("size") if isinstance(v.get("size"), dict) else {}
total_bytes = (
v.get("total")
or v.get("total_size")
or v.get("size_total")
or v.get("capacity")
or size_obj.get("total")
)
used_bytes = (
v.get("used")
or v.get("used_size")
or v.get("size_used")
or v.get("used_space")
or size_obj.get("used")
)
free_bytes = (
v.get("free")
or v.get("free_size")
or v.get("size_free")
or v.get("available")
or size_obj.get("free")
)
total_i = _to_int(total_bytes)
used_i = _to_int(used_bytes)
free_i = _to_int(free_bytes)
if free_i is None and total_i is not None and used_i is not None:
free_i = max(total_i - used_i, 0)
if total_i is not None or used_i is not None or free_i is not None:
disk_size_usage[name] = {
"total_bytes": total_i,
"used_bytes": used_i,
"free_bytes": free_i,
"total_gib": _to_gib(total_i),
"used_gib": _to_gib(used_i),
"free_gib": _to_gib(free_i),
}
elif isinstance(volume_obj, dict) and volume_obj:
volumes = list(volume_obj.keys())
disk_health = hdata.get("disk", hdata.get("disks", hdata))
if not disk_health:
disk_health = {}
for d in scdata.get("disks", []) if isinstance(scdata, dict) else []:
if not isinstance(d, dict):
continue
name = d.get("name") or d.get("device") or d.get("id")
if not name:
continue
vendor = d.get("vendor")
if isinstance(vendor, str):
vendor = vendor.strip()
disk_health[name] = {
"status": d.get("status"),
"smart_status": d.get("smart_status"),
"temp_c": d.get("temp"),
"model": d.get("model"),
"vendor": vendor,
"disk_type": d.get("diskType"),
}
if isinstance(disk_health, dict):
for _, info in disk_health.items():
if isinstance(info, dict) and isinstance(info.get("vendor"), str):
info["vendor"] = info["vendor"].strip()
primary_ip = None
if isinstance(ndata, list):
for nic in ndata:
if not isinstance(nic, dict):
continue
if nic.get("ifname") == "eth0" and nic.get("ip"):
primary_ip = nic.get("ip")
break
if primary_ip is None:
for nic in ndata:
if isinstance(nic, dict) and nic.get("ip") and nic.get("ip") != "":
primary_ip = nic.get("ip")
break
serial = sdata.get("serial")
result = {
"model": sdata.get("model"),
"serial": serial,
"ip": primary_ip,
"dsm_version": sdata.get("firmware_ver"),
"cpu_usage_pct": cpu_usage_pct,
"cpu_clock_mhz": sdata.get("cpu_clock_speed"),
"memory_total_mb": sdata.get("ram_size"),
"memory_usage_pct": mem_usage_pct,
"temperature_c": sdata.get("sys_temp", _find_key(sdata, ("temperature", "temp"))),
"volumes": volumes,
"disk_size_usage": disk_size_usage,
"disk_health": disk_health,
"time": sdata.get("time"),
"uptime": sdata.get("up_time"),
}
return result
def _safe_call(fn: Callable[[], dict[str, Any]], name: str) -> dict[str, Any]:
try:
return fn()
except Exception as exc: # noqa: BLE001
return {
"success": False,
"error": f"{name} failed: {exc}",
"response": {"data": {}},
}
def _load_cached_sid(session_file: str, expected_base: str) -> str:
p = Path(session_file)
if not p.exists():
return ""
try:
data = json.loads(p.read_text(encoding="utf-8"))
except Exception: # noqa: BLE001
return ""
base = str(data.get("base", "")).strip().rstrip("/")
sid = str(data.get("sid", "")).strip()
if not sid:
return ""
if base and base != expected_base.rstrip("/"):
return ""
return sid
def _is_sid_valid(client: DSMApiClient) -> bool:
probe = _safe_call(client.get_system_info, "get_system_info")
return bool(probe.get("success") is True)
def _print_summary(bundle: dict[str, Any]) -> None:
c = _build_compact_summary(bundle)
vol_summary = ", ".join(c.get("volumes", [])) if c.get("volumes") else "N/A"
print("[SUMMARY] System Information")
print(f"- CPU 使用率: {str(c.get('cpu_usage_pct')) + '%' if c.get('cpu_usage_pct') is not None else 'N/A'}")
print(f"- CPU 頻率: {str(c.get('cpu_clock_mhz')) + ' MHz' if c.get('cpu_clock_mhz') is not None else 'N/A'}")
print(f"- 記憶體(總量): {str(c.get('memory_total_mb')) + ' MB' if c.get('memory_total_mb') is not None else 'N/A'}")
print(f"- 記憶體(使用): {str(c.get('memory_usage_pct')) + '%' if c.get('memory_usage_pct') is not None else 'N/A'}")
print(f"- 溫度: {str(c.get('temperature_c')) + '°C' if c.get('temperature_c') is not None else 'N/A'}")
print(f"- Volume: {vol_summary}")
size_usage = c.get("disk_size_usage", {})
if isinstance(size_usage, dict) and size_usage:
print("- 容量:")
for vname, vinfo in size_usage.items():
if not isinstance(vinfo, dict):
continue
t = vinfo.get("total_gib")
u = vinfo.get("used_gib")
f = vinfo.get("free_gib")
print(f" - {vname}: 總 {t} GiB / 已用 {u} GiB / 可用 {f} GiB")
def main() -> int:
parser = argparse.ArgumentParser(description="DSM API login (no Playwright)")
parser.add_argument("--profile", help="主機設定名稱(讀 references/hosts.json")
parser.add_argument("--url", default=DEFAULT_URL, help="DSM URL")
parser.add_argument("--user", default=DEFAULT_USER, help="帳號(預設 DS_USER")
parser.add_argument("--password", default=DEFAULT_PASSWORD, help="密碼(預設 DS_PASSWORD")
parser.add_argument("--system-info", action="store_true", help="登入後抓取 System Information")
parser.add_argument("--summary", action="store_true", help="搭配 --system-info 顯示人類可讀摘要")
parser.add_argument("--json-compact", action="store_true", help="搭配 --system-info 輸出精簡 JSON")
args = parser.parse_args()
if args.profile:
args.url, args.user, args.password = resolve_profile(args.profile)
client = DSMApiClient(base_url=args.url)
Path("tmp").mkdir(parents=True, exist_ok=True)
used_cached_sid = False
cached_sid = _load_cached_sid(OUT_SESSION, client.base_url)
if cached_sid:
client.sid = cached_sid
if _is_sid_valid(client):
used_cached_sid = True
result_obj: dict[str, Any]
if used_cached_sid:
result_obj = {
"success": True,
"meta": {
"base": client.base_url,
"source": "cached_sid",
"session_file": OUT_SESSION,
},
"data": {"sid": client.sid},
}
print("[✓] DSM API Session 可用(沿用既有 SID")
print(f"[i] SID: {client.sid[:8]}... (來源 {OUT_SESSION})")
else:
if not args.user or not args.password:
print("[x] 缺少帳密:請設定 DS_USER / DS_PASSWORD 或傳 --user/--password", file=sys.stderr)
return 2
result = client.login(user=args.user, password=args.password)
result_obj = {
"success": result.success,
"meta": result.meta,
"data": result.data,
}
if not result.success:
Path(OUT_RESULT).write_text(json.dumps(result_obj, ensure_ascii=False, indent=2), encoding="utf-8")
print("[x] DSM API 登入失敗", file=sys.stderr)
print(f"[i] 詳細結果:{OUT_RESULT}", file=sys.stderr)
return 1
session_obj = client.export_session(result)
Path(OUT_SESSION).write_text(json.dumps(session_obj, ensure_ascii=False, indent=2), encoding="utf-8")
sid = session_obj.get("sid", "")
print("[✓] DSM API 登入成功")
print(f"[i] SID: {sid[:8]}... (已寫入 {OUT_SESSION})")
Path(OUT_RESULT).write_text(json.dumps(result_obj, ensure_ascii=False, indent=2), encoding="utf-8")
if args.system_info:
sysinfo = _safe_call(client.get_system_info, "get_system_info")
util = _safe_call(client.get_utilization, "get_utilization")
vols = _safe_call(client.get_volumes, "get_volumes")
hms_disk = _safe_call(client.get_storage_hms_disk, "get_storage_hms_disk")
storage_cgi = _safe_call(client.get_storage_cgi_storage, "get_storage_cgi_storage")
network_ethernet = _safe_call(
lambda: client.api_get(
"/webapi/entry.cgi",
{"api": "SYNO.Core.Network.Ethernet", "version": "1", "method": "list"},
),
"get_network_ethernet",
)
bundle = {
"system_info": sysinfo,
"utilization": util,
"volumes": vols,
"storage_hms_disk": hms_disk,
"storage_cgi_storage": storage_cgi,
"network_ethernet": network_ethernet,
}
Path(OUT_SYSINFO).write_text(json.dumps(bundle, ensure_ascii=False, indent=2), encoding="utf-8")
if sysinfo.get("success"):
print(f"[i] System Information 已寫入:{OUT_SYSINFO}")
if args.summary:
_print_summary(bundle)
if args.json_compact:
print(json.dumps(_build_compact_summary(bundle), ensure_ascii=False))
else:
print(f"[!] System Information 取得失敗,已寫入:{OUT_SYSINFO}")
print(f"[i] 詳細結果:{OUT_RESULT}")
return 0
if __name__ == "__main__":
raise SystemExit(main())