公网IP变更邮箱提醒工具

再贱就再见

找人写要450元,AI手搓了个公网IP变更邮箱提醒,设置完要点保存配置,可以最小化到托盘,软件同目录要有app.ico图标,才能最小化到系统托盘.

https://wwbmj.lanzouu.com/ibNUA43nnk5e

[Python] 纯文本查看 复制代码

import requests
import smtplib
import time
import logging
import json
import os
import threading
import queue
import sys
import ctypes
from email.mime.text import MIMEText
from email.header import Header
import tkinter as tk
from tkinter import ttk, scrolledtext, messagebox
import winreg

# 延迟导入pystray，避免直接运行py报错
try:
    from pystray import Icon, MenuItem, Menu
    from PIL import Image
    HAS_TRAY = True
except ModuleNotFoundError:
    HAS_TRAY = False
    print("警告：未安装pystray，托盘功能不可用，请执行 pip install pystray pillow")

# Windows互斥锁，防止重复启动
MUTEX_NAME = "IPMonitor_Tool_20260820_UniqueMutex"
if sys.platform == "win32":
    kernel32 = ctypes.windll.kernel32
    h_mutex = kernel32.CreateMutexW(None, False, MUTEX_NAME)
    last_err = kernel32.GetLastError()
    if last_err == 183:  # ERROR_ALREADY_EXISTS 已存在实例
        messagebox.showinfo("提示", "软件已经在运行中，请勿重复启动！")
        sys.exit(0)

CONFIG_FILE = "config.json"
IP_SAVE_FILE = "last_ip.txt"
LOG_FILE = "ip_log.txt"

logging.basicConfig(
    filename=LOG_FILE,
    level=logging.INFO,
    format="%(asctime)s - %(levelname)s - %(message)s",
    encoding="utf-8"
)


class IPMonitorGUI:
    def __init__(self, root):
        self.root = root
        self.root.title("公网IP变更邮件提醒工具")
        self.root.geometry("680x580")
        self.running = False
        self.monitor_thread = None
        self.tray_icon = None
        self.autostart_key_name = "IPMonitorTool"

        self.log_queue = queue.Queue()
        self._poll_log_queue()

        # 邮箱配置
        frame_cfg = ttk.LabelFrame(root, text="邮箱配置")
        frame_cfg.pack(fill="x", padx=10, pady=5)

        ttk.Label(frame_cfg, text="SMTP服务器:").grid(row=0, column=0, padx=5, pady=3, sticky="w")
        self.var_smtp = tk.StringVar(value="smtp.qq.com")
        ttk.Entry(frame_cfg, textvariable=self.var_smtp, width=32).grid(row=0, column=1, padx=5, pady=3)

        ttk.Label(frame_cfg, text="端口:").grid(row=0, column=2, padx=5, pady=3, sticky="w")
        self.var_port = tk.StringVar(value="465")
        ttk.Entry(frame_cfg, textvariable=self.var_port, width=10).grid(row=0, column=3, padx=5, pady=3)

        ttk.Label(frame_cfg, text="发件邮箱:").grid(row=1, column=0, padx=5, pady=3, sticky="w")
        self.var_sender = tk.StringVar()
        ttk.Entry(frame_cfg, textvariable=self.var_sender, width=32).grid(row=1, column=1, padx=5, pady=3)

        ttk.Label(frame_cfg, text="授权码:").grid(row=1, column=2, padx=5, pady=3, sticky="w")
        self.var_pwd = tk.StringVar()
        ttk.Entry(frame_cfg, textvariable=self.var_pwd, width=18, show="*").grid(row=1, column=3, padx=5, pady=3)

        ttk.Label(frame_cfg, text="接收邮箱:").grid(row=2, column=0, padx=5, pady=3, sticky="w")
        self.var_receiver = tk.StringVar()
        ttk.Entry(frame_cfg, textvariable=self.var_receiver, width=32).grid(row=2, column=1, padx=5, pady=3)

        ttk.Label(frame_cfg, text="检测间隔(秒):").grid(row=2, column=2, padx=5, pady=3, sticky="w")
        self.var_interval = tk.StringVar(value="300")
        ttk.Entry(frame_cfg, textvariable=self.var_interval, width=10).grid(row=2, column=3, padx=5, pady=3)

        # 高级设置
        frame_adv = ttk.LabelFrame(root, text="高级设置")
        frame_adv.pack(fill="x", padx=10, pady=5)

        ttk.Label(frame_adv, text="日志保留天数:").grid(row=0, column=0, padx=5, pady=3, sticky="w")
        self.var_log_days = tk.StringVar(value="30")
        ttk.Entry(frame_adv, textvariable=self.var_log_days, width=3).grid(row=0, column=1, padx=5, pady=3)
        ttk.Label(frame_adv, text="(到期自动清理旧日志)").grid(row=0, column=2, padx=2, pady=3, sticky="w")

        frame_boot = ttk.Frame(frame_adv)
        frame_boot.grid(row=0, column=3, columnspan=4, padx=5, pady=3)

        # 按钮顺序：手动清理日志 → 开始监控 → 开机自启 → 保存配置
        self.btn_clean_log = ttk.Button(frame_boot, text="手动清理日志", command=self.manual_clean_log)
        self.btn_clean_log.grid(row=0, column=0, padx=3)

        self.btn_toggle_monitor = ttk.Button(frame_boot, text="开始监控", command=self.toggle_monitor)
        self.btn_toggle_monitor.grid(row=0, column=1, padx=3)

        self.btn_autostart = ttk.Button(frame_boot, text="开启开机自启", command=self.toggle_auto_start)
        self.btn_autostart.grid(row=0, column=2, padx=3)

        self.btn_save = ttk.Button(frame_boot, text="保存配置", command=self.save_config)
        self.btn_save.grid(row=0, column=3, padx=3)

        # 日志框
        frame_log = ttk.LabelFrame(root, text="运行日志")
        frame_log.pack(fill="both", expand=True, padx=10, pady=5)
        self.log_text = scrolledtext.ScrolledText(frame_log, height=12)
        self.log_text.pack(fill="both", expand=True, padx=5, pady=5)

        self.update_autostart_btn_text()
        self.update_monitor_button_text()
        self.init_tray()
        self.root.protocol("WM_DELETE_WINDOW", self.hide_window_to_tray)
        self.load_config()

        # 如果配置标记监控开启，程序启动自动运行监控
        cfg = self.read_raw_config()
        if cfg.get("monitor_running", False):
            self.log("检测到上次监控为开启状态，自动启动监控")
            self.root.after(300, self.start_monitor_inner)

    def read_raw_config(self):
        if os.path.exists(CONFIG_FILE):
            try:
                with open(CONFIG_FILE, "r", encoding="utf-8") as f:
                    return json.load(f)
            except:
                return {}
        return {}

    def is_autostart_enabled(self):
        try:
            key = winreg.OpenKey(winreg.HKEY_CURRENT_USER, r"Software\Microsoft\Windows\CurrentVersion\Run", 0, winreg.KEY_READ)
            winreg.QueryValueEx(key, self.autostart_key_name)
            winreg.CloseKey(key)
            return True
        except FileNotFoundError:
            return False

    def update_autostart_btn_text(self):
        if self.is_autostart_enabled():
            self.btn_autostart.config(text="关闭开机自启")
        else:
            self.btn_autostart.config(text="开启开机自启")

    def update_monitor_button_text(self):
        if self.running:
            self.btn_toggle_monitor.config(text="停止监控")
        else:
            self.btn_toggle_monitor.config(text="开始监控")

    def toggle_auto_start(self):
        exe_path = self.get_exe_path()
        try:
            if self.is_autostart_enabled():
                key = winreg.OpenKey(winreg.HKEY_CURRENT_USER, r"Software\Microsoft\Windows\CurrentVersion\Run", 0, winreg.KEY_WRITE)
                winreg.DeleteValue(key, self.autostart_key_name)
                winreg.CloseKey(key)
                messagebox.showinfo("提示", "已关闭开机自启")
                self.log("✅已关闭Windows开机自启")
            else:
                key = winreg.OpenKey(winreg.HKEY_CURRENT_USER, r"Software\Microsoft\Windows\CurrentVersion\Run", 0, winreg.KEY_WRITE)
                winreg.SetValueEx(key, self.autostart_key_name, 0, winreg.REG_SZ, exe_path)
                winreg.CloseKey(key)
                messagebox.showinfo("提示", "已开启开机自启！\n源码py运行无效，请打包exe")
                self.log("✅已开启Windows开机自启")
            self.update_autostart_btn_text()
        except Exception as e:
            messagebox.showerror("错误", f"操作开机自启失败：{e}")

    def toggle_monitor(self):
        if self.running:
            self.stop_monitor_inner()
        else:
            self.start_monitor_inner()

    def start_monitor_inner(self):
        if self.running:
            messagebox.showwarning("提示", "监控已经在运行！")
            return
        try:
            port = int(self.var_port.get())
            interval = int(self.var_interval.get())
            log_days = int(self.var_log_days.get())
            if interval <= 0 or log_days <= 0:
                raise ValueError
        except:
            messagebox.showerror("错误", "端口、检测间隔、日志保留天数必须填写正整数！")
            return

        self.running = True
        self.save_config(monitor_running=True)
        self.monitor_thread = threading.Thread(target=self.monitor_loop, daemon=True)
        self.monitor_thread.start()
        self.update_monitor_button_text()

    def stop_monitor_inner(self):
        self.running = False
        self.save_config(monitor_running=False)
        self.update_monitor_button_text()

    def init_tray(self):
        if not HAS_TRAY:
            self.log("pystray未安装，托盘功能禁用")
            return
        try:
            # 加载自定义托盘图标 app.ico
            icon_img = Image.open("app.ico")
            menu = Menu(
                MenuItem("显示主窗口", self.show_window, default=True),
                MenuItem("退出程序", self.quit_app)
            )
            self.tray_icon = Icon("IPMonitor", icon_img, "IP监控工具", menu)
            tray_thread = threading.Thread(target=self.tray_icon.run, daemon=True)
            tray_thread.start()
        except Exception as e:
            self.log(f"托盘初始化失败:{e}")

    def hide_window_to_tray(self):
        if HAS_TRAY and self.tray_icon:
            self.root.withdraw()
        else:
            self.quit_app()

    def show_window(self, icon=None, item=None):
        self.root.after(0, lambda: (self.root.deiconify(), self.root.lift()))

    def quit_app(self):
        self.running = False
        time.sleep(0.4)
        if HAS_TRAY and self.tray_icon:
            self.tray_icon.stop()
        self.root.after(0, lambda: self.root.destroy())

    def _poll_log_queue(self):
        while not self.log_queue.empty():
            msg = self.log_queue.get()
            self.log_text.insert(tk.END, f"{time.strftime('%Y-%m-%d %H:%M:%S')} {msg}\n")
            self.log_text.see(tk.END)
        self.root.after(100, self._poll_log_queue)

    def log(self, msg):
        self.log_queue.put(msg)
        logging.info(msg)

    def auto_clean_log(self, keep_days: int):
        if keep_days <= 0:
            return
        if not os.path.exists(LOG_FILE):
            return
        now_ts = time.time()
        keep_sec = keep_days * 86400
        new_lines = []
        try:
            with open(LOG_FILE, "r", encoding="utf-8") as f:
                lines = f.readlines()
            for line in lines:
                try:
                    time_str = line.split(" - ", 1)[0]
                    log_time = time.strptime(time_str, "%Y-%m-%d %H:%M:%S")
                    log_ts = time.mktime(log_time)
                    if now_ts - log_ts <= keep_sec:
                        new_lines.append(line)
                except Exception:
                    new_lines.append(line)
            with open(LOG_FILE, "w", encoding="utf-8") as f:
                f.writelines(new_lines)
            return True
        except Exception as e:
            self.log(f"&#9888;&#65039;日志清理异常：{e}")
            return False

    def manual_clean_log(self):
        try:
            days = int(self.var_log_days.get().strip())
            if days <= 0:
                messagebox.showerror("错误", "日志保留天数必须大于0")
                return
            self.log_text.delete(1.0, tk.END)
            ok = self.auto_clean_log(days)
            if ok:
                messagebox.showinfo("完成", f"已清理，保留最近 {days} 天日志")
                self.log(f"&#9989;手动执行日志清理，保留{days}天")
        except Exception as e:
            messagebox.showerror("错误", f"清理失败:{e}")

    def get_exe_path(self):
        if getattr(sys, 'frozen', False):
            return sys.executable
        else:
            return sys.argv[0]

    def load_config(self):
        if os.path.exists(CONFIG_FILE):
            try:
                with open(CONFIG_FILE, "r", encoding="utf-8") as f:
                    cfg = json.load(f)
                self.var_smtp.set(cfg.get("smtp_server", "smtp.qq.com"))
                self.var_port.set(cfg.get("smtp_port", "465"))
                self.var_sender.set(cfg.get("sender_email", ""))
                self.var_pwd.set(cfg.get("sender_pwd", ""))
                self.var_receiver.set(cfg.get("receiver_email", ""))
                self.var_interval.set(cfg.get("check_interval", "300"))
                self.var_log_days.set(cfg.get("log_keep_days", "30"))
                self.log("已读取本地保存的配置")
            except Exception as e:
                self.log(f"读取配置失败: {e}")

    def save_config(self, monitor_running=None):
        try:
            port = int(self.var_port.get().strip())
            interval = int(self.var_interval.get().strip())
            log_days = int(self.var_log_days.get().strip())
            if interval <= 0 or log_days <= 0:
                raise ValueError("数值必须大于0")
            old_cfg = self.read_raw_config()
            cfg = {
                "smtp_server": self.var_smtp.get().strip(),
                "smtp_port": port,
                "sender_email": self.var_sender.get().strip(),
                "sender_pwd": self.var_pwd.get().strip(),
                "receiver_email": self.var_receiver.get().strip(),
                "check_interval": interval,
                "log_keep_days": log_days,
                "monitor_running": monitor_running if monitor_running is not None else old_cfg.get("monitor_running", False)
            }
            with open(CONFIG_FILE, "w", encoding="utf-8") as f:
                json.dump(cfg, f, ensure_ascii=False, indent=2)
            self.log("&#9989; 配置保存成功！")
        except Exception as e:
            self.log(f"&#10060; 保存配置失败: {e}")
            messagebox.showerror("错误", f"保存失败：{e}")

    def get_public_ip(self):
        ip_api_list = [
            "https://ifconfig.me/ip",
            "https://api.ipify.org",
            "https://ip.3322.net",
            "https://ipv4.icanhazip.com"
        ]
        for api in ip_api_list:
            try:
                res = requests.get(api, timeout=8)
                ip = res.text.strip()
                if ip:
                    return ip
            except Exception:
                continue
        self.log("所有公网IP接口全部访问失败")
        return None

    def send_start_test_email(self, current_ip, cfg):
        subject = "【IP监控启动通知】当前公网IP"
        content = f"IP监控程序已成功启动！\n当前公网IP：{current_ip}\n后续IP发生变动将会推送邮件通知"
        msg = MIMEText(content, "plain", "utf-8")
        msg["Subject"] = Header(subject, "utf-8")
        msg["From"] = cfg["sender_email"]
        msg["To"] = cfg["receiver_email"]
        try:
            server = smtplib.SMTP_SSL(cfg["smtp_server"], cfg["smtp_port"])
            server.login(cfg["sender_email"], cfg["sender_pwd"])
            server.sendmail(cfg["sender_email"], cfg["receiver_email"], msg.as_string())
            server.quit()
            self.log("&#9989; 启动测试邮件发送成功，请查收邮箱！")
            return True
        except Exception as e:
            self.log(f"&#10060; 启动测试邮件发送失败：{e}")
            return False

    def send_change_email(self, old_ip, new_ip, cfg):
        subject = "【公网IP变更提醒】"
        content = f"检测到公网IP发生变更！\n旧IP：{old_ip}\n新IP：{new_ip}"
        msg = MIMEText(content, "plain", "utf-8")
        msg["Subject"] = Header(subject, "utf-8")
        msg["From"] = cfg["sender_email"]
        msg["To"] = cfg["receiver_email"]
        try:
            server = smtplib.SMTP_SSL(cfg["smtp_server"], cfg["smtp_port"])
            server.login(cfg["sender_email"], cfg["sender_pwd"])
            server.sendmail(cfg["sender_email"], cfg["receiver_email"], msg.as_string())
            server.quit()
            self.log("&#9989; IP变更通知邮件发送成功！")
        except Exception as e:
            self.log(f"&#10060; IP变更邮件发送失败：{e}")

    def read_last_ip(self):
        try:
            with open(IP_SAVE_FILE, "r", encoding="utf-8") as f:
                return f.read().strip()
        except FileNotFoundError:
            return None

    def save_current_ip(self, ip):
        with open(IP_SAVE_FILE, "w", encoding="utf-8") as f:
            f.write(ip)

    def monitor_loop(self):
        cfg = {
            "smtp_server": self.var_smtp.get().strip(),
            "smtp_port": int(self.var_port.get().strip()),
            "sender_email": self.var_sender.get().strip(),
            "sender_pwd": self.var_pwd.get().strip(),
            "receiver_email": self.var_receiver.get().strip(),
            "check_interval": int(self.var_interval.get().strip()),
            "log_keep_days": int(self.var_log_days.get().strip())
        }
        if not cfg["sender_email"] or not cfg["sender_pwd"] or not cfg["receiver_email"]:
            self.log("&#10060; 邮箱配置不全，请填写完整邮箱、授权码！")
            self.running = False
            self.root.after(0, self.update_monitor_button_text)
            return

        self.log("==== IP监控已启动 ====")
        self.auto_clean_log(cfg["log_keep_days"])

        current_ip = self.get_public_ip()
        if current_ip:
            self.send_start_test_email(current_ip, cfg)
            last_ip = self.read_last_ip()
            if last_ip is None:
                self.save_current_ip(current_ip)
                self.log("首次运行，记录当前IP")
        else:
            self.log("&#9888;&#65039; 启动时获取公网IP失败！")

        while self.running:
            current_ip = self.get_public_ip()
            if not current_ip:
                self.log("获取IP失败，等待下一轮检测")
                for _ in range(cfg["check_interval"]):
                    if not self.running:
                        break
                    time.sleep(1)
                continue

            last_ip = self.read_last_ip()
            self.log(f"当前IP: {current_ip} | 上次IP: {last_ip}")

            if last_ip is not None and current_ip != last_ip:
                self.log("&#9888;&#65039; 公网IP发生变更！")
                self.send_change_email(last_ip, current_ip, cfg)
                self.save_current_ip(current_ip)

            for _ in range(cfg["check_interval"]):
                if not self.running:
                    break
                time.sleep(1)

        self.log("==== IP监控已停止 ====")
        self.root.after(0, self.update_monitor_button_text)


if __name__ == "__main__":
    root = tk.Tk()
    # 设置窗口左上角图标，缺失则跳过
    try:
        root.iconbitmap("app.ico")
    except Exception:
        pass
    app = IPMonitorGUI(root)
    root.mainloop()

fyufx

感谢，这授权码是多少来着