import tkinter as tk from tkinter import ttk, messagebox import urllib.request import hashlib import threading import time from datetime import datetime try: from plyer import notification as plyer_notification # 任意: 入っていれば OS の通知を使う except ImportError: plyer_notification = None class App063: """Webページ監視ツール""" def __init__(self, root): self.root = root self.root.title("Webページ監視ツール") self.root.geometry("920x664") self.root.minsize(920, 664) self.root.configure(bg="#1e1e1e") self._jobs = {} # url -> {"interval", "last_hash", "after_id", ...} self._log = [] self._build_ui() def _build_ui(self): header = tk.Frame(self.root, bg="#252526", pady=6) header.pack(fill=tk.X) tk.Label(header, text="🌐 Webページ監視ツール", font=("Noto Sans JP", 12, "bold"), bg="#252526", fg="#4fc3f7").pack(side=tk.LEFT, padx=12) # URL追加（試すときはローカルで `python -m http.server 8000` を起動） add_f = tk.Frame(self.root, bg="#1e1e1e", pady=4) add_f.pack(fill=tk.X, padx=8) tk.Label(add_f, text="URL:", bg="#1e1e1e", fg="#ccc", font=("Arial", 9)).pack(side=tk.LEFT) self.url_var = tk.StringVar(value="http://127.0.0.1:8000/") ttk.Entry(add_f, textvariable=self.url_var, width=40).pack(side=tk.LEFT, padx=4) tk.Label(add_f, text="間隔(秒):", bg="#1e1e1e", fg="#ccc", font=("Arial", 9)).pack(side=tk.LEFT, padx=(8, 2)) self.interval_var = tk.IntVar(value=60) ttk.Spinbox(add_f, from_=10, to=3600, textvariable=self.interval_var, width=6).pack(side=tk.LEFT) ttk.Button(add_f, text="▶ 監視開始", command=self._add_job).pack(side=tk.LEFT, padx=8) ttk.Button(add_f, text="⏹ 選択停止", command=self._stop_job).pack(side=tk.LEFT) ttk.Button(add_f, text="🗑 削除", command=self._remove_job).pack(side=tk.LEFT, padx=4) self.notify_var = tk.BooleanVar(value=True) tk.Checkbutton(add_f, text="🔔 変化時に通知", variable=self.notify_var, bg="#1e1e1e", fg="#ccc", selectcolor="#3c3c3c", activebackground="#1e1e1e").pack(side=tk.LEFT, padx=4) # 監視リスト jobs_f = tk.Frame(self.root, bg="#1e1e1e") jobs_f.pack(fill=tk.X, padx=8, pady=4) tk.Label(jobs_f, text="監視中のURL", bg="#1e1e1e", fg="#888", font=("Arial", 9)).pack(anchor="w") cols = ("url", "interval", "last_check", "status", "changes") self.job_tree = ttk.Treeview(jobs_f, columns=cols, show="headings", height=6, selectmode="browse") self.job_tree.heading("url", text="URL") self.job_tree.heading("interval", text="間隔") self.job_tree.heading("last_check", text="最終確認") self.job_tree.heading("status", text="状態") self.job_tree.heading("changes", text="変化回数") self.job_tree.column("url", width=300, anchor="w") self.job_tree.column("interval", width=60, anchor="center") self.job_tree.column("last_check", width=140, anchor="w") self.job_tree.column("status", width=80, anchor="center") self.job_tree.column("changes", width=70, anchor="center") jsb = ttk.Scrollbar(jobs_f, command=self.job_tree.yview) self.job_tree.configure(yscrollcommand=jsb.set) jsb.pack(side=tk.RIGHT, fill=tk.Y) self.job_tree.pack(fill=tk.X) self.job_tree.tag_configure("changed", foreground="#f48771") self.job_tree.tag_configure("ok", foreground="#4ec9b0") self.job_tree.tag_configure("error", foreground="#ffd700") # 変化ログ log_f = tk.Frame(self.root, bg="#1e1e1e") log_f.pack(fill=tk.BOTH, expand=True, padx=8, pady=4) tk.Label(log_f, text="変化ログ", bg="#1e1e1e", fg="#888", font=("Arial", 9)).pack(anchor="w") self.log_text = tk.Text(log_f, bg="#0d1117", fg="#d4d4d4", font=("Courier New", 9), relief=tk.FLAT, state=tk.DISABLED, wrap=tk.WORD) lsb = ttk.Scrollbar(log_f, command=self.log_text.yview) self.log_text.configure(yscrollcommand=lsb.set) lsb.pack(side=tk.RIGHT, fill=tk.Y) self.log_text.pack(fill=tk.BOTH, expand=True) self.log_text.tag_configure("changed", foreground="#f48771") self.log_text.tag_configure("ok", foreground="#4ec9b0") self.log_text.tag_configure("error", foreground="#ffd700") self.log_text.tag_configure("info", foreground="#4fc3f7") self.status_var = tk.StringVar(value="URLを追加して監視を開始してください") tk.Label(self.root, textvariable=self.status_var, bg="#252526", fg="#858585", font=("Arial", 9), anchor="w", padx=8).pack(fill=tk.X, side=tk.BOTTOM) def _add_job(self): url = self.url_var.get().strip() if not url.startswith("http"): messagebox.showerror("エラー", "有効なURLを入力してください") return if url in self._jobs: messagebox.showinfo("情報", "既に監視中のURLです") return interval = max(10, self.interval_var.get()) self._jobs[url] = { "interval": interval, "last_hash": None, "last_check": "", "changes": 0, "running": True, "after_id": None, } # Treeviewに追加 self.job_tree.insert("", tk.END, iid=url, values=(url, f"{interval}s", "-", "監視中", 0), tags=("ok",)) self._log_msg(f"監視開始: {url}", "info") self._schedule_check(url) def _schedule_check(self, url): job = self._jobs.get(url) if not job or not job["running"]: return threading.Thread(target=self._check, args=(url,), daemon=True).start() def _check(self, url): job = self._jobs.get(url) if not job: return try: req = urllib.request.Request( url, headers={"User-Agent": "Mozilla/5.0"}) with urllib.request.urlopen(req, timeout=15) as resp: content = resp.read() h = hashlib.md5(content).hexdigest() now = datetime.now().strftime("%Y-%m-%d %H:%M:%S") if job["last_hash"] is None: job["last_hash"] = h job["last_check"] = now self.root.after(0, self._update_job_row, url, "初回取得", "ok") elif h != job["last_hash"]: job["last_hash"] = h job["last_check"] = now job["changes"] += 1 self.root.after(0, self._update_job_row, url, "変化あり!", "changed") self.root.after(0, self._log_msg, f"⚠ 変化検出: {url} ({now})", "changed") self.root.after(0, self._notify, url, now) else: job["last_check"] = now self.root.after(0, self._update_job_row, url, "変化なし", "ok") except Exception as e: self.root.after(0, self._update_job_row, url, "エラー", "error") self.root.after(0, self._log_msg, f"✖ エラー: {url}: {e}", "error") # 次回スケジュール if job["running"]: after_id = self.root.after( job["interval"] * 1000, lambda u=url: self._schedule_check(u)) job["after_id"] = after_id def _update_job_row(self, url, status, tag): job = self._jobs.get(url) if not job: return try: self.job_tree.item(url, values=( url, f"{job['interval']}s", job["last_check"], status, job["changes"]), tags=(tag,)) except Exception: pass def _log_msg(self, text, tag=""): self.log_text.configure(state=tk.NORMAL) ts = datetime.now().strftime("%H:%M:%S") self.log_text.insert(tk.END, f"[{ts}] {text}\n", tag) self.log_text.see(tk.END) self.log_text.configure(state=tk.DISABLED) def _notify(self, url, when): """変化検出をデスクトップ通知で知らせる（plyer があれば OS の通知・無ければダイアログ）""" if not self.notify_var.get(): return title = "Webページ監視: 変化を検出" message = f"{url}\n{when}" if plyer_notification is not None: try: plyer_notification.notify(title=title, message=message, app_name="Webページ監視ツール", timeout=10) return except Exception: pass # 通知の仕組みが無い環境ではダイアログに切り替える messagebox.showinfo(title, message) def _stop_job(self): sel = self.job_tree.selection() if not sel: return url = sel[0] job = self._jobs.get(url) if job: job["running"] = False if job["after_id"]: self.root.after_cancel(job["after_id"]) self.job_tree.item(url, values=( url, f"{job['interval']}s", job["last_check"], "停止中", job["changes"])) self._log_msg(f"監視停止: {url}", "info") def _remove_job(self): sel = self.job_tree.selection() if not sel: return url = sel[0] self._stop_job() del self._jobs[url] self.job_tree.delete(url) self._log_msg(f"削除: {url}", "info") if __name__ == "__main__": root = tk.Tk() app = App063(root) root.mainloop()