Shujakuin
857 lines
32 KiB
Python
857 lines
32 KiB
Python
#!/usr/bin/env python3
|
|
"""Textual client for proxy-port-monitor and mihomo selector groups."""
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
from concurrent.futures import ThreadPoolExecutor, as_completed
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import sys
|
|
from typing import Any, Iterator
|
|
import urllib.error
|
|
import urllib.parse
|
|
import urllib.request
|
|
|
|
|
|
APP_NAME = "proxy-port-monitor-tui"
|
|
DEFAULT_API_URL = "https://proxy.tailbeb9ad.ts.net"
|
|
DEFAULT_CONFIG_FILE = "~/.config/proxy-port-monitor-tui/api-url"
|
|
DEFAULT_REFRESH_SECONDS = 15
|
|
DEFAULT_TIMEOUT_SECONDS = 10
|
|
DEFAULT_TEST_URL = "https://www.gstatic.com/generate_204"
|
|
DEFAULT_TEST_TIMEOUT_MS = 5000
|
|
DEFAULT_DELAY_CONCURRENCY = 4
|
|
MAX_JSON_RESPONSE_BYTES = 8 * 1024 * 1024
|
|
|
|
|
|
class ApiError(RuntimeError):
|
|
"""A bounded, user-facing API failure."""
|
|
|
|
|
|
def env_int(name: str, default: int, *, minimum: int = 1, maximum: int | None = None) -> int:
|
|
try:
|
|
value = int(os.environ.get(name, default))
|
|
except (TypeError, ValueError):
|
|
value = default
|
|
value = max(minimum, value)
|
|
return min(value, maximum) if maximum is not None else value
|
|
|
|
|
|
def configured_url(env_names: tuple[str, ...], config_path: str, default: str) -> str:
|
|
for env_name in env_names:
|
|
value = os.environ.get(env_name, "").strip()
|
|
if value:
|
|
return value
|
|
path = Path(config_path).expanduser()
|
|
try:
|
|
for line in path.read_text(encoding="utf-8").splitlines():
|
|
value = line.strip()
|
|
if value and not value.startswith("#"):
|
|
return value
|
|
except OSError:
|
|
pass
|
|
return default
|
|
|
|
|
|
def default_api_url() -> str:
|
|
return configured_url(
|
|
("PROXY_MONITOR_TUI_API_URL",),
|
|
os.environ.get("PROXY_MONITOR_TUI_API_URL_FILE", DEFAULT_CONFIG_FILE),
|
|
DEFAULT_API_URL,
|
|
)
|
|
|
|
|
|
def config_file_path() -> Path:
|
|
return Path(os.environ.get("PROXY_MONITOR_TUI_API_URL_FILE", DEFAULT_CONFIG_FILE)).expanduser()
|
|
|
|
|
|
def write_api_url_config(api_url: str) -> Path:
|
|
value = normalize_base_url(api_url)
|
|
path = config_file_path()
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
try:
|
|
path.parent.chmod(0o700)
|
|
except OSError:
|
|
pass
|
|
path.write_text(value + "\n", encoding="utf-8")
|
|
try:
|
|
path.chmod(0o600)
|
|
except OSError:
|
|
pass
|
|
return path
|
|
|
|
|
|
def normalize_base_url(value: str) -> str:
|
|
candidate = str(value or "").strip()
|
|
parsed = urllib.parse.urlparse(candidate)
|
|
if parsed.scheme not in {"http", "https"} or not parsed.netloc:
|
|
raise ValueError("api url must be an absolute http(s) URL")
|
|
if parsed.params or parsed.query or parsed.fragment:
|
|
raise ValueError("api url must not include params, query, or fragment")
|
|
return urllib.parse.urlunparse(parsed._replace(path=parsed.path.rstrip("/")))
|
|
|
|
|
|
def validate_test_url(value: str) -> str:
|
|
candidate = str(value or "").strip()
|
|
parsed = urllib.parse.urlparse(candidate)
|
|
if parsed.scheme not in {"http", "https"} or not parsed.netloc:
|
|
raise ValueError("test url must be an absolute http(s) URL")
|
|
return candidate
|
|
|
|
|
|
def as_mapping(value: object) -> dict[str, Any]:
|
|
return value if isinstance(value, dict) else {}
|
|
|
|
|
|
def as_list(value: object) -> list[Any]:
|
|
return value if isinstance(value, list) else []
|
|
|
|
|
|
def as_int(value: object) -> int | None:
|
|
if isinstance(value, bool):
|
|
return None
|
|
try:
|
|
return int(value) if value is not None and str(value).strip() else None
|
|
except (TypeError, ValueError):
|
|
return None
|
|
|
|
|
|
def quote_path_component(value: object) -> str:
|
|
return urllib.parse.quote(str(value), safe="")
|
|
|
|
|
|
class ProxyMonitorClient:
|
|
"""Narrow client for the existing proxy-port-monitor API surface."""
|
|
|
|
def __init__(self, base_url: str, timeout_seconds: int) -> None:
|
|
self.base_url = normalize_base_url(base_url)
|
|
self.timeout_seconds = max(1, int(timeout_seconds))
|
|
self._base = urllib.parse.urlparse(self.base_url)
|
|
|
|
def endpoint(self, path: str, query: dict[str, object] | None = None) -> str:
|
|
normalized_path = "/" + str(path).lstrip("/")
|
|
base_path = self._base.path.rstrip("/")
|
|
encoded_query = urllib.parse.urlencode(query or {}, doseq=True)
|
|
return urllib.parse.urlunparse(
|
|
self._base._replace(path=base_path + normalized_path, query=encoded_query)
|
|
)
|
|
|
|
def _request(
|
|
self,
|
|
method: str,
|
|
path: str,
|
|
*,
|
|
query: dict[str, object] | None = None,
|
|
payload: dict[str, object] | None = None,
|
|
expect_json: bool = True,
|
|
) -> Any:
|
|
body = None if payload is None else json.dumps(payload, ensure_ascii=False).encode("utf-8")
|
|
request = urllib.request.Request(
|
|
self.endpoint(path, query),
|
|
data=body,
|
|
method=method,
|
|
headers={"Accept": "application/json", **({"Content-Type": "application/json"} if body else {})},
|
|
)
|
|
try:
|
|
with urllib.request.urlopen(request, timeout=self.timeout_seconds) as response:
|
|
try:
|
|
content_length = int(response.headers.get("Content-Length", "0"))
|
|
except ValueError:
|
|
content_length = 0
|
|
if content_length > MAX_JSON_RESPONSE_BYTES:
|
|
raise ApiError("response exceeds 8 MiB limit")
|
|
raw = response.read(MAX_JSON_RESPONSE_BYTES + 1)
|
|
except urllib.error.HTTPError as exc:
|
|
try:
|
|
detail = exc.read(1024).decode("utf-8", errors="replace").strip()
|
|
except Exception:
|
|
detail = ""
|
|
suffix = f": {detail}" if detail else ""
|
|
raise ApiError(f"HTTP {exc.code} {exc.reason}{suffix}") from exc
|
|
except urllib.error.URLError as exc:
|
|
raise ApiError(f"request failed: {exc.reason}") from exc
|
|
except OSError as exc:
|
|
raise ApiError(f"request failed: {exc}") from exc
|
|
|
|
if not expect_json:
|
|
return None
|
|
if len(raw) > MAX_JSON_RESPONSE_BYTES:
|
|
raise ApiError("response exceeds 8 MiB limit")
|
|
if not raw:
|
|
raise ApiError("response did not contain JSON")
|
|
try:
|
|
return json.loads(raw.decode("utf-8"))
|
|
except (UnicodeDecodeError, json.JSONDecodeError) as exc:
|
|
raise ApiError("response did not contain valid JSON") from exc
|
|
|
|
def fetch_status(self) -> dict[str, Any]:
|
|
payload = self._request("GET", "/status.json")
|
|
if not isinstance(payload, dict):
|
|
raise ApiError("status response must be an object")
|
|
return payload
|
|
|
|
def fetch_proxies(self, target_name: str) -> dict[str, Any]:
|
|
payload = self._request(
|
|
"GET",
|
|
f"/admin/{quote_path_component(target_name)}/api/proxies",
|
|
)
|
|
if not isinstance(payload, dict):
|
|
raise ApiError("proxies response must be an object")
|
|
return payload
|
|
|
|
def measure_delay(
|
|
self,
|
|
target_name: str,
|
|
proxy_name: str,
|
|
*,
|
|
test_url: str,
|
|
timeout_ms: int,
|
|
) -> int | None:
|
|
payload = self._request(
|
|
"GET",
|
|
f"/admin/{quote_path_component(target_name)}/api/proxies/{quote_path_component(proxy_name)}/delay",
|
|
query={"url": test_url, "timeout": max(100, int(timeout_ms))},
|
|
)
|
|
delay = as_int(as_mapping(payload).get("delay"))
|
|
return delay if delay is not None and delay > 0 else None
|
|
|
|
def select_proxy(self, target_name: str, group_name: str, proxy_name: str) -> None:
|
|
self._request(
|
|
"PUT",
|
|
f"/admin/{quote_path_component(target_name)}/api/proxies/{quote_path_component(group_name)}",
|
|
payload={"name": proxy_name},
|
|
expect_json=False,
|
|
)
|
|
|
|
|
|
def normalize_machine_rows(payload: dict[str, Any]) -> list[dict[str, Any]]:
|
|
rows: list[dict[str, Any]] = []
|
|
for item in as_list(payload.get("targets")):
|
|
target = as_mapping(item)
|
|
name = str(target.get("name") or "").strip()
|
|
if not name:
|
|
continue
|
|
ok = target.get("ok")
|
|
if ok is True:
|
|
state = "online"
|
|
elif ok is False:
|
|
state = "offline"
|
|
else:
|
|
state = "unknown"
|
|
rows.append(
|
|
{
|
|
"name": name,
|
|
"state": state,
|
|
"ok": ok is True,
|
|
"detail": str(target.get("detail") or "-"),
|
|
"checked_at": str(target.get("checked_at") or "-"),
|
|
"consecutive_failures": as_int(target.get("consecutive_failures")) or 0,
|
|
"exit_country": str(target.get("exit_country") or "-"),
|
|
"exit_country_code": str(target.get("exit_country_code") or ""),
|
|
"note": str(target.get("note") or ""),
|
|
}
|
|
)
|
|
return rows
|
|
|
|
|
|
def normalize_selector_groups(payload: dict[str, Any]) -> list[dict[str, Any]]:
|
|
groups: list[dict[str, Any]] = []
|
|
for raw_name, raw_proxy in as_mapping(payload.get("proxies")).items():
|
|
proxy = as_mapping(raw_proxy)
|
|
if str(proxy.get("type") or "").casefold() != "selector":
|
|
continue
|
|
candidates: list[str] = []
|
|
for candidate in as_list(proxy.get("all")):
|
|
value = str(candidate or "").strip()
|
|
if value and value not in candidates:
|
|
candidates.append(value)
|
|
if not candidates:
|
|
continue
|
|
groups.append(
|
|
{
|
|
"name": str(raw_name),
|
|
"now": str(proxy.get("now") or ""),
|
|
"candidates": candidates,
|
|
}
|
|
)
|
|
return sorted(groups, key=lambda item: str(item["name"]).casefold())
|
|
|
|
|
|
def recommend_fastest(delay_results: dict[str, tuple[int | None, str]]) -> str | None:
|
|
successful = [
|
|
(delay, name)
|
|
for name, (delay, _detail) in delay_results.items()
|
|
if delay is not None and delay > 0
|
|
]
|
|
return min(successful)[1] if successful else None
|
|
|
|
|
|
def iter_measure_candidates(
|
|
client: ProxyMonitorClient,
|
|
target_name: str,
|
|
candidates: list[str],
|
|
*,
|
|
test_url: str,
|
|
timeout_ms: int,
|
|
concurrency: int,
|
|
) -> Iterator[tuple[str, tuple[int | None, str]]]:
|
|
"""Yield each candidate result as soon as its controller request finishes."""
|
|
unique_candidates = list(dict.fromkeys(candidate for candidate in candidates if candidate))
|
|
|
|
def measure(candidate: str) -> tuple[int | None, str]:
|
|
try:
|
|
delay = client.measure_delay(
|
|
target_name,
|
|
candidate,
|
|
test_url=test_url,
|
|
timeout_ms=timeout_ms,
|
|
)
|
|
except Exception as exc: # noqa: BLE001
|
|
return None, str(exc)
|
|
return (delay, f"{delay} ms") if delay is not None else (None, "no delay result")
|
|
|
|
with ThreadPoolExecutor(max_workers=max(1, min(int(concurrency), len(unique_candidates) or 1))) as executor:
|
|
futures = {executor.submit(measure, candidate): candidate for candidate in unique_candidates}
|
|
for future in as_completed(futures):
|
|
candidate = futures[future]
|
|
try:
|
|
yield candidate, future.result()
|
|
except Exception as exc: # noqa: BLE001
|
|
yield candidate, (None, str(exc))
|
|
|
|
|
|
def measure_candidates(
|
|
client: ProxyMonitorClient,
|
|
target_name: str,
|
|
candidates: list[str],
|
|
*,
|
|
test_url: str,
|
|
timeout_ms: int,
|
|
concurrency: int,
|
|
) -> dict[str, tuple[int | None, str]]:
|
|
"""Collect all candidate results for non-interactive callers."""
|
|
unique_candidates = list(dict.fromkeys(candidate for candidate in candidates if candidate))
|
|
results = dict(
|
|
iter_measure_candidates(
|
|
client,
|
|
target_name,
|
|
unique_candidates,
|
|
test_url=test_url,
|
|
timeout_ms=timeout_ms,
|
|
concurrency=concurrency,
|
|
)
|
|
)
|
|
|
|
return {candidate: results.get(candidate, (None, "not run")) for candidate in unique_candidates}
|
|
|
|
|
|
def format_machine_exit(row: dict[str, Any]) -> str:
|
|
code = str(row.get("exit_country_code") or "").strip()
|
|
country = str(row.get("exit_country") or "-").strip()
|
|
return f"{country} {code}".strip()
|
|
|
|
|
|
def print_once(payload: dict[str, Any]) -> None:
|
|
rows = normalize_machine_rows(payload)
|
|
up = as_int(payload.get("targets_up"))
|
|
total = as_int(payload.get("targets_total"))
|
|
print(f"proxy targets: {up if up is not None else '-'} / {total if total is not None else len(rows)}")
|
|
print("name state detail exit checked failures")
|
|
for row in rows:
|
|
print(
|
|
f"{row['name'][:15]:<15} "
|
|
f"{row['state']:<8} "
|
|
f"{row['detail'][:22]:<22} "
|
|
f"{format_machine_exit(row)[:13]:<13} "
|
|
f"{row['checked_at'][:20]:<20} "
|
|
f"{row['consecutive_failures']}"
|
|
)
|
|
|
|
|
|
def run_textual(
|
|
client: ProxyMonitorClient,
|
|
*,
|
|
refresh_seconds: int,
|
|
test_url: str,
|
|
test_timeout_ms: int,
|
|
delay_concurrency: int,
|
|
) -> int:
|
|
try:
|
|
from textual import work
|
|
from textual.app import App, ComposeResult
|
|
from textual.screen import ModalScreen, Screen
|
|
from textual.widgets import DataTable, Footer, Header, Static
|
|
except ImportError:
|
|
print(
|
|
"Textual is required. Run with: uv run --with textual python proxy_monitor_tui.py",
|
|
file=sys.stderr,
|
|
)
|
|
return 2
|
|
|
|
class SwitchConfirmation(ModalScreen):
|
|
CSS = """
|
|
SwitchConfirmation {
|
|
align: center middle;
|
|
}
|
|
#switch-confirmation {
|
|
width: 72;
|
|
height: auto;
|
|
padding: 2 3;
|
|
border: tall $warning;
|
|
background: $surface;
|
|
}
|
|
"""
|
|
BINDINGS = [
|
|
("y", "confirm", "Confirm"),
|
|
("n", "cancel", "Cancel"),
|
|
("escape", "cancel", "Cancel"),
|
|
]
|
|
|
|
def __init__(self, target_name: str, group_name: str, current_name: str, next_name: str) -> None:
|
|
super().__init__()
|
|
self.message = (
|
|
"Runtime-only mihomo selector change\n\n"
|
|
f"machine: {target_name}\n"
|
|
f"group: {group_name}\n"
|
|
f"current: {current_name or '-'}\n"
|
|
f"next: {next_name}\n\n"
|
|
"Press y to switch, or n / Esc to cancel."
|
|
)
|
|
|
|
def compose(self) -> ComposeResult:
|
|
yield Static(self.message, id="switch-confirmation")
|
|
|
|
def action_confirm(self) -> None:
|
|
self.dismiss(True)
|
|
|
|
def action_cancel(self) -> None:
|
|
self.dismiss(False)
|
|
|
|
class MachineScreen(Screen):
|
|
CSS = """
|
|
#machine-summary, #machine-operation {
|
|
height: 2;
|
|
padding: 0 1;
|
|
}
|
|
#groups {
|
|
height: 1fr;
|
|
}
|
|
#nodes {
|
|
height: 1fr;
|
|
}
|
|
"""
|
|
BINDINGS = [
|
|
("escape", "back", "Back"),
|
|
("r", "reload_groups", "Reload"),
|
|
("t", "test_group", "Test delays"),
|
|
("s", "switch_node", "Switch node"),
|
|
("q", "quit", "Quit"),
|
|
]
|
|
|
|
def __init__(self, target: dict[str, Any]) -> None:
|
|
super().__init__()
|
|
self.target = target
|
|
self.groups_by_name: dict[str, dict[str, Any]] = {}
|
|
self.delay_results: dict[str, dict[str, tuple[int | None, str]]] = {}
|
|
self.delay_test_runs: dict[str, int] = {}
|
|
self.selected_group = ""
|
|
self.selected_node = ""
|
|
self.pending_switch: tuple[str, str] | None = None
|
|
|
|
def compose(self) -> ComposeResult:
|
|
yield Header(show_clock=True)
|
|
yield Static("", id="machine-summary")
|
|
yield DataTable(id="groups")
|
|
yield DataTable(id="nodes")
|
|
yield Static("", id="machine-operation")
|
|
yield Footer()
|
|
|
|
def on_mount(self) -> None:
|
|
groups = self.query_one("#groups", DataTable)
|
|
groups.cursor_type = "row"
|
|
groups.zebra_stripes = True
|
|
groups.add_columns("Selector group", "Current", "Candidates")
|
|
nodes = self.query_one("#nodes", DataTable)
|
|
nodes.cursor_type = "row"
|
|
nodes.zebra_stripes = True
|
|
nodes.add_columns("Candidate", "Current", "Delay", "Recommendation")
|
|
self.query_one("#machine-summary", Static).update(
|
|
f"{self.target['name']} | {self.target['state']} | "
|
|
f"{self.target['detail']} | exit {format_machine_exit(self.target)}"
|
|
)
|
|
self.load_groups()
|
|
|
|
def _set_operation(self, message: str) -> None:
|
|
self.query_one("#machine-operation", Static).update(message)
|
|
|
|
@work(thread=True, exclusive=True)
|
|
def load_groups(self) -> None:
|
|
try:
|
|
groups = normalize_selector_groups(client.fetch_proxies(str(self.target["name"])))
|
|
except Exception as exc: # noqa: BLE001
|
|
self.app.call_from_thread(self._set_operation, f"controller error: {exc}")
|
|
return
|
|
self.app.call_from_thread(self._finish_groups, groups, self.selected_group)
|
|
|
|
def _finish_groups(self, groups: list[dict[str, Any]], preferred_group: str = "") -> None:
|
|
self.groups_by_name = {str(group["name"]): group for group in groups}
|
|
table = self.query_one("#groups", DataTable)
|
|
table.clear()
|
|
for group in groups:
|
|
table.add_row(
|
|
str(group["name"]),
|
|
str(group.get("now") or "-"),
|
|
str(len(as_list(group.get("candidates")))),
|
|
key=str(group["name"]),
|
|
)
|
|
if not groups:
|
|
self.selected_group = ""
|
|
self.selected_node = ""
|
|
self.query_one("#nodes", DataTable).clear()
|
|
self._set_operation("no Selector groups returned by this controller")
|
|
return
|
|
selected = preferred_group if preferred_group in self.groups_by_name else str(groups[0]["name"])
|
|
self._show_group(selected)
|
|
self._set_operation(
|
|
"Select a group and candidate. t tests delay; s opens a runtime-only switch confirmation."
|
|
)
|
|
|
|
def _show_group(self, group_name: str) -> None:
|
|
group = self.groups_by_name.get(group_name)
|
|
if not group:
|
|
return
|
|
previous_node = self.selected_node if self.selected_group == group_name else ""
|
|
self.selected_group = group_name
|
|
table = self.query_one("#nodes", DataTable)
|
|
table.clear()
|
|
results = self.delay_results.get(group_name, {})
|
|
recommended = recommend_fastest(results)
|
|
candidates = as_list(group.get("candidates"))
|
|
self.selected_node = (
|
|
previous_node
|
|
if previous_node in candidates
|
|
else (str(candidates[0]) if candidates else "")
|
|
)
|
|
for candidate in candidates:
|
|
node = str(candidate)
|
|
delay, detail = results.get(node, (None, "-"))
|
|
table.add_row(
|
|
node,
|
|
"yes" if node == str(group.get("now") or "") else "",
|
|
f"{delay} ms" if delay is not None else detail,
|
|
"recommended" if node == recommended else "",
|
|
key=node,
|
|
)
|
|
|
|
def on_data_table_row_highlighted(self, event: Any) -> None:
|
|
table_id = str(event.data_table.id or "")
|
|
key = str(event.row_key.value)
|
|
if table_id == "groups":
|
|
self._show_group(key)
|
|
elif table_id == "nodes":
|
|
self.selected_node = key
|
|
|
|
def action_reload_groups(self) -> None:
|
|
self._set_operation("loading selector groups...")
|
|
self.load_groups()
|
|
|
|
def action_test_group(self) -> None:
|
|
group = self.groups_by_name.get(self.selected_group)
|
|
if not group:
|
|
self._set_operation("choose a Selector group first")
|
|
return
|
|
group_name = str(group["name"])
|
|
candidates = list(dict.fromkeys(str(candidate) for candidate in as_list(group.get("candidates")) if candidate))
|
|
run_id = self.delay_test_runs.get(group_name, 0) + 1
|
|
self.delay_test_runs[group_name] = run_id
|
|
self.delay_results[group_name] = {}
|
|
self._show_group(group_name)
|
|
self._set_operation(
|
|
f"testing {len(candidates)} candidates; each completed delay appears immediately..."
|
|
)
|
|
self.test_selected_group(group_name, candidates, run_id)
|
|
|
|
@work(thread=True, exclusive=True)
|
|
def test_selected_group(self, group_name: str, candidates: list[str], run_id: int) -> None:
|
|
total = len(candidates)
|
|
for candidate, result in iter_measure_candidates(
|
|
client,
|
|
str(self.target["name"]),
|
|
candidates,
|
|
test_url=test_url,
|
|
timeout_ms=test_timeout_ms,
|
|
concurrency=delay_concurrency,
|
|
):
|
|
self.app.call_from_thread(
|
|
self._record_delay_result,
|
|
group_name,
|
|
run_id,
|
|
candidate,
|
|
result,
|
|
total,
|
|
)
|
|
self.app.call_from_thread(self._finish_delay_test, group_name, run_id)
|
|
|
|
def _record_delay_result(
|
|
self,
|
|
group_name: str,
|
|
run_id: int,
|
|
candidate: str,
|
|
result: tuple[int | None, str],
|
|
total: int,
|
|
) -> None:
|
|
if self.delay_test_runs.get(group_name) != run_id:
|
|
return
|
|
results = self.delay_results.setdefault(group_name, {})
|
|
results[candidate] = result
|
|
completed = len(results)
|
|
recommended = recommend_fastest(results)
|
|
if group_name == self.selected_group:
|
|
self._show_group(group_name)
|
|
delay, detail = result
|
|
latest = f"{delay} ms" if delay is not None else detail
|
|
self._set_operation(
|
|
f"testing {completed}/{total}; {candidate}: {latest}; "
|
|
f"best so far: {recommended or 'none'}"
|
|
)
|
|
|
|
def _finish_delay_test(
|
|
self,
|
|
group_name: str,
|
|
run_id: int,
|
|
) -> None:
|
|
if self.delay_test_runs.get(group_name) != run_id:
|
|
return
|
|
results = self.delay_results.get(group_name, {})
|
|
recommended = recommend_fastest(results)
|
|
if group_name == self.selected_group:
|
|
self._show_group(group_name)
|
|
self._set_operation(
|
|
f"delay test complete; recommendation: {recommended or 'no reachable candidate'}; "
|
|
"no node was switched."
|
|
)
|
|
|
|
def action_switch_node(self) -> None:
|
|
group = self.groups_by_name.get(self.selected_group)
|
|
node = self.selected_node
|
|
if not group or not node:
|
|
self._set_operation("choose a Selector group and candidate first")
|
|
return
|
|
current = str(group.get("now") or "")
|
|
if node == current:
|
|
self._set_operation("selected node is already active")
|
|
return
|
|
self.pending_switch = (str(group["name"]), node)
|
|
self.app.push_screen(
|
|
SwitchConfirmation(str(self.target["name"]), str(group["name"]), current, node),
|
|
self._handle_switch_confirmation,
|
|
)
|
|
|
|
def _handle_switch_confirmation(self, confirmed: object) -> None:
|
|
pending = self.pending_switch
|
|
self.pending_switch = None
|
|
if not confirmed or not pending:
|
|
self._set_operation("switch cancelled")
|
|
return
|
|
group_name, node_name = pending
|
|
self._set_operation(f"switching {group_name} to {node_name}...")
|
|
self.switch_selected_node(group_name, node_name)
|
|
|
|
@work(thread=True, exclusive=True)
|
|
def switch_selected_node(self, group_name: str, node_name: str) -> None:
|
|
try:
|
|
client.select_proxy(str(self.target["name"]), group_name, node_name)
|
|
groups = normalize_selector_groups(client.fetch_proxies(str(self.target["name"])))
|
|
except Exception as exc: # noqa: BLE001
|
|
self.app.call_from_thread(self._set_operation, f"switch failed: {exc}")
|
|
return
|
|
self.app.call_from_thread(self._finish_switch, group_name, node_name, groups)
|
|
|
|
def _finish_switch(
|
|
self,
|
|
group_name: str,
|
|
node_name: str,
|
|
groups: list[dict[str, Any]],
|
|
) -> None:
|
|
selected = next((group for group in groups if group["name"] == group_name), {})
|
|
if str(as_mapping(selected).get("now") or "") != node_name:
|
|
self._set_operation(
|
|
f"switch request completed but controller reports {as_mapping(selected).get('now') or '-'}"
|
|
)
|
|
self._finish_groups(groups, group_name)
|
|
return
|
|
self._finish_groups(groups, group_name)
|
|
self._set_operation(f"switched {group_name} to {node_name}; verified by controller readback")
|
|
|
|
def action_back(self) -> None:
|
|
self.app.pop_screen()
|
|
|
|
class ProxyMonitorApp(App):
|
|
CSS = """
|
|
#summary, #status {
|
|
height: 2;
|
|
padding: 0 1;
|
|
}
|
|
#machines {
|
|
height: 1fr;
|
|
}
|
|
"""
|
|
BINDINGS = [
|
|
("q", "quit", "Quit"),
|
|
("r", "refresh", "Refresh"),
|
|
]
|
|
|
|
def __init__(self) -> None:
|
|
super().__init__()
|
|
self.rows_by_name: dict[str, dict[str, Any]] = {}
|
|
self.selected_machine = ""
|
|
|
|
def compose(self) -> ComposeResult:
|
|
yield Header(show_clock=True)
|
|
yield Static("", id="summary")
|
|
yield DataTable(id="machines")
|
|
yield Static("", id="status")
|
|
yield Footer()
|
|
|
|
def on_mount(self) -> None:
|
|
table = self.query_one("#machines", DataTable)
|
|
table.cursor_type = "row"
|
|
table.zebra_stripes = True
|
|
table.add_columns("Machine", "State", "17890 check", "Exit", "Checked", "Failures")
|
|
self.refresh_status()
|
|
self.set_interval(max(1, refresh_seconds), self.refresh_status)
|
|
|
|
@work(thread=True, exclusive=True)
|
|
def refresh_status(self) -> None:
|
|
try:
|
|
payload = client.fetch_status()
|
|
rows = normalize_machine_rows(payload)
|
|
except Exception as exc: # noqa: BLE001
|
|
self.app.call_from_thread(self._set_status, f"status refresh failed: {exc}")
|
|
return
|
|
self.app.call_from_thread(self._finish_status, payload, rows)
|
|
|
|
def _set_status(self, message: str) -> None:
|
|
self.query_one("#status", Static).update(message)
|
|
|
|
def _finish_status(self, payload: dict[str, Any], rows: list[dict[str, Any]]) -> None:
|
|
self.rows_by_name = {str(row["name"]): row for row in rows}
|
|
table = self.query_one("#machines", DataTable)
|
|
table.clear()
|
|
for row in rows:
|
|
table.add_row(
|
|
str(row["name"]),
|
|
str(row["state"]),
|
|
str(row["detail"]),
|
|
format_machine_exit(row),
|
|
str(row["checked_at"]),
|
|
str(row["consecutive_failures"]),
|
|
key=str(row["name"]),
|
|
)
|
|
if rows and self.selected_machine not in self.rows_by_name:
|
|
self.selected_machine = str(rows[0]["name"])
|
|
up = payload.get("targets_up", "-")
|
|
total = payload.get("targets_total", len(rows))
|
|
self.query_one("#summary", Static).update(
|
|
f"proxy-port-monitor | reachable {up}/{total} | "
|
|
"Enter opens a reachable machine; r refreshes."
|
|
)
|
|
self.query_one("#status", Static).update(
|
|
f"source {client.base_url} | last summary: {payload.get('last_summary') or '-'}"
|
|
)
|
|
|
|
def on_data_table_row_highlighted(self, event: Any) -> None:
|
|
if str(event.data_table.id or "") == "machines":
|
|
self.selected_machine = str(event.row_key.value)
|
|
|
|
def on_data_table_row_selected(self, event: Any) -> None:
|
|
if str(event.data_table.id or "") == "machines":
|
|
self.selected_machine = str(event.row_key.value)
|
|
self.action_open_machine()
|
|
|
|
def action_refresh(self) -> None:
|
|
self.query_one("#status", Static).update("refreshing...")
|
|
self.refresh_status()
|
|
|
|
def action_open_machine(self) -> None:
|
|
target = self.rows_by_name.get(self.selected_machine)
|
|
if not target:
|
|
self.query_one("#status", Static).update("choose a machine first")
|
|
return
|
|
if not target.get("ok"):
|
|
self.query_one("#status", Static).update(
|
|
f"{target['name']} is not reachable; controller actions are disabled"
|
|
)
|
|
return
|
|
self.push_screen(MachineScreen(target))
|
|
|
|
ProxyMonitorApp().run()
|
|
return 0
|
|
|
|
|
|
def build_parser() -> argparse.ArgumentParser:
|
|
parser = argparse.ArgumentParser(description="Proxy Port Monitor Textual TUI")
|
|
parser.add_argument("--api-url", default=default_api_url(), help="proxy-port-monitor base URL")
|
|
parser.add_argument("--save-config", action="store_true", help="save --api-url to local config before starting")
|
|
parser.add_argument(
|
|
"--refresh-seconds",
|
|
type=int,
|
|
default=env_int("PROXY_MONITOR_TUI_REFRESH_SECONDS", DEFAULT_REFRESH_SECONDS),
|
|
)
|
|
parser.add_argument(
|
|
"--timeout",
|
|
type=int,
|
|
default=env_int("PROXY_MONITOR_TUI_TIMEOUT", DEFAULT_TIMEOUT_SECONDS),
|
|
help="HTTP API timeout in seconds",
|
|
)
|
|
parser.add_argument(
|
|
"--test-url",
|
|
default=os.environ.get("PROXY_MONITOR_TUI_TEST_URL", DEFAULT_TEST_URL),
|
|
help="mihomo delay probe URL",
|
|
)
|
|
parser.add_argument(
|
|
"--test-timeout-ms",
|
|
type=int,
|
|
default=env_int(
|
|
"PROXY_MONITOR_TUI_TEST_TIMEOUT_MS",
|
|
DEFAULT_TEST_TIMEOUT_MS,
|
|
minimum=100,
|
|
),
|
|
help="mihomo delay timeout in milliseconds",
|
|
)
|
|
parser.add_argument(
|
|
"--delay-concurrency",
|
|
type=int,
|
|
default=env_int(
|
|
"PROXY_MONITOR_TUI_DELAY_CONCURRENCY",
|
|
DEFAULT_DELAY_CONCURRENCY,
|
|
maximum=16,
|
|
),
|
|
)
|
|
parser.add_argument("--once", action="store_true", help="print the connection snapshot and exit")
|
|
return parser
|
|
|
|
|
|
def main(argv: list[str] | None = None) -> int:
|
|
args = build_parser().parse_args(argv)
|
|
api_url = normalize_base_url(args.api_url)
|
|
test_url = validate_test_url(args.test_url)
|
|
if args.save_config:
|
|
print(f"saved api url to {write_api_url_config(api_url)}")
|
|
client = ProxyMonitorClient(api_url, max(1, args.timeout))
|
|
if args.once:
|
|
print_once(client.fetch_status())
|
|
return 0
|
|
return run_textual(
|
|
client,
|
|
refresh_seconds=max(1, args.refresh_seconds),
|
|
test_url=test_url,
|
|
test_timeout_ms=max(100, args.test_timeout_ms),
|
|
delay_concurrency=max(1, min(16, args.delay_concurrency)),
|
|
)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|