#!/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())