706 lines
28 KiB
Python
Executable File
706 lines
28 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""Local production verifier for SHD2YPP2 installs.
|
|
|
|
This is intentionally stdlib-only so it can run immediately after install.sh on
|
|
fresh Debian/Ubuntu/Armbian hosts. It checks the real delivery path, not only a
|
|
single process: files, systemd, API, edge proxy, Docker/Neko, Tampermonkey
|
|
provisioning, and Android screenshot/control readiness when devices exist.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import os
|
|
import shutil
|
|
import subprocess
|
|
import sys
|
|
import time
|
|
import urllib.error
|
|
import urllib.parse
|
|
import urllib.request
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
|
|
DEFAULT_INSTALL_DIR = Path("/opt/live/d2ypp")
|
|
PNG_SIG = b"\x89PNG\r\n\x1a\n"
|
|
|
|
|
|
def run(cmd: list[str], *, timeout: float = 8.0) -> tuple[int, str, str]:
|
|
try:
|
|
proc = subprocess.run(
|
|
cmd,
|
|
capture_output=True,
|
|
text=True,
|
|
timeout=timeout,
|
|
errors="replace",
|
|
)
|
|
except FileNotFoundError as exc:
|
|
return 127, "", str(exc)
|
|
except subprocess.TimeoutExpired as exc:
|
|
out = exc.stdout if isinstance(exc.stdout, str) else ""
|
|
err = exc.stderr if isinstance(exc.stderr, str) else ""
|
|
return 124, out, err or f"timeout after {timeout}s"
|
|
return proc.returncode, proc.stdout or "", proc.stderr or ""
|
|
|
|
|
|
def read_env(path: Path) -> dict[str, str]:
|
|
env: dict[str, str] = {}
|
|
if not path.is_file():
|
|
return env
|
|
for raw in path.read_text(encoding="utf-8", errors="replace").splitlines():
|
|
line = raw.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:
|
|
env[key] = value
|
|
return env
|
|
|
|
|
|
def env_flag(env: dict[str, str], key: str, default: bool = True) -> bool:
|
|
raw = env.get(key)
|
|
if raw is None or raw == "":
|
|
return default
|
|
return raw.strip().lower() not in {"0", "false", "no", "off", "disable", "disabled"}
|
|
|
|
|
|
def env_int(env: dict[str, str], key: str, default: int) -> int:
|
|
try:
|
|
return int(str(env.get(key, default)).strip())
|
|
except (TypeError, ValueError):
|
|
return default
|
|
|
|
|
|
def check(
|
|
checks: list[dict[str, Any]],
|
|
check_id: str,
|
|
label: str,
|
|
ok: bool,
|
|
*,
|
|
required: bool = True,
|
|
detail: str = "",
|
|
hint: str = "",
|
|
data: dict[str, Any] | None = None,
|
|
) -> dict[str, Any]:
|
|
item = {
|
|
"id": check_id,
|
|
"label": label,
|
|
"ok": bool(ok),
|
|
"required": bool(required),
|
|
"detail": detail,
|
|
}
|
|
if hint:
|
|
item["hint"] = hint
|
|
if data:
|
|
item.update(data)
|
|
checks.append(item)
|
|
return item
|
|
|
|
|
|
def http_bytes(url: str, *, timeout: float = 5.0) -> tuple[bool, int, bytes, str, float]:
|
|
start = time.monotonic()
|
|
try:
|
|
with urllib.request.urlopen(url, timeout=timeout) as resp:
|
|
body = resp.read()
|
|
return True, int(resp.status), body, "", round((time.monotonic() - start) * 1000, 1)
|
|
except urllib.error.HTTPError as exc:
|
|
body = exc.read() if hasattr(exc, "read") else b""
|
|
return False, int(exc.code), body, str(exc), round((time.monotonic() - start) * 1000, 1)
|
|
except (OSError, urllib.error.URLError) as exc:
|
|
return False, 0, b"", str(exc), round((time.monotonic() - start) * 1000, 1)
|
|
|
|
|
|
def http_json(url: str, *, timeout: float = 5.0) -> tuple[bool, int, dict[str, Any], str, float]:
|
|
ok, code, body, err, latency = http_bytes(url, timeout=timeout)
|
|
if not body:
|
|
return ok, code, {}, err, latency
|
|
try:
|
|
return ok and 200 <= code < 300, code, json.loads(body.decode("utf-8", errors="replace")), err, latency
|
|
except json.JSONDecodeError as exc:
|
|
return False, code, {}, f"invalid JSON: {exc}", latency
|
|
|
|
|
|
def http_json_post(url: str, payload: dict[str, Any], *, timeout: float = 5.0) -> tuple[bool, int, dict[str, Any], str, float]:
|
|
start = time.monotonic()
|
|
body = json.dumps(payload).encode("utf-8")
|
|
request = urllib.request.Request(
|
|
url,
|
|
data=body,
|
|
headers={"Content-Type": "application/json"},
|
|
method="POST",
|
|
)
|
|
try:
|
|
with urllib.request.urlopen(request, timeout=timeout) as resp:
|
|
raw = resp.read()
|
|
latency = round((time.monotonic() - start) * 1000, 1)
|
|
data = json.loads(raw.decode("utf-8", errors="replace")) if raw else {}
|
|
return True, int(resp.status), data, "", latency
|
|
except urllib.error.HTTPError as exc:
|
|
raw = exc.read() if hasattr(exc, "read") else b""
|
|
latency = round((time.monotonic() - start) * 1000, 1)
|
|
try:
|
|
data = json.loads(raw.decode("utf-8", errors="replace")) if raw else {}
|
|
except json.JSONDecodeError:
|
|
data = {}
|
|
return False, int(exc.code), data, str(exc), latency
|
|
except (OSError, urllib.error.URLError, json.JSONDecodeError) as exc:
|
|
return False, 0, {}, str(exc), round((time.monotonic() - start) * 1000, 1)
|
|
|
|
|
|
def systemd_active(unit: str) -> tuple[bool, str]:
|
|
code, out, err = run(["systemctl", "is-active", unit], timeout=5)
|
|
state = (out or err).strip()
|
|
return code == 0 and state == "active", state or f"exit={code}"
|
|
|
|
|
|
def docker_ps_names() -> tuple[bool, list[str], str]:
|
|
if not shutil.which("docker"):
|
|
return False, [], "docker missing"
|
|
code, out, err = run(["docker", "ps", "--format", "{{.Names}}"], timeout=8)
|
|
if code != 0:
|
|
return False, [], err.strip() or out.strip() or f"exit={code}"
|
|
return True, [line.strip() for line in out.splitlines() if line.strip()], ""
|
|
|
|
|
|
def docker_exec(container: str, command: str, *, timeout: float = 8.0) -> tuple[int, str, str]:
|
|
return run(["docker", "exec", container, "sh", "-lc", command], timeout=timeout)
|
|
|
|
|
|
def verify_files(checks: list[dict[str, Any]], install_dir: Path) -> None:
|
|
check(
|
|
checks,
|
|
"install_dir",
|
|
"Runtime install directory",
|
|
install_dir.is_dir(),
|
|
detail=str(install_dir),
|
|
hint="Run install.sh again; it syncs the project to /opt/live/d2ypp by default.",
|
|
)
|
|
check(
|
|
checks,
|
|
"stack_env",
|
|
"system-stack.env",
|
|
(install_dir / "config" / "system-stack.env").is_file(),
|
|
detail=str(install_dir / "config" / "system-stack.env"),
|
|
hint="Copy config/system-stack.env.example or rerun the installer.",
|
|
)
|
|
check(
|
|
checks,
|
|
"web_assets",
|
|
"Exported web-console assets",
|
|
(install_dir / "web-console" / "out" / "index.html").is_file(),
|
|
detail=str(install_dir / "web-console" / "out" / "index.html"),
|
|
hint="Run sudo bash upgrade-live.sh or rebuild web-console.",
|
|
)
|
|
check(
|
|
checks,
|
|
"start_script",
|
|
"start.sh executable",
|
|
os.access(install_dir / "start.sh", os.X_OK),
|
|
detail=str(install_dir / "start.sh"),
|
|
hint="Run chmod +x start.sh and refresh live-console.service.",
|
|
)
|
|
|
|
|
|
def verify_systemd(checks: list[dict[str, Any]], env: dict[str, str]) -> None:
|
|
ok, detail = systemd_active("live-console.service")
|
|
check(
|
|
checks,
|
|
"live_console_service",
|
|
"live-console.service",
|
|
ok,
|
|
detail=detail,
|
|
hint="Inspect: journalctl -u live-console.service -n 120 --no-pager",
|
|
)
|
|
if env_flag(env, "ENABLE_LIVE_EDGE_PROXY", True):
|
|
ok, detail = systemd_active("live-edge.service")
|
|
check(
|
|
checks,
|
|
"live_edge_service",
|
|
"live-edge.service",
|
|
ok,
|
|
detail=detail,
|
|
hint="Run scripts/linux/reinstall_live_edge.sh or restart live-edge.service.",
|
|
)
|
|
ok, detail = systemd_active("live-edge-watch.service")
|
|
check(
|
|
checks,
|
|
"live_edge_watch_service",
|
|
"live-edge-watch.service",
|
|
ok,
|
|
detail=detail,
|
|
hint="Run scripts/linux/reinstall_live_edge.sh to install the live.local watchdog.",
|
|
)
|
|
|
|
|
|
def verify_http(checks: list[dict[str, Any]], env: dict[str, str], *, timeout: float) -> None:
|
|
port = env.get("PORT", os.environ.get("PORT", "8001"))
|
|
mdns_host = f"{env.get('HOSTNAME_ALIAS', 'live')}.local"
|
|
ok, code, data, err, latency = http_json(f"http://127.0.0.1:{port}/health", timeout=timeout)
|
|
check(
|
|
checks,
|
|
"api_health",
|
|
"FastAPI /health",
|
|
ok and data.get("status") == "ok",
|
|
detail=f"HTTP {code}, {latency} ms" if code else err,
|
|
hint="live-console is not serving the control plane.",
|
|
data={"url": f"http://127.0.0.1:{port}/health"},
|
|
)
|
|
ok, code, data, err, latency = http_json(f"http://127.0.0.1:{port}/deploy/check", timeout=timeout + 4)
|
|
check(
|
|
checks,
|
|
"deploy_check",
|
|
"FastAPI /deploy/check",
|
|
ok and bool(data.get("ready")),
|
|
detail=(
|
|
f"{'READY' if data.get('ready') else 'DEGRADED'} "
|
|
f"({data.get('pass_count', 0)}/{data.get('total_count', 0)}), {latency} ms"
|
|
if code
|
|
else err
|
|
),
|
|
hint="Use deploy/check failing_checks for the exact live.local or edge proxy failure.",
|
|
data={"url": f"http://127.0.0.1:{port}/deploy/check", "report": data},
|
|
)
|
|
if env_flag(env, "ENABLE_LIVE_EDGE_PROXY", True):
|
|
ok_b, code_b, _body, err_b, latency_b = http_bytes("http://127.0.0.1/health", timeout=timeout)
|
|
check(
|
|
checks,
|
|
"edge_health",
|
|
"live-edge /health",
|
|
ok_b and code_b == 200,
|
|
detail=f"HTTP {code_b}, {latency_b} ms" if code_b else err_b,
|
|
hint="Port 80 should serve http://live.local through live-edge.service.",
|
|
data={"url": "http://127.0.0.1/health", "mdns_url": f"http://{mdns_host}/"},
|
|
)
|
|
|
|
|
|
def neko_required_count(env: dict[str, str]) -> int:
|
|
max_count = max(2, min(64, env_int(env, "NEKO_MAX_INSTANCE_COUNT", 64)))
|
|
return max(2, min(max_count, env_int(env, "NEKO_INSTANCE_COUNT", 2)))
|
|
|
|
|
|
def verify_neko(checks: list[dict[str, Any]], env: dict[str, str], *, timeout: float) -> None:
|
|
if not env_flag(env, "ENABLE_NEKO", True):
|
|
check(checks, "neko_enabled", "Neko enabled", True, required=False, detail="disabled")
|
|
return
|
|
managed_policies = env_flag(env, "NEKO_MANAGED_POLICIES", True)
|
|
docker_ok, names, err = docker_ps_names()
|
|
check(
|
|
checks,
|
|
"docker_daemon",
|
|
"Docker daemon",
|
|
docker_ok,
|
|
detail="running" if docker_ok else err,
|
|
hint="Start Docker before Neko/Redroid/SRS services.",
|
|
)
|
|
if not docker_ok:
|
|
return
|
|
ok, detail = systemd_active("live-neko-ip-watch.service")
|
|
check(
|
|
checks,
|
|
"neko_ip_watch_service",
|
|
"live-neko-ip-watch.service",
|
|
ok,
|
|
detail=detail,
|
|
hint="Run scripts/linux/upgrade_live_console.sh or reinstall the Neko LAN IP watcher.",
|
|
)
|
|
required = neko_required_count(env)
|
|
running = [name for name in names if name.startswith("live-neko-")]
|
|
check(
|
|
checks,
|
|
"neko_containers",
|
|
"Neko containers",
|
|
len(running) >= required,
|
|
detail=f"running={len(running)} required={required}: {', '.join(running) or '-'}",
|
|
hint="Run sudo bash scripts/linux/service_ctl.sh neko restart.",
|
|
)
|
|
for idx in range(1, required + 1):
|
|
name = f"live-neko-{idx}"
|
|
if name not in running:
|
|
continue
|
|
code, out, err2 = docker_exec(
|
|
name,
|
|
"supervisorctl status google-chrome 2>/dev/null || supervisorctl status chromium 2>/dev/null",
|
|
timeout=timeout,
|
|
)
|
|
check(
|
|
checks,
|
|
f"neko_{idx}_browser",
|
|
f"{name} browser process",
|
|
code == 0 and "RUNNING" in out,
|
|
detail=(out or err2).strip(),
|
|
hint="Make services/neko/browser-launch.sh executable and restart Neko.",
|
|
)
|
|
code, out, err2 = docker_exec(
|
|
name,
|
|
"if [ \"${NEKO_MEMBER_PROVIDER:-}\" = multiuser ] "
|
|
"&& [ -n \"${NEKO_MEMBER_MULTIUSER_ADMIN_PASSWORD:-}\" ] "
|
|
"&& [ -n \"${NEKO_MEMBER_MULTIUSER_USER_PASSWORD:-}\" ]; then "
|
|
"printf 'provider=%s admin_password=set user_password=set\\n' \"$NEKO_MEMBER_PROVIDER\"; "
|
|
"else echo 'NEKO member env invalid'; exit 1; fi",
|
|
timeout=timeout,
|
|
)
|
|
check(
|
|
checks,
|
|
f"neko_{idx}_multiuser_admin",
|
|
f"{name} multiuser admin enabled",
|
|
code == 0,
|
|
detail=(out or err2).strip(),
|
|
hint="Neko must run with multiuser provider and a non-empty admin password.",
|
|
)
|
|
code, out, err2 = docker_exec(
|
|
name,
|
|
"awk '$5==\"/home/neko/.config/google-chrome\" || $5==\"/home/neko/.config/chromium\" {print $5}' "
|
|
"/proc/self/mountinfo | sort -u",
|
|
timeout=timeout,
|
|
)
|
|
mounted_profiles = {line.strip() for line in out.splitlines() if line.strip()}
|
|
check(
|
|
checks,
|
|
f"neko_{idx}_persistent_profile",
|
|
f"{name} persistent browser profile",
|
|
code == 0
|
|
and {
|
|
"/home/neko/.config/google-chrome",
|
|
"/home/neko/.config/chromium",
|
|
}.issubset(mounted_profiles),
|
|
detail=out.strip() or err2.strip() or "profile mounts missing",
|
|
hint="Neko profile directories must be bind-mounted so Google login, Tampermonkey, and channel state survive restarts.",
|
|
)
|
|
code, out, err2 = docker_exec(
|
|
name,
|
|
"python3 - <<'PY'\n"
|
|
"import json\n"
|
|
"from pathlib import Path\n"
|
|
"for path in (Path('/home/neko/.config/google-chrome/Default/Preferences'), Path('/home/neko/.config/chromium/Default/Preferences'), Path('/home/neko/.config/google-chrome/Local State'), Path('/home/neko/.config/chromium/Local State')):\n"
|
|
" if not path.exists():\n"
|
|
" continue\n"
|
|
" try:\n"
|
|
" data = json.loads(path.read_text(encoding='utf-8'))\n"
|
|
" except Exception:\n"
|
|
" continue\n"
|
|
" if data.get('extensions', {}).get('ui', {}).get('developer_mode') is True:\n"
|
|
" print(f'{path}: developer_mode=true')\n"
|
|
" raise SystemExit(0)\n"
|
|
"print('Developer mode preference not found')\n"
|
|
"raise SystemExit(1)\n"
|
|
"PY",
|
|
timeout=timeout,
|
|
)
|
|
check(
|
|
checks,
|
|
f"neko_{idx}_developer_mode",
|
|
f"{name} Chrome developer mode",
|
|
code == 0,
|
|
detail=(out or err2).strip(),
|
|
hint="browser-launch.sh must set extensions.ui.developer_mode in Preferences/Local State before Chrome starts.",
|
|
)
|
|
code, out, err2 = docker_exec(
|
|
name,
|
|
"find /etc/chromium/policies/managed /etc/chromium-browser/policies/managed /etc/opt/chrome/policies/managed "
|
|
"-maxdepth 1 -type f -name '*.json' -print 2>/dev/null | head -1",
|
|
timeout=timeout,
|
|
)
|
|
policy_file = out.strip()
|
|
check(
|
|
checks,
|
|
f"neko_{idx}_personal_policy_mode",
|
|
f"{name} Chrome policy mode",
|
|
bool(policy_file) if managed_policies else not bool(policy_file),
|
|
required=True,
|
|
detail=(
|
|
f"managed policy enabled: {policy_file}"
|
|
if managed_policies and policy_file
|
|
else ("no managed policy JSON" if not policy_file else f"unexpected managed policy: {policy_file}")
|
|
),
|
|
hint=(
|
|
"NEKO_MANAGED_POLICIES=1 should mount services/neko/policies into Chrome policy dirs."
|
|
if managed_policies
|
|
else "Personal mode should mount services/neko/no-managed-policies over Chrome policy dirs."
|
|
),
|
|
)
|
|
code, out, err2 = docker_exec(name, "supervisorctl status tampermonkey-provisioning", timeout=timeout)
|
|
check(
|
|
checks,
|
|
f"neko_{idx}_tm_server",
|
|
f"{name} Tampermonkey provisioning server",
|
|
code == 0 and "RUNNING" in out,
|
|
detail=(out or err2).strip(),
|
|
hint="The Neko supervisor must run tampermonkey-provisioning on 127.0.0.1:18080.",
|
|
)
|
|
code, out, err2 = docker_exec(
|
|
name,
|
|
"curl -fsS http://127.0.0.1:18080/tampermonkey-provisioning.json | grep -q 'YouTube Studio Auto Dismiss'",
|
|
timeout=timeout,
|
|
)
|
|
check(
|
|
checks,
|
|
f"neko_{idx}_tm_payload",
|
|
f"{name} Tampermonkey default script payload",
|
|
code == 0,
|
|
detail="payload contains YouTube Studio Auto Dismiss" if code == 0 else (err2 or out).strip(),
|
|
hint="Regenerate services/neko/provisioning/tampermonkey-provisioning.json and restart Neko.",
|
|
)
|
|
code, out, err2 = docker_exec(
|
|
name,
|
|
"python3 - <<'PY'\n"
|
|
"import json\n"
|
|
"from pathlib import Path\n"
|
|
"ext = 'dhdgffkkebhmkfjojejmpbldmpobfkfo'\n"
|
|
"for base in ('/home/neko/.config/google-chrome/Default', '/home/neko/.config/chromium/Default'):\n"
|
|
" path = Path(base) / 'Preferences'\n"
|
|
" if not path.exists():\n"
|
|
" continue\n"
|
|
" try:\n"
|
|
" data = json.loads(path.read_text(encoding='utf-8'))\n"
|
|
" except Exception:\n"
|
|
" continue\n"
|
|
" state = data.get('extensions', {}).get('settings', {}).get(ext, {})\n"
|
|
" if state:\n"
|
|
" path_value = str(state.get('path') or '')\n"
|
|
" enabled = (\n"
|
|
" state.get('state') == 1\n"
|
|
" and not state.get('disable_reasons')\n"
|
|
" and state.get('from_webstore') is True\n"
|
|
" and not path_value.startswith('/')\n"
|
|
" and state.get('location') != 8\n"
|
|
" )\n"
|
|
" print(\n"
|
|
" f\"{base}: state={state.get('state')} disable_reasons={state.get('disable_reasons')} \"\n"
|
|
" f\"from_webstore={state.get('from_webstore')} location={state.get('location')} path={path_value}\"\n"
|
|
" )\n"
|
|
" raise SystemExit(0 if enabled else 1)\n"
|
|
"print('Tampermonkey preference not found')\n"
|
|
"raise SystemExit(1)\n"
|
|
"PY",
|
|
timeout=timeout,
|
|
)
|
|
check(
|
|
checks,
|
|
f"neko_{idx}_tm_enabled",
|
|
f"{name} Tampermonkey enabled",
|
|
code == 0,
|
|
detail=(out or err2).strip(),
|
|
hint="Tampermonkey must be a normal enabled extension, not an unpacked forced-disabled extension.",
|
|
)
|
|
code, out, err2 = docker_exec(
|
|
name,
|
|
"python3 - <<'PY'\n"
|
|
"import os\n"
|
|
"skip = {os.getpid(), os.getppid()}\n"
|
|
"for name in os.listdir('/proc'):\n"
|
|
" if not name.isdigit() or int(name) in skip:\n"
|
|
" continue\n"
|
|
" try:\n"
|
|
" cmdline = open(f'/proc/{name}/cmdline', 'rb').read()\n"
|
|
" except OSError:\n"
|
|
" continue\n"
|
|
" args = [arg for arg in cmdline.split(b'\\0') if arg]\n"
|
|
" has_unpacked_ext = any(arg.startswith(b'--load-extension=') and arg.split(b'=', 1)[1].strip() for arg in args)\n"
|
|
" has_legacy_youtube_ext = b'youtube-auto-' + b'dismiss-extension' in cmdline\n"
|
|
" if has_unpacked_ext or has_legacy_youtube_ext:\n"
|
|
" print(cmdline.replace(b'\\0', b' ').decode('utf-8', errors='replace'))\n"
|
|
" raise SystemExit(1)\n"
|
|
"print('no standalone youtube extension')\n"
|
|
"PY",
|
|
timeout=timeout,
|
|
)
|
|
check(
|
|
checks,
|
|
f"neko_{idx}_no_standalone_youtube_extension",
|
|
f"{name} no standalone YouTube extension",
|
|
code == 0,
|
|
detail=(out or err2).strip(),
|
|
hint="The YouTube Studio Auto Dismiss automation must live inside Tampermonkey, not as a separate --load-extension.",
|
|
)
|
|
code, out, err2 = docker_exec(
|
|
name,
|
|
"for base in /home/neko/.config/google-chrome/Default /home/neko/.config/chromium/Default; do "
|
|
"for area in 'Local Extension Settings' 'Managed Extension Settings'; do "
|
|
"dir=\"$base/$area/dhdgffkkebhmkfjojejmpbldmpobfkfo\"; "
|
|
"if [ -d \"$dir\" ] && grep -R -a -l 'YouTube Studio Auto Dismiss' \"$dir\" 2>/dev/null | head -1; then exit 0; fi; "
|
|
"done; done; exit 1",
|
|
timeout=timeout,
|
|
)
|
|
check(
|
|
checks,
|
|
f"neko_{idx}_tm_storage",
|
|
f"{name} Tampermonkey YouTube script imported",
|
|
code == 0 and bool(out.strip()),
|
|
required=not managed_policies,
|
|
detail=out.strip() or err2.strip() or "not found",
|
|
hint=(
|
|
"Open Neko once or restart the browser so Chrome policy imports the provisioning payload."
|
|
if managed_policies
|
|
else "Restart Neko so browser-launch.sh copies services/neko/tampermonkey-seed into the personal profile."
|
|
),
|
|
)
|
|
|
|
|
|
def verify_youtube_pro_capacity(
|
|
checks: list[dict[str, Any]],
|
|
env: dict[str, str],
|
|
*,
|
|
port: str,
|
|
timeout: float,
|
|
) -> None:
|
|
ok, code, data, err, latency = http_json(f"http://127.0.0.1:{port}/youtube_pro_channels", timeout=timeout)
|
|
channels = data.get("channels") if isinstance(data, dict) else []
|
|
if not isinstance(channels, list):
|
|
channels = []
|
|
desired = max(2, min(64, len(channels) or 2))
|
|
current = neko_required_count(env)
|
|
check(
|
|
checks,
|
|
"youtube_pro_neko_capacity",
|
|
"YouTube PRO Neko capacity",
|
|
ok and current >= desired,
|
|
required=env_flag(env, "ENABLE_NEKO", True),
|
|
detail=(
|
|
f"HTTP {code}, channels={len(channels)}, neko={current}, required={desired}, {latency} ms"
|
|
if code
|
|
else err
|
|
),
|
|
hint="Saving YouTube PRO channels should raise NEKO_INSTANCE_COUNT and restart Neko.",
|
|
)
|
|
|
|
|
|
def verify_android(checks: list[dict[str, Any]], env: dict[str, str], *, port: str, timeout: float) -> None:
|
|
ok, code, data, err, latency = http_json(f"http://127.0.0.1:{port}/android/devices", timeout=timeout)
|
|
devices = data.get("devices") if isinstance(data, dict) else []
|
|
if not isinstance(devices, list):
|
|
devices = []
|
|
panel_required = env_flag(env, "ENABLE_ANDROID_PANEL", True)
|
|
check(
|
|
checks,
|
|
"android_devices",
|
|
"Android device API",
|
|
ok and 200 <= code < 300 and isinstance(data, dict),
|
|
required=panel_required,
|
|
detail=f"HTTP {code}, devices={len(devices)}, {latency} ms" if code else err,
|
|
hint="The control plane must expose /android/devices for Redroid and physical ADB devices.",
|
|
)
|
|
check(
|
|
checks,
|
|
"android_device_presence",
|
|
"ADB Android device presence",
|
|
bool(data.get("available")) and bool(devices),
|
|
required=False,
|
|
detail=f"devices={len(devices)}",
|
|
hint="Run adb devices -l; for Redroid run scripts/linux/service_ctl.sh redroid start.",
|
|
)
|
|
if not devices:
|
|
return
|
|
serial = str(devices[0].get("serial") or "")
|
|
if not serial:
|
|
return
|
|
query = urllib.parse.urlencode({"serial": serial, "fast": "1"})
|
|
ok_b, code_b, body, err_b, latency_b = http_bytes(
|
|
f"http://127.0.0.1:{port}/android/screenshot?{query}",
|
|
timeout=max(timeout, 12.0),
|
|
)
|
|
check(
|
|
checks,
|
|
"android_screenshot",
|
|
"Android screenshot PNG",
|
|
ok_b and code_b == 200 and body.startswith(PNG_SIG),
|
|
required=True,
|
|
detail=f"HTTP {code_b}, bytes={len(body)}, {latency_b} ms" if code_b else err_b,
|
|
hint="The Web panel needs a valid PNG fallback even when scrcpy/WebCodecs fails.",
|
|
data={"serial": serial},
|
|
)
|
|
ok_a, code_a, action_data, err_a, latency_a = http_json_post(
|
|
f"http://127.0.0.1:{port}/android/action",
|
|
{"serial": serial, "action": "wake", "payload": {}},
|
|
timeout=max(timeout, 12.0),
|
|
)
|
|
check(
|
|
checks,
|
|
"android_action",
|
|
"Android web control action",
|
|
ok_a and code_a == 200 and isinstance(action_data, dict) and not action_data.get("error"),
|
|
required=True,
|
|
detail=(
|
|
f"HTTP {code_a}, {action_data.get('message', 'action ok')}, {latency_a} ms"
|
|
if code_a
|
|
else err_a
|
|
),
|
|
hint="The Web panel touch path requires /android/action to send ADB input events.",
|
|
data={"serial": serial},
|
|
)
|
|
|
|
|
|
def summarize(checks: list[dict[str, Any]]) -> dict[str, Any]:
|
|
required = [item for item in checks if item.get("required", True)]
|
|
failed = [item for item in required if not item.get("ok")]
|
|
return {
|
|
"ready": not failed,
|
|
"pass_count": sum(1 for item in checks if item.get("ok")),
|
|
"total_count": len(checks),
|
|
"required_total": len(required),
|
|
"checks": checks,
|
|
"failing_checks": failed,
|
|
"hints": [item["hint"] for item in failed if item.get("hint")],
|
|
}
|
|
|
|
|
|
def print_human(report: dict[str, Any]) -> None:
|
|
status = "READY" if report["ready"] else "DEGRADED"
|
|
print(f"SHD2YPP2 local verification: {status} ({report['pass_count']}/{report['total_count']})")
|
|
for item in report["checks"]:
|
|
mark = "OK" if item.get("ok") else ("FAIL" if item.get("required", True) else "WARN")
|
|
req = "required" if item.get("required", True) else "optional"
|
|
print(f"{mark:4} {item['id']:28} {req:8} {item['detail']}")
|
|
if report["hints"]:
|
|
print("Hints:")
|
|
for hint in report["hints"]:
|
|
print(f"- {hint}")
|
|
|
|
|
|
def build_report(args: argparse.Namespace) -> dict[str, Any]:
|
|
install_dir = Path(args.install_dir).resolve()
|
|
env = read_env(install_dir / "config" / "system-stack.env")
|
|
port = str(args.port or env.get("PORT") or os.environ.get("PORT") or "8001")
|
|
checks: list[dict[str, Any]] = []
|
|
verify_files(checks, install_dir)
|
|
verify_systemd(checks, env)
|
|
verify_http(checks, env, timeout=args.timeout)
|
|
if not args.skip_neko:
|
|
verify_neko(checks, env, timeout=args.timeout)
|
|
verify_youtube_pro_capacity(checks, env, port=port, timeout=args.timeout)
|
|
if not args.skip_android:
|
|
verify_android(checks, env, port=port, timeout=args.timeout)
|
|
report = summarize(checks)
|
|
report.update(
|
|
{
|
|
"install_dir": str(install_dir),
|
|
"port": port,
|
|
"mdns_host": f"{env.get('HOSTNAME_ALIAS', 'live')}.local",
|
|
}
|
|
)
|
|
return report
|
|
|
|
|
|
def main() -> int:
|
|
parser = argparse.ArgumentParser(description="Verify a local SHD2YPP2 installation")
|
|
parser.add_argument("--install-dir", default=str(DEFAULT_INSTALL_DIR), help="Runtime project directory")
|
|
parser.add_argument("--port", default="", help="Control plane port; defaults to system-stack.env PORT")
|
|
parser.add_argument("--timeout", type=float, default=5.0, help="HTTP and command timeout seconds")
|
|
parser.add_argument("--skip-neko", action="store_true", help="Skip Docker/Neko/Tampermonkey checks")
|
|
parser.add_argument("--skip-android", action="store_true", help="Skip Android/ADB screenshot checks")
|
|
parser.add_argument("--json", action="store_true", help="Print JSON report")
|
|
args = parser.parse_args()
|
|
|
|
report = build_report(args)
|
|
if args.json:
|
|
print(json.dumps(report, ensure_ascii=False, indent=2))
|
|
else:
|
|
print_human(report)
|
|
return 0 if report["ready"] else 1
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|