commit 5c53e5bade0ec73af2c96f804f0a51a69cd074d6 Author: Shujakuin Date: Mon Jul 20 13:49:39 2026 +0800 Initial proxy port monitor TUI diff --git a/README.md b/README.md new file mode 100644 index 0000000..ed51700 --- /dev/null +++ b/README.md @@ -0,0 +1,126 @@ +# proxy-port-monitor-tui + +- 文档层级：客户端工具 +- 应用性质：客户端 TUI +- 源码来源：repo 原生 +- 运行关系：客户端工具 +- 部署模型：无 +- 对应服务：无 +- 对应 stack：无 +- Secret 边界：无 secret +- 分发方式：`uv run` / `uv tool install`。 +- 包名：`proxy-port-monitor-tui` +- 命令名：`proxy-monitor-tui` +- 本地状态路径：`~/.config/proxy-port-monitor-tui/` +- 目标 API：server2 的 proxy-port-monitor Tailscale Service。 +- 备注：客户端不保存 SSH 凭据、mihomo controller secret、订阅 URL 或节点密码。 + +## 项目定位 + +用于在本机终端查看 Proxy Port Monitor 的目标连通性，并通过既有的 +同源 controller 转发查看 mihomo Selector 组、做低流量延迟探测，以及 +手动切换单个 Selector 的运行态当前节点。 + +它不是通用代理客户端，不直接 SSH 到目标机，不编辑 subscription、mixin、 +host_vars 或远端 mihomo 配置文件。 + +## 文档边界 + +本 README 说明本地客户端；服务入口、Tailscale Service 和 controller secret +边界仍由 proxy-port-monitor 的 service README 维护。 + +## 源码结构 + +- `proxy_monitor_tui.py`：CLI、HTTP API client、Textual 视图和运行态切换确认。 +- `tests/test_proxy_monitor_tui.py`：mock HTTP server 覆盖状态、控制器路径、测速和切换请求。 +- `pyproject.toml`：Python 包与命令入口。 + +## 对应服务文档 + +- `services/server2/systemd/proxy-port-monitor/current/README.md` + +## 运行与部署模型 + +这是本机客户端工具，没有 systemd、Compose 或 Ansible 部署模型。开发时可直接运行： + +``BT@@text +cd apps/proxy-port-monitor-tui +uv run proxy-monitor-tui +``BT@@ + +安装为当前用户命令： + +``BT@@text +uv tool install git+https://gitea.shujk.top/shujakuin/proxy-port-monitor-tui.git +proxy-monitor-tui +``BT@@ + +升级到仓库最新版本： + +``BT@@text +uv tool install --upgrade git+https://gitea.shujk.top/shujakuin/proxy-port-monitor-tui.git +``BT@@ + +若要把它作为某个 Python 项目的依赖（而非安装为当前用户命令），在该项目目录运行： + +``BT@@text +uv add git+https://gitea.shujk.top/shujakuin/proxy-port-monitor-tui.git +uv run proxy-monitor-tui +``BT@@ + +默认 API 地址是 `https://proxy.tailbeb9ad.ts.net`。可使用 `--api-url` 覆盖，或以 +`--save-config` 保存到本地状态路径。支持的环境变量： + +- `PROXY_MONITOR_TUI_API_URL` +- `PROXY_MONITOR_TUI_API_URL_FILE` +- `PROXY_MONITOR_TUI_REFRESH_SECONDS` +- `PROXY_MONITOR_TUI_TIMEOUT` +- `PROXY_MONITOR_TUI_TEST_URL` +- `PROXY_MONITOR_TUI_TEST_TIMEOUT_MS` +- `PROXY_MONITOR_TUI_DELAY_CONCURRENCY` + +## 配置与 Secret 边界 + +客户端只读取 `/status.json`，并限于调用每台机器的 `proxies`、`delay` 和 Selector +切换 controller API；不会调用 Proxy Port Monitor 的 `/config` 写接口。 + +controller secret 始终由 server2 的 proxy-port-monitor 服务注入。客户端不读取 +SOPS 文件、运行时 env、SSH key 或 mihomo 配置。 + +## 使用方式 + +主列表显示机器的 17890 可达性、连续失败数、出口位置和最近检查时间： + +- `Enter`：仅在目标在线时进入其 controller 详情。 +- `r`：刷新连通性状态。 +- `t`：对当前 Selector 组的候选节点做 mihomo delay 探测；默认使用 + `https://www.gstatic.com/generate_204`，最多并发 4 个。 +- `s`：对当前选择的候选节点发起运行态切换。TUI 会显示机器、组、旧节点和新节点， + 必须按 `y` 二次确认；成功后立即读取 controller 验证当前节点。 + +测速只推荐最低延迟的成功节点，绝不自动切换。切换不会写回订阅、mixin、受管模型 +或仓库真源，服务重启或上游配置刷新后的持久化行为以目标 mihomo 实现为准。 + +可用 `--once` 只打印当前机器连通性后退出，适用于脚本或只读排障。 + +## 测试与发布 + +本地验证： + +``BT@@text +cd apps/proxy-port-monitor-tui +uv run python -m unittest discover -s tests -v +``BT@@ + +测试使用本地 mock HTTP server，不访问 Tailscale、controller 或任何远端服务。 + +首次连接真实环境时，先执行 `--once` 或只做读取和延迟探测。任何节点切换都属于 +运行态控制操作，必须在当次获得明确授权，并限定为一台机器、一个 Selector 组和 +一个候选节点。 + +## Files + +- `apps/proxy-port-monitor-tui/proxy_monitor_tui.py` +- `apps/proxy-port-monitor-tui/tests/test_proxy_monitor_tui.py` +- `apps/proxy-port-monitor/README.md` +- `services/server2/systemd/proxy-port-monitor/current/README.md` diff --git a/__pycache__/proxy_monitor_tui.cpython-313.pyc b/__pycache__/proxy_monitor_tui.cpython-313.pyc new file mode 100644 index 0000000..a0d5db4 Binary files /dev/null and b/__pycache__/proxy_monitor_tui.cpython-313.pyc differ diff --git a/__pycache__/proxy_monitor_tui.cpython-314.pyc b/__pycache__/proxy_monitor_tui.cpython-314.pyc new file mode 100644 index 0000000..954832e Binary files /dev/null and b/__pycache__/proxy_monitor_tui.cpython-314.pyc differ diff --git a/proxy_monitor_tui.py b/proxy_monitor_tui.py new file mode 100644 index 0000000..adc6aec --- /dev/null +++ b/proxy_monitor_tui.py @@ -0,0 +1,781 @@ +#!/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 +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 + + +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: + raw = response.read(1_048_576) + 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 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 measure_candidates( + client: ProxyMonitorClient, + target_name: str, + candidates: list[str], + *, + test_url: str, + timeout_ms: int, + concurrency: int, +) -> dict[str, tuple[int | None, str]]: + unique_candidates = list(dict.fromkeys(candidate for candidate in candidates if candidate)) + results: dict[str, tuple[int | None, str]] = {} + + 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: + results[candidate] = future.result() + except Exception as exc: # noqa: BLE001 + results[candidate] = (None, str(exc)) + + 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.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 + self._set_operation(f"testing {len(as_list(group.get('candidates')))} candidates...") + self.test_selected_group(str(group["name"]), list(as_list(group.get("candidates")))) + + @work(thread=True, exclusive=True) + def test_selected_group(self, group_name: str, candidates: list[str]) -> None: + results = 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._finish_delay_test, group_name, results) + + def _finish_delay_test( + self, + group_name: str, + results: dict[str, tuple[int | None, str]], + ) -> None: + self.delay_results[group_name] = results + 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()) diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..cf0e1f1 --- /dev/null +++ b/pyproject.toml @@ -0,0 +1,29 @@ +[project] +name = "proxy-port-monitor-tui" +version = "0.1.0" +description = "Terminal UI for proxy-port-monitor and mihomo selector groups" +readme = "README.md" +requires-python = ">=3.11" +license = { text = "MIT" } +authors = [ + { name = "Shujakuin" }, +] +keywords = ["proxy", "mihomo", "clash", "tui", "textual", "tailscale"] +dependencies = [ + "textual>=0.89.1", +] + +[build-system] +requires = ["setuptools>=68"] +build-backend = "setuptools.build_meta" + +[project.scripts] +proxy-monitor-tui = "proxy_monitor_tui:main" + +[project.urls] +Homepage = "https://gitea.shujk.top/shujakuin/proxy-port-monitor-tui" +Repository = "https://gitea.shujk.top/shujakuin/proxy-port-monitor-tui.git" +Issues = "https://gitea.shujk.top/shujakuin/proxy-port-monitor-tui/issues" + +[tool.setuptools] +py-modules = ["proxy_monitor_tui"] diff --git a/tests/__pycache__/test_proxy_monitor_tui.cpython-313.pyc b/tests/__pycache__/test_proxy_monitor_tui.cpython-313.pyc new file mode 100644 index 0000000..29a42a1 Binary files /dev/null and b/tests/__pycache__/test_proxy_monitor_tui.cpython-313.pyc differ diff --git a/tests/__pycache__/test_proxy_monitor_tui.cpython-314.pyc b/tests/__pycache__/test_proxy_monitor_tui.cpython-314.pyc new file mode 100644 index 0000000..7e8c366 Binary files /dev/null and b/tests/__pycache__/test_proxy_monitor_tui.cpython-314.pyc differ diff --git a/tests/test_proxy_monitor_tui.py b/tests/test_proxy_monitor_tui.py new file mode 100644 index 0000000..2823bb3 --- /dev/null +++ b/tests/test_proxy_monitor_tui.py @@ -0,0 +1,166 @@ +from __future__ import annotations + +import importlib.util +import json +from pathlib import Path +import sys +import threading +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +from typing import Any +import unittest +import urllib.parse + + +def load_module(): + module_path = Path(__file__).resolve().parents[1] / "proxy_monitor_tui.py" + spec = importlib.util.spec_from_file_location("proxy_monitor_tui", module_path) + assert spec and spec.loader + module = importlib.util.module_from_spec(spec) + sys.modules[spec.name] = module + spec.loader.exec_module(module) + return module + + +class ApiFixtureHandler(BaseHTTPRequestHandler): + requests: list[dict[str, Any]] = [] + + def _json(self, status: int, payload: object) -> None: + body = json.dumps(payload).encode("utf-8") + self.send_response(status) + self.send_header("Content-Type", "application/json") + self.send_header("Content-Length", str(len(body))) + self.end_headers() + self.wfile.write(body) + + def do_GET(self) -> None: # noqa: N802 + parsed = urllib.parse.urlparse(self.path) + type(self).requests.append({"method": "GET", "path": parsed.path, "query": urllib.parse.parse_qs(parsed.query)}) + if parsed.path == "/base/status.json": + self._json(200, {"targets_up": 1, "targets_total": 1, "targets": [{"name": "pc", "ok": True}]}) + return + if parsed.path == "/base/admin/pc%20win/api/proxies": + self._json( + 200, + { + "proxies": { + "Selector A": { + "type": "Selector", + "now": "Node A", + "all": ["Node A", "Node B"], + } + } + }, + ) + return + if parsed.path == "/base/admin/pc%20win/api/proxies/Node%20A/delay": + self._json(200, {"delay": 42}) + return + self._json(404, {"error": "not found"}) + + def do_PUT(self) -> None: # noqa: N802 + parsed = urllib.parse.urlparse(self.path) + length = int(self.headers.get("Content-Length", "0")) + payload = json.loads(self.rfile.read(length).decode("utf-8")) + type(self).requests.append({"method": "PUT", "path": parsed.path, "payload": payload}) + if parsed.path == "/base/admin/pc%20win/api/proxies/Selector%20A": + self.send_response(204) + self.send_header("Content-Length", "0") + self.end_headers() + return + self._json(404, {"error": "not found"}) + + def log_message(self, format: str, *args: object) -> None: # noqa: A003 + return + + +class ProxyMonitorTUITests(unittest.TestCase): + @classmethod + def setUpClass(cls) -> None: + cls.server = ThreadingHTTPServer(("127.0.0.1", 0), ApiFixtureHandler) + cls.thread = threading.Thread(target=cls.server.serve_forever, daemon=True) + cls.thread.start() + + @classmethod + def tearDownClass(cls) -> None: + cls.server.shutdown() + cls.thread.join(timeout=5) + cls.server.server_close() + + def setUp(self) -> None: + ApiFixtureHandler.requests = [] + self.mod = load_module() + self.base_url = f"http://127.0.0.1:{self.server.server_port}/base" + self.client = self.mod.ProxyMonitorClient(self.base_url, 2) + + def test_status_and_controller_paths_keep_the_base_prefix_and_quote_names(self) -> None: + status = self.client.fetch_status() + proxies = self.client.fetch_proxies("pc win") + delay = self.client.measure_delay( + "pc win", + "Node A", + test_url="https://www.gstatic.com/generate_204", + timeout_ms=5000, + ) + self.client.select_proxy("pc win", "Selector A", "Node B") + + self.assertEqual(status["targets_up"], 1) + self.assertIn("Selector A", proxies["proxies"]) + self.assertEqual(delay, 42) + self.assertEqual( + ApiFixtureHandler.requests, + [ + {"method": "GET", "path": "/base/status.json", "query": {}}, + {"method": "GET", "path": "/base/admin/pc%20win/api/proxies", "query": {}}, + { + "method": "GET", + "path": "/base/admin/pc%20win/api/proxies/Node%20A/delay", + "query": {"url": ["https://www.gstatic.com/generate_204"], "timeout": ["5000"]}, + }, + { + "method": "PUT", + "path": "/base/admin/pc%20win/api/proxies/Selector%20A", + "payload": {"name": "Node B"}, + }, + ], + ) + + def test_selector_groups_only_include_switchable_groups(self) -> None: + groups = self.mod.normalize_selector_groups( + { + "proxies": { + "url-test": {"type": "URLTest", "all": ["A"]}, + "selector": {"type": "Selector", "now": "A", "all": ["A", "B", "A"]}, + "empty": {"type": "Selector", "now": "", "all": []}, + } + } + ) + self.assertEqual(groups, [{"name": "selector", "now": "A", "candidates": ["A", "B"]}]) + + def test_recommendation_ignores_failed_or_zero_delay_candidates(self) -> None: + recommended = self.mod.recommend_fastest( + { + "failure": (None, "timed out"), + "zero": (0, "no delay result"), + "slow": (340, "340 ms"), + "fast": (81, "81 ms"), + } + ) + self.assertEqual(recommended, "fast") + + def test_machine_rows_preserve_monitor_order_and_status(self) -> None: + rows = self.mod.normalize_machine_rows( + { + "targets": [ + {"name": "first", "ok": True, "detail": "latency=1ms", "exit_country": "Japan"}, + {"name": "second", "ok": False, "consecutive_failures": 3}, + ] + } + ) + self.assertEqual([row["name"] for row in rows], ["first", "second"]) + self.assertEqual(rows[0]["state"], "online") + self.assertEqual(rows[1]["state"], "offline") + self.assertEqual(rows[1]["consecutive_failures"], 3) + + def test_invalid_api_url_is_rejected_before_network_access(self) -> None: + with self.assertRaises(ValueError): + self.mod.ProxyMonitorClient("proxy.tailbeb9ad.ts.net", 2) diff --git a/uv.lock b/uv.lock new file mode 100644 index 0000000..d3790ba --- /dev/null +++ b/uv.lock @@ -0,0 +1,130 @@ +version = 1 +revision = 3 +requires-python = ">=3.11" + +[[package]] +name = "linkify-it-py" +version = "2.1.0" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "uc-micro-py" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/2e/c9/06ea13676ef354f0af6169587ae292d3e2406e212876a413bf9eece4eb23/linkify_it_py-2.1.0.tar.gz", hash = "sha256:43360231720999c10e9328dc3691160e27a718e280673d444c38d7d3aaa3b98b", size = 29158, upload-time = "2026-03-01T07:48:47.683Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/b4/de/88b3be5c31b22333b3ca2f6ff1de4e863d8fe45aaea7485f591970ec1d3e/linkify_it_py-2.1.0-py3-none-any.whl", hash = "sha256:0d252c1594ecba2ecedc444053db5d3a9b7ec1b0dd929c8f1d74dce89f86c05e", size = 19878, upload-time = "2026-03-01T07:48:46.098Z" }, +] + +[[package]] +name = "markdown-it-py" +version = "4.2.0" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "mdurl" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/06/ff/7841249c247aa650a76b9ee4bbaeae59370dc8bfd2f6c01f3630c35eb134/markdown_it_py-4.2.0.tar.gz", hash = "sha256:04a21681d6fbb623de53f6f364d352309d4094dd4194040a10fd51833e418d49", size = 82454, upload-time = "2026-05-07T12:08:28.36Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/b3/81/4da04ced5a082363ecfa159c010d200ecbd959ae410c10c0264a38cac0f5/markdown_it_py-4.2.0-py3-none-any.whl", hash = "sha256:9f7ebbcd14fe59494226453aed97c1070d83f8d24b6fc3a3bcf9a38092641c4a", size = 91687, upload-time = "2026-05-07T12:08:27.182Z" }, +] + +[package.optional-dependencies] +linkify = [ + { name = "linkify-it-py" }, +] + +[[package]] +name = "mdit-py-plugins" +version = "0.6.1" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "markdown-it-py" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/59/fc/f8d0863f8862f25602c0404d75568e89fb6b4109804645e5cdfb1be5cf56/mdit_py_plugins-0.6.1.tar.gz", hash = "sha256:a2bca0f039f39dbd35fb74ae1b5f998608c437463371f0ff7f49a19a17a114d0", size = 56114, upload-time = "2026-05-13T09:03:38.91Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/a5/69/6da5581c6a7fede7dc261bf4e67d6adca4196f176b43288b55b3db395b6e/mdit_py_plugins-0.6.1-py3-none-any.whl", hash = "sha256:214c82fb2ac524472ab6a5bcab1de80f73b50443e187f401bfd77efbc7c6481d", size = 66663, upload-time = "2026-05-13T09:03:37.76Z" }, +] + +[[package]] +name = "mdurl" +version = "0.1.2" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/d6/54/cfe61301667036ec958cb99bd3efefba235e65cdeb9c84d24a8293ba1d90/mdurl-0.1.2.tar.gz", hash = "sha256:bb413d29f5eea38f31dd4754dd7377d4465116fb207585f97bf925588687c1ba", size = 8729, upload-time = "2022-08-14T12:40:10.846Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/b3/38/89ba8ad64ae25be8de66a6d463314cf1eb366222074cfda9ee839c56a4b4/mdurl-0.1.2-py3-none-any.whl", hash = "sha256:84008a41e51615a49fc9966191ff91509e3c40b939176e643fd50a5c2196b8f8", size = 9979, upload-time = "2022-08-14T12:40:09.779Z" }, +] + +[[package]] +name = "platformdirs" +version = "4.10.1" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/52/cd/4f25b2f95b23f5d2c9c1fe43e49841bff5800562149b2666afc09309aa8f/platformdirs-4.10.1.tar.gz", hash = "sha256:ceab4084426fe6319ce18e86deada8ab1b7487c7aee7040c55e277c9ae793695", size = 31678, upload-time = "2026-07-18T03:53:43.808Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/ec/73/6fd0bb9ce84138c3857f12e9de63bc901852975a092d545f18087a204aa2/platformdirs-4.10.1-py3-none-any.whl", hash = "sha256:0e4eff26be2d75293977f7cddc153fd9b8eaa7fb0c7b64ffe4076cb443117443", size = 22906, upload-time = "2026-07-18T03:53:42.576Z" }, +] + +[[package]] +name = "proxy-port-monitor-tui" +version = "0.1.0" +source = { editable = "." } +dependencies = [ + { name = "textual" }, +] + +[package.metadata] +requires-dist = [{ name = "textual", specifier = ">=0.89.1" }] + +[[package]] +name = "pygments" +version = "2.20.0" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/c3/b2/bc9c9196916376152d655522fdcebac55e66de6603a76a02bca1b6414f6c/pygments-2.20.0.tar.gz", hash = "sha256:6757cd03768053ff99f3039c1a36d6c0aa0b263438fcab17520b30a303a82b5f", size = 4955991, upload-time = "2026-03-29T13:29:33.898Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/f4/7e/a72dd26f3b0f4f2bf1dd8923c85f7ceb43172af56d63c7383eb62b332364/pygments-2.20.0-py3-none-any.whl", hash = "sha256:81a9e26dd42fd28a23a2d169d86d7ac03b46e2f8b59ed4698fb4785f946d0176", size = 1231151, upload-time = "2026-03-29T13:29:30.038Z" }, +] + +[[package]] +name = "rich" +version = "15.0.0" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "markdown-it-py" }, + { name = "pygments" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/c0/8f/0722ca900cc807c13a6a0c696dacf35430f72e0ec571c4275d2371fca3e9/rich-15.0.0.tar.gz", hash = "sha256:edd07a4824c6b40189fb7ac9bc4c52536e9780fbbfbddf6f1e2502c31b068c36", size = 230680, upload-time = "2026-04-12T08:24:00.75Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/82/3b/64d4899d73f91ba49a8c18a8ff3f0ea8f1c1d75481760df8c68ef5235bf5/rich-15.0.0-py3-none-any.whl", hash = "sha256:33bd4ef74232fb73fe9279a257718407f169c09b78a87ad3d296f548e27de0bb", size = 310654, upload-time = "2026-04-12T08:24:02.83Z" }, +] + +[[package]] +name = "textual" +version = "8.2.8" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "markdown-it-py", extra = ["linkify"] }, + { name = "mdit-py-plugins" }, + { name = "platformdirs" }, + { name = "pygments" }, + { name = "rich" }, + { name = "typing-extensions" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/00/21/39a76b01bd5eea82a04baaca7580e105d8c59450df03998345bb2cfb307b/textual-8.2.8.tar.gz", hash = "sha256:3f106a9fbc73e39dd266c9712432087de78a6d644084c7c241d6a25c3169115b", size = 1860502, upload-time = "2026-06-30T06:51:24.495Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/fb/be/35261223d9416a0751cdff1c7b4a6f881387218a12d439fe22fefebc8c04/textual-8.2.8-py3-none-any.whl", hash = "sha256:267375fd402dc8d981457212efa71f0e3365fd17bba144ba9bb3ed7563cb374a", size = 731418, upload-time = "2026-06-30T06:51:26.364Z" }, +] + +[[package]] +name = "typing-extensions" +version = "4.16.0" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/f6/cc/6253133b5bb138fc3306cebfbda2c520f545d36b5be2c7255cc528bb45d6/typing_extensions-4.16.0.tar.gz", hash = "sha256:dc983d19a509c94dba722ee6abd33940f7c05a89e243c47e907eb4db6f1a43e5", size = 113555, upload-time = "2026-07-02T08:40:05.92Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/49/d3/b8441a820a491ddfc024b0b0cf0393375b75ea13866d9c66727e54c2fc80/typing_extensions-4.16.0-py3-none-any.whl", hash = "sha256:481caa481374e813c1b176ada14e97f1f67a4539ce9cfeb3f350d78d6370c2e8", size = 45571, upload-time = "2026-07-02T08:40:04.659Z" }, +] + +[[package]] +name = "uc-micro-py" +version = "2.0.0" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/78/67/9a363818028526e2d4579334460df777115bdec1bb77c08f9db88f6389f2/uc_micro_py-2.0.0.tar.gz", hash = "sha256:c53691e495c8db60e16ffc4861a35469b0ba0821fe409a8a7a0a71864d33a811", size = 6611, upload-time = "2026-03-01T06:31:27.526Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/61/73/d21edf5b204d1467e06500080a50f79d49ef2b997c79123a536d4a17d97c/uc_micro_py-2.0.0-py3-none-any.whl", hash = "sha256:3603a3859af53e5a39bc7677713c78ea6589ff188d70f4fee165db88e22b242c", size = 6383, upload-time = "2026-03-01T06:31:26.257Z" }, +]