《Snipaste命令行模式实现网站监控自动截图与视觉变化对比》

snipaste --help

import win32gui

def find_window_hwnd(window_title):
    """根据窗口标题（可模糊匹配）查找窗口句柄"""
    def callback(hwnd, hwnds):
        if win32gui.IsWindowVisible(hwnd) and window_title.lower() in win32gui.GetWindowText(hwnd).lower():
            hwnds.append(hwnd)
        return True
    hwnds = []
    win32gui.EnumWindows(callback, hwnds)
    return hwnds[0] if hwnds else None

# 示例：查找标题包含“Chrome”的窗口
chrome_hwnd = find_window_hwnd("Chrome")
if chrome_hwnd:
    print(f"找到Chrome窗口，句柄为：{chrome_hwnd}")
else:
    print("未找到指定窗口。")

pip install pillow opencv-python numpy

import subprocess
import os
import time
from pathlib import Path

class SnipasteScreenshotter:
    def __init__(self, snipaste_path="snipaste"):
        self.snipaste_cmd = snipaste_path

    def capture_by_hwnd(self, hwnd, save_path, delay_ms=1000):
        """通过窗口句柄截图"""
        # 确保目录存在
        Path(save_path).parent.mkdir(parents=True, exist_ok=True)
        # 构建命令
        cmd = [
            self.snipaste_cmd,
            "snip",
            "--hwnd", str(hwnd),
            "--file", save_path,
            "--delay", str(delay_ms)
        ]
        try:
            result = subprocess.run(cmd, capture_output=True, text=True, timeout=30)
            if result.returncode == 0 and os.path.exists(save_path):
                print(f"[成功] 截图已保存至：{save_path}")
                return True, save_path
            else:
                print(f"[失败] 截图命令错误：{result.stderr}")
                return False, None
        except subprocess.TimeoutExpired:
            print("[失败] 截图超时")
            return False, None

# 使用示例
if __name__ == "__main__":
    snipper = SnipasteScreenshotter()
    # 假设已通过之前的函数获取到hwnd
    test_hwnd = 123456 # 替换为实际句柄
    success, path = snipper.capture_by_hwnd(test_hwnd, "screenshots/monitor_{}.png".format(int(time.time())))

from PIL import Image, ImageChops, ImageStat
import cv2
import numpy as np
from pathlib import Path

class ImageComparator:
    @staticmethod
    def pixel_diff(image_path_a, image_path_b, diff_save_path=None, threshold=10):
        """像素差异比对，返回差异比例和差异图"""
        img_a = Image.open(image_path_a).convert('RGB')
        img_b = Image.open(image_path_b).convert('RGB')
        
        # 确保图像尺寸一致
        if img_a.size != img_b.size:
            # 尝试调整第二张图的大小以匹配第一张
            img_b = img_b.resize(img_a.size, Image.Resampling.LANCZOS)
        
        # 计算绝对差异
        diff = ImageChops.difference(img_a, img_b)
        # 将差异转换为灰度并二值化
        diff_gray = diff.convert('L')
        # 阈值处理：超过阈值的像素视为有差异
        diff_mask = diff_gray.point(lambda p: 255 if p > threshold else 0)
        
        # 计算差异像素比例
        stat = ImageStat.Stat(diff_mask)
        diff_pixel_count = stat.sum[0] / 255  # 白色像素点数量
        total_pixels = img_a.size[0] * img_a.size[1]
        diff_ratio = diff_pixel_count / total_pixels
        
        # 保存差异图（可选）
        if diff_save_path:
            diff_display = Image.new('RGB', img_a.size)
            # 将差异部分用红色高亮在原图上
            img_a_copy = img_a.copy()
            diff_red = Image.new('RGB', img_a.size, (255, 0, 0))
            img_a_copy.paste(diff_red, mask=diff_mask)
            img_a_copy.save(diff_save_path)
            print(f"差异图已保存至：{diff_save_path}")
        
        return diff_ratio, diff_mask

    @staticmethod
    def ssim_diff(image_path_a, image_path_b):
        """使用结构相似性指数(SSIM)进行比对，返回相似度分数 (0-1，越接近1越相似)"""
        # 读取图像
        img_a = cv2.imread(image_path_a)
        img_b = cv2.imread(image_path_b)
        
        if img_a is None or img_b is None:
            raise ValueError("无法读取图像文件")
        
        # 转换为灰度图
        gray_a = cv2.cvtColor(img_a, cv2.COLOR_BGR2GRAY)
        gray_b = cv2.cvtColor(img_b, cv2.COLOR_BGR2GRAY)
        
        # 确保尺寸一致
        if gray_a.shape != gray_b.shape:
            gray_b = cv2.resize(gray_b, (gray_a.shape[1], gray_a.shape[0]))
        
        # 计算SSIM
        from skimage.metrics import structural_similarity as ssim
        # 注意：需要安装 scikit-image: pip install scikit-image
        score, diff = ssim(gray_a, gray_b, full=True)
        return score, diff

# 使用示例
if __name__ == "__main__":
    comparator = ImageComparator()
    ratio, mask = comparator.pixel_diff("baseline.png", "latest.png", "diff_highlight.png", threshold=20)
    print(f"像素差异比例：{ratio:.4%}")
    if ratio > 0.001: # 如果差异超过0.1%
        print("检测到显著视觉变化！")

import time
import json
from datetime import datetime
from screenshot_agent import SnipasteScreenshotter
from image_comparator import ImageComparator
from win_utils import find_window_hwnd  # 假设这是之前写的窗口查找函数
import smtplib
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart
from email.mime.image import MIMEImage

class WebsiteVisualMonitor:
    def __init__(self, config_path="config.json"):
        with open(config_path, 'r') as f:
            self.config = json.load(f)
        self.snipper = SnipasteScreenshotter()
        self.comparator = ImageComparator()
        self.base_dir = Path(self.config.get("base_dir", "./monitor_data"))
        self.base_dir.mkdir(exist_ok=True)
        
    def run_monitoring_cycle(self):
        """执行一次完整的监控周期"""
        cycle_time = datetime.now().strftime("%Y%m%d_%H%M%S")
        print(f"\n=== 开始监控周期 {cycle_time} ===")
        
        # 1. 获取目标窗口句柄
        target_title = self.config["window_title"]
        hwnd = find_window_hwnd(target_title)
        if not hwnd:
            self.send_alert(f"警告：未找到目标窗口 '{target_title}'", is_error=True)
            return
        
        # 2. 执行截图
        latest_path = self.base_dir / "latest" / f"screenshot_{cycle_time}.png"
        success, screenshot_path = self.snipper.capture_by_hwnd(
            hwnd, 
            str(latest_path), 
            delay_ms=self.config.get("delay_ms", 2000)
        )
        if not success:
            self.send_alert("警告：截图失败", is_error=True)
            return
        
        # 3. 与基线对比
        baseline_path = self.base_dir / "baseline.png"
        if baseline_path.exists():
            diff_ratio, _ = self.comparator.pixel_diff(
                str(baseline_path), 
                screenshot_path,
                diff_save_path=str(self.base_dir / "diff" / f"diff_{cycle_time}.png"),
                threshold=self.config.get("diff_threshold", 15)
            )
            
            # 4. 根据差异判断
            alert_threshold = self.config.get("alert_threshold", 0.001) # 0.1%
            if diff_ratio > alert_threshold:
                alert_msg = f"检测到显著视觉变化！差异比例：{diff_ratio:.4%}\n最新截图：{screenshot_path}"
                self.send_alert(alert_msg, attachment_path=screenshot_path)
                # 可以选择不更新基线，以便人工审查
                if self.config.get("auto_update_baseline_on_alert", False):
                    self.update_baseline(screenshot_path)
            else:
                print(f"视觉状态正常，差异比例：{diff_ratio:.4%}")
                # 可选的自动基线更新策略：定期更新或始终更新
                if self.config.get("auto_update_baseline_always", False):
                    self.update_baseline(screenshot_path)
        else:
            # 首次运行，建立基线
            print("首次运行，创建基线截图。")
            self.update_baseline(screenshot_path)
        
        # 5. 归档管理（示例：保留最近7天的截图）
        self.cleanup_old_files(self.base_dir / "latest", days=7)
        self.cleanup_old_files(self.base_dir / "diff", days=7)
    
    def update_baseline(self, new_baseline_path):
        """更新基线图片"""
        baseline_path = self.base_dir / "baseline.png"
        import shutil
        shutil.copy2(new_baseline_path, baseline_path)
        print(f"基线已更新为：{new_baseline_path}")
    
    def send_alert(self, message, attachment_path=None, is_error=False):
        """发送警报（示例为邮件，可替换为其他通知方式）"""
        print(f"[警报] {message}")
        # 这里简化实现，实际应配置SMTP等信息
        # 可扩展为发送钉钉、Slack、企业微信消息
        if self.config.get("email_alerts", False):
            # 实现邮件发送逻辑...
            pass
    
    def cleanup_old_files(self, directory, days):
        """清理指定目录中超过天数的文件"""
        from datetime import datetime, timedelta
        cutoff_time = datetime.now() - timedelta(days=days)
        for file in directory.glob("*"):
            if file.is_file():
                file_time = datetime.fromtimestamp(file.stat().st_mtime)
                if file_time < cutoff_time:
                    file.unlink()
                    print(f"已清理旧文件：{file.name}")

if __name__ == "__main__":
    monitor = WebsiteVisualMonitor("monitor_config.json")
    monitor.run_monitoring_cycle()

{
    "window_title": "My Website - Chrome",
    "base_dir": "./data/website_monitor",
    "delay_ms": 3000,
    "diff_threshold": 20,
    "alert_threshold": 0.001,
    "auto_update_baseline_always": false,
    "auto_update_baseline_on_alert": false,
    "email_alerts": false,
    "smtp_server": "smtp.example.com",
    "smtp_port": 587,
    "alert_recipients": ["team@example.com"]
}

from selenium import webdriver
from selenium.webdriver.chrome.options import Options

def prepare_browser(url, window_title_prefix="My Website"):
    """使用Selenium打开并准备浏览器窗口"""
    chrome_options = Options()
    chrome_options.add_argument("--start-maximized") # 最大化窗口，确保布局一致
    # 可添加无头模式选项，但截图需要非无头模式
    # chrome_options.add_argument("--headless") 
    
    driver = webdriver.Chrome(options=chrome_options)
    driver.get(url)
    time.sleep(5) # 等待页面初步加载
    # 可以在这里执行登录、点击等操作
    # driver.find_element(...).click()
    
    # 设置独特的窗口标题，便于后续查找
    original_title = driver.title
    driver.execute_script(f"document.title = '{window_title_prefix} - {original_title}';")
    
    return driver  # 注意：需要保持driver实例不关闭

@echo off
cd /d "C:\path\to\your\script\directory"
python website_monitor.py