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()