MCPcopy Create free account
hub / github.com/CJackHwang/AIstudioProxyAPI / find_pids_on_port

Function find_pids_on_port

launcher/utils.py:26–88  ·  view source on GitHub ↗
(port: int)

Source from the content-addressed store, hash-verified

24
25
26def find_pids_on_port(port: int) -> List[int]:
27 pids: List[int] = []
28 system_platform = platform.system()
29 command = ""
30 try:
31 if system_platform == "Linux" or system_platform == "Darwin":
32 command = f"lsof -ti :{port} -sTCP:LISTEN"
33 process = subprocess.Popen(
34 command,
35 shell=True,
36 stdout=subprocess.PIPE,
37 stderr=subprocess.PIPE,
38 text=True,
39 close_fds=True,
40 )
41 stdout, stderr = process.communicate(timeout=5)
42 if process.returncode == 0 and stdout:
43 pids = [int(pid) for pid in stdout.strip().split("\n") if pid.isdigit()]
44 elif process.returncode != 0 and (
45 "command not found" in stderr.lower() or "未找到命令" in stderr
46 ):
47 logger.error("Command 'lsof' not found. Please ensure it is installed.")
48 elif process.returncode not in [0, 1]: # lsof returns 1 when not found
49 logger.warning(
50 f"Failed to execute lsof command (return code {process.returncode}): {stderr.strip()}"
51 )
52 elif system_platform == "Windows":
53 command = f'netstat -ano -p TCP | findstr "LISTENING" | findstr ":{port} "'
54 process = subprocess.Popen(
55 command,
56 shell=True,
57 stdout=subprocess.PIPE,
58 stderr=subprocess.PIPE,
59 text=True,
60 )
61 stdout, stderr = process.communicate(timeout=10)
62 if process.returncode == 0 and stdout:
63 for line in stdout.strip().split("\n"):
64 parts = line.split()
65 if (
66 len(parts) >= 4
67 and parts[0].upper() == "TCP"
68 and f":{port}" in parts[1]
69 ):
70 if parts[-1].isdigit():
71 pids.append(int(parts[-1]))
72 pids = list(set(pids)) # Deduplicate
73 elif process.returncode not in [0, 1]: # findstr returns 1 when not found
74 logger.warning(
75 f"Failed to execute netstat/findstr command (return code {process.returncode}): {stderr.strip()}"
76 )
77 else:
78 logger.warning(
79 f"Unsupported operating system '{system_platform}' for finding processes on port."
80 )
81 except FileNotFoundError:
82 cmd_name = command.split()[0] if command else "required tool"
83 logger.error(f"Command '{cmd_name}' not found.")

Calls 2

errorMethod · 0.80
warningMethod · 0.80