commit 321456d4e69505d0afdefb773b0848f0bd8bab10 Author: Timmy Date: Thu Mar 26 17:55:43 2026 +0800 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) diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..fa0555f --- /dev/null +++ b/.gitignore @@ -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 diff --git a/CLAUDE.md b/CLAUDE.md new file mode 100644 index 0000000..92264c6 --- /dev/null +++ b/CLAUDE.md @@ -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` diff --git a/SKILL.md b/SKILL.md new file mode 100644 index 0000000..89172e4 --- /dev/null +++ b/SKILL.md @@ -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`。 diff --git a/references/env.example b/references/env.example new file mode 100644 index 0000000..fc0bc77 --- /dev/null +++ b/references/env.example @@ -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 diff --git a/references/hosts.json b/references/hosts.json new file mode 100644 index 0000000..29dbf3e --- /dev/null +++ b/references/hosts.json @@ -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" + } +} diff --git a/references/usage.md b/references/usage.md new file mode 100644 index 0000000..32b6cd7 --- /dev/null +++ b/references/usage.md @@ -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/`」為前提。 diff --git a/scripts/synology_api.py b/scripts/synology_api.py new file mode 100644 index 0000000..e7cdb81 --- /dev/null +++ b/scripts/synology_api.py @@ -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 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"}}, + ] + ) diff --git a/scripts/synology_api_login.py b/scripts/synology_api_login.py new file mode 100644 index 0000000..84b5503 --- /dev/null +++ b/scripts/synology_api_login.py @@ -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())