#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ check-web-port — 本地 Web 服务器管理器（单文件版） ================================================ 扫描本机所有处于 LISTEN 状态的 TCP 端口，探测其中哪些是 HTTP 服务， 在本地网页面板中展示，支持一键打开 / 关闭。 ✅ 单文件、零依赖（仅 Python 标准库）、纯本地运行、不联网。 ✅ HTML 面板已内嵌，运行后自动打开浏览器。 用法： python3 check-web-port.py # 默认端口 18443 python3 check-web-port.py 9000 # 指定端口 或 macOS 双击 start.command 启动。 说明：扫描端口 / 杀进程 / 开浏览器等系统级操作必须由本机程序完成， 浏览器沙箱无法直接做这些事，所以这个 Python 程序无法省略。 """ import json import os import re import signal import socket import subprocess import sys import threading from concurrent.futures import ThreadPoolExecutor from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from urllib.parse import urlparse # --------------------------------------------------------------------------- # 配置 # --------------------------------------------------------------------------- DEFAULT_PORT = 18443 HTTP_PROBE_TIMEOUT = 0.4 # 探测 HTTP 时的连接/读取超时（秒） PROBE_WORKERS = 32 # 并行探测线程数 # --------------------------------------------------------------------------- # 扫描逻辑 # --------------------------------------------------------------------------- def list_listeners(): """用 lsof 获取所有 TCP LISTEN 端口。""" try: out = subprocess.check_output( ["lsof", "-i", "-P", "-n", "-sTCP:LISTEN"], stderr=subprocess.DEVNULL, ).decode("utf-8", errors="ignore") except Exception: return [] results = [] for line in out.splitlines()[1:]: parts = line.split() if len(parts) < 9: continue command = parts[0] try: pid = int(parts[1]) except ValueError: continue user = parts[2] name = parts[8] # 形如 127.0.0.1:3000 / *:3000 / [::1]:3000 m = re.match(r"^(.+):(\d+)$", name) if not m: continue addr = m.group(1) port = int(m.group(2)) results.append( {"command": command, "pid": pid, "user": user, "addr": addr, "port": port} ) return results def probe_http(host, port, timeout=HTTP_PROBE_TIMEOUT): """探测 (host, port) 是否为 HTTP 服务，返回状态行与 Server 头。""" sock_host = host if host.startswith("[") and host.endswith("]"): sock_host = host.strip("[]") try: s = socket.create_connection((sock_host, port), timeout=timeout) s.settimeout(timeout) s.sendall(b"HEAD / HTTP/1.0\r\nHost: localhost\r\n\r\n") data = s.recv(2048) s.close() except Exception: return None text = data.decode("latin-1", errors="ignore") if not text.startswith("HTTP/"): return {"http": False} lines = text.splitlines() status_line = lines[0] if lines else "" server = "" for l in lines: if l.lower().startswith("server:"): server = l.split(":", 1)[1].strip() break return {"http": True, "status": status_line, "server": server} def host_for_probe(addr): if addr in ("*", "*:*"): return "127.0.0.1" if addr == "*": return "127.0.0.1" return addr def collect_servers(manager_port, manager_pid): """收集所有监听服务，并行探测 HTTP。""" listeners = [] seen = set() for l in list_listeners(): key = (l["pid"], l["port"], l["addr"]) if key in seen: continue seen.add(key) l["host"] = host_for_probe(l["addr"]) listeners.append(l) def probe_one(l): return (l, probe_http(l["host"], l["port"])) probe_results = {} with ThreadPoolExecutor(max_workers=PROBE_WORKERS) as ex: for l, probe in ex.map(probe_one, listeners): probe_results[(l["pid"], l["port"], l["addr"])] = probe servers = [] for l in listeners: probe = probe_results.get((l["pid"], l["port"], l["addr"])) is_web = bool(probe and probe.get("http")) host = l["host"] servers.append( { "command": l["command"], "pid": l["pid"], "user": l["user"], "addr": l["addr"], "port": l["port"], "host": host, "is_web": is_web, "server_header": probe.get("server", "") if probe else "", "status_line": probe.get("status", "") if probe else "", "url": f"http://{host}:{l['port']}", "is_self": (l["port"] == manager_port and l["pid"] == manager_pid), } ) servers.sort(key=lambda s: (not s["is_web"], s["port"])) return servers # --------------------------------------------------------------------------- # 操作 # --------------------------------------------------------------------------- def open_in_browser(url): try: if sys.platform == "darwin": subprocess.run(["open", url], check=False) elif os.name == "nt": os.startfile(url) # type: ignore[attr-defined] else: subprocess.run(["xdg-open", url], check=False) return True, "" except Exception as e: return False, str(e) def kill_process(pid): if pid <= 0: return False, "无效的 PID" try: os.kill(pid, signal.SIGTERM) return True, f"已向 PID {pid} 发送终止信号" except ProcessLookupError: return False, f"进程 {pid} 不存在" except PermissionError: try: subprocess.run( ["sudo", "-n", "kill", str(pid)], check=True, stderr=subprocess.DEVNULL, stdout=subprocess.DEVNULL, ) return True, f"已通过 sudo 终止 PID {pid}" except Exception as e: return False, f"无权限终止 PID {pid}：{e}" except Exception as e: return False, f"终止 PID {pid} 失败：{e}" # --------------------------------------------------------------------------- # 内嵌前端页面（原始字符串，保留 JS 中的 \n 等转义） # --------------------------------------------------------------------------- INDEX_HTML = r""" 本地 Web 服务器管理器

本地 Web 服务器管理器

扫描本机 LISTEN 端口 · 探测 HTTP · 一键打开 / 关闭
-
未连接 提示：本地运行时地址留空即可（同源）。若用云端页面，则填本机后端地址。
正在扫描本机端口…
""" # --------------------------------------------------------------------------- # HTTP 服务 # --------------------------------------------------------------------------- class Handler(BaseHTTPRequestHandler): manager_port = DEFAULT_PORT manager_pid = os.getpid() def log_message(self, *args): # 静默 pass def _cors(self): self.send_header("Access-Control-Allow-Origin", "*") self.send_header("Access-Control-Allow-Methods", "GET, POST, OPTIONS") self.send_header("Access-Control-Allow-Headers", "Content-Type") def do_OPTIONS(self): self.send_response(204) self._cors() self.send_header("Content-Length", "0") self.end_headers() def _json(self, obj, code=200): body = json.dumps(obj, ensure_ascii=False).encode("utf-8") self.send_response(code) self.send_header("Content-Type", "application/json; charset=utf-8") self.send_header("Content-Length", str(len(body))) self.send_header("Cache-Control", "no-store") self._cors() self.end_headers() self.wfile.write(body) def _html(self, text): body = text.encode("utf-8") self.send_response(200) self.send_header("Content-Type", "text/html; charset=utf-8") self.send_header("Content-Length", str(len(body))) self.send_header("Cache-Control", "no-store") self._cors() self.end_headers() self.wfile.write(body) def do_GET(self): parsed = urlparse(self.path) if parsed.path == "/": self._html(INDEX_HTML) elif parsed.path == "/api/servers": self._json({"ok": True, "servers": collect_servers(self.manager_port, self.manager_pid)}) else: self._json({"ok": False, "error": "not found"}, 404) def do_POST(self): parsed = urlparse(self.path) length = int(self.headers.get("Content-Length", 0) or 0) raw = self.rfile.read(length) if length else b"" try: payload = json.loads(raw.decode("utf-8")) if raw else {} except Exception: payload = {} if parsed.path == "/api/open": url = payload.get("url") if not url: self._json({"ok": False, "error": "缺少 url"}, 400) return ok, err = open_in_browser(url) self._json({"ok": ok, "error": err}) elif parsed.path == "/api/kill": pid = payload.get("pid") if pid in ("self", None): pid_val = self.manager_pid else: try: pid_val = int(pid) except (TypeError, ValueError): self._json({"ok": False, "error": "无效的 pid"}, 400) return ok, msg = kill_process(pid_val) if ok: self._json({"ok": True, "message": msg}) if pid_val == self.manager_pid: threading.Timer(0.4, lambda: os._exit(0)).start() else: self._json({"ok": False, "error": msg}) else: self._json({"ok": False, "error": "not found"}, 404) def find_free_port(preferred): for port in range(preferred, preferred + 50): with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s: try: s.bind(("127.0.0.1", port)) return port except OSError: continue return preferred def main(): port = DEFAULT_PORT if len(sys.argv) > 1: try: port = int(sys.argv[1]) except ValueError: print("参数应为端口号，例如: python3 check-web-port.py 9000") sys.exit(1) port = find_free_port(port) Handler.manager_port = port Handler.manager_pid = os.getpid() server = ThreadingHTTPServer(("127.0.0.1", port), Handler) url = f"http://localhost:{port}" print("=" * 56) print(" check-web-port · 本地 Web 服务器管理器") print(f" 面板地址: {url}") print(" 浏览器将自动打开，纯本地运行，不联网") print(" 按 Ctrl+C 退出") print("=" * 56) threading.Timer(0.6, lambda: open_in_browser(url)).start() try: server.serve_forever() except KeyboardInterrupt: print("\n已退出。") finally: server.server_close() if __name__ == "__main__": main()