Python实现Windows端口监听程序

pip install tabulate

#!/usr/bin/env python
# -*- coding: utf-8 -*-
 
"""
Windows端口监听程序
显示本机处于监听状态的端口，进程ID和程序名称
"""
 
import subprocess
import re
import os
import sys
from tabulate import tabulate
 
def get_listening_ports():
    """
    获取所有处于监听状态的端口信息
    返回包含端口、PID和程序名称的列表
    """
    try:
        # 使用netstat命令获取所有TCP监听端口
        netstat_output = subprocess.check_output(
            'netstat -ano -p tcp | findstr "LISTENING"',
            shell=True, 
            text=True
        )
        
        # 解析netstat输出
        port_info = []
        for line in netstat_output.splitlines():
            # 清理并分割行
            parts = re.split(r'\s+', line.strip())
            if len(parts) >= 5:
                # 提取本地地址和PID
                local_address = parts[1]
                pid = parts[4]
                
                # 从本地地址中提取端口
                if ':' in local_address:
                    port = local_address.split(':')[-1]
                    
                    # 获取进程名称
                    try:
                        process_info = subprocess.check_output(
                            f'tasklist /fi "PID eq {pid}" /fo csv /nh',
                            shell=True,
                            text=True
                        )
                        
                        # 解析进程信息
                        if process_info and '","' in process_info:
                            process_name = process_info.split('","')[0].strip('"')
                            port_info.append({
                                'port': port,
                                'pid': pid,
                                'program': process_name
                            })
                    except subprocess.SubprocessError:
                        port_info.append({
                            'port': port,
                            'pid': pid,
                            'program': 'Unknown'
                        })
        
        return port_info
    
    except subprocess.SubprocessError as e:
        print(f"获取端口信息时出错: {e}")
        return []
 
def display_port_info(port_info):
    """
    以表格形式显示端口信息
    """
    if not port_info:
        print("未找到监听中的端口")
        return
    
    # 准备表格数据
    table_data = []
    for info in port_info:
        table_data.append([
            info['port'],
            info['pid'],
            info['program']
        ])
    
    # 按端口号排序
    table_data.sort(key=lambda x: int(x[0]) if x[0].isdigit() else float('inf'))
    
    # 显示表格
    headers = ["端口", "进程ID", "程序名称"]
    print(tabulate(table_data, headers=headers, tablefmt="grid"))
 
def main():
    """
    主函数
    """
    print("正在获取本机监听端口信息...\n")
    port_info = get_listening_ports()
    display_port_info(port_info)
 
if __name__ == "__main__":
    try:
        main()
    except KeyboardInterrupt:
        print("\n程序被用户中断")
    except Exception as e:
        print(f"程序执行出错: {e}")
        sys.exit(1)

######################
# 根据端口自动重启脚本~ #
######################
import os
import time

port = input("\n>>>请输入端口: \n")
# C:/Users/syp1293/Desktop/configfile/center.bat
bat_path = input("\n>>>请输入bat文件路径: \n")
time_u = input("\n>>>请输入间隔时间: \n")
while True:
    with os.popen('netstat -aon|findstr "' + port + '"') as res:
        res = res.read().split('\n')
    result = []
    if res[0] == '':
        # 不存在端口占用，启动服务
        os.system('start ' + bat_path)
        time.sleep(120)
        continue
    else:
        for line in res:
            # print(line)
            temp = [i for i in line.split(' ') if i != '']
            if len(temp) > 4:
                #  result.append({'pid': temp[4], 'address': temp[1], 'state': temp[3]})
                print('PID:', temp[4])
    # print('sleeping...')
    time.sleep(int(time_u))

pyinstaller -F osport.py

import struct
import socket
import time
import multiprocessing as mp
SERVER_IP = "192.168.1.33"
PORT_BASE = 30000
import os
def recvall(conn, n):
    # Helper function to recv n bytes or return None if EOF is hit
    data = b''
 
    while len(data) < n:
        packet = conn.recv(n - len(data))
        if not packet:
            #print("hahahah","come here")
            break
        data += packet
        #print(len(data))
    #print("result ======",len(data))
    return data
 
def recv_data(conn):
    # result = result.decode("utf8")
    data = b''
    counts = 0
    while True:
        packet = conn.recv(1024)
        counts += len(packet)
        if not packet:
            break
        data += packet
    print("++++++++++++++++",len(data),counts)
    return data
 
def listen_client(port):
    #server_port = PORT_BASE + port
    # try its best to listen to bind the ports to receive data
    while True:
        try:
            recv_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
            recv_socket.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)
            recv_socket.bind((SERVER_IP,port))
            print(" listening ",SERVER_IP,port)
            recv_socket.listen()
            break
        except Exception as e:
            print("bind port fails ", e, SERVER_IP)
            os.system('fuser -k -n tcp ' + str(port))
            time.sleep(3)
 
    while True:
        try:
            conn, client_addr = recv_socket.accept()
            recv_data(conn)
        except Exception as e:
            print(e)
 
for i in range(6):
    print("================",PORT_BASE + 2 * i)
    mp.Process(target=listen_client, args=(PORT_BASE + 2 * i,)).start()