最新国产好看的视频,伊人天堂AV在线,国产Aaaaaa视频,蜜臀视频在线观看一区,人妻av色图,密臀久久久精品影片,青青视频免费观看毛片,久草在线观看视,国产三级精品色情在线

使用Python實(shí)現(xiàn)IP地址和端口狀態(tài)檢測(cè)與監(jiān)控

 更新時(shí)間:2025年04月30日 15:28:15   作者:創(chuàng)客白澤  
在網(wǎng)絡(luò)運(yùn)維和服務(wù)器管理中,IP地址和端口的可用性監(jiān)控是保障業(yè)務(wù)連續(xù)性的基礎(chǔ)需求,本文將帶你用Python從零打造一個(gè)高可用IP監(jiān)控系統(tǒng),感興趣的小伙伴可以了解下

概述:為什么需要IP監(jiān)控系統(tǒng)

在網(wǎng)絡(luò)運(yùn)維和服務(wù)器管理中,IP地址和端口的可用性監(jiān)控是保障業(yè)務(wù)連續(xù)性的基礎(chǔ)需求。傳統(tǒng)的人工巡檢方式效率低下,而商業(yè)監(jiān)控工具又往往價(jià)格昂貴。本文將帶你用Python從零打造一個(gè)高可用IP監(jiān)控系統(tǒng),具備以下核心功能:

  • 多目標(biāo)監(jiān)控:同時(shí)監(jiān)測(cè)多個(gè)IP+端口組合狀態(tài)
  • 智能告警:異常狀態(tài)自動(dòng)觸發(fā)郵件通知(支持多收件人)
  • 可視化界面:基于Tkinter的現(xiàn)代化UI,操作直觀
  • 配置持久化:所有設(shè)置自動(dòng)保存,重啟不丟失
  • 斷線重試機(jī)制:避免網(wǎng)絡(luò)抖動(dòng)導(dǎo)致的誤報(bào)

技術(shù)棧:Python 3 + Tkinter + smtplib + socket + 多線程

使用步驟說(shuō)明

1. 環(huán)境準(zhǔn)備

# 所需庫(kù)(Python內(nèi)置,無(wú)需額外安裝)
import tkinter
import threading
import smtplib
import socket

2. 系統(tǒng)部署

下載完整代碼(文末提供)

配置config.json(首次運(yùn)行會(huì)自動(dòng)生成)

運(yùn)行主程序:

 python ip_monitor.py

3. 核心功能配置

目標(biāo)配置

通過(guò)表格形式管理監(jiān)控目標(biāo)

復(fù)選框控制是否啟用監(jiān)測(cè)

支持雙擊編輯現(xiàn)有條目

郵件設(shè)置

服務(wù)商SMTP地址端口授權(quán)碼獲取方式
網(wǎng)易163smtp.163.com465/994郵箱設(shè)置→POP3/SMTP服務(wù)
QQ郵箱smtp.qq.com465設(shè)置→賬戶→POP3服務(wù)
Gmailsmtp.gmail.com587Google賬號(hào)→應(yīng)用密碼

# 配置示例(支持SSL/TLS)
SMTP服務(wù)器: smtp.163.com:465
郵箱賬戶: yourname@163.com
授權(quán)碼: xxxxxx  # 需在郵箱設(shè)置中獲取
接收郵箱: admin@company.com,backup@company.com

注意事項(xiàng):

必須開(kāi)啟SMTP服務(wù)

部分郵箱需要使用授權(quán)碼而非密碼

Gmail需開(kāi)啟"低安全性應(yīng)用訪問(wèn)"

監(jiān)控參數(shù)

檢測(cè)模式說(shuō)明

純IP檢測(cè)模式

  • 留空端口字段
  • 使用ICMP協(xié)議Ping檢測(cè)
  • 適用場(chǎng)景:網(wǎng)絡(luò)設(shè)備監(jiān)控

組合檢測(cè)模式

# 同時(shí)驗(yàn)證IP可達(dá)性和端口開(kāi)放狀態(tài)
if ping_success and port_open:
    return ONLINE
參數(shù)名默認(rèn)值說(shuō)明
監(jiān)測(cè)間隔10秒兩次檢測(cè)的時(shí)間間隔
超時(shí)時(shí)間2秒判定離線的超時(shí)閾值
重試次數(shù)3次連續(xù)失敗次數(shù)觸發(fā)告警
重試間隔5秒失敗后的快速重試間隔

系統(tǒng)效果展示

實(shí)時(shí)監(jiān)控面板

郵件告警示例

主題:[告警] IP狀態(tài)變更: 192.168.1.1:80 - 離線

IP地址: 192.168.1.1
端口: 80
備注: 主數(shù)據(jù)庫(kù)服務(wù)器
狀態(tài)變更為: 離線
檢測(cè)時(shí)間: 2023-08-20 14:30:45

日志記錄

[2023-08-20 14:30:45] 檢測(cè)到 192.168.1.1:80 狀態(tài)變更為離線
[2023-08-20 14:31:00] 已發(fā)送告警郵件給3個(gè)收件人
[2023-08-20 15:00:00] 已發(fā)送每日狀態(tài)報(bào)告

核心源碼解析

1. 狀態(tài)檢測(cè)引擎

def check_target(self, ip, port=None, timeout=2):
    """智能檢測(cè)IP/端口狀態(tài)"""
    try:
        if port:  # 端口檢測(cè)模式
            with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
                s.settimeout(timeout)
                s.connect((ip, port))
                return IPStatus.ONLINE
        else:     # 純Ping模式
            # 使用ICMP協(xié)議實(shí)現(xiàn)Ping
            sock = socket.socket(socket.AF_INET, 
                               socket.SOCK_RAW, 
                               socket.IPPROTO_ICMP)
            sock.settimeout(timeout)
            sock.connect((ip, 1))
            return IPStatus.ONLINE
    except socket.timeout:
        return IPStatus.OFFLINE
    except Exception:
        return IPStatus.UNKNOWN

2. 多線程監(jiān)控架構(gòu)

class MonitorThread(threading.Thread):
    def __init__(self, app):
        super().__init__(daemon=True)
        self.app = app
        self.running = True
        
    def run(self):
        while self.running:
            # 1. 執(zhí)行所有目標(biāo)的檢測(cè)
            # 2. 觸發(fā)狀態(tài)變更通知
            # 3. 智能休眠控制CPU占用
            time.sleep(self.calculate_sleep_time())
            
    def stop(self):
        self.running = False

3. 配置持久化實(shí)現(xiàn)

def save_config(self):
    """JSON格式保存所有配置"""
    config = {
        'targets': [
            self.tree.item(item, 'values')
            for item in self.tree.get_children()
        ],
        'email_settings': {
            'smtp': self.smtp_entry.get(),
            'user': self.user_entry.get(),
            'pass': self.pass_entry.get(),
            'receivers': self.receiver_entry.get()
        }
    }
    with open('config.json', 'w') as f:
        json.dump(config, f, indent=2)

高級(jí)功能擴(kuò)展建議

1. 微信/釘釘機(jī)器人告警

# 示例:企業(yè)微信機(jī)器人API
import requests
def send_wechat_alert(message):
    webhook = "https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=xxx"
    requests.post(webhook, json={"text": {"content": message}})

2. 數(shù)據(jù)庫(kù)存儲(chǔ)歷史記錄

# 使用SQLite記錄狀態(tài)變化
import sqlite3
conn = sqlite3.connect('monitor.db')
c = conn.cursor()
c.execute('''CREATE TABLE IF NOT EXISTS status_log
             (ip TEXT, port INT, status TEXT, check_time TIMESTAMP)''')

3. 可視化圖表展示

# 使用Matplotlib繪制可用率曲線
plt.plot(dates, availability_rates)
plt.title('月度服務(wù)可用率')
plt.ylabel('百分比(%)')
plt.savefig('report.png')

總結(jié)與資源下載

項(xiàng)目亮點(diǎn)

  • 工業(yè)級(jí)可靠性:斷線重試+智能休眠機(jī)制
  • 開(kāi)箱即用:無(wú)需復(fù)雜配置,5分鐘快速部署
  • 高度可擴(kuò)展:代碼結(jié)構(gòu)清晰,便于二次開(kāi)發(fā)

完整源碼下載

import tkinter as tk
from tkinter import ttk, messagebox, scrolledtext
import threading
import time
import smtplib
import ipaddress
import re
import json
import socket
from email.message import EmailMessage
from enum import Enum


class IPStatus(Enum):
    ONLINE = 1
    OFFLINE = 2
    UNKNOWN = 3


class IPMonitorApp(tk.Tk):
    def __init__(self):
        super().__init__()
        self.title("IP狀態(tài)監(jiān)測(cè)系統(tǒng)")
        self.geometry("1300x850")
        
        try:
            self.iconbitmap('monitor.ico')
        except:
            pass

        self.style = ttk.Style()
        self.configure_style()
        
        self.stop_event = threading.Event()
        self.monitor_thread = None
        self.ip_status = {}
        self.next_report_time = self.calculate_next_report_time()
        
        self.create_widgets()
        self.load_config()

    def configure_style(self):
        self.style.theme_create('ipmonitor', parent='alt', settings={
            'TFrame': {'configure': {'background': '#f5f5f5'}},
            'TLabelFrame': {
                'configure': {
                    'background': '#f5f5f5',
                    'foreground': '#1e3d59',
                    'font': ('微軟雅黑', 10, 'bold')
                }
            },
            'TLabel': {
                'configure': {
                    'background': '#f5f5f5',
                    'foreground': '#1e3d59',
                    'font': ('微軟雅黑', 10)
                }
            },
            'TButton': {
                'configure': {
                    'background': '#1e3d59',
                    'foreground': 'white',
                    'font': ('微軟雅黑', 10),
                    'padding': 5
                },
                'map': {
                    'background': [('active', '#3a6ea5')],
                    'foreground': [('disabled', '#888888')]
                }
            },
            'TCheckbutton': {
                'configure': {
                    'background': '#f5f5f5',
                    'font': ('微軟雅黑', 10)
                }
            }
        })
        self.style.theme_use('ipmonitor')

    def create_widgets(self):
        # 主容器
        main_frame = ttk.Frame(self, padding=10)
        main_frame.grid(row=0, column=0, sticky="nsew")
        self.grid_columnconfigure(0, weight=1)
        self.grid_rowconfigure(0, weight=1)
        
        # 配置區(qū)域
        config_frame = ttk.Frame(main_frame)
        config_frame.grid(row=0, column=0, sticky="nsew")
        
        # IP配置區(qū)域
        ip_frame = ttk.LabelFrame(config_frame, text="監(jiān)測(cè)目標(biāo)配置", padding=10)
        ip_frame.grid(row=0, column=0, padx=5, pady=5, sticky="nsew")
        ip_frame.columnconfigure(0, weight=1)
        
        # IP列表Treeview
        self.ip_tree = ttk.Treeview(
            ip_frame, 
            columns=("check", "ip", "port", "remark"), 
            show="headings",
            height=8,
            selectmode="browse"
        )
        self.ip_tree.grid(row=0, column=0, sticky="nsew")
        
        # 配置列
        self.ip_tree.heading("check", text="檢測(cè)")
        self.ip_tree.heading("ip", text="IP地址")
        self.ip_tree.heading("port", text="端口")
        self.ip_tree.heading("remark", text="備注")
        
        self.ip_tree.column("check", width=50, anchor="center")
        self.ip_tree.column("ip", width=150, anchor="w")
        self.ip_tree.column("port", width=80, anchor="center")
        self.ip_tree.column("remark", width=250, anchor="w")
        
        # 添加復(fù)選框
        self.ip_tree.tag_configure("checked", background="#e6f7ff")
        self.ip_tree.tag_configure("unchecked", background="#f5f5f5")
        
        # 編輯區(qū)域
        edit_frame = ttk.Frame(ip_frame)
        edit_frame.grid(row=1, column=0, sticky="ew", pady=(5, 0))
        
        ttk.Label(edit_frame, text="IP:").grid(row=0, column=0, sticky="e")
        self.ip_entry = ttk.Entry(edit_frame, width=18)
        self.ip_entry.grid(row=0, column=1, padx=2, sticky="w")
        
        ttk.Label(edit_frame, text="端口:").grid(row=0, column=2, sticky="e")
        self.port_entry = ttk.Entry(edit_frame, width=8)
        self.port_entry.grid(row=0, column=3, padx=2, sticky="w")
        
        ttk.Label(edit_frame, text="備注:").grid(row=0, column=4, sticky="e")
        self.remark_entry = ttk.Entry(edit_frame, width=20)
        self.remark_entry.grid(row=0, column=5, padx=2, sticky="w")
        
        # 操作按鈕
        btn_frame = ttk.Frame(edit_frame)
        btn_frame.grid(row=0, column=6, padx=5)
        
        ttk.Button(btn_frame, text="添加", command=self.add_ip).grid(row=0, column=0, padx=2)
        ttk.Button(btn_frame, text="更新", command=self.update_ip).grid(row=0, column=1, padx=2)
        ttk.Button(btn_frame, text="刪除", command=self.remove_ip).grid(row=0, column=2, padx=2)
        
        # 郵件配置區(qū)域
        mail_frame = ttk.LabelFrame(config_frame, text="郵件通知配置", padding=10)
        mail_frame.grid(row=0, column=1, padx=5, pady=5, sticky="nsew")
        
        # 郵件配置項(xiàng)
        mail_config = [
            ("SMTP服務(wù)器:端口", "smtp_server", "smtp.163.com:465"),
            ("郵箱賬戶", "email_user", "yourname@163.com"),
            ("郵箱授權(quán)碼", "email_pass", ""),
            ("接收郵箱", "email_receiver", "多個(gè)郵箱用逗號(hào)分隔")
        ]
        
        for i, (label, attr, ph) in enumerate(mail_config):
            ttk.Label(mail_frame, text=label).grid(row=i, column=0, sticky="e", pady=3)
            entry = ttk.Entry(mail_frame, width=25)
            entry.grid(row=i, column=1, padx=5, pady=3, sticky="ew")
            setattr(self, attr+"_entry", entry)
            ttk.Label(mail_frame, text=ph, foreground="#888888").grid(row=i, column=2, sticky="w", padx=5)
        
        # 監(jiān)控參數(shù)
        param_frame = ttk.LabelFrame(config_frame, text="監(jiān)控參數(shù)", padding=10)
        param_frame.grid(row=1, column=0, columnspan=2, sticky="ew", pady=(5, 0))
        
        params = [
            ("監(jiān)測(cè)間隔(秒):", "interval", "10"),
            ("超時(shí)(秒):", "timeout", "2"),
            ("重試次數(shù):", "retry", "3"),
            ("重試間隔(秒):", "retry_interval", "5")
        ]
        
        for i, (label, attr, default) in enumerate(params):
            ttk.Label(param_frame, text=label).grid(row=0, column=i*2, sticky="e")
            entry = ttk.Entry(param_frame, width=8)
            entry.insert(0, default)
            entry.grid(row=0, column=i*2+1, padx=5, sticky="w")
            setattr(self, attr+"_entry", entry)
        
        # 控制按鈕
        ctrl_frame = ttk.Frame(config_frame)
        ctrl_frame.grid(row=2, column=0, columnspan=2, pady=(10, 0), sticky="e")
        
        ttk.Button(ctrl_frame, text="開(kāi)始監(jiān)測(cè)", command=self.start_monitoring).grid(row=0, column=0, padx=5)
        ttk.Button(ctrl_frame, text="停止監(jiān)測(cè)", command=self.stop_monitoring, state=tk.DISABLED).grid(row=0, column=1, padx=5)
        self.start_btn = ctrl_frame.grid_slaves(row=0, column=0)[0]
        self.stop_btn = ctrl_frame.grid_slaves(row=0, column=1)[0]
        
        ttk.Button(ctrl_frame, text="保存配置", command=self.save_config).grid(row=0, column=2, padx=5)
        ttk.Button(ctrl_frame, text="清空日志", command=self.clear_logs).grid(row=0, column=3, padx=5)
        
        # 狀態(tài)監(jiān)控區(qū)域
        status_frame = ttk.LabelFrame(main_frame, text="狀態(tài)監(jiān)控", padding=10)
        status_frame.grid(row=1, column=0, sticky="nsew", pady=(5, 0))
        
        self.status_tree = ttk.Treeview(
            status_frame, 
            columns=("ip", "port", "remark", "status", "last_check"), 
            show="headings",
            height=8
        )
        self.status_tree.grid(row=0, column=0, sticky="nsew")
        
        # 狀態(tài)列配置
        for col, width in [("ip", 150), ("port", 80), ("remark", 200), ("status", 100), ("last_check", 180)]:
            self.status_tree.column(col, width=width, anchor="center")
            self.status_tree.heading(col, text=col)
        
        # 日志區(qū)域
        log_frame = ttk.LabelFrame(main_frame, text="系統(tǒng)日志", padding=10)
        log_frame.grid(row=2, column=0, sticky="nsew", pady=(5, 0))
        
        self.log_text = scrolledtext.ScrolledText(
            log_frame, 
            wrap=tk.WORD, 
            font=('微軟雅黑', 9), 
            bg='white', 
            fg='#333333',
            height=10
        )
        self.log_text.pack(fill=tk.BOTH, expand=True)
        
        # 配置權(quán)重
        main_frame.columnconfigure(0, weight=1)
        main_frame.rowconfigure(1, weight=1)
        main_frame.rowconfigure(2, weight=1)
        
        config_frame.columnconfigure(1, weight=1)
        
        ip_frame.rowconfigure(0, weight=1)
        ip_frame.columnconfigure(0, weight=1)
        
        status_frame.rowconfigure(0, weight=1)
        status_frame.columnconfigure(0, weight=1)
        
        # 綁定事件
        self.ip_tree.bind("<Button-1>", self.on_tree_click)
        self.ip_tree.bind("<Double-1>", self.on_tree_double_click)

    def on_tree_click(self, event):
        """處理樹(shù)狀圖點(diǎn)擊事件"""
        region = self.ip_tree.identify("region", event.x, event.y)
        if region == "cell":
            column = self.ip_tree.identify_column(event.x)
            item = self.ip_tree.identify_row(event.y)
            if column == "#1":  # 復(fù)選框列
                current_val = self.ip_tree.item(item, "values")[0]
                new_val = "?" if current_val == "" else ""
                values = list(self.ip_tree.item(item, "values"))
                values[0] = new_val
                self.ip_tree.item(item, values=values, tags=("checked" if new_val else "unchecked"))

    def on_tree_double_click(self, event):
        """處理樹(shù)狀圖雙擊事件"""
        item = self.ip_tree.selection()
        if item:
            values = self.ip_tree.item(item, "values")
            self.ip_entry.delete(0, tk.END)
            self.ip_entry.insert(0, values[1])
            self.port_entry.delete(0, tk.END)
            self.port_entry.insert(0, values[2])
            self.remark_entry.delete(0, tk.END)
            self.remark_entry.insert(0, values[3])

    def add_ip(self):
        """添加IP到列表"""
        ip = self.ip_entry.get().strip()
        port = self.port_entry.get().strip()
        remark = self.remark_entry.get().strip()
        
        if not ip:
            messagebox.showwarning("錯(cuò)誤", "IP地址不能為空")
            return
            
        if not self.is_valid_ip(ip):
            messagebox.showwarning("錯(cuò)誤", "無(wú)效的IP地址格式")
            return
            
        if port:
            try:
                port = int(port)
                if not (1 <= port <= 65535):
                    messagebox.showwarning("錯(cuò)誤", "端口必須在1-65535之間")
                    return
            except ValueError:
                messagebox.showwarning("錯(cuò)誤", "端口必須是數(shù)字")
                return
        
        self.ip_tree.insert("", tk.END, values=("?", ip, port, remark), tags=("checked",))
        
        # 清空輸入框
        self.ip_entry.delete(0, tk.END)
        self.port_entry.delete(0, tk.END)
        self.remark_entry.delete(0, tk.END)

    def update_ip(self):
        """更新選中的IP"""
        item = self.ip_tree.selection()
        if not item:
            messagebox.showwarning("錯(cuò)誤", "請(qǐng)先選擇要更新的項(xiàng)")
            return
            
        ip = self.ip_entry.get().strip()
        port = self.port_entry.get().strip()
        remark = self.remark_entry.get().strip()
        
        if not ip:
            messagebox.showwarning("錯(cuò)誤", "IP地址不能為空")
            return
            
        if not self.is_valid_ip(ip):
            messagebox.showwarning("錯(cuò)誤", "無(wú)效的IP地址格式")
            return
            
        if port:
            try:
                port = int(port)
                if not (1 <= port <= 65535):
                    messagebox.showwarning("錯(cuò)誤", "端口必須在1-65535之間")
                    return
            except ValueError:
                messagebox.showwarning("錯(cuò)誤", "端口必須是數(shù)字")
                return
        
        # 保留原來(lái)的復(fù)選框狀態(tài)
        current_check = self.ip_tree.item(item, "values")[0]
        self.ip_tree.item(item, values=(current_check, ip, port, remark))

    def remove_ip(self):
        """刪除選中的IP"""
        items = self.ip_tree.selection()
        if not items:
            messagebox.showwarning("錯(cuò)誤", "請(qǐng)先選擇要?jiǎng)h除的項(xiàng)")
            return
            
        for item in items:
            self.ip_tree.delete(item)

    def is_valid_ip(self, ip):
        """驗(yàn)證IP地址格式"""
        try:
            ipaddress.ip_address(ip)
            return True
        except ValueError:
            return False

    def start_monitoring(self):
        """開(kāi)始監(jiān)控"""
        if not self.validate_inputs():
            return
            
        # 獲取要監(jiān)控的目標(biāo)
        targets = []
        for item in self.ip_tree.get_children():
            check, ip, port, remark = self.ip_tree.item(item, "values")
            if check == "?":  # 只監(jiān)控選中的項(xiàng)
                targets.append((ip, int(port) if port else None, remark))
        
        if not targets:
            messagebox.showwarning("錯(cuò)誤", "沒(méi)有選中要監(jiān)控的目標(biāo)")
            return
            
        # 初始化狀態(tài)字典
        self.ip_status = {}
        for ip, port, remark in targets:
            self.ip_status[(ip, port)] = {
                "status": IPStatus.UNKNOWN,
                "last_notified": None,
                "remark": remark,
                "retries": 0,
                "next_retry_time": 0,
                "last_check": "從未檢測(cè)"
            }
        
        # 更新?tīng)顟B(tài)顯示
        self.update_status_display()
        
        # 更新按鈕狀態(tài)
        self.start_btn.config(state=tk.DISABLED)
        self.stop_btn.config(state=tk.NORMAL)
        self.stop_event.clear()
        
        # 獲取監(jiān)控參數(shù)
        interval = int(self.interval_entry.get())
        timeout = int(self.timeout_entry.get())
        max_retries = int(self.retry_entry.get())
        retry_interval = int(self.retry_interval_entry.get())
        
        # 創(chuàng)建監(jiān)控線程
        self.monitor_thread = threading.Thread(
            target=self.monitor_targets,
            args=(interval, timeout, max_retries, retry_interval),
            daemon=True
        )
        self.monitor_thread.start()
        
        self.log_message("監(jiān)控已啟動(dòng)")

    def stop_monitoring(self):
        """停止監(jiān)控"""
        self.stop_event.set()
        self.start_btn.config(state=tk.NORMAL)
        self.stop_btn.config(state=tk.DISABLED)
        self.log_message("監(jiān)控已停止")

    def monitor_targets(self, interval, timeout, max_retries, retry_interval):
        """監(jiān)控主循環(huán)"""
        while not self.stop_event.is_set():
            current_time = time.time()

            # 每日?qǐng)?bào)告檢查
            if current_time >= self.next_report_time:
                self.send_daily_report()
                self.next_report_time = self.calculate_next_report_time()

            # 檢查所有目標(biāo)
            for (ip, port), data in list(self.ip_status.items()):
                if self.stop_event.is_set():
                    break

                if current_time < data['next_retry_time']:
                    continue

                # 執(zhí)行檢測(cè)
                if port:  # 如果有端口則檢測(cè)端口
                    current_status = self.check_port(ip, port, timeout)
                else:  # 否則只ping IP
                    current_status = self.ping_ip(ip, timeout)
                
                previous_status = data["status"]
                data["last_check"] = time.strftime("%Y-%m-%d %H:%M:%S")
                
                # 處理狀態(tài)變化
                if current_status == IPStatus.OFFLINE:
                    if data['retries'] < max_retries:
                        data['retries'] += 1
                        data['next_retry_time'] = current_time + retry_interval
                        self.log_message(f"{ip}:{port or '無(wú)端口'} 檢測(cè)失敗,正在進(jìn)行第 {data['retries']} 次重試...")
                    else:
                        if previous_status != current_status:
                            self.send_alert(ip, port, data['remark'], current_status)
                        data["status"] = current_status
                        data['retries'] = 0
                        data['next_retry_time'] = current_time + interval
                else:
                    if previous_status != current_status:
                        self.send_alert(ip, port, data['remark'], current_status)
                    data["status"] = current_status
                    data['retries'] = 0
                    data['next_retry_time'] = current_time + interval
                
                # 更新UI
                self.after(0, self.update_status_display)

            # 計(jì)算睡眠時(shí)間
            next_checks = [data['next_retry_time'] for data in self.ip_status.values()]
            next_check = min(next_checks) if next_checks else current_time + interval
            sleep_time = max(min(next_check - time.time(), interval), 0.1)
            time.sleep(sleep_time)

    def ping_ip(self, ip, timeout):
        """Ping IP地址"""
        try:
            # 使用socket創(chuàng)建ICMP ping
            sock = socket.socket(socket.AF_INET, socket.SOCK_RAW, socket.IPPROTO_ICMP)
            sock.settimeout(timeout)
            sock.connect((ip, 1))  # 端口號(hào)不重要
            return IPStatus.ONLINE
        except socket.timeout:
            return IPStatus.OFFLINE
        except Exception:
            return IPStatus.UNKNOWN
        finally:
            try:
                sock.close()
            except:
                pass

    def check_port(self, ip, port, timeout):
        """檢查端口"""
        try:
            with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
                s.settimeout(timeout)
                s.connect((ip, port))
                return IPStatus.ONLINE
        except socket.timeout:
            return IPStatus.OFFLINE
        except ConnectionRefusedError:
            return IPStatus.OFFLINE
        except Exception as e:
            self.log_message(f"檢測(cè)異常: {str(e)}")
            return IPStatus.UNKNOWN

    def update_status_display(self):
        """更新?tīng)顟B(tài)顯示"""
        # 清空現(xiàn)有顯示
        for item in self.status_tree.get_children():
            self.status_tree.delete(item)
        
        # 添加新?tīng)顟B(tài)
        for (ip, port), data in self.ip_status.items():
            status = data["status"]
            status_text = "在線" if status == IPStatus.ONLINE else "離線" if status == IPStatus.OFFLINE else "未知"
            status_color = "green" if status == IPStatus.ONLINE else "red" if status == IPStatus.OFFLINE else "orange"
            
            item = self.status_tree.insert("", tk.END, values=(
                ip, 
                port if port else "無(wú)", 
                data["remark"], 
                status_text, 
                data["last_check"]
            ))
            self.status_tree.tag_configure(status_color, foreground=status_color)
            self.status_tree.item(item, tags=(status_color,))

    def send_alert(self, ip, port, remark, status):
        """發(fā)送狀態(tài)變更通知"""
        status_text = "在線" if status == IPStatus.ONLINE else "離線"
        subject = f"IP狀態(tài)變更: {remark or ip}:{port if port else '無(wú)端口'} - {status_text}"
        
        content = "\n".join([
            f"IP地址: {ip}",
            f"端口: {port if port else '無(wú)'}",
            f"備注: {remark or '無(wú)備注信息'}",
            f"狀態(tài)變更為: {status_text}",
            f"檢測(cè)時(shí)間: {time.strftime('%Y-%m-%d %H:%M:%S')}"
        ])
        
        try:
            self.send_email(subject, content)
            self.log_message(f"已發(fā)送狀態(tài)變更通知: {ip}:{port if port else '無(wú)'} -> {status_text}")
        except Exception as e:
            self.log_message(f"郵件發(fā)送失敗: {str(e)}")

    def send_email(self, subject, content):
        """發(fā)送郵件"""
        smtp_server = self.smtp_server_entry.get().strip()
        user = self.email_user_entry.get().strip()
        password = self.email_pass_entry.get().strip()
        receiver_str = self.email_receiver_entry.get().strip()
        
        if not all([smtp_server, user, password, receiver_str]):
            raise Exception("郵件配置不完整")
        
        receivers = [addr.strip() for addr in receiver_str.split(',') if addr.strip()]
        
        msg = EmailMessage()
        msg['Subject'] = subject
        msg['From'] = user
        msg['To'] = receivers
        msg.set_content(content)
        
        server, port = smtp_server.split(":")
        port = int(port)
        
        try:
            if port == 465:
                with smtplib.SMTP_SSL(server, port) as smtp:
                    smtp.login(user, password)
                    smtp.send_message(msg)
            else:
                with smtplib.SMTP(server, port) as smtp:
                    smtp.starttls()
                    smtp.login(user, password)
                    smtp.send_message(msg)
        except Exception as e:
            raise Exception(f"SMTP錯(cuò)誤: {str(e)}")

    def send_daily_report(self):
        """發(fā)送每日?qǐng)?bào)告"""
        report_lines = []
        for (ip, port), data in self.ip_status.items():
            status = data["status"]
            status_text = "在線" if status == IPStatus.ONLINE else "離線" if status == IPStatus.OFFLINE else "未知"
            report_lines.append(f"IP: {ip}:{port if port else '無(wú)'} 備注: {data['remark']} 狀態(tài): {status_text}")
        
        current_time = time.strftime("%Y-%m-%d %H:%M:%S")
        content = "每日IP狀態(tài)報(bào)告\n\n" + "\n".join(report_lines) + f"\n\n報(bào)告時(shí)間: {current_time}"
        subject = "每日IP狀態(tài)報(bào)告"
        
        try:
            self.send_email(subject, content)
            self.log_message("已發(fā)送每日狀態(tài)報(bào)告")
        except Exception as e:
            self.log_message(f"發(fā)送每日?qǐng)?bào)告失敗: {str(e)}")

    def calculate_next_report_time(self):
        """計(jì)算下次報(bào)告時(shí)間"""
        now = time.localtime()
        today_10am = time.mktime((now.tm_year, now.tm_mon, now.tm_mday, 10, 0, 0, 0, 0, -1))
        current_time = time.time()
        
        return today_10am + 86400 if current_time >= today_10am else today_10am

    def log_message(self, message):
        """記錄日志"""
        timestamp = time.strftime("%Y-%m-%d %H:%M:%S")
        log_line = f"[{timestamp}] {message}\n"
        
        self.log_text.insert(tk.END, log_line)
        self.log_text.see(tk.END)
        
        with open("monitor.log", "a", encoding="utf-8") as f:
            f.write(log_line)

    def clear_logs(self):
        """清空日志"""
        self.log_text.delete(1.0, tk.END)

    def validate_inputs(self):
        """驗(yàn)證輸入"""
        # 檢查郵件配置
        smtp_server = self.smtp_server_entry.get().strip()
        user = self.email_user_entry.get().strip()
        password = self.email_pass_entry.get().strip()
        receiver_str = self.email_receiver_entry.get().strip()
        
        if not smtp_server:
            messagebox.showwarning("錯(cuò)誤", "SMTP服務(wù)器不能為空")
            return False
            
        if ":" not in smtp_server:
            messagebox.showwarning("錯(cuò)誤", "SMTP服務(wù)器格式應(yīng)為 host:port")
            return False
            
        if not user:
            messagebox.showwarning("錯(cuò)誤", "郵箱賬戶不能為空")
            return False
            
        if not password:
            messagebox.showwarning("錯(cuò)誤", "郵箱授權(quán)碼不能為空")
            return False
            
        if not receiver_str:
            messagebox.showwarning("錯(cuò)誤", "接收郵箱不能為空")
            return False
            
        # 驗(yàn)證監(jiān)控參數(shù)
        try:
            interval = int(self.interval_entry.get())
            timeout = int(self.timeout_entry.get())
            retries = int(self.retry_entry.get())
            retry_interval = int(self.retry_interval_entry.get())
            
            if interval < 5:
                messagebox.showwarning("錯(cuò)誤", "監(jiān)測(cè)間隔不能小于5秒")
                return False
                
            if timeout < 1:
                messagebox.showwarning("錯(cuò)誤", "超時(shí)時(shí)間不能小于1秒")
                return False
                
            if retries < 0:
                messagebox.showwarning("錯(cuò)誤", "重試次數(shù)不能為負(fù)數(shù)")
                return False
                
            if retry_interval < 1:
                messagebox.showwarning("錯(cuò)誤", "重試間隔不能小于1秒")
                return False
                
        except ValueError:
            messagebox.showwarning("錯(cuò)誤", "請(qǐng)輸入有效的數(shù)字")
            return False
            
        return True

    def save_config(self):
        """保存配置"""
        config = {
            "targets": [
                self.ip_tree.item(item, "values")
                for item in self.ip_tree.get_children()
            ],
            "smtp_server": self.smtp_server_entry.get(),
            "email_user": self.email_user_entry.get(),
            "email_pass": self.email_pass_entry.get(),
            "email_receiver": self.email_receiver_entry.get(),
            "interval": self.interval_entry.get(),
            "timeout": self.timeout_entry.get(),
            "retry": self.retry_entry.get(),
            "retry_interval": self.retry_interval_entry.get()
        }
        
        try:
            with open("config.json", "w", encoding="utf-8") as f:
                json.dump(config, f, indent=2)
            self.log_message("配置已保存")
            messagebox.showinfo("成功", "配置已保存到config.json")
        except Exception as e:
            self.log_message(f"保存配置失敗: {str(e)}")
            messagebox.showerror("錯(cuò)誤", f"保存配置失敗: {str(e)}")

    def load_config(self):
        """加載配置"""
        try:
            with open("config.json", encoding="utf-8") as f:
                config = json.load(f)
            
            # 加載目標(biāo)
            for item in self.ip_tree.get_children():
                self.ip_tree.delete(item)
                
            for values in config.get("targets", []):
                self.ip_tree.insert("", tk.END, values=values, tags=("checked" if values[0] == "?" else "unchecked"))
            
            # 加載郵件配置
            self.smtp_server_entry.delete(0, tk.END)
            self.smtp_server_entry.insert(0, config.get("smtp_server", ""))
            
            self.email_user_entry.delete(0, tk.END)
            self.email_user_entry.insert(0, config.get("email_user", ""))
            
            self.email_pass_entry.delete(0, tk.END)
            self.email_pass_entry.insert(0, config.get("email_pass", ""))
            
            self.email_receiver_entry.delete(0, tk.END)
            self.email_receiver_entry.insert(0, config.get("email_receiver", ""))
            
            # 加載監(jiān)控參數(shù)
            self.interval_entry.delete(0, tk.END)
            self.interval_entry.insert(0, config.get("interval", "10"))
            
            self.timeout_entry.delete(0, tk.END)
            self.timeout_entry.insert(0, config.get("timeout", "2"))
            
            self.retry_entry.delete(0, tk.END)
            self.retry_entry.insert(0, config.get("retry", "3"))
            
            self.retry_interval_entry.delete(0, tk.END)
            self.retry_interval_entry.insert(0, config.get("retry_interval", "5"))
            
            self.log_message("配置已加載")
        except FileNotFoundError:
            self.log_message("未找到配置文件,使用默認(rèn)配置")
        except Exception as e:
            self.log_message(f"加載配置失敗: {str(e)}")


if __name__ == "__main__":
    app = IPMonitorApp()
    app.mainloop()

適合人群:網(wǎng)絡(luò)管理員、運(yùn)維工程師、Python中級(jí)開(kāi)發(fā)者

常見(jiàn)問(wèn)題解答

Q:如何修改每日?qǐng)?bào)告發(fā)送時(shí)間?

A:修改calculate_next_report_time()方法中的小時(shí)數(shù)(默認(rèn)10:00)

Q:支持監(jiān)控IPv6地址嗎?

A:當(dāng)前版本需要稍作修改,建議使用ipaddress庫(kù)進(jìn)行驗(yàn)證

Q:最大支持監(jiān)控多少個(gè)IP?

A:理論上無(wú)限制,但建議不超過(guò)100個(gè)以保證性能

到此這篇關(guān)于使用Python實(shí)現(xiàn)IP地址和端口狀態(tài)檢測(cè)與監(jiān)控的文章就介紹到這了,更多相關(guān)Python IP地址檢測(cè)與監(jiān)控內(nèi)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家!

相關(guān)文章

  • 詳解如何使用python創(chuàng)建和結(jié)束線程

    詳解如何使用python創(chuàng)建和結(jié)束線程

    線程的創(chuàng)建和結(jié)束是多線程編程中的核心概念之一,在本文中,我們將學(xué)習(xí)如何使用 Python 創(chuàng)建線程,并探討如何優(yōu)雅地結(jié)束線程,需要的朋友可以參考下
    2024-04-04
  • Python中的數(shù)據(jù)類(lèi)dataclass解讀

    Python中的數(shù)據(jù)類(lèi)dataclass解讀

    這篇文章主要介紹了Python中的數(shù)據(jù)類(lèi)dataclass使用,具有很好的參考價(jià)值,希望對(duì)大家有所幫助,如有錯(cuò)誤或未考慮完全的地方,望不吝賜教
    2024-01-01
  • Python爬蟲(chóng)庫(kù)urllib的使用教程詳解

    Python爬蟲(chóng)庫(kù)urllib的使用教程詳解

    Python?給人的印象是抓取網(wǎng)頁(yè)非常方便,提供這種生產(chǎn)力的,主要依靠的就是?urllib、requests這兩個(gè)模塊。本文主要給大家介紹一下urllib的使用,感興趣的可以了解一下
    2022-11-11
  • Django配置文件代碼說(shuō)明

    Django配置文件代碼說(shuō)明

    在本篇文章里小編給大家整理了關(guān)于Django配置文件代碼說(shuō)明知識(shí)點(diǎn),有需要的朋友們學(xué)習(xí)下。
    2019-12-12
  • Python中__name__的具體使用

    Python中__name__的具體使用

    在Python中,__name__是一個(gè)內(nèi)置變量,用于表示當(dāng)前模塊的名稱(chēng),文中通過(guò)示例代碼介紹的非常詳細(xì),對(duì)大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值,需要的朋友們下面隨著小編來(lái)一起學(xué)習(xí)學(xué)習(xí)吧
    2024-09-09
  • 為Python程序添加圖形化界面的教程

    為Python程序添加圖形化界面的教程

    這篇文章主要介紹了為Python程序添加圖形化界面的教程,圖形化界面編程是開(kāi)發(fā)Python本地應(yīng)用的必備,需要的朋友可以參考下
    2015-04-04
  • python中py文件與pyc文件相互轉(zhuǎn)換的方法實(shí)例

    python中py文件與pyc文件相互轉(zhuǎn)換的方法實(shí)例

    pyc是一種二進(jìn)制文件,是由py文件經(jīng)過(guò)編譯后,生成的文件,下面這篇文章主要給大家介紹了關(guān)于python中py文件與pyc文件相互轉(zhuǎn)換的相關(guān)資料,需要的朋友可以參考下
    2022-05-05
  • Python包裝之對(duì)象處理

    Python包裝之對(duì)象處理

    這篇文章主要介紹了Python包裝之對(duì)象處理,文章圍繞主題展開(kāi)詳細(xì)的內(nèi)容介紹,具有一定的參考價(jià)值,需要的小伙伴可以參考一下
    2022-06-06
  • Python數(shù)據(jù)類(lèi)型詳解(四)字典:dict

    Python數(shù)據(jù)類(lèi)型詳解(四)字典:dict

    本文給大家分享的知識(shí)是Python數(shù)據(jù)類(lèi)型中的字典(dict)的基本概念,常用操作以及示例,非常的實(shí)用,對(duì)于大家理解字典dict非常有幫助,希望大家能夠喜歡
    2016-05-05
  • Python中定時(shí)器用法詳解之Timer定時(shí)器和schedule庫(kù)

    Python中定時(shí)器用法詳解之Timer定時(shí)器和schedule庫(kù)

    目前所在的項(xiàng)目組需要經(jīng)常執(zhí)行一些定時(shí)任務(wù),于是選擇使用 Python 的定時(shí)器,下面這篇文章主要給大家介紹了關(guān)于Python中定時(shí)器用法詳解之Timer定時(shí)器和schedule庫(kù)的相關(guān)資料,需要的朋友可以參考下
    2024-02-02

最新評(píng)論

舟曲县| 霸州市| 河南省| 科技| 任丘市| 称多县| 巧家县| 苏尼特右旗| 自贡市| 剑川县| 自贡市| 鄂伦春自治旗| 美姑县| 壤塘县| 壤塘县| 朝阳区| 天柱县| 连山| 晋宁县| 鹿泉市| 偃师市| 嘉义县| 五河县| 祁阳县| 西乡县| 罗甸县| 将乐县| 云和县| 沙雅县| 韩城市| 雷山县| 阿拉善左旗| 达拉特旗| 洛浦县| 镇安县| 赞皇县| 县级市| 会理县| 陇南市| 青田县| 天长市|