Python脚本实现定时监控端口

import os
import sys
import threading
import time
 
import socket
import subprocess
 
 
def check_port_and_run_script(port, script_path):
    # 创建一个socket对象
    sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
 
    # 设置超时时间
    sock.settimeout(1)
 
    # 尝试连接到指定的端口
    result = sock.connect_ex(('localhost', port))
 
    # 如果连接成功，端口正在被使用
    if result == 0:
        print(f"端口 {port} 正在被使用。")
    else:
        print(f"端口 {port} 未被使用，正在执行 {script_path} 脚本...")
        # 执行a.sh脚本
        subprocess.run(['D:/software/Git/bin/bash.exe', script_path])
        # subprocess.run(['bash', script_path])
 
    # 关闭socket
    sock.close()
 
 
def read_cof_file():
    current_directory = os.getcwd()
    # 使用glob查找所有.doc文件
    if os.path.exists(os.path.join(current_directory, 'shany.txt')):
        with open(os.path.join(current_directory, 'shany.txt'), 'r', encoding='utf-8') as file:
            # 逐行读取文件内容
            for line in file:
                if 'shany-end' in line:
                    print('定时任务已结束')
                    sys.exit(0)
                elif ':=>' in line:
                    parts = line.split(":=>")
                    check_port_and_run_script(int(parts[0]), parts[1])
 
 
def timer_task():
    print("定时器任务执行了！")
    # check_port_and_run_script(8686, 'D:/WorkSpace/shany/simulator/target/start.sh')
    read_cof_file()
    set_timer(20)  
 
 
def set_timer(interval):
    timer = threading.Timer(interval, timer_task)
    timer.start()
 
 
if __name__ == '__main__':
    # 首次设置定时器
    set_timer(20)
 
    # 主线程可以继续执行其他任务，或者简单地等待定时器执行
    try:
        while True:
            time.sleep(1)  # 防止主线程立即退出，可以按需调整或添加其他逻辑
    except KeyboardInterrupt:
        print("程序被用户中断")

8686:=>D:/WorkSpace/shany/simulator/target/start.sh

import socket
 
def monitor_port(port_number):
    # 创建一个 socket 对象
    sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
 
    try:
        # 尝试连接到指定的端口
        sock.connect(('localhost', port_number))
 
        print(f"Port {port_number} is active.")
 
    except socket.error as e:
        print(f"Error occurred while monitoring port {port_number} - {str(e)}")
 
    finally:
        # 关闭 socket
        sock.close()
 
# 监控 8001 端口
monitor_port(8001)

import tkinter as tk
from tkinter import messagebox
import socket
import time
def check_port(host, port):
    sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    sock.settimeout(5)  # 设置超时时间
    result = sock.connect_ex((host, port))
    sock.close()  # 在检查后关闭 socket
    if result == 0:
        return f"端口 {port} 是开放的。"
    else:
        return f"端口 {port} 是关闭的。"
def on_check():
    global host, ports, time_interval, port_success_counts, port_failure_counts
    host = entry_host.get()
    ports = entry_port.get().split(",")  # 以逗号分隔多个端口号
    try:
        time_interval = int(entry_timeset.get())
        if time_interval <= 0:
            raise ValueError("时间间隔必须大于0")
        for port in ports:
            port = int(port)
            if not (0 <= port <= 65535):
                raise ValueError(f"端口号 {port} 必须在 0 到 65535 之间")
        # 初始化成功和失败计数
        port_success_counts = {int(port): 0 for port in ports}
        port_failure_counts = {int(port): 0 for port in ports}
        start_monitoring()
    except ValueError as e:
        messagebox.showerror("错误", str(e))
def start_monitoring():
    global host, ports, time_interval, port_success_counts, port_failure_counts
    for port in ports:
        port = int(port)
        result = check_port(host, port)
        timestamp = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())
        text_result.insert(tk.END, f"{timestamp}: {result}\n")
        # 更新成功和失败计数
        if "开放" in result:
            port_success_counts[port] += 1
        else:
            port_failure_counts[port] += 1
    update_port_counts()
    root.after(time_interval * 1000, start_monitoring)  # 定时监测
def update_port_counts():
    text_port_counts.delete(1.0, tk.END)  # 清空文本框
    for port, success_count in port_success_counts.items():
        failure_count = port_failure_counts[port]
        text_port_counts.insert(tk.END, f"端口 {port} 成功: {success_count}, 失败: {failure_count}\n")
# 创建主窗口
root = tk.Tk()
root.title("端口监测工具")
root.geometry("500x400")  # 设置窗口大小
# 创建标签
label_host = tk.Label(root, text="主机:")
label_host.grid(row=0, column=0)
label_port = tk.Label(root, text="端口:")
label_port.grid(row=1, column=0)
label_port2 = tk.Label(root, text="(0~65535, 用逗号分隔)")
label_port2.grid(row=1, column=2)
label_timeset = tk.Label(root, text="时间 (间隔):")
label_timeset.grid(row=2, column=0)
label_timeset2 = tk.Label(root, text="(单位：秒)")
label_timeset2.grid(row=2, column=2)
# 创建输入框
entry_host = tk.Entry(root)
entry_host.grid(row=0, column=1)
entry_port = tk.Entry(root)
entry_port.grid(row=1, column=1)
entry_timeset = tk.Entry(root)
entry_timeset.grid(row=2, column=1)
# 创建按钮
button_check = tk.Button(root, text="检查端口", command=on_check)
button_check.grid(row=3, columnspan=3)
# 创建文本框显示端口成功和失败次数
text_port_counts = tk.Text(root, height=5, width=50)
text_port_counts.grid(row=5, column=0, columnspan=3, sticky=tk.NSEW)
# 创建滚动条
scrollbar = tk.Scrollbar(root)
scrollbar.grid(row=4, column=3, sticky=tk.NS)
# 创建文本框并关联滚动条
text_result = tk.Text(root, height=10, width=50, yscrollcommand=scrollbar.set)
text_result.grid(row=4, column=0, columnspan=3, sticky=tk.NSEW)
# 配置滚动条
scrollbar.config(command=text_result.yview)
# 初始化全局计数器
port_success_counts = {}
port_failure_counts = {}
# 启动主循环
root.mainloop()