diff --git a/.claude/skills/port/SKILL.md b/.claude/skills/port/SKILL.md new file mode 100644 index 0000000..7a5eb2f --- /dev/null +++ b/.claude/skills/port/SKILL.md @@ -0,0 +1,45 @@ +--- +name: port +description: 列出当前电脑正在监听的本地 web 服务(端口、可点击网址、页面 title、进程/PID),并支持按列表 ID kill 掉对应服务。当用户说 /port、"看看本地起了哪些服务"、"哪些端口在监听"、"把第 N 个服务杀掉"、"kill 5173" 时使用。 +--- + +# port — 本地 web 服务列表与按 ID kill + +所有逻辑都在 `scripts/port.py`(本 skill 目录下),不要用内联 bash 拼 lsof/curl 管道。 + +## list(默认动作) + +用户说 `/port` 或 `/port list` 时: + +```bash +python3 .claude/skills/port/scripts/port.py list +``` + +脚本会: + +1. 用 `lsof` 扫出所有 LISTEN 中的 TCP 端口(当前用户可见的进程) +2. 并发对每个端口探测 `http://localhost:PORT/`(失败再试 https),能响应 HTTP 的才算"有效 web 服务" +3. 抓取页面 `` +4. 输出 markdown 表格:`ID | Title | URL | Port | 进程 | PID` +5. 把 ID→pid/port 的映射写入 `~/.cache/claude-port/state.json`,供后续 kill 使用 + +**把脚本输出的表格原样展示给用户**(markdown 表格里的 URL 在终端可点击),不要改写或省略行。表格下方如提示"另有 N 个非 HTTP 监听端口",一并转告;用户想看全部时加 `--all` 重跑。 + +## kill + +用户说 `/port kill 1`、"把第 1 个杀掉"、"kill 3" 时(数字指上次 list 表格里的 ID 列)。支持一次杀多个,含区间写法: + +```bash +python3 .claude/skills/port/scripts/port.py kill 1 +python3 .claude/skills/port/scripts/port.py kill 2 4 6-9 +``` + +- 脚本 kill 前会**重新扫描核对**该端口的 pid 是否仍与记录一致,不一致会拒绝执行——此时转告用户并重新 list,绝不猜测目标。 +- 默认发 SIGTERM 并等待确认;脚本提示进程仍存活时,询问用户后再用 `kill <ID> -9` 强杀。 +- 用户如果说的是端口号而不是 ID(如 "kill 5173" 且 5173 不像小序号),先对照上次表格找到对应 ID 再执行;拿不准就重新 list 让用户确认。 +- 杀完后把脚本输出(杀掉了哪个 title/url/pid)转告用户。 + +## 注意 + +- 距上次 list 超过几分钟再 kill 时,建议先重新 list 一次刷新编号。 +- 脚本只列当前用户的进程;需要 root 进程时用户需自行 sudo,本 skill 不主动提权。 diff --git a/.claude/skills/port/scripts/port.py b/.claude/skills/port/scripts/port.py new file mode 100644 index 0000000..a175adf --- /dev/null +++ b/.claude/skills/port/scripts/port.py @@ -0,0 +1,206 @@ +#!/usr/bin/env python3 +"""port skill — 列出本机在监听的 web 服务并支持按 ID kill。 + +用法: + port.py list [--all] 扫描监听端口, 探测 HTTP, 抓取 title, 输出 markdown 表格 + port.py kill <ID...> [-9] 按上次 list 的 ID 杀掉对应进程 (kill 前重新核对 pid/端口) + 支持多个 ID 和区间, 如: kill 1 3 5-8 + +状态文件: ~/.cache/claude-port/state.json (list 时写入, kill 时读取) +""" + +import json +import os +import re +import signal +import ssl +import subprocess +import sys +import time +import urllib.request +from concurrent.futures import ThreadPoolExecutor + +STATE_DIR = os.path.expanduser("~/.cache/claude-port") +STATE_FILE = os.path.join(STATE_DIR, "state.json") +HTTP_TIMEOUT = 2.5 +TITLE_READ_LIMIT = 65536 + + +def scan_listeners(): + """返回 [{port, pid, name}], 按端口去重排序。只扫当前用户可见的进程。""" + out = subprocess.run( + ["lsof", "-nP", "-iTCP", "-sTCP:LISTEN"], + capture_output=True, text=True, + ).stdout + seen = {} + for line in out.splitlines()[1:]: + parts = line.split() + if len(parts) < 9: + continue + name, pid, addr = parts[0], parts[1], parts[8] + m = re.search(r":(\d+)$", addr) + if not m: + continue + port = int(m.group(1)) + # 同一端口 IPv4/IPv6 重复行只保留一条 + if port not in seen: + seen[port] = {"port": port, "pid": int(pid), "name": name} + return sorted(seen.values(), key=lambda x: x["port"]) + + +def probe(entry): + """探测端口是否为 web 服务; 是则填入 url 和 title, 否则 url=None。""" + port = entry["port"] + ctx = ssl.create_default_context() + ctx.check_hostname = False + ctx.verify_mode = ssl.CERT_NONE + for scheme in ("http", "https"): + url = f"{scheme}://localhost:{port}/" + req = urllib.request.Request(url, headers={"User-Agent": "port-skill/1.0"}) + try: + with urllib.request.urlopen(req, timeout=HTTP_TIMEOUT, context=ctx) as resp: + body = resp.read(TITLE_READ_LIMIT) + entry["url"] = url + entry["status"] = resp.status + entry["title"] = extract_title(body) or "(无 title)" + return entry + except urllib.error.HTTPError as e: + # 4xx/5xx 也是 HTTP 服务在响应 + entry["url"] = url + entry["status"] = e.code + try: + entry["title"] = extract_title(e.read(TITLE_READ_LIMIT)) or f"(HTTP {e.code})" + except Exception: + entry["title"] = f"(HTTP {e.code})" + return entry + except Exception: + continue + entry["url"] = None + entry["title"] = None + return entry + + +def extract_title(body): + for enc in ("utf-8", "gbk", "latin-1"): + try: + text = body.decode(enc, errors="ignore") + break + except Exception: + continue + m = re.search(r"<title[^>]*>(.*?)", text, re.IGNORECASE | re.DOTALL) + if not m: + return None + title = re.sub(r"\s+", " ", m.group(1)).strip() + return title[:80] if title else None + + +def cmd_list(show_all=False): + listeners = scan_listeners() + if not listeners: + print("没有发现任何处于 LISTEN 状态的 TCP 端口。") + return + with ThreadPoolExecutor(max_workers=16) as ex: + results = list(ex.map(probe, listeners)) + + web = [r for r in results if r["url"]] + other = [r for r in results if not r["url"]] + + state = [] + print("| ID | Title | URL | Port | 进程 | PID |") + print("|---:|---|---|---:|---|---:|") + for i, r in enumerate(web, 1): + title = (r["title"] or "").replace("|", "\\|") + print(f"| {i} | {title} | {r['url']} | {r['port']} | {r['name']} | {r['pid']} |") + state.append({"id": i, "port": r["port"], "pid": r["pid"], + "name": r["name"], "title": r["title"], "url": r["url"]}) + + os.makedirs(STATE_DIR, exist_ok=True) + with open(STATE_FILE, "w") as f: + json.dump({"ts": time.time(), "entries": state}, f, ensure_ascii=False, indent=2) + + if other: + if show_all: + print("\n非 HTTP 的监听端口 (不参与 kill 编号):") + print("| Port | 进程 | PID |") + print("|---:|---|---:|") + for r in other: + print(f"| {r['port']} | {r['name']} | {r['pid']} |") + else: + ports = ", ".join(str(r["port"]) for r in other[:20]) + print(f"\n另有 {len(other)} 个非 HTTP 监听端口未列出 ({ports}),加 --all 查看。") + + +def kill_one(state, target_id, force=False): + entry = next((e for e in state["entries"] if e["id"] == target_id), None) + if entry is None: + print(f"跳过 [{target_id}]: 不在上次 list 的列表里 (共 {len(state['entries'])} 项),请重新 list。") + return + + # kill 前重新核对: 该 pid 是否仍在监听该端口, 防止 pid 被复用误杀 + current = {e["port"]: e for e in scan_listeners()} + live = current.get(entry["port"]) + if live is None or live["pid"] != entry["pid"]: + print( + f"跳过 [{target_id}] 端口 {entry['port']}: 进程已变化 (记录 pid={entry['pid']}, " + f"当前 {'无监听' if live is None else 'pid=' + str(live['pid'])})。" + "为避免误杀未执行,请重新 list。" + ) + return + + sig = signal.SIGKILL if force else signal.SIGTERM + try: + os.kill(entry["pid"], sig) + except OSError as e: + print(f"跳过 [{target_id}] pid {entry['pid']}: kill 失败 ({e})") + return + desc = f"[{entry['id']}] {entry['title']} — {entry['url']} (pid {entry['pid']}, {entry['name']})" + if force: + print(f"已发送 SIGKILL: {desc}") + return + # SIGTERM 后等待确认 + for _ in range(10): + time.sleep(0.3) + try: + os.kill(entry["pid"], 0) + except OSError: + print(f"已终止: {desc}") + return + print(f"已发送 SIGTERM 但进程仍在运行: {desc}\n可用 `kill {entry['id']} -9` 强制终止。") + + +def parse_ids(tokens): + """把 ["1", "3", "5-8"] 解析成 [1, 3, 5, 6, 7, 8]。""" + ids = [] + for t in tokens: + m = re.fullmatch(r"(\d+)-(\d+)", t) + if m: + ids.extend(range(int(m.group(1)), int(m.group(2)) + 1)) + elif t.isdigit(): + ids.append(int(t)) + return sorted(set(ids)) + + +def cmd_kill(ids, force=False): + if not os.path.exists(STATE_FILE): + sys.exit("没有找到上次 list 的记录,请先运行 list。") + with open(STATE_FILE) as f: + state = json.load(f) + for target_id in ids: + kill_one(state, target_id, force=force) + + +def main(): + args = sys.argv[1:] + if not args or args[0] == "list": + cmd_list(show_all="--all" in args) + elif args[0] == "kill": + ids = parse_ids(args[1:]) + if not ids: + sys.exit("用法: port.py kill [-9] (支持区间, 如 kill 1 3 5-8)") + cmd_kill(ids, force="-9" in args or "--force" in args) + else: + sys.exit(__doc__) + + +if __name__ == "__main__": + main()