liumangmang
95 lines
2.8 KiB
Bash
Executable File
95 lines
2.8 KiB
Bash
Executable File
#!/usr/bin/env bash
|
|
|
|
set -euo pipefail
|
|
|
|
SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
|
|
# shellcheck disable=SC1091
|
|
source "$SCRIPT_DIR/common.sh"
|
|
|
|
load_config
|
|
ensure_command python3
|
|
ensure_runtime_dir
|
|
|
|
print_runtime_status() {
|
|
local label="$1"
|
|
local service_key="$2"
|
|
|
|
local pid_file
|
|
pid_file="$(pid_file_for "$service_key")"
|
|
|
|
if pid="$(active_pid_from_file "$pid_file" 2>/dev/null)"; then
|
|
printf '%s: running (PID %s)\n' "$label" "$pid"
|
|
else
|
|
printf '%s: stopped\n' "$label"
|
|
fi
|
|
}
|
|
|
|
|
|
probe_service() {
|
|
local label="$1"
|
|
local provider="$2"
|
|
local url="$3"
|
|
local keywords="$4"
|
|
local limit="$5"
|
|
|
|
python3 - "$label" "$provider" "$url" "$keywords" "$limit" <<'PY'
|
|
import json
|
|
import sys
|
|
import urllib.parse
|
|
import urllib.request
|
|
|
|
label, provider, base_url, keywords, limit = sys.argv[1:]
|
|
|
|
|
|
def request_json(url: str):
|
|
req = urllib.request.Request(url, headers={"User-Agent": "MusicWorkshop/metadata-status"})
|
|
with urllib.request.urlopen(req, timeout=3) as response:
|
|
return response.status, response.read().decode("utf-8", errors="replace")
|
|
|
|
|
|
def parse_root(base: str):
|
|
try:
|
|
status, _ = request_json(base)
|
|
return f"root: ok (HTTP {status})"
|
|
except urllib.error.HTTPError as exc:
|
|
if exc.code < 500:
|
|
return f"root: ok (HTTP {exc.code})"
|
|
return f"root: failed (HTTP {exc.code})"
|
|
except Exception as exc:
|
|
return f"root: failed ({exc})"
|
|
|
|
|
|
def parse_search(provider_name: str, base: str):
|
|
params = {"keywords": keywords, "limit": limit}
|
|
if provider_name == "netease":
|
|
params["type"] = "1"
|
|
search_url = f"{base.rstrip('/')}/search?{urllib.parse.urlencode(params)}"
|
|
|
|
try:
|
|
_, body = request_json(search_url)
|
|
payload = json.loads(body)
|
|
except Exception as exc:
|
|
return f"search: failed ({exc})"
|
|
|
|
if provider_name == "netease":
|
|
songs = ((payload.get("result") or {}).get("songs")) or []
|
|
return f"search: ok ({len(songs)} song candidates)" if isinstance(songs, list) else "search: failed (result.songs 缺失)"
|
|
|
|
song_list = (
|
|
((payload.get("data") or {}).get("song") or {}).get("list")
|
|
or ((payload.get("result") or {}).get("list") or [])
|
|
)
|
|
return f"search: ok ({len(song_list)} song candidates)" if isinstance(song_list, list) else "search: failed (data.song.list/result.list 缺失)"
|
|
|
|
|
|
print(f"{label} probes")
|
|
print(f" {parse_root(base_url)}")
|
|
print(f" {parse_search(provider, base_url)}")
|
|
PY
|
|
}
|
|
|
|
print_runtime_status "网易云音乐 API" "netease"
|
|
probe_service "网易云音乐 API" "netease" "http://$NETEASE_HOST:$NETEASE_PORT" "$SEARCH_SMOKE_KEYWORDS" "$SEARCH_SMOKE_LIMIT"
|
|
print_runtime_status "QQ 音乐 API" "qq"
|
|
probe_service "QQ 音乐 API" "qq" "http://$QQ_HOST:$QQ_PORT" "$SEARCH_SMOKE_KEYWORDS" "$SEARCH_SMOKE_LIMIT"
|