| 24 | |
| 25 | |
| 26 | def 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.") |