shujakuin/proxy-port-monitor-tui
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Initial proxy port monitor TUI

Browse Source
This commit is contained in:
Shujakuin
2026-07-20 13:49:39 +08:00
commit 5c53e5bade
9 changed files with 1232 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
README.md
+126
View File
@@ -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`
__pycache__/proxy_monitor_tui.cpython-313.pyc
BIN
View File
Binary file not shown.
__pycache__/proxy_monitor_tui.cpython-314.pyc
BIN
View File
Binary file not shown.
proxy_monitor_tui.py
+781
View File
@@ -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())
pyproject.toml
+29
View File
@@ -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"]
tests/__pycache__/test_proxy_monitor_tui.cpython-313.pyc
BIN
View File
Binary file not shown.
tests/__pycache__/test_proxy_monitor_tui.cpython-314.pyc
BIN
View File
Binary file not shown.
tests/test_proxy_monitor_tui.py
+166
View File
@@ -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)
uv.lock
Generated
+130
View File
@@ -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" },
]