Python scripts implement timing monitoring ports

import os
import sys
import threading
import time
 
import socket
import subprocess
 
 
def check_port_and_run_script(port, script_path):
    # Create a socket object    sock = (socket.AF_INET, socket.SOCK_STREAM)
 
    # Set timeout    (1)
 
    # Try to connect to the specified port    result = sock.connect_ex(('localhost', port))
 
    # If the connection is successful, the port is being used    if result == 0:
        print(f"port {port} In use。")
    else:
        print(f"port {port} Not used，Execution is underway {script_path} script...")
        # Execute scripts        (['D:/software/Git/bin/', script_path])
        # (['bash', script_path])
 
    # Close socket    ()
 
 
def read_cof_file():
    current_directory = ()
    # Use glob to find all .doc files    if ((current_directory, '')):
        with open((current_directory, ''), 'r', encoding='utf-8') as file:
            # Read file content line by line            for line in file:
                if 'shany-end' in line:
                    print('The timing task has ended')
                    (0)
                elif ':=>' in line:
                    parts = (":=>")
                    check_port_and_run_script(int(parts[0]), parts[1])
 
 
def timer_task():
    print("The timer task has been executed!")
    # check_port_and_run_script(8686, 'D:/WorkSpace/shany/simulator/target/')
    read_cof_file()
    set_timer(20)  
 
 
def set_timer(interval):
    timer = (interval, timer_task)
    ()
 
 
if __name__ == '__main__':
    # Set the timer for the first time    set_timer(20)
 
    # The main thread can continue to execute other tasks, or simply wait for the timer to execute    try:
        while True:
            (1)  # Prevent the main thread from exiting immediately, and other logic can be adjusted or added as needed.    except KeyboardInterrupt:
        print("Program interrupted by user")

8686:=>D:/WorkSpace/shany/simulator/target/

import socket
 
def monitor_port(port_number):
    # Create a socket object    sock = (socket.AF_INET, socket.SOCK_STREAM)
 
    try:
        # Try to connect to the specified port        (('localhost', port_number))
 
        print(f"Port {port_number} is active.")
 
    except  as e:
        print(f"Error occurred while monitoring port {port_number} - {str(e)}")
 
    finally:
        # Close socket        ()
 
# Monitor port 8001monitor_port(8001)

import tkinter as tk
from tkinter import messagebox
import socket
import time
def check_port(host, port):
    sock = (socket.AF_INET, socket.SOCK_STREAM)
    (5)  # Set timeout    result = sock.connect_ex((host, port))
    ()  # Close socket after checking    if result == 0:
        return f"port {port} It's open。"
    else:
        return f"port {port} It's closed。"
def on_check():
    global host, ports, time_interval, port_success_counts, port_failure_counts
    host = entry_host.get()
    ports = entry_port.get().split(",")  # Separate multiple port numbers by commas    try:
        time_interval = int(entry_timeset.get())
        if time_interval <= 0:
            raise ValueError("The time interval must be greater than 0")
        for port in ports:
            port = int(port)
            if not (0 <= port <= 65535):
                raise ValueError(f"port号 {port} Must be in 0 arrive 65535 between")
        # Initialization success and failure count        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:
        ("mistake", 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 = ("%Y-%m-%d %H:%M:%S", ())
        text_result.insert(, f"{timestamp}: {result}\n")
        # Update success and failure count        if "open" in result:
            port_success_counts[port] += 1
        else:
            port_failure_counts[port] += 1
    update_port_counts()
    (time_interval * 1000, start_monitoring)  # Timed monitoringdef update_port_counts():
    text_port_counts.delete(1.0, )  # Clear the text box    for port, success_count in port_success_counts.items():
        failure_count = port_failure_counts[port]
        text_port_counts.insert(, f"port {port} success: {success_count}, fail: {failure_count}\n")
# Create the main windowroot = ()
("Port Monitoring Tool")
("500x400")  # Set the window size# Create tagslabel_host = (root, text="Host:")
label_host.grid(row=0, column=0)
label_port = (root, text="port:")
label_port.grid(row=1, column=0)
label_port2 = (root, text="(0~65535, Separated by commas)")
label_port2.grid(row=1, column=2)
label_timeset = (root, text="time (interval):")
label_timeset.grid(row=2, column=0)
label_timeset2 = (root, text="(unit：Second)")
label_timeset2.grid(row=2, column=2)
# Create an input boxentry_host = (root)
entry_host.grid(row=0, column=1)
entry_port = (root)
entry_port.grid(row=1, column=1)
entry_timeset = (root)
entry_timeset.grid(row=2, column=1)
# Create buttonbutton_check = (root, text="Check the port", command=on_check)
button_check.grid(row=3, columnspan=3)
# Create text box to display the number of successes and failures of portstext_port_counts = (root, height=5, width=50)
text_port_counts.grid(row=5, column=0, columnspan=3, sticky=)
# Create scrollbarscrollbar = (root)
(row=4, column=3, sticky=)
# Create a text box and associate a scrollbartext_result = (root, height=10, width=50, yscrollcommand=)
text_result.grid(row=4, column=0, columnspan=3, sticky=)
# Configure scrollbars(command=text_result.yview)
# Initialize the global counterport_success_counts = {}
port_failure_counts = {}
# Start the main loop()