feat(screenshot): 定时任务截图瘦身（三态模式 + JPEG 压缩 + 过期自动清理）

- 默认 failure_only 模式（仅失败步骤截图，质量 60）
- _should_capture_step 三态判定 + screenshot=False 总开关优先
- start() 自动 _cleanup_expired_screenshots（SCREENSHOT_MAX_AGE_DAYS=14）
- scheduler_service 注入 screenshot_mode=failure_only

测试：pytest tests/test_screenshot_mode.py 15/15 pass + 后端全量 403/403 pass
Co-Authored-By: Claude <noreply@anthropic.com>

backend/app/config.py

@@ -25,6 +25,9 @@ class Settings:
         CORS_ORIGINS (list): 允许的跨域来源
         PLAYWRIGHT_HEADLESS (bool): Playwright 是否无头模式
         SCREENSHOT_DIR (str): 截图存储目录
+        SCREENSHOT_MODE (str): 截图模式 full/failure_only/off，默认 failure_only
+        SCREENSHOT_QUALITY (int): JPEG 压缩质量，默认 60
+        SCREENSHOT_MAX_AGE_DAYS (int): 截图保留天数（自动清理），默认 14
     """
 
     # 基础配置
@@ -51,6 +54,13 @@ class Settings:
     # 文件存储配置
     SCREENSHOT_DIR: str = os.getenv("SCREENSHOT_DIR", "./data/screenshots")
     REPORT_DIR: str = os.getenv("REPORT_DIR", "./data/reports")
+    # 截图瘦身配置（定时任务/批量执行磁盘防护）：
+    # SCREENSHOT_MODE: full=每步截图(PNG) / failure_only=仅失败步骤截图(JPEG压缩,默认) / off=不截图
+    SCREENSHOT_MODE: str = os.getenv("SCREENSHOT_MODE", "failure_only")
+    # SCREENSHOT_QUALITY: JPEG 压缩质量（仅 failure_only 模式生效，建议 40~80）
+    SCREENSHOT_QUALITY: int = int(os.getenv("SCREENSHOT_QUALITY", "60"))
+    # SCREENSHOT_MAX_AGE_DAYS: 截图保留天数，执行启动时自动清理超期文件（0=不清理）
+    SCREENSHOT_MAX_AGE_DAYS: int = int(os.getenv("SCREENSHOT_MAX_AGE_DAYS", "14"))
 
     # 设备模拟配置
     DEVICE_SIM_MAX_INSTANCES: int = int(os.getenv("DEVICE_SIM_MAX_INSTANCES", "50"))

backend/app/executors/playwright_executor.py

@@ -101,7 +101,9 @@ class PlaywrightExecutor:
     Playwright 测试执行器（同步版本）
 
     负责执行测试用例中的步骤，支持多种浏览器操作和断言验证。
-    每个步骤自动截图，支持实时回调通知。
+    步骤级截图按 screenshot_mode 三态控制：
+    full=每步PNG / failure_only=仅失败步骤JPEG压缩（默认，磁盘防护） / off=不截图。
+    支持实时回调通知。
 
     使用同步 Playwright API 以避免 Windows 上 asyncio 子进程的限制。
 
@@ -150,6 +152,8 @@ class PlaywrightExecutor:
             config (Optional[dict]): 执行配置
                 - timeout (int): 超时时间（毫秒），默认30000
                 - screenshot (bool): 是否截图，默认True
+                - screenshot_mode (str): 截图模式 full/failure_only/off，默认取全局配置（failure_only）
+                - screenshot_quality (int): JPEG 压缩质量（仅 failure_only 模式），默认60
                 - headless (bool): 是否无头模式，默认False
                 - screenshot_dir (str): 截图目录
                 - retry (int): 失败重试次数，默认0
@@ -166,6 +170,19 @@ class PlaywrightExecutor:
         ).rstrip("/") or self.LOGIN_CONFIG["base_url"]
         self.timeout: int = self.config.get("timeout", 30000)
         self.screenshot_enabled: bool = self.config.get("screenshot", True)
+        # 截图瘦身（定时任务/批量执行磁盘防护）：
+        # screenshot_mode: full=每步PNG / failure_only=仅失败步骤JPEG压缩(默认) / off=不截图
+        # 默认值取自全局配置 settings.SCREENSHOT_MODE（环境变量 SCREENSHOT_MODE 可覆盖）
+        self.screenshot_mode: str = str(
+            self.config.get("screenshot_mode", settings.SCREENSHOT_MODE)
+            or "failure_only"
+        ).strip().lower()
+        # JPEG 压缩质量（仅 failure_only 模式生效；Playwright 的 quality 参数只对 JPEG 有效）
+        self.screenshot_quality: int = int(
+            self.config.get("screenshot_quality", settings.SCREENSHOT_QUALITY)
+        )
+        # 过期截图自动清理标记（start() 时执行一次，防重复）
+        self._screenshot_cleanup_done: bool = False
         # headless 默认值取自全局配置 settings.PLAYWRIGHT_HEADLESS（由环境变量 PLAYWRIGHT_HEADLESS 控制）。
         # 容器内无 X Server，必须 headless=True；原默认 False（headed）会导致容器内启动崩溃。
         self.headless: bool = self.config.get("headless", settings.PLAYWRIGHT_HEADLESS)
@@ -238,6 +255,9 @@ class PlaywrightExecutor:
             RuntimeError: 当启动失败时抛出
         """
         try:
+            # 过期截图自动兜底清理（SCREENSHOT_MAX_AGE_DAYS，容错不中断启动）
+            self._cleanup_expired_screenshots()
+
             # 在 Windows 上，当从 run_in_executor 调用时，需要绕过 asyncio 检测
             # Playwright sync_playwright().start() 内部会检测是否有事件循环
             # 关键：必须确保当前线程没有关联任何 asyncio 事件循环
@@ -306,6 +326,72 @@ class PlaywrightExecutor:
             logger.error(f"执行器启动失败: {str(e)}")
             raise RuntimeError(f"执行器启动失败: {str(e)}")
 
+    def _cleanup_expired_screenshots(self) -> None:
+        """
+        清理超过 SCREENSHOT_MAX_AGE_DAYS 的历史截图（磁盘兜底防护）
+
+        在 start() 时执行一次（_screenshot_cleanup_done 防同实例重复执行）。
+        纯防御性逻辑：目录不存在/无权限/单文件删除失败均只告警，绝不中断执行启动。
+
+        配置 SCREENSHOT_MAX_AGE_DAYS <= 0 时跳过（关闭自动清理）。
+        """
+        if self._screenshot_cleanup_done:
+            return
+        self._screenshot_cleanup_done = True
+
+        max_age_days = int(getattr(settings, "SCREENSHOT_MAX_AGE_DAYS", 14) or 0)
+        if max_age_days <= 0:
+            return
+
+        try:
+            if not os.path.isdir(self.screenshot_dir):
+                return
+            cutoff = time.time() - max_age_days * 86400
+            removed = 0
+            for name in os.listdir(self.screenshot_dir):
+                path = os.path.join(self.screenshot_dir, name)
+                try:
+                    if not os.path.isfile(path):
+                        continue
+                    if os.path.getmtime(path) >= cutoff:
+                        continue
+                    os.remove(path)
+                    removed += 1
+                except Exception as e:
+                    logger.debug(f"过期截图清理跳过 {name}: {e}")
+            if removed:
+                logger.info(
+                    f"过期截图自动清理完成: 删除 {removed} 个文件"
+                    f"（保留 {max_age_days} 天内）"
+                )
+        except Exception as e:
+            logger.warning(f"过期截图清理失败（忽略，不影响执行）: {e}")
+
+    def _should_capture_step(self, step_result) -> bool:
+        """
+        三态截图模式判定（screenshot_mode）
+
+        - off: 不截图
+        - failure_only: 仅步骤最终失败时截图（cancelled/skipped 等一律不截）
+        - full: 每步截图（旧行为）
+        screenshot_enabled=False（用例级总开关）时任何模式都不截图。
+
+        Args:
+            step_result: 当前步骤结果对象（含 status 字段）
+
+        Returns:
+            bool: 是否需要截图
+        """
+        if not self.screenshot_enabled:
+            return False
+        mode = (self.screenshot_mode or "failure_only").strip().lower()
+        if mode == "off":
+            return False
+        if mode == "full":
+            return True
+        # failure_only（默认）：仅失败步骤截图
+        return step_result.status == "failed"
+
     def do_login(self) -> bool:
         """
         执行自动登录
@@ -1988,12 +2074,24 @@ class PlaywrightExecutor:
         if step_result.status == "passed":
             self._post_action_wait(action, wait_after)
 
-        # 截图（无论成功与否，如果开启了截图）
-        if self.screenshot_enabled:
+        # 截图（按 screenshot_mode 三态判定：full=每步PNG / failure_only=仅失败步骤JPEG / off=不截图）
+        # 注意：Playwright 的 quality 参数仅对 JPEG 生效，故 failure_only 用 .jpg
+        if self.screenshot_enabled and self._should_capture_step(step_result):
             try:
-                screenshot_name = f"step_{step_result.order}_{datetime.now().strftime('%H%M%S')}.png"
-                screenshot_path = os.path.join(self.screenshot_dir, screenshot_name)
-                self._page.screenshot(path=screenshot_path)
+                if self.screenshot_mode == "failure_only":
+                    screenshot_name = (
+                        f"step_{step_result.order}_{datetime.now().strftime('%H%M%S')}.jpg"
+                    )
+                    screenshot_path = os.path.join(self.screenshot_dir, screenshot_name)
+                    self._page.screenshot(
+                        path=screenshot_path,
+                        type="jpeg",
+                        quality=self.screenshot_quality,
+                    )
+                else:  # full 模式保持无损 PNG（调试场景）
+                    screenshot_name = f"step_{step_result.order}_{datetime.now().strftime('%H%M%S')}.png"
+                    screenshot_path = os.path.join(self.screenshot_dir, screenshot_name)
+                    self._page.screenshot(path=screenshot_path)
                 step_result.screenshot = screenshot_path
             except Exception as e:
                 logger.error(f"✗ 截图失败: {str(e)}")

backend/app/services/scheduler_service.py

@@ -346,7 +346,12 @@ async def run_task_once(task_id: str, manual: bool = False) -> Optional[str]:
                 trigger_type="scheduled",
                 trigger_by=task.name,
                 environment="default",
-                config={"auto_report": task.auto_report},
+                config={
+                    "auto_report": task.auto_report,
+                    # 定时任务全量执行默认瘦身：仅失败步骤截图（JPEG），避免磁盘被
+                    # 全量 PNG 截图撑满；需要全量调试时手动执行并显式传 screenshot_mode=full
+                    "screenshot_mode": "failure_only",
+                },
             )
             await db.commit()
             logger.info(

backend/tests/test_screenshot_mode.py

+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
+"""
+模块名称：test_screenshot_mode.py
+模块描述：截图瘦身（三态模式）单元测试
+
+覆盖：
+- PlaywrightExecutor 三态截图模式（full/failure_only/off）判定逻辑
+- 用例级 screenshot 总开关优先
+- _cleanup_expired_screenshots 过期截图自动清理
+- 定时任务触发时注入 screenshot_mode=failure_only（磁盘防护）
+
+作者：czj
+创建日期：2026-09-07
+"""
+
+import os
+import time
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+
+from app.config import settings
+from app.executors.playwright_executor import PlaywrightExecutor, StepResult
+
+
+def _step(status: str, order: int = 1) -> StepResult:
+    return StepResult(order=order, action="click", status=status)
+
+
+class TestScreenshotModeDefaults:
+    """默认配置应为 failure_only + quality 60（磁盘防护默认开启）"""
+
+    def test_default_mode_from_settings(self):
+        executor = PlaywrightExecutor()
+        assert executor.screenshot_mode == settings.SCREENSHOT_MODE
+        assert executor.screenshot_mode == "failure_only"
+
+    def test_default_quality_from_settings(self):
+        executor = PlaywrightExecutor()
+        assert executor.screenshot_quality == settings.SCREENSHOT_QUALITY
+
+    def test_mode_normalized(self):
+        """传入大写/带空白模式应归一化，保证格式分支与判定分支一致"""
+        executor = PlaywrightExecutor({"screenshot_mode": "  FAILURE_ONLY "})
+        assert executor.screenshot_mode == "failure_only"
+        assert executor._should_capture_step(_step("failed")) is True
+
+    def test_config_override(self):
+        executor = PlaywrightExecutor(
+            {"screenshot_mode": "off", "screenshot_quality": 80}
+        )
+        assert executor.screenshot_mode == "off"
+        assert executor.screenshot_quality == 80
+
+
+class TestShouldCaptureStep:
+    """三态判定矩阵：off 不截 / failure_only 仅 failed / full 全截"""
+
+    def setup_method(self):
+        self.passed = _step("passed", 1)
+        self.failed = _step("failed", 2)
+        self.skipped = _step("skipped", 3)
+
+    def test_failure_only_default(self):
+        e = PlaywrightExecutor()
+        assert e._should_capture_step(self.failed) is True
+        assert e._should_capture_step(self.passed) is False
+        assert e._should_capture_step(self.skipped) is False
+
+    def test_full_captures_everything(self):
+        e = PlaywrightExecutor({"screenshot_mode": "full"})
+        assert e._should_capture_step(self.failed) is True
+        assert e._should_capture_step(self.passed) is True
+        assert e._should_capture_step(self.skipped) is True
+
+    def test_off_captures_nothing(self):
+        e = PlaywrightExecutor({"screenshot_mode": "off"})
+        assert e._should_capture_step(self.failed) is False
+        assert e._should_capture_step(self.passed) is False
+
+    def test_screenshot_disabled_overrides_mode(self):
+        """用例级 screenshot=False 总开关优先，任何模式都不截图"""
+        e = PlaywrightExecutor({"screenshot": False, "screenshot_mode": "full"})
+        assert e._should_capture_step(self.failed) is False
+
+    def test_unknown_mode_falls_back_to_failure_only(self):
+        e = PlaywrightExecutor({"screenshot_mode": "bogus"})
+        assert e._should_capture_step(self.failed) is True
+        assert e._should_capture_step(self.passed) is False
+
+
+class TestCleanupExpiredScreenshots:
+    """过期截图自动清理（SCREENSHOT_MAX_AGE_DAYS 兜底）"""
+
+    def _make_executor(self, screenshot_dir):
+        return PlaywrightExecutor({"screenshot_dir": str(screenshot_dir)})
+
+    def test_deletes_expired_keeps_fresh(self, tmp_path):
+        old_file = tmp_path / "old.png"
+        new_file = tmp_path / "new.png"
+        old_file.write_bytes(b"x")
+        new_file.write_bytes(b"y")
+        # 人为把 old.png 的 mtime 拨到 20 天前
+        two_days_ago = time.time() - 20 * 86400
+        os.utime(old_file, (two_days_ago, two_days_ago))
+
+        executor = self._make_executor(tmp_path)
+        executor._cleanup_expired_screenshots()
+
+        assert not old_file.exists(), "超过 SCREENSHOT_MAX_AGE_DAYS 的截图应被删除"
+        assert new_file.exists(), "未过期的截图应保留"
+
+    def test_only_removes_files_not_dirs(self, tmp_path):
+        sub_dir = tmp_path / "sub"
+        sub_dir.mkdir()
+        old_dir = tmp_path / "olddir"
+        old_dir.mkdir()
+        ancient = time.time() - 30 * 86400
+        os.utime(old_dir, (ancient, ancient))
+
+        executor = self._make_executor(tmp_path)
+        executor._cleanup_expired_screenshots()
+
+        assert sub_dir.exists()
+        assert old_dir.exists(), "目录不应被误删"
+
+    def test_missing_dir_is_silent(self, tmp_path):
+        executor = self._make_executor(tmp_path / "no_such_dir")
+        # 不抛异常即通过
+        executor._cleanup_expired_screenshots()
+
+    def test_zero_max_age_disables_cleanup(self, tmp_path, monkeypatch):
+        old_file = tmp_path / "old.png"
+        old_file.write_bytes(b"x")
+        ancient = time.time() - 30 * 86400
+        os.utime(old_file, (ancient, ancient))
+
+        monkeypatch.setattr(settings, "SCREENSHOT_MAX_AGE_DAYS", 0)
+        executor = self._make_executor(tmp_path)
+        executor._cleanup_expired_screenshots()
+
+        assert old_file.exists(), "SCREENSHOT_MAX_AGE_DAYS<=0 应关闭自动清理"
+
+    def test_runs_once_per_executor(self, tmp_path):
+        """_screenshot_cleanup_done 标记防同实例重复执行"""
+        executor = self._make_executor(tmp_path)
+        executor._cleanup_expired_screenshots()
+        assert executor._screenshot_cleanup_done is True
+        # 第二次调用直接短路（无副作用，不抛异常）
+        executor._cleanup_expired_screenshots()
+
+
+class TestScheduledTaskInjection:
+    """定时任务触发时必须注入 screenshot_mode=failure_only"""
+
+    def _make_task(self):
+        from app.models.scheduled_task import ScheduledTask
+
+        task = ScheduledTask()
+        task.id = "task_test_001"
+        task.name = "每日定时自动化测试"
+        task.case_type = "ui"
+        task.module_ids = ["module_x"]
+        task.enabled = True
+        task.auto_report = False
+        task.schedule_type = "interval"
+        task.interval_unit = "hours"
+        task.interval_value = 6
+        return task
+
+    @patch("app.services.scheduler_service.ExecutionService")
+    @patch(
+        "app.services.scheduler_service.is_execution_running",
+        return_value=(False, None),
+    )
+    @patch("app.services.scheduler_service.collect_module_cases", new_callable=AsyncMock)
+    async def test_ui_task_injects_failure_only(
+        self, mock_collect, mock_is_running, mock_exec_cls
+    ):
+        from app.services import scheduler_service
+
+        task = self._make_task()
+        mock_collect.return_value = [MagicMock(id="case_1")]
+
+        mock_svc = MagicMock()
+        mock_svc.create_execution = AsyncMock(
+            return_value=MagicMock(id="exec_test_001")
+        )
+        mock_svc.run_execution = AsyncMock(return_value=None)
+        mock_exec_cls.return_value = mock_svc
+
+        # 最小 DB 假件：第一次 execute 返回任务，第二次（DB 级运行守卫）返回 None
+        task_result = MagicMock()
+        task_result.scalar_one_or_none = MagicMock(return_value=task)
+        none_result = MagicMock()
+        none_result.scalar_one_or_none = MagicMock(return_value=None)
+        fake_db = MagicMock()
+        fake_db.execute = AsyncMock(side_effect=[task_result, none_result])
+        fake_db.flush = AsyncMock()
+        fake_db.commit = AsyncMock()
+
+        # 会话工厂 → 返回假件的异步上下文管理器
+        session_ctx = MagicMock()
+        session_ctx.__aenter__ = AsyncMock(return_value=fake_db)
+        session_ctx.__aexit__ = AsyncMock(return_value=False)
+
+        with patch.object(
+            scheduler_service, "async_session_maker", return_value=session_ctx
+        ):
+            exec_id = await scheduler_service.run_task_once(task.id, manual=True)
+
+        assert exec_id == "exec_test_001"
+        mock_svc.create_execution.assert_awaited_once()
+        kwargs = mock_svc.create_execution.await_args.kwargs
+        assert kwargs["config"].get("screenshot_mode") == "failure_only", (
+            "定时任务创建执行时必须注入 screenshot_mode=failure_only"
+        )
+        assert kwargs["trigger_type"] == "scheduled"