深度解析Python自动化截图和录屏的3大方案(PIL/mss/ffmpeg)

from PIL import Image, ImageDraw, ImageFont
import pyautogui  # PIL的ImageGrab在部分Windows版本有兼容性问题，用pyautogui辅助截图
import datetime
import os

class TestScreenshot:
    def __init__(self, module_name):
        # 初始化模块名称和保存目录
        self.module_name = module_name
        self.save_dir = self._create_save_dir()
        # 加载水印字体（需提前准备字体文件，避免中文乱码）
        self.font = ImageFont.truetype("simhei.ttf", 20)  # 黑体，20号字

    def _create_save_dir(self):
        # 按“模块名/年-月-日”创建保存目录
        today = datetime.datetime.now().strftime("%Y-%m-%d")
        save_dir = os.path.join("test_screenshots", self.module_name, today)
        if not os.path.exists(save_dir):
            os.makedirs(save_dir)
        return save_dir

    def take_screenshot(self, case_id, case_result="pass"):
        """
        执行截图并保存
        :param case_id: 测试用例ID
        :param case_result: 测试结果（pass/fail）
        :return: 截图保存路径
        """
        # 1. 系统级全屏截图（pyautogui本质是封装了PIL，兼容性更好）
        screenshot = pyautogui.screenshot()  # 返回PIL.Image对象
        # 2. 添加水印（用例ID+执行时间+测试结果）
        draw = ImageDraw.Draw(screenshot)
        now_time = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")
        watermark_text = f"用例ID：{case_id} | 执行时间：{now_time} | 结果：{case_result}"
        # 水印位置：右下角，距边缘20px
        text_width, text_height = draw.textbbox((0, 0), watermark_text, font=self.font)[2:]
        screen_width, screen_height = screenshot.size
        text_x = screen_width - text_width - 20
        text_y = screen_height - text_height - 20
        # 绘制水印（黑色字体，半透明背景，增强可读性）
        draw.rectangle(
            [text_x - 10, text_y - 5, text_x + text_width + 10, text_y + text_height + 5],
            fill=(255, 255, 255, 128)  # 白色半透明背景
        )
        draw.text((text_x, text_y), watermark_text, font=self.font, fill=(0, 0, 0))  # 黑色字体
        # 3. 按规则保存截图（PNG格式，支持透明，画质清晰）
        file_name = f"{case_id}_{now_time.replace(' ', '_').replace(':', '-')}_{case_result}.png"
        save_path = os.path.join(self.save_dir, file_name)
        screenshot.save(save_path, format="PNG", quality=95)  # quality=95保证画质
        return save_path

# 测试使用
if __name__ == "__main__":
    screenshot_tool = TestScreenshot(module_name="商品详情页模块")
    save_path = screenshot_tool.take_screenshot(case_id="SPXQ-001", case_result="pass")
    print(f"截图保存成功：{save_path}")
    

import mss
import mss.tools
import imageio
import datetime
import os
import threading
import queue

class HighFpsScreenRecorder:
    def __init__(self, fps=30, resolution=(1920, 1080), save_dir="monitor_recordings"):
        self.fps = fps
        self.resolution = resolution  # (宽, 高)
        self.save_dir = self._create_save_dir()
        self.is_recording = False
        self.frame_queue = queue.Queue(maxsize=10)  # 帧缓存队列，避免截图与编码阻塞
        self.sct = mss.mss()  # 初始化mss截图对象
        # 配置截图区域（全屏：从(0,0)到分辨率大小）
        self.monitor = {"top": 0, "left": 0, "width": resolution[0], "height": resolution[1]}

    def _create_save_dir(self):
        # 创建保存目录
        if not os.path.exists(self.save_dir):
            os.makedirs(self.save_dir)
        return self.save_dir

    def _capture_frames(self):
        """截图线程：持续采集图像帧，放入缓存队列"""
        while self.is_recording:
            # 1. mss高速截图，获取原始RGB数据
            frame = self.sct.grab(self.monitor)
            # 2. 转换为imageio可处理的格式（numpy数组）
            frame_np = mss.tools.to_numpy(frame)
            # 3. 放入队列（非阻塞，避免截图线程阻塞）
            try:
                self.frame_queue.put_nowait(frame_np)
            except queue.Full:
                # 队列满时丢弃最旧的帧，保证实时性
                self.frame_queue.get_nowait()
                self.frame_queue.put_nowait(frame_np)

    def _encode_video(self):
        """编码线程：从缓存队列获取帧，编码为MP4"""
        # 生成保存文件名（按时间戳命名）
        timestamp = datetime.datetime.now().strftime("%Y%m%d_%H%M%S")
        save_path = os.path.join(self.save_dir, f"monitor_{timestamp}.mp4")
        # 初始化imageio写入器，使用ffmpeg编码，设置帧率和编码格式
        # 选用libx264编码，平衡压缩比和CPU占用
        writer = imageio.get_writer(
            save_path,
            fps=self.fps,
            codec="libx264",
            pixelformat="yuv420p",  # 兼容大多数播放器
            quality=8  # 质量1-10，8为平衡值
        )

        while self.is_recording or not self.frame_queue.empty():
            # 从队列获取帧（阻塞，直到有帧或录制结束）
            try:
                frame_np = self.frame_queue.get(timeout=1)
                # 写入视频
                writer.append_data(frame_np)
                self.frame_queue.task_done()
            except queue.Empty:
                continue

        # 关闭写入器，完成视频保存
        writer.close()
        print(f"录屏保存成功：{save_path}")

    def start_recording(self):
        """开始录屏"""
        if self.is_recording:
            print("已处于录屏状态！")
            return
        self.is_recording = True
        # 启动截图线程和编码线程（分离线程，避免阻塞主线程）
        capture_thread = threading.Thread(target=self._capture_frames)
        encode_thread = threading.Thread(target=self._encode_video)
        capture_thread.daemon = True
        encode_thread.daemon = True
        capture_thread.start()
        encode_thread.start()
        print(f"录屏已启动，帧率：{self.fps} FPS，保存目录：{self.save_dir}")

    def stop_recording(self):
        """停止录屏"""
        self.is_recording = True
        print("正在停止录屏...")
        # 等待队列处理完成
        self.frame_queue.join()
        print("录屏已停止")

# 测试使用
if __name__ == "__main__":
    # 初始化录屏工具（30 FPS，1920×1080分辨率）
    recorder = HighFpsScreenRecorder(fps=30, resolution=(1920, 1080))
    try:
        recorder.start_recording()
        # 模拟录制10秒（实际使用中可根据业务逻辑控制录制时长）
        input("按Enter键停止录屏...\n")
    finally:
        recorder.stop_recording()