chore: 更新2026-08-31股票估值每日台账

Cai

2026-08-31 3fdf9063cb61ff10d9f54aeed6eaf0eca1777844

 dev/project-dev/test/test_extract_ppt_slides.py

@@ -1,8 +1,11 @@
from __future__ import annotations

import csv
from datetime import datetime, timedelta, timezone
import hashlib
import importlib.util
import io
import json
import os
from pathlib import Path
import re
@@ -10,6 +13,7 @@
import subprocess
import sys
import tempfile
import textwrap
import unittest
from unittest import mock
import zlib
@@ -26,21 +30,63 @@
SPEC.loader.exec_module(slides)


def feature(index: int, array: np.ndarray, score: float = 0.2, sharpness: float | None = None):
def palette(marker_bin: int | None = None, coverage: float = 0.20):
    histogram = np.full(slides.PALETTE_BIN_COUNT, 1 / slides.PALETTE_BIN_COUNT, np.float32)
    if marker_bin is not None:
        histogram.fill((1.0 - coverage) / (slides.PALETTE_BIN_COUNT - 1))
        histogram[marker_bin] = coverage
    histogram /= float(np.sum(histogram, dtype=np.float64))
    return histogram


def feature(
    index: int,
    array: np.ndarray,
    score: float = 0.2,
    sharpness: float | None = None,
    page_palette: np.ndarray | None = None,
):
    value = int(np.mean(array))
    phash = int.from_bytes(hashlib.sha256(array.tobytes()).digest()[:8], "big")
    return slides.FrameFeature(
        sample_index=index,
        relative_seconds=float(index),
        gray=array.astype(np.uint8).copy(),
        palette_histogram=(palette() if page_palette is None else page_palette.copy()),
        perceptual_hash=phash,
        content_score=score,
        sharpness=float(value if sharpness is None else sharpness),
    )


def candidate(timestamp: float, array: np.ndarray, score: float, phash: int = 0):
    return slides.PageCandidate(timestamp, array.astype(np.uint8).copy(), phash, score, 1.0)
def candidate(
    timestamp: float,
    array: np.ndarray,
    score: float,
    phash: int = 0,
    page_palette: np.ndarray | None = None,
):
    return slides.PageCandidate(
        timestamp,
        array.astype(np.uint8).copy(),
        palette() if page_palette is None else page_palette.copy(),
        phash,
        score,
        1.0,
    )


def logical_page(order: int, page_palette: np.ndarray | None = None):
    page_feature = np.full((8, 8), order % 256, np.uint8)
    return slides.LogicalPage(
        logical_anchor_feature=page_feature.copy(),
        current_feature=page_feature,
        current_palette=palette() if page_palette is None else page_palette.copy(),
        current_hash=order,
        current_timestamp=float(order),
        current_score=0.2,
        first_seen_order=order,
    )


def sha256(path: Path) -> str:
@@ -148,20 +194,915 @@
        recent = np.full((8, 8), 4, np.uint8)
        classifier.consume(candidate(1, historical, 0.20, 0))
        classifier.pages.append(
            slides.LogicalPage(recent.copy(), recent.copy(), 0, 2.0, 0.20, 1)
            slides.LogicalPage(recent.copy(), recent.copy(), palette(), 0, 2.0, 0.20, 1)
        )
        action = classifier.consume(candidate(3, historical, 0.22, 0))
        self.assertEqual(action, "animation_updated")
        self.assertEqual(len(classifier.pages), 2)


class MainSequenceTests(unittest.TestCase):
    def test_palette_histogram_distinguishes_dark_blue_from_black(self):
        image = np.array(
            [
                [[16, 16, 48], [16, 16, 48]],
                [[16, 16, 16], [16, 16, 16]],
            ],
            dtype=np.uint8,
        )
        histogram = slides._palette_histogram(image)
        self.assertEqual(histogram.shape, (512,))
        self.assertTrue(np.all(np.isfinite(histogram)))
        self.assertAlmostEqual(float(np.sum(histogram)), 1.0, places=6)
        self.assertEqual(float(histogram[1]), 0.5)
        self.assertEqual(float(histogram[0]), 0.5)

    def test_selects_old_pages_002_through_038_and_keeps_two_gaps(self):
        marker_pages = {2, *range(4, 9), *range(10, 39)}
        pages = [
            logical_page(index - 1, palette(1) if index in marker_pages else palette())
            for index in range(1, 54)
        ]
        selected = slides.select_main_sequence(pages)
        self.assertEqual(selected, pages[1:38])
        self.assertIs(selected[1], pages[2])  # old slide_003: first one-page gap
        self.assertIs(selected[7], pages[8])  # old slide_009: second one-page gap
        self.assertEqual([page.first_seen_order for page in selected], list(range(1, 38)))

    def test_conservative_gates_keep_all_candidates(self):
        cases = {
            "markers": [palette(1) if index < 7 else palette() for index in range(10)],
            "share": [palette(1) if 3 <= index <= 10 else palette() for index in range(20)],
            "density": [
                palette(1) if index in {0, 2, 4, 6, 7, 8, 9, 10} else palette()
                for index in range(11)
            ],
        }
        for name, histograms in cases.items():
            with self.subTest(name=name):
                pages = [logical_page(index, value) for index, value in enumerate(histograms)]
                selected = slides.select_main_sequence(pages)
                self.assertEqual(selected, pages)
                self.assertTrue(all(left is right for left, right in zip(selected, pages)))

    def test_all_theme_pages_are_preserved(self):
        pages = [logical_page(index, palette(7)) for index in range(12)]
        selected = slides.select_main_sequence(pages)
        self.assertEqual(selected, pages)
        self.assertTrue(all(left is right for left, right in zip(selected, pages)))

    def test_invalid_palette_is_explicit_failure(self):
        invalid = [
            np.zeros(5, np.float32),
            np.full(512, np.nan, np.float32),
            np.zeros(512, np.float32),
        ]
        for value in invalid:
            with self.subTest(shape=value.shape, finite=bool(np.all(np.isfinite(value)))):
                with self.assertRaises(slides.SlideExtractionError):
                    slides.select_main_sequence([logical_page(0, value)])


class ProcessAndCommandTests(unittest.TestCase):
    def test_command_freezes_time_mapping(self):
        scan = slides._scan_command("ffmpeg", Path("x.mp4"), 320, 180)
        self.assertIn("setpts=PTS-STARTPTS,fps=fps=1:start_time=0,scale=320:180:flags=bilinear", scan)
        input_index = scan.index("-i")
        self.assertLess(scan.index("-threads"), input_index)
        self.assertEqual(scan[scan.index("-threads") + 1], str(slides.MEDIA_DECODE_THREADS))
        self.assertEqual(scan[scan.index("-filter_threads") + 1], str(slides.MEDIA_FILTER_THREADS))
        extract = slides._extract_command("ffmpeg", Path("x.mp4"), 12.0, Path("x.png"))
        self.assertEqual(extract[extract.index("-ss") + 1], "12.000")
        self.assertLess(extract.index("-threads"), extract.index("-i"))
        self.assertEqual(feature(9, np.zeros((2, 2), np.uint8)).relative_seconds, 9.0)

    def test_hardware_commands_place_decoder_and_threads_before_input(self):
        for codec, decoder in (("h264", "h264_cuvid"), ("hevc", "hevc_cuvid")):
            with self.subTest(codec=codec):
                self.assertEqual(slides._decoder_for_codec(codec, "yuv420p"), decoder)
                command = slides._scan_command(
                    "ffmpeg", Path(f"{codec}.mp4"), 320, 180, decoder
                )
                input_index = command.index("-i")
                for option in ("-hwaccel", "-hwaccel_output_format", "-c:v", "-threads"):
                    self.assertLess(command.index(option), input_index)
                self.assertEqual(command[command.index("-hwaccel") + 1], "cuda")
                self.assertEqual(command[command.index("-c:v") + 1], decoder)
                self.assertEqual(
                    command[command.index("-threads") + 1],
                    str(slides.MEDIA_DECODE_THREADS),
                )
                self.assertIn("hwdownload,format=nv12", command[command.index("-vf") + 1])
        self.assertEqual(slides._decoder_for_codec("h264", "gbrp"), "h264_cuvid")
        self.assertEqual(slides._decoder_for_codec("hevc", "p010le"), "hevc_cuvid")
        self.assertIsNone(slides._decoder_for_codec("vp9", "yuv420p"))

    def test_hardware_external_failures_have_typed_boundary(self):
        class FakeStdout:
            def __init__(self, payloads):
                self.payloads = list(payloads)

            def read(self, size):
                return self.payloads.pop(0) if self.payloads else b""

            def close(self):
                return None

        class FakeProcess:
            def __init__(self, payloads, code):
                self.stdout = FakeStdout(payloads)
                self.code = code

            def wait(self, timeout=None):
                return self.code

            def poll(self):
                return self.code

            def terminate(self):
                return None

            def kill(self):
                return None

        info = slides.VideoInfo(320, 180, 10.0, "h264", "yuv420p")
        cases = (
            ("nonzero", FakeProcess([], 7), "exit=7"),
            ("zero", FakeProcess([], 0), "未输出任何完整"),
            ("partial", FakeProcess([b"x", b""], 0), "输出半帧"),
        )
        with tempfile.TemporaryDirectory() as raw, mock.patch.object(
            slides, "_which_or_error", return_value="ffmpeg"
        ):
            root = Path(raw)
            with mock.patch.object(slides.subprocess, "Popen", side_effect=OSError("start denied")):
                with self.assertRaisesRegex(slides.HardwareDecodePathError, "启动失败"):
                    slides._scan_video_once(Path("x.mp4"), root, info, decoder="h264_cuvid")
            for name, process, pattern in cases:
                with self.subTest(name=name), mock.patch.object(
                    slides.subprocess, "Popen", return_value=process
                ):
                    with self.assertRaisesRegex(slides.HardwareDecodePathError, pattern):
                        slides._scan_video_once(
                            Path("x.mp4"), root, info, decoder="h264_cuvid"
                        )

    def test_hardware_cleanup_failure_does_not_trigger_cpu_fallback(self):
        info = slides.VideoInfo(320, 180, 10.0, "h264", "yuv420p")
        hardware_error = slides.HardwareDecodePathError("hardware-root")
        slides._attach_cleanup_note(hardware_error, "cleanup-failed")
        with tempfile.TemporaryDirectory() as raw, mock.patch.object(
            slides, "_scan_video_once", side_effect=hardware_error
        ) as scan:
            with self.assertRaises(slides.HardwareDecodePathError) as caught:
                slides.scan_video(Path("x.mp4"), Path(raw), info)
            self.assertIs(caught.exception, hardware_error)
            self.assertEqual(scan.call_count, 1)
            self.assertIn("cleanup-failed", "\n".join(caught.exception.__notes__))

    def test_active_errors_keep_identity_when_process_cleanup_fails(self):
        frame_bytes = 320 * 180 * 3

        class Stream:
            def __init__(self, error=None, partial=False):
                self.error = error
                self.partial = partial
                self.calls = 0

            def read(self, size):
                self.calls += 1
                if self.error is not None:
                    raise self.error
                if self.partial:
                    return b"x" if self.calls == 1 else b""
                return b"\0" * frame_bytes if self.calls == 1 else b""

            def close(self):
                return None

        class FakeProcess:
            def __init__(self, stream):
                self.stdout = stream

            def wait(self, timeout=None):
                return 0

            def poll(self):
                return None

            def terminate(self):
                return None

        cleanup = OSError("cleanup-failed")
        info = slides.VideoInfo(320, 180, 10.0, "h264", "yuv420p")
        cases = (
            ("business", None, ValueError("business-root")),
            ("cpu", None, slides.SlideExtractionError("cpu-root")),
            ("base", KeyboardInterrupt(), None),
        )
        for name, read_error, feature_error in cases:
            root_error = read_error or feature_error
            stream = Stream(error=read_error)
            patches = [
                mock.patch.object(slides, "_which_or_error", return_value="ffmpeg"),
                mock.patch.object(slides.subprocess, "Popen", return_value=FakeProcess(stream)),
                mock.patch.object(slides, "_attempt_reap", return_value=cleanup),
            ]
            if feature_error is not None:
                patches.append(mock.patch.object(slides, "_gray_feature", side_effect=feature_error))
            with self.subTest(name=name), tempfile.TemporaryDirectory() as raw:
                for item in patches:
                    item.start()
                try:
                    with self.assertRaises(type(root_error)) as caught:
                        slides._scan_video_once(
                            Path("x.mp4"), Path(raw), info, decoder=None
                        )
                    self.assertIs(caught.exception, root_error)
                    self.assertIn("cleanup-failed", "\n".join(caught.exception.__notes__))
                finally:
                    for item in reversed(patches):
                        item.stop()

        partial_process = FakeProcess(Stream(partial=True))
        with tempfile.TemporaryDirectory() as raw, mock.patch.object(
            slides, "_which_or_error", return_value="ffmpeg"
        ), mock.patch.object(slides.subprocess, "Popen", return_value=partial_process), mock.patch.object(
            slides, "_attempt_reap", return_value=cleanup
        ):
            with self.assertRaises(slides.HardwareDecodePathError) as caught:
                slides._scan_video_once(
                    Path("x.mp4"), Path(raw), info, decoder="h264_cuvid"
                )
            hardware_error = caught.exception
            self.assertTrue(getattr(hardware_error, "_media_cleanup_failed", False))
            self.assertIn("cleanup-failed", "\n".join(hardware_error.__notes__))
        with tempfile.TemporaryDirectory() as raw, mock.patch.object(
            slides, "_scan_video_once", side_effect=hardware_error
        ) as scan:
            with self.assertRaises(slides.HardwareDecodePathError) as caught_again:
                slides.scan_video(Path("x.mp4"), Path(raw), info)
            self.assertIs(caught_again.exception, hardware_error)
            self.assertEqual(scan.call_count, 1)

    def test_product_popen_uses_below_normal_creation_flags(self):
        class FakeProcess:
            stdout = None

            def wait(self, timeout=None):
                return 0

            def poll(self):
                return 0

        with tempfile.TemporaryDirectory() as raw, mock.patch.object(
            slides, "_subprocess_creation_flags", return_value=12345
        ), mock.patch.object(slides.subprocess, "Popen", return_value=FakeProcess()) as popen:
            slides._run_file_command(["tool"], Path(raw) / "stderr.log", "stage")
        self.assertEqual(popen.call_args.kwargs["creationflags"], 12345)

    def test_fallback_only_handles_hardware_error_and_never_retries_cpu(self):
        info = slides.VideoInfo(320, 180, 10.0, "h264", "yuv420p")
        page = logical_page(0)
        success = slides.ScanResult((page,), 10, 2, (("new_page", 1),), None)
        hardware_error = slides.HardwareDecodePathError("decoder failed")
        with tempfile.TemporaryDirectory() as raw, mock.patch.object(
            slides, "_scan_video_once", side_effect=[hardware_error, success]
        ) as scan, mock.patch("sys.stderr", new_callable=io.StringIO) as stderr:
            self.assertEqual(slides.scan_video(Path("x.mp4"), Path(raw), info), [page])
            self.assertEqual([call.kwargs["decoder"] for call in scan.call_args_list], [
                "h264_cuvid", None
            ])
            self.assertIn("decoder failed", stderr.getvalue())

        cpu_error = slides.SlideExtractionError("cpu root cause")
        with tempfile.TemporaryDirectory() as raw, mock.patch.object(
            slides, "_scan_video_once", side_effect=[hardware_error, cpu_error]
        ) as scan:
            with self.assertRaises(slides.SlideExtractionError) as caught:
                slides.scan_video(Path("x.mp4"), Path(raw), info)
            self.assertIs(caught.exception, cpu_error)
            self.assertEqual(scan.call_count, 2)

    def test_algorithm_empty_and_baseexception_never_fall_back(self):
        info = slides.VideoInfo(320, 180, 10.0, "h264", "yuv420p")
        empty = slides.ScanResult((), 10, 0, (), "h264_cuvid")
        with tempfile.TemporaryDirectory() as raw, mock.patch.object(
            slides, "_scan_video_once", return_value=empty
        ) as scan:
            with self.assertRaisesRegex(slides.SlideExtractionError, "未发现"):
                slides.scan_video(Path("x.mp4"), Path(raw), info)
            scan.assert_called_once()
        for error in (RuntimeError("classifier failed"), KeyboardInterrupt(), SystemExit(9)):
            with self.subTest(error=type(error).__name__), tempfile.TemporaryDirectory() as raw, mock.patch.object(
                slides, "_scan_video_once", side_effect=error
            ) as scan:
                with self.assertRaises(type(error)) as caught:
                    slides.scan_video(Path("x.mp4"), Path(raw), info)
                self.assertIs(caught.exception, error)
                scan.assert_called_once()

    def test_algorithm_exception_inside_hardware_scan_preserves_original_type(self):
        frame_bytes = 320 * 180 * 3

        class FullFrameStream:
            def __init__(self):
                self.done = False

            def read(self, size):
                if self.done:
                    return b""
                self.done = True
                return b"\0" * frame_bytes

            def close(self):
                return None

        class FakeProcess:
            stdout = FullFrameStream()

            def wait(self, timeout=None):
                return 0

            def poll(self):
                return 0

        root_error = ValueError("feature root cause")
        info = slides.VideoInfo(320, 180, 10.0, "h264", "yuv420p")
        with tempfile.TemporaryDirectory() as raw, mock.patch.object(
            slides, "_which_or_error", return_value="ffmpeg"
        ), mock.patch.object(slides.subprocess, "Popen", return_value=FakeProcess()), mock.patch.object(
            slides, "_gray_feature", side_effect=root_error
        ):
            with self.assertRaises(ValueError) as caught:
                slides._scan_video_once(
                    Path("x.mp4"), Path(raw), info, decoder="h264_cuvid"
                )
            self.assertIs(caught.exception, root_error)

    @unittest.skipUnless(shutil.which("powershell"), "Windows PowerShell is required")
    def test_resource_sampler_parser_rejects_cpu_boundaries_and_empty_windows(self):
        sampler = Path(__file__).resolve().parent / "monitor_media_host_responsiveness.ps1"

        fields = (
            "timestamp_utc", "previous_valid_timestamp_utc", "interval_seconds",
            "counter_warmup", "cpu_valid", "host_cpu_percent",
            "scan_phase_active", "scan_ffmpeg_present", "scan_ffmpeg_exit_observed",
            "scan_ffmpeg_pid", "scan_ffmpeg_command_line", "decoder",
            "logical_processor_count", "task_cpu_percent", "task_working_set_bytes",
            "task_pids", "root_priority", "ffmpeg_priorities", "ffmpeg_command_lines",
            "gpu_query_status", "gpu_query_raw_rows", "gpu_utilization_percent",
            "gpu_decoder_utilization_percent", "gpu_memory_used_mib",
            "nvidia_smi_compute_query_status", "nvidia_smi_compute_raw_rows",
            "nvidia_smi_compute_task_rows",
            "gpu_process_memory_query_status", "gpu_process_memory_instances",
            "task_gpu_dedicated_memory_mib",
        )
        scan_command = (
            "ffmpeg -hwaccel cuda -hwaccel_output_format cuda "
            "-c:v h264_cuvid -threads 4 "
            "-filter_threads 2 -filter_complex_threads 2 -i x.mp4 -map 0:v:0 "
            "-vf hwdownload,format=nv12,fps=fps=1:start_time=0 "
            "-pix_fmt rgb24 -f rawvideo -"
        )

        def make_row(current, previous, interval, cpu, *, warmup, present, exited):
            return {
                "timestamp_utc": current.isoformat().replace("+00:00", "Z"),
                "previous_valid_timestamp_utc": (
                    "" if previous is None else previous.isoformat().replace("+00:00", "Z")
                ),
                "interval_seconds": f"{interval:.6f}",
                "counter_warmup": str(warmup).lower(),
                "cpu_valid": str(not warmup).lower(),
                "host_cpu_percent": "" if warmup else f"{cpu:.6f}",
                "scan_phase_active": "true",
                "scan_ffmpeg_present": str(present).lower(),
                "scan_ffmpeg_exit_observed": str(exited).lower(),
                "scan_ffmpeg_pid": "4242",
                "scan_ffmpeg_command_line": scan_command,
                "decoder": "h264_cuvid",
                "logical_processor_count": "44",
                "task_cpu_percent": "" if warmup else "12.5",
                "task_working_set_bytes": "123456789",
                "task_pids": "101;4242",
                "root_priority": "BelowNormal",
                "ffmpeg_priorities": (
                    "[]" if not present else '[{"pid":4242,"priority":"BelowNormal"}]'
                ),
                "ffmpeg_command_lines": (
                    '[{"pid":4242,"command_line":' + json.dumps(scan_command) + "}]"
                ),
                "gpu_query_status": "ok",
                "gpu_query_raw_rows": '["20, 31, 2048"]',
                "gpu_utilization_percent": "20",
                "gpu_decoder_utilization_percent": "31",
                "gpu_memory_used_mib": "2048",
                "nvidia_smi_compute_query_status": "ok",
                "nvidia_smi_compute_raw_rows": '["4242, N/A"]',
                "nvidia_smi_compute_task_rows": '["4242, N/A"]',
                "gpu_process_memory_query_status": "ok",
                "gpu_process_memory_instances": (
                    '[{"pid":4242,"instance_name":"pid_4242_luid_0",'
                    '"counter_path":"path","dedicated_usage_bytes":1048576}]'
                ),
                "task_gpu_dedicated_memory_mib": "1",
            }

        def analyze_case(root, name, intervals, cpu_values, mutate=None, omit_field=None):
            case = root / name
            case.mkdir()
            samples = case / "samples.csv"
            current = datetime(2026, 8, 4, tzinfo=timezone.utc)
            rows = [make_row(current, None, 0, 0, warmup=True, present=True, exited=False)]
            previous = current
            for interval, cpu in zip(intervals, cpu_values):
                current += timedelta(seconds=interval)
                rows.append(make_row(
                    current, previous, interval, cpu,
                    warmup=False, present=True, exited=False,
                ))
                previous = current
            current += timedelta(seconds=0.5)
            rows.append(make_row(
                current, previous, 0.5, 20.0,
                warmup=False, present=False, exited=True,
            ))
            if mutate is not None:
                mutate(rows)
            output_fields = tuple(field for field in fields if field != omit_field)
            with samples.open("w", newline="", encoding="utf-8") as handle:
                writer = csv.DictWriter(handle, fieldnames=output_fields, extrasaction="ignore")
                writer.writeheader()
                writer.writerows(rows)
            evidence = case / "evidence"
            result = subprocess.run(
                [
                    shutil.which("powershell"), "-NoProfile", "-File", str(sampler),
                    "-AnalyzeSamplesPath", str(samples), "-EvidenceDirectory", str(evidence),
                ],
                capture_output=True,
                text=True,
                timeout=60,
            )
            self.assertEqual(result.returncode, 0, result.stderr + result.stdout)
            return json.loads((evidence / "resource_summary.json").read_text(encoding="utf-8-sig"))

        with tempfile.TemporaryDirectory() as raw:
            root = Path(raw)
            low = analyze_case(root, "low", [0.5] * 22, [84.9] * 22)
            self.assertTrue(low["sampling_valid"])
            self.assertGreater(low["window_count"], 0)
            self.assertTrue(low["cpu_gate_pass"])

            equal = analyze_case(root, "equal", [0.5] * 22, [85.0] * 22)
            self.assertTrue(equal["sampling_valid"])
            self.assertFalse(equal["all_windows_host_cpu_strictly_below_85"])
            self.assertFalse(equal["cpu_gate_pass"])

            exact_ten = analyze_case(
                root, "exact-ten", [0.5] * 21, [85.0] * 20 + [20.0]
            )
            self.assertEqual(exact_ten["maximum_high_cpu_segment_seconds"], 10.0)
            self.assertFalse(exact_ten["cpu_gate_pass"])

            below_ten = analyze_case(
                root, "below-ten", [0.49995] * 20 + [0.5], [85.0] * 20 + [20.0]
            )
            self.assertEqual(below_ten["maximum_high_cpu_segment_seconds"], 9.999)

            sparse = analyze_case(root, "sparse", [1.0] * 12, [20.0] * 12)
            self.assertFalse(sparse["sampling_valid"])
            self.assertTrue(any("window_density_invalid" in item for item in sparse["sampling_problems"]))

            zero_window = analyze_case(root, "zero-window", [0.5] * 8, [20.0] * 8)
            self.assertFalse(zero_window["sampling_valid"])
            self.assertEqual(zero_window["window_count"], 0)

            gap = analyze_case(
                root, "gap", [0.5] * 10 + [1.6] + [0.5] * 10, [20.0] * 21
            )
            self.assertFalse(gap["sampling_valid"])
            self.assertGreater(gap["maximum_observed_interval_seconds"], 1.5)

            reversed_time = analyze_case(
                root, "time-reversal", [0.5] * 12, [20.0] * 12,
                mutate=lambda rows: rows[5].__setitem__("timestamp_utc", rows[3]["timestamp_utc"]),
            )
            self.assertFalse(reversed_time["sampling_valid"])
            self.assertTrue(any(
                "timestamp_not_strictly_increasing" in item
                for item in reversed_time["sampling_problems"]
            ))

            interval_mismatch = analyze_case(
                root, "interval-mismatch", [0.5] * 12, [20.0] * 12,
                mutate=lambda rows: rows[4].__setitem__("interval_seconds", "9.0"),
            )
            self.assertFalse(interval_mismatch["sampling_valid"])
            self.assertTrue(any(
                "interval_mismatch" in item for item in interval_mismatch["sampling_problems"]
            ))

            previous_mismatch = analyze_case(
                root, "previous-mismatch", [0.5] * 12, [20.0] * 12,
                mutate=lambda rows: rows[4].__setitem__(
                    "previous_valid_timestamp_utc", rows[1]["timestamp_utc"]
                ),
            )
            self.assertFalse(previous_mismatch["sampling_valid"])
            self.assertTrue(any(
                "previous_timestamp_mismatch" in item
                for item in previous_mismatch["sampling_problems"]
            ))

            for omitted in (
                "gpu_query_raw_rows", "task_pids", "ffmpeg_command_lines",
                "nvidia_smi_compute_task_rows",
            ):
                with self.subTest(omitted=omitted):
                    missing_field = analyze_case(
                        root, f"missing-{omitted}", [0.5] * 12, [20.0] * 12,
                        omit_field=omitted,
                    )
                    self.assertFalse(missing_field["sampling_valid"])
                    self.assertIn(
                        f"required_column_missing_{omitted}",
                        missing_field["sampling_problems"],
                    )

            first_not_present = analyze_case(
                root, "first-coverage", [0.5] * 12, [20.0] * 12,
                mutate=lambda rows: rows[0].__setitem__("scan_ffmpeg_present", "false"),
            )
            self.assertFalse(first_not_present["sampling_valid"])
            self.assertIn("scan_phase_first_row_not_present", first_not_present["sampling_problems"])

            last_not_exit = analyze_case(
                root, "last-coverage", [0.5] * 12, [20.0] * 12,
                mutate=lambda rows: rows[-1].__setitem__("scan_ffmpeg_exit_observed", "false"),
            )
            self.assertFalse(last_not_exit["sampling_valid"])
            self.assertIn("scan_phase_last_row_not_exit_observation", last_not_exit["sampling_problems"])

            argv_rewrites = {
                "missing-output-format": lambda value: value.replace(
                    "-hwaccel_output_format cuda ", ""
                ),
                "wrong-output-format": lambda value: value.replace(
                    "-hwaccel_output_format cuda", "-hwaccel_output_format qsv"
                ),
                "late-output-format": lambda value: value.replace(
                    "-hwaccel_output_format cuda ", ""
                ).replace("-i x.mp4", "-i x.mp4 -hwaccel_output_format cuda"),
                "missing-hwdownload": lambda value: value.replace(
                    "hwdownload,format=nv12,", ""
                ),
            }
            for name, rewrite in argv_rewrites.items():
                with self.subTest(argv_case=name):
                    invalid_argv = analyze_case(
                        root, f"invalid-{name}", [0.5] * 12, [20.0] * 12,
                        mutate=lambda rows, transform=rewrite: [
                            row.__setitem__(
                                "scan_ffmpeg_command_line",
                                transform(row["scan_ffmpeg_command_line"]),
                            )
                            for row in rows
                        ],
                    )
                    self.assertFalse(invalid_argv["sampling_valid"])
                    self.assertIn("nvdec_scan_argv_invalid", invalid_argv["sampling_problems"])

    @unittest.skipUnless(
        os.name == "nt" and shutil.which("powershell") and shutil.which("nvidia-smi"),
        "Windows PowerShell and nvidia-smi are required",
    )
    def test_resource_sampler_run_rejects_forged_nvdec_and_excludes_page_ffmpeg(self):
        sampler = Path(__file__).resolve().parent / "monitor_media_host_responsiveness.ps1"
        sampler_text = sampler.read_text(encoding="utf-8")
        process_start = sampler_text.index("[void]$Process.Start()")
        self.assertLess(
            sampler_text.index("$HostCpuCounter = New-Object"), process_start
        )
        self.assertLess(sampler_text.index("$HostCpuCounter.NextValue()"), process_start)
        self.assertLess(
            sampler_text.index("$GpuMemoryCategory = New-Object"), process_start
        )
        self.assertLess(sampler_text.index("$GpuMemoryCategory.GetInstanceNames()"), process_start)
        self.assertLess(
            sampler_text.index('$StartInfo.EnvironmentVariables["MBX_MEDIA_SAMPLER_READY"]'),
            process_start,
        )
        with tempfile.TemporaryDirectory() as raw:
            root = Path(raw)
            fake_ffmpeg = root / "ffmpeg.exe"
            shutil.copy2(sys.executable, fake_ffmpeg)
            child = root / "fake_ffmpeg_child.py"
            child.write_text(
                "import sys, time\n"
                "time.sleep(float(sys.argv[sys.argv.index('--sleep') + 1]))\n",
                encoding="utf-8",
            )
            target = root / "stub_media_target.py"
            target.write_text(
                textwrap.dedent(
                    f"""
                    import ctypes
                    import os
                    from pathlib import Path
                    import subprocess
                    import sys
                    import time

                    sys.stdout.reconfigure(encoding="utf-8")
                    sys.stderr.reconfigure(encoding="utf-8")
                    kernel32 = ctypes.WinDLL("kernel32", use_last_error=True)
                    kernel32.SetPriorityClass(kernel32.GetCurrentProcess(), 0x00004000)
                    creationflags = 0x00004000
                    if os.environ.get("MBX_MEDIA_SAMPLER_READY") != "1":
                        raise RuntimeError("sampler-ready handshake missing")
                    fake_ffmpeg = Path({str(fake_ffmpeg)!r})
                    child = Path({str(child)!r})
                    scan = subprocess.Popen([
                        str(fake_ffmpeg), str(child), "--sleep", "7.0",
                        "-hwaccel", "cuda", "-c:v", "h264_cuvid",
                        "-threads", "4", "-filter_threads", "2",
                        "-filter_complex_threads", "2", "-i", "fake.mp4",
                        "-map", "0:v:0", "-vf",
                        "hwdownload,format=nv12,setpts=PTS-STARTPTS,"
                        "fps=fps=1:start_time=0,scale=320:180",
                        "-pix_fmt", "rgb24", "-f", "rawvideo", "-",
                    ], creationflags=creationflags)
                    scan.wait()
                    print("警告：NVDEC 不可用；改用受限 CPU。", file=sys.stderr, flush=True)
                    print(
                        "扫描解码路径：NVDEC/CUDA (h264_cuvid)；解码线程=4；滤镜线程=2",
                        flush=True,
                    )
                    time.sleep(1.5)
                    for index in range(24):
                        churn = subprocess.Popen([
                            str(fake_ffmpeg), str(child), "--sleep", "0.04",
                            "-threads", "4", "-i", "fake.mp4",
                            "-frames:v", "1", "-n", f"race-{{index}}.png",
                        ], creationflags=creationflags)
                        churn.wait()
                    page = subprocess.Popen([
                        str(fake_ffmpeg), str(child), "--sleep", "4.0",
                        "-threads", "4", "-i", "fake.mp4",
                        "-frames:v", "1", "-n", "fake.png",
                    ], creationflags=creationflags)
                    page.wait()
                    """
                ).lstrip(),
                encoding="utf-8",
            )
            source = root / "source.stub"
            source.write_bytes(b"read-only source fingerprint")
            evidence = root / "evidence"
            result = subprocess.run(
                [
                    shutil.which("powershell"), "-NoProfile", "-File", str(sampler),
                    "-PythonPath", sys.executable,
                    "-ScriptPath", str(target),
                    "-VideoPath", str(source),
                    "-OutputPath", str(root / "unused-output"),
                    "-EvidenceDirectory", str(evidence),
                ],
                capture_output=True,
                text=True,
                encoding="utf-8",
                errors="replace",
                env={
                    **os.environ,
                    "PYTHONIOENCODING": "utf-8",
                    "PYTHONUTF8": "1",
                    "MBX_MEDIA_SAMPLER_TEST_CIM_DELAY_MS": "100",
                    "MBX_MEDIA_SAMPLER_TEST_FAIL_SAMPLE_ONCE": "1",
                },
                timeout=90,
            )
            self.assertEqual(result.returncode, 3, result.stderr + result.stdout)
            run_summary = json.loads(
                (evidence / "run_summary.json").read_text(encoding="utf-8-sig")
            )
            self.assertEqual(run_summary["target_exit_code"], 0)
            self.assertFalse(run_summary["resource_acceptance_pass"])
            self.assertTrue(
                run_summary["nvdec_command_log_evidence"],
                json.dumps(
                    {
                        "run_summary": run_summary,
                        "stdout": (evidence / "stdout.log").read_text("utf-8"),
                    },
                    ensure_ascii=False,
                ),
            )
            self.assertFalse(run_summary["nvdec_scan_argv_evidence"])
            self.assertGreater(run_summary["command_line_disappearance_race_count"], 0)
            self.assertEqual(run_summary["injected_sample_failure_count"], 1)
            self.assertEqual(
                len(run_summary["sampling_errors"]), 1, run_summary["sampling_errors"]
            )
            self.assertIn("injected post-scan sample failure", run_summary["sampling_errors"][0])
            self.assertTrue(run_summary["source_unchanged"])
            self.assertIn("NVDEC/CUDA (h264_cuvid)", (evidence / "stdout.log").read_text("utf-8"))
            self.assertIn("改用受限 CPU", (evidence / "stderr.log").read_text("utf-8"))

            with (evidence / "resource_samples.csv").open(
                newline="", encoding="utf-8-sig"
            ) as handle:
                rows = list(csv.DictReader(handle))
            resource_summary = json.loads(
                (evidence / "resource_summary.json").read_text(encoding="utf-8-sig")
            )
            self.assertFalse(resource_summary["sampling_valid"], resource_summary)
            self.assertEqual(
                resource_summary["sampling_problems"], ["nvdec_scan_argv_invalid"]
            )
            self.assertTrue(resource_summary["all_windows_host_cpu_strictly_below_85"])
            self.assertTrue(resource_summary["window_count"] > 0)
            self.assertFalse(any(
                "previous_timestamp" in problem or "interval_mismatch" in problem
                for problem in resource_summary["sampling_problems"]
            ))
            for index, row in enumerate(rows):
                if index == 0:
                    self.assertEqual(row["previous_valid_timestamp_utc"], "")
                    self.assertAlmostEqual(float(row["interval_seconds"]), 0.0, places=6)
                    continue
                current = datetime.fromisoformat(row["timestamp_utc"].replace("Z", "+00:00"))
                previous = datetime.fromisoformat(
                    rows[index - 1]["timestamp_utc"].replace("Z", "+00:00")
                )
                reported_previous = datetime.fromisoformat(
                    row["previous_valid_timestamp_utc"].replace("Z", "+00:00")
                )
                self.assertLess(abs((reported_previous - previous).total_seconds()), 0.001)
                self.assertLess(
                    abs(float(row["interval_seconds"]) - (current - previous).total_seconds()),
                    0.001,
                )
            active_indexes = [
                index for index, row in enumerate(rows)
                if row["scan_phase_active"].lower() == "true"
            ]
            self.assertGreaterEqual(
                len(active_indexes), 2,
                json.dumps({"rows": rows, "run_summary": run_summary}, ensure_ascii=False),
            )
            self.assertEqual(len({rows[index]["scan_ffmpeg_pid"] for index in active_indexes}), 1)
            last_active = active_indexes[-1]
            self.assertEqual(rows[last_active]["scan_ffmpeg_present"].lower(), "false")
            self.assertEqual(rows[last_active]["scan_ffmpeg_exit_observed"].lower(), "true")
            later_rows = rows[last_active + 1 :]
            self.assertTrue(later_rows)
            self.assertTrue(any("-frames:v" in row["ffmpeg_command_lines"] for row in later_rows))
            self.assertTrue(all(row["scan_phase_active"].lower() == "false" for row in later_rows))

    @unittest.skipUnless(shutil.which("ffmpeg") and shutil.which("ffprobe"), "FFmpeg is required")
    def test_h264_h265_nvdec_and_cpu_product_chains_are_equivalent(self):
        ffmpeg = shutil.which("ffmpeg")
        capabilities = subprocess.check_output(
            [ffmpeg, "-hide_banner", "-encoders"], text=True, errors="replace"
        ) + subprocess.check_output(
            [ffmpeg, "-hide_banner", "-decoders"], text=True, errors="replace"
        )
        required = ("libx264", "libx265", "h264_cuvid", "hevc_cuvid")
        missing = [name for name in required if name not in capabilities]
        if missing:
            self.skipTest(f"missing FFmpeg capabilities: {missing}")

        with tempfile.TemporaryDirectory() as raw:
            root = Path(raw)
            frames_dir = root / "frames"
            frames_dir.mkdir()
            size = (320, 180)

            def page(kind, animated=False):
                colors = {"a": (245, 245, 240), "b": (20, 45, 85), "c": (232, 210, 70)}
                image = Image.new("RGB", size, colors[kind])
                draw = ImageDraw.Draw(image)
                draw.rectangle((0, 0, 319, 30), fill=(25, 90, 160))
                draw.rectangle((30, 55, 285, 70), fill=(30, 30, 30))
                if kind != "a" or animated:
                    draw.rectangle((30, 95, 250, 110), fill=(35, 35, 35))
                if animated:
                    draw.rectangle((260, 140, 280, 160), fill=(210, 30, 30))
                return image

            transition_colors = ((100, 100, 100), (120, 25, 130), (20, 150, 90), (180, 65, 20))
            frames = [page("a")] * 3 + [Image.new("RGB", size, transition_colors[0])]
            frames += [page("a", True)] * 3 + [Image.new("RGB", size, transition_colors[1])]
            frames += [page("b")] * 3 + [Image.new("RGB", size, transition_colors[2])]
            frames += [page("a", True)] * 3 + [Image.new("RGB", size, transition_colors[3])]
            frames += [page("c")] * 3
            for index, image in enumerate(frames):
                image.save(frames_dir / f"frame_{index:03d}.png")

            codec_settings = {
                "h264": ("libx264", "h264_cuvid"),
                "hevc": ("libx265", "hevc_cuvid"),
            }
            evidence = []
            for codec, (encoder, decoder) in codec_settings.items():
                source = root / f"sequence-{codec}.mp4"
                command = [
                    ffmpeg, "-hide_banner", "-loglevel", "error", "-framerate", "1",
                    "-i", str(frames_dir / "frame_%03d.png"), "-threads", "4",
                    "-c:v", encoder, "-preset", "ultrafast",
                ]
                if codec == "h264":
                    command.extend(("-qp", "0"))
                else:
                    command.extend(("-x265-params", "lossless=1:pools=4"))
                command.extend(("-pix_fmt", "yuv420p", "-n", str(source)))
                subprocess.run(command, check=True, timeout=120)
                probe_staging = root / f"probe-{codec}"
                probe_staging.mkdir()
                info = slides.probe_video(source, probe_staging)
                self.assertEqual(info.codec_name, codec)
                self.assertEqual(info.pixel_format, "yuv420p")

                hardware_staging = root / f"scan-{codec}-hardware"
                cpu_staging = root / f"scan-{codec}-cpu"
                hardware_staging.mkdir()
                cpu_staging.mkdir()
                hardware = slides._scan_video_once(
                    source, hardware_staging, info, decoder=decoder
                )
                cpu = slides._scan_video_once(source, cpu_staging, info, decoder=None)
                self.assertEqual(hardware.sample_count, cpu.sample_count)
                self.assertEqual(hardware.stable_candidate_count, cpu.stable_candidate_count)
                self.assertEqual(hardware.classification_counts, cpu.classification_counts)
                self.assertEqual(len(hardware.pages), len(cpu.pages))
                hardware_selected = slides.select_main_sequence(hardware.pages)
                cpu_selected = slides.select_main_sequence(cpu.pages)
                self.assertEqual(
                    [page.first_seen_order for page in hardware_selected],
                    [page.first_seen_order for page in cpu_selected],
                )
                self.assertEqual(len(hardware_selected), len(cpu_selected))
                self.assertGreaterEqual(len(hardware_selected), 3)
                for hardware_page, cpu_page in zip(hardware_selected, cpu_selected):
                    self.assertAlmostEqual(
                        hardware_page.current_timestamp,
                        cpu_page.current_timestamp,
                        delta=0.001,
                    )

                product_summaries = []
                for label, selected in (("hardware", hardware_selected), ("cpu", cpu_selected)):
                    output = root / f"product-{codec}-{label}"
                    output.mkdir()
                    pngs = slides.extract_original_pages(source, output, info, selected)
                    pdf = output / "slides.pdf"
                    slides.write_image_pdf(pngs, pdf)
                    png_summary = []
                    for path in pngs:
                        with Image.open(path) as image:
                            png_summary.append((path.name, image.mode, image.size))
                    product_summaries.append((png_summary, pdf_rgb_pages(pdf)))
                self.assertEqual(product_summaries[0][0], product_summaries[1][0])
                self.assertEqual(
                    [page[0] for page in product_summaries[0][1]],
                    [page[0] for page in product_summaries[1][1]],
                )
                self.assertEqual(
                    len(product_summaries[0][1]), len(product_summaries[1][1])
                )
                hardware_command = slides._scan_command(
                    ffmpeg, source, *slides._scan_size(info), decoder
                )
                cpu_command = slides._scan_command(
                    ffmpeg, source, *slides._scan_size(info), None
                )
                evidence.extend((
                    {
                        "codec": codec,
                        "path": "nvdec",
                        "decoder": decoder,
                        "argv": hardware_command,
                        "samples": hardware.sample_count,
                        "stable_candidates": hardware.stable_candidate_count,
                        "classifications": dict(hardware.classification_counts),
                        "selected_pages": len(hardware_selected),
                        "timestamps": [page.current_timestamp for page in hardware_selected],
                    },
                    {
                        "codec": codec,
                        "path": "cpu",
                        "decoder": None,
                        "argv": cpu_command,
                        "samples": cpu.sample_count,
                        "stable_candidates": cpu.stable_candidate_count,
                        "classifications": dict(cpu.classification_counts),
                        "selected_pages": len(cpu_selected),
                        "timestamps": [page.current_timestamp for page in cpu_selected],
                    },
                ))
            print("MEDIA_HOST_SYNTHETIC_EVIDENCE=" + json.dumps(evidence, ensure_ascii=False))

    @unittest.skipUnless(shutil.which("ffmpeg") and shutil.which("ffprobe"), "FFmpeg is required")
    def test_vfr_nonzero_pts_scan_and_exact_seek_share_relative_time(self):
@@ -281,10 +1222,9 @@
            source = root / "source.mp4"
            source.write_bytes(b"stub")
            output = root / "result"
            page_feature = np.zeros((8, 8), np.uint8)
            logical_page = slides.LogicalPage(page_feature, page_feature, 0, 1.0, 0.2, 0)
            page = logical_page(0)
            with mock.patch.object(slides, "probe_video", return_value=slides.VideoInfo(640, 360, 10)), mock.patch.object(
                slides, "scan_video", return_value=[logical_page]
                slides, "scan_video", return_value=[page]
            ), mock.patch.object(slides, "_which_or_error", return_value="ffmpeg"), mock.patch.object(
                slides.subprocess, "Popen", side_effect=make_process
            ):
@@ -575,6 +1515,46 @@
            self.assertFalse(output.exists())
            self.assertFalse(any(root.glob(".result.staging-*")))

    def test_main_sequence_filter_runs_once_before_original_extraction(self):
        with tempfile.TemporaryDirectory() as raw:
            root = Path(raw)
            source = root / "source.mp4"
            source.write_bytes(b"stub")
            output = root / "result"
            candidates = [logical_page(index, palette(1)) for index in range(10)]
            selected = candidates[1:9]
            with mock.patch.object(
                slides, "probe_video", return_value=slides.VideoInfo(640, 360, 10)
            ), mock.patch.object(
                slides, "scan_video", return_value=candidates
            ), mock.patch.object(
                slides, "select_main_sequence", return_value=selected
            ) as sequence_filter, mock.patch.object(
                slides, "extract_original_pages", return_value=[]
            ) as original_extraction, mock.patch.object(slides, "write_image_pdf"):
                self.assertEqual(slides.extract_slides(str(source), str(output)), output)
            sequence_filter.assert_called_once_with(candidates)
            self.assertIs(original_extraction.call_args.args[3], selected)

    def test_selected_pages_are_renumbered_continuously(self):
        with tempfile.TemporaryDirectory() as raw:
            staging = Path(raw)
            selected = [logical_page(index, palette(1)) for index in range(3, 6)]

            def write_png(command, stderr_path, stage):
                destination = Path(command[-1])
                Image.new("RGB", (64, 36), (20, 40, 80)).save(destination)

            with mock.patch.object(slides, "_which_or_error", return_value="ffmpeg"), mock.patch.object(
                slides, "_run_file_command", side_effect=write_png
            ):
                outputs = slides.extract_original_pages(
                    Path("source.mp4"), staging, slides.VideoInfo(64, 36, 10), selected
                )
            self.assertEqual([path.name for path in outputs], [
                "slide_001.png", "slide_002.png", "slide_003.png"
            ])

    def test_commit_conflict_is_wrapped_and_staging_is_removed(self):
        with tempfile.TemporaryDirectory() as raw:
            root = Path(raw)
@@ -582,7 +1562,7 @@
            source.write_bytes(b"stub")
            output = root / "result"
            with mock.patch.object(slides, "probe_video", return_value=slides.VideoInfo(640, 360, 10)), mock.patch.object(
                slides, "scan_video", return_value=[]
                slides, "scan_video", return_value=[logical_page(0)]
            ), mock.patch.object(slides, "extract_original_pages", return_value=[]), mock.patch.object(
                slides, "write_image_pdf"
            ), mock.patch.object(slides.os, "rename", side_effect=FileExistsError("race")):