89 lines
2.9 KiB
Python
89 lines
2.9 KiB
Python
from __future__ import annotations
|
|
|
|
from typing import Any, Callable, Iterable
|
|
|
|
|
|
def singbox_clients_netns_state(clients: Iterable[Any]) -> tuple[bool, bool]:
|
|
flags: list[bool] = []
|
|
for client in clients:
|
|
cfg = getattr(client, "config", {}) or {}
|
|
if not isinstance(cfg, dict):
|
|
cfg = {}
|
|
flags.append(bool(cfg.get("netns_enabled", False)))
|
|
if not flags:
|
|
return False, False
|
|
return all(flags), any(flags)
|
|
|
|
|
|
def singbox_netns_toggle_button(all_enabled: bool, any_enabled: bool) -> tuple[str, str]:
|
|
if all_enabled:
|
|
return "Debug netns: ON", "green"
|
|
if any_enabled:
|
|
return "Debug netns: MIXED", "orange"
|
|
return "Debug netns: OFF", "gray"
|
|
|
|
|
|
def apply_singbox_netns_toggle(
|
|
controller: Any,
|
|
clients: Iterable[Any],
|
|
target_enabled: bool,
|
|
log_line: Callable[[str], None],
|
|
) -> list[str]:
|
|
failures: list[str] = []
|
|
client_ids: list[str] = []
|
|
for client in clients:
|
|
cid = str(getattr(client, "id", "") or "").strip()
|
|
if cid:
|
|
client_ids.append(cid)
|
|
|
|
if not client_ids:
|
|
return ["no SingBox clients selected"]
|
|
|
|
target = bool(target_enabled)
|
|
result = controller.transport_netns_toggle(
|
|
enabled=target,
|
|
client_ids=client_ids,
|
|
provision=True,
|
|
restart_running=True,
|
|
)
|
|
|
|
summary = (result.message or "").strip()
|
|
if summary:
|
|
log_line(summary)
|
|
|
|
for item in list(result.items or []):
|
|
cid = str(getattr(item, "client_id", "") or "").strip() or "unknown"
|
|
msg = str(getattr(item, "message", "") or "").strip()
|
|
code = str(getattr(item, "code", "") or "").strip()
|
|
status_before = str(getattr(item, "status_before", "") or "").strip().lower()
|
|
status_after = str(getattr(item, "status_after", "") or "").strip().lower()
|
|
config_updated = bool(getattr(item, "config_updated", False))
|
|
provisioned = bool(getattr(item, "provisioned", False))
|
|
restarted = bool(getattr(item, "restarted", False))
|
|
ok = bool(getattr(item, "ok", False))
|
|
|
|
steps: list[str] = []
|
|
if config_updated:
|
|
steps.append("config")
|
|
if provisioned:
|
|
steps.append("provision")
|
|
if restarted:
|
|
steps.append("restart")
|
|
step_text = ",".join(steps) if steps else "noop"
|
|
|
|
parts = [f"{cid}: {'ok' if ok else 'fail'}", f"steps={step_text}"]
|
|
if status_before or status_after:
|
|
parts.append(f"status {status_before or '-'}->{status_after or '-'}")
|
|
if msg:
|
|
parts.append(msg)
|
|
elif code:
|
|
parts.append(code)
|
|
log_line(" | ".join(parts))
|
|
|
|
if not ok:
|
|
failures.append(f"{cid}: {msg or code or 'toggle failed'}")
|
|
|
|
if not bool(getattr(result, "ok", False)) and not failures:
|
|
failures.append((summary or "netns toggle failed").strip())
|
|
return failures
|