import os
import sys
import threading
import time
 
import socket
import subprocess
 
 
def check_port_and_run_script(port, script_path):
    # 創(chuàng)建一個(gè)socket對(duì)象
    sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
 
    # 設(shè)置超時(shí)時(shí)間
    sock.settimeout(1)
 
    # 嘗試連接到指定的端口
    result = sock.connect_ex(('localhost', port))
 
    # 如果連接成功，端口正在被使用
    if result == 0:
        print(f"端口 {port} 正在被使用。")
    else:
        print(f"端口 {port} 未被使用，正在執(zhí)行 {script_path} 腳本...")
        # 執(zhí)行a.sh腳本
        subprocess.run(['D:/software/Git/bin/bash.exe', script_path])
        # subprocess.run(['bash', script_path])
 
    # 關(guān)閉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:
            # 逐行讀取文件內(nèi)容
            for line in file:
                if 'shany-end' in line:
                    print('定時(shí)任務(wù)已結(jié)束')
                    sys.exit(0)
                elif ':=>' in line:
                    parts = line.split(":=>")
                    check_port_and_run_script(int(parts[0]), parts[1])
 
 
def timer_task():
    print("定時(shí)器任務(wù)執(zhí)行了！")
    # 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__':
    # 首次設(shè)置定時(shí)器
    set_timer(20)
 
    # 主線程可以繼續(xù)執(zhí)行其他任務(wù)，或者簡(jiǎn)單地等待定時(shí)器執(zhí)行
    try:
        while True:
            time.sleep(1)  # 防止主線程立即退出，可以按需調(diào)整或添加其他邏輯
    except KeyboardInterrupt:
        print("程序被用戶中斷")

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

import socket
 
def monitor_port(port_number):
    # 創(chuàng)建一個(gè) socket 對(duì)象
    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:
        # 關(guān)閉 socket
        sock.close()
 
# 監(jiān)控 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)  # 設(shè)置超時(shí)時(shí)間
    result = sock.connect_ex((host, port))
    sock.close()  # 在檢查后關(guān)閉 socket
    if result == 0:
        return f"端口 {port} 是開(kāi)放的。"
    else:
        return f"端口 {port} 是關(guān)閉的。"
def on_check():
    global host, ports, time_interval, port_success_counts, port_failure_counts
    host = entry_host.get()
    ports = entry_port.get().split(",")  # 以逗號(hào)分隔多個(gè)端口號(hào)
    try:
        time_interval = int(entry_timeset.get())
        if time_interval <= 0:
            raise ValueError("時(shí)間間隔必須大于0")
        for port in ports:
            port = int(port)
            if not (0 <= port <= 65535):
                raise ValueError(f"端口號(hào) {port} 必須在 0 到 65535 之間")
        # 初始化成功和失敗計(jì)數(shù)
        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("錯(cuò)誤", 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")
        # 更新成功和失敗計(jì)數(shù)
        if "開(kāi)放" in result:
            port_success_counts[port] += 1
        else:
            port_failure_counts[port] += 1
    update_port_counts()
    root.after(time_interval * 1000, start_monitoring)  # 定時(shí)監(jiān)測(cè)
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")
# 創(chuàng)建主窗口
root = tk.Tk()
root.title("端口監(jiān)測(cè)工具")
root.geometry("500x400")  # 設(shè)置窗口大小
# 創(chuàng)建標(biāo)簽
label_host = tk.Label(root, text="主機(jī):")
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, 用逗號(hào)分隔)")
label_port2.grid(row=1, column=2)
label_timeset = tk.Label(root, text="時(shí)間 (間隔):")
label_timeset.grid(row=2, column=0)
label_timeset2 = tk.Label(root, text="(單位：秒)")
label_timeset2.grid(row=2, column=2)
# 創(chuàng)建輸入框
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)
# 創(chuàng)建按鈕
button_check = tk.Button(root, text="檢查端口", command=on_check)
button_check.grid(row=3, columnspan=3)
# 創(chuàng)建文本框顯示端口成功和失敗次數(shù)
text_port_counts = tk.Text(root, height=5, width=50)
text_port_counts.grid(row=5, column=0, columnspan=3, sticky=tk.NSEW)
# 創(chuàng)建滾動(dòng)條
scrollbar = tk.Scrollbar(root)
scrollbar.grid(row=4, column=3, sticky=tk.NS)
# 創(chuàng)建文本框并關(guān)聯(lián)滾動(dòng)條
text_result = tk.Text(root, height=10, width=50, yscrollcommand=scrollbar.set)
text_result.grid(row=4, column=0, columnspan=3, sticky=tk.NSEW)
# 配置滾動(dòng)條
scrollbar.config(command=text_result.yview)
# 初始化全局計(jì)數(shù)器
port_success_counts = {}
port_failure_counts = {}
# 啟動(dòng)主循環(huán)
root.mainloop()