chore(project-info): archive Bilibili dynamic 9b4b6b19b895

MB-X Bilibili Pipeline

6 days ago 856d836cce5b57bda9d5aa9313cc9040ea56f02f

 dev/project-dev/extract_ppt_slides.py

@@ -20,6 +20,18 @@
import numpy as np
from PIL import Image

MODULE_DIRECTORY = str(Path(__file__).resolve().parent)
if MODULE_DIRECTORY not in sys.path:
    sys.path.insert(0, MODULE_DIRECTORY)

from transcribe_media import (
    MEDIA_DECODE_THREADS,
    MEDIA_FILTER_THREADS,
    MediaHostGuardError,
    _subprocess_creation_flags,
    media_host_guard,
)


SCAN_MAX_WIDTH = 320
FEATURE_WIDTH = 64
@@ -40,6 +52,13 @@
DEDUPE_HASH_DISTANCE = 7
EDGE_THRESHOLD = 18
CONTENT_SCORE_TOLERANCE = 0.012
PALETTE_BINS_PER_CHANNEL = 8
PALETTE_BIN_COUNT = PALETTE_BINS_PER_CHANNEL ** 3
SEQUENCE_THEME_COVERAGE = 0.12
SEQUENCE_THEME_MAX_GAP = 1
SEQUENCE_THEME_MIN_MARKERS = 8
SEQUENCE_THEME_MIN_SHARE = 0.50
SEQUENCE_THEME_MIN_DENSITY = 0.80
PROCESS_WAIT_SECONDS = 5.0
STDERR_TAIL_BYTES = 8192

@@ -48,11 +67,30 @@
    """Expected, user-facing extraction failure."""


class PartialFrameError(SlideExtractionError):
    """A rawvideo stream ended in the middle of one frame."""


class HardwareDecodePathError(SlideExtractionError):
    """The NVDEC decoder/transfer pipe did not yield a complete frame stream."""


@dataclass(frozen=True)
class VideoInfo:
    width: int
    height: int
    duration: float | None
    codec_name: str | None = None
    pixel_format: str | None = None


@dataclass(frozen=True)
class ScanResult:
    pages: tuple[LogicalPage, ...]
    sample_count: int
    stable_candidate_count: int
    classification_counts: tuple[tuple[str, int], ...]
    decoder: str | None


@dataclass(frozen=True)
@@ -60,6 +98,7 @@
    sample_index: int
    relative_seconds: float
    gray: np.ndarray
    palette_histogram: np.ndarray
    perceptual_hash: int
    content_score: float
    sharpness: float
@@ -69,6 +108,7 @@
class PageCandidate:
    timestamp: float
    feature: np.ndarray
    palette_histogram: np.ndarray
    perceptual_hash: int
    content_score: float
    sharpness: float
@@ -78,6 +118,7 @@
class LogicalPage:
    logical_anchor_feature: np.ndarray
    current_feature: np.ndarray
    current_palette: np.ndarray
    current_hash: int
    current_timestamp: float
    current_score: float
@@ -96,12 +137,18 @@
            if not chunks:
                return None
            actual = size - remaining
            raise SlideExtractionError(
            raise PartialFrameError(
                f"rawvideo 半帧 EOF：期望 {size} bytes，实际 {actual} bytes"
            )
        chunks.append(chunk)
        remaining -= len(chunk)
    return b"".join(chunks)


def _attach_cleanup_note(error: BaseException, detail: str) -> None:
    error.add_note(f"清理诊断：{detail}")
    if isinstance(error, HardwareDecodePathError):
        setattr(error, "_media_cleanup_failed", True)


def _tail(path: Path, limit: int = STDERR_TAIL_BYTES) -> str:
@@ -153,6 +200,7 @@
            stdout=subprocess.DEVNULL,
            stderr=stderr_handle,
            shell=False,
            creationflags=_subprocess_creation_flags(),
        )
        return_code = process.wait()
        if return_code != 0:
@@ -200,7 +248,7 @@
        "-select_streams",
        "v:0",
        "-show_entries",
        "stream=index,width,height,codec_type:format=duration",
        "stream=index,width,height,codec_name,pix_fmt,codec_type:format=duration",
        "-of",
        "json",
        "-o",
@@ -214,13 +262,21 @@
        stream = streams[0]
        width = int(stream["width"])
        height = int(stream["height"])
        codec_name = str(stream.get("codec_name") or "").strip().lower() or None
        pixel_format = str(stream.get("pix_fmt") or "").strip().lower() or None
        duration_raw = (payload.get("format") or {}).get("duration")
        duration = float(duration_raw) if duration_raw not in (None, "N/A") else None
    except (OSError, ValueError, TypeError, KeyError, IndexError, json.JSONDecodeError) as exc:
        raise SlideExtractionError(f"无法解析第一条视频流信息：{exc}") from exc
    if width <= 0 or height <= 0:
        raise SlideExtractionError("第一条视频流宽高无效")
    return VideoInfo(width=width, height=height, duration=duration)
    return VideoInfo(
        width=width,
        height=height,
        duration=duration,
        codec_name=codec_name,
        pixel_format=pixel_format,
    )


def _scan_size(info: VideoInfo) -> tuple[int, int]:
@@ -229,8 +285,28 @@
    return width, height


def _palette_histogram(rgb_array: np.ndarray) -> np.ndarray:
    """Return a normalized 8x8x8 RGB histogram without retaining the RGB frame."""
    if rgb_array.ndim != 3 or rgb_array.shape[2] != 3 or rgb_array.size == 0:
        raise SlideExtractionError("颜色主题特征要求非空 RGB 图像")
    if rgb_array.dtype != np.uint8:
        raise SlideExtractionError("颜色主题特征要求 uint8 RGB 图像")
    quantized = rgb_array >> 5
    indexes = (
        quantized[:, :, 0].astype(np.int16) * PALETTE_BINS_PER_CHANNEL ** 2
        + quantized[:, :, 1].astype(np.int16) * PALETTE_BINS_PER_CHANNEL
        + quantized[:, :, 2].astype(np.int16)
    )
    histogram = np.bincount(indexes.reshape(-1), minlength=PALETTE_BIN_COUNT).astype(
        np.float32
    )
    histogram /= float(indexes.size)
    return histogram


def _gray_feature(rgb: bytes, width: int, height: int, index: int) -> FrameFeature:
    array = np.frombuffer(rgb, dtype=np.uint8).reshape((height, width, 3))
    palette_histogram = _palette_histogram(array)
    with Image.fromarray(array, mode="RGB") as image:
        feature_height = max(1, round(height * FEATURE_WIDTH / width))
        gray_image = image.convert("L").resize(
@@ -260,6 +336,7 @@
        sample_index=index,
        relative_seconds=float(index),
        gray=gray,
        palette_histogram=palette_histogram,
        perceptual_hash=perceptual_hash,
        content_score=float(edge_density + 0.25 * contrast),
        sharpness=sharpness,
@@ -305,6 +382,7 @@
        return PageCandidate(
            timestamp=best.relative_seconds,
            feature=best.gray,
            palette_histogram=best.palette_histogram.copy(),
            perceptual_hash=best.perceptual_hash,
            content_score=best.content_score,
            sharpness=best.sharpness,
@@ -412,6 +490,7 @@
            if self._animation(candidate, recent):
                if candidate.content_score >= recent.current_score - CONTENT_SCORE_TOLERANCE:
                    recent.current_feature = candidate.feature
                    recent.current_palette = candidate.palette_histogram.copy()
                    recent.current_hash = candidate.perceptual_hash
                    recent.current_timestamp = candidate.timestamp
                    recent.current_score = candidate.content_score
@@ -424,6 +503,7 @@
            LogicalPage(
                logical_anchor_feature=candidate.feature.copy(),
                current_feature=candidate.feature,
                current_palette=candidate.palette_histogram.copy(),
                current_hash=candidate.perceptual_hash,
                current_timestamp=candidate.timestamp,
                current_score=candidate.content_score,
@@ -433,57 +513,213 @@
        return "new_page"


def _scan_command(ffmpeg: str, source: Path, scan_width: int, scan_height: int) -> list[str]:
    filter_graph = (
        f"setpts=PTS-STARTPTS,fps=fps=1:start_time=0,"
        f"scale={scan_width}:{scan_height}:flags=bilinear"
def _validated_page_palettes(pages: Sequence[LogicalPage]) -> np.ndarray:
    palettes: list[np.ndarray] = []
    for index, page in enumerate(pages, 1):
        palette = np.asarray(page.current_palette)
        if palette.shape != (PALETTE_BIN_COUNT,):
            raise SlideExtractionError(
                f"第 {index} 个稳定候选颜色主题维度错误：{palette.shape}"
            )
        if not np.all(np.isfinite(palette)) or np.any(palette < 0):
            raise SlideExtractionError(f"第 {index} 个稳定候选颜色主题包含非法数值")
        total = float(np.sum(palette, dtype=np.float64))
        if not math.isclose(total, 1.0, rel_tol=1e-4, abs_tol=1e-4):
            raise SlideExtractionError(
                f"第 {index} 个稳定候选颜色主题未归一化：sum={total:.6f}"
            )
        palettes.append(palette.astype(np.float32, copy=False))
    return np.stack(palettes)


def select_main_sequence(pages: Sequence[LogicalPage]) -> list[LogicalPage]:
    """Conservatively keep the dominant contiguous visual-theme sequence."""
    if not pages:
        raise SlideExtractionError("主课件序列过滤没有稳定候选页面")
    palettes = _validated_page_palettes(pages)
    total_pages = len(pages)
    best_score: tuple[int, int, float, int, int] | None = None
    best_bounds: tuple[int, int] | None = None

    for color_bin in range(PALETTE_BIN_COUNT):
        marker_indexes = np.flatnonzero(
            palettes[:, color_bin] >= SEQUENCE_THEME_COVERAGE
        ).tolist()
        if not marker_indexes:
            continue
        groups: list[list[int]] = []
        current_group = [int(marker_indexes[0])]
        for marker_index in marker_indexes[1:]:
            marker_index = int(marker_index)
            if marker_index - current_group[-1] <= SEQUENCE_THEME_MAX_GAP + 1:
                current_group.append(marker_index)
            else:
                groups.append(current_group)
                current_group = [marker_index]
        groups.append(current_group)

        for group in groups:
            marker_count = len(group)
            start, end = group[0], group[-1]
            span = end - start + 1
            share = marker_count / total_pages
            density = marker_count / span
            if (
                marker_count < SEQUENCE_THEME_MIN_MARKERS
                or share < SEQUENCE_THEME_MIN_SHARE
                or density < SEQUENCE_THEME_MIN_DENSITY
            ):
                continue
            coverage_sum = float(
                np.sum(palettes[start : end + 1, color_bin], dtype=np.float64)
            )
            score = (marker_count, span, coverage_sum, -start, -color_bin)
            if best_score is None or score > best_score:
                best_score = score
                best_bounds = (start, end)

    if best_bounds is None:
        return list(pages)
    start, end = best_bounds
    if start == 0 and end == total_pages - 1:
        return list(pages)
    return list(pages[start : end + 1])


def _decoder_for_codec(
    codec_name: str | None,
    pixel_format: str | None = None,
) -> str | None:
    del pixel_format  # evidence only; codec decides whether the single NVDEC attempt is made
    return {"h264": "h264_cuvid", "hevc": "hevc_cuvid"}.get(codec_name or "")


def _scan_command(
    ffmpeg: str,
    source: Path,
    scan_width: int,
    scan_height: int,
    decoder: str | None = None,
) -> list[str]:
    filters = []
    if decoder is not None:
        filters.extend(("hwdownload", "format=nv12"))
    filters.extend(
        (
            "setpts=PTS-STARTPTS",
            "fps=fps=1:start_time=0",
            f"scale={scan_width}:{scan_height}:flags=bilinear",
        )
    )
    return [
    command = [
        ffmpeg,
        "-hide_banner",
        "-loglevel",
        "error",
        "-i",
        str(source),
    ]
    if decoder is not None:
        command.extend(
            (
                "-xerror",
                "-hwaccel",
                "cuda",
                "-hwaccel_output_format",
                "cuda",
                "-c:v",
                decoder,
            )
        )
    command.extend(
        (
            "-threads",
            str(MEDIA_DECODE_THREADS),
            "-filter_threads",
            str(MEDIA_FILTER_THREADS),
            "-filter_complex_threads",
            str(MEDIA_FILTER_THREADS),
            "-i",
            str(source),
        )
    )
    command.extend(
        (
        "-map",
        "0:v:0",
        "-vf",
        filter_graph,
        ",".join(filters),
        "-pix_fmt",
        "rgb24",
        "-f",
        "rawvideo",
        "-",
    ]
        )
    )
    return command


def scan_video(source: Path, staging: Path, info: VideoInfo) -> list[LogicalPage]:
def _scan_video_once(
    source: Path,
    staging: Path,
    info: VideoInfo,
    *,
    decoder: str | None,
) -> ScanResult:
    scan_width, scan_height = _scan_size(info)
    frame_bytes = scan_width * scan_height * 3
    stderr_path = staging / ".scan.stderr.log"
    path_label = decoder or "cpu"
    stderr_path = staging / f".scan-{path_label}.stderr.log"
    stderr_handle: BinaryIO | None = None
    process: subprocess.Popen[bytes] | None = None
    active_error: BaseException | None = None
    detector = StableSegmentDetector()
    classifier = PageClassifier()
    samples = 0
    stable_candidates = 0
    classifications: dict[str, int] = {}
    try:
        stderr_handle = stderr_path.open("wb")
        process = subprocess.Popen(
            _scan_command(_which_or_error("ffmpeg"), source, scan_width, scan_height),
            stdin=subprocess.DEVNULL,
            stdout=subprocess.PIPE,
            stderr=stderr_handle,
            shell=False,
        try:
            stderr_handle = stderr_path.open("wb")
        except OSError as exc:
            raise SlideExtractionError(f"FFmpeg 稳定帧扫描打开 stderr 失败：{exc}") from exc
        command = _scan_command(
            _which_or_error("ffmpeg"),
            source,
            scan_width,
            scan_height,
            decoder,
        )
        try:
            process = subprocess.Popen(
                command,
                stdin=subprocess.DEVNULL,
                stdout=subprocess.PIPE,
                stderr=stderr_handle,
                shell=False,
                creationflags=_subprocess_creation_flags(),
            )
        except BaseException as exc:
            if not isinstance(exc, Exception):
                raise
            if decoder is not None:
                raise HardwareDecodePathError(
                    f"{decoder} 启动失败：{exc}"
                ) from exc
            raise SlideExtractionError(f"受限 CPU 扫描启动失败：{exc}") from exc
        assert process.stdout is not None
        while True:
            raw = read_exact(process.stdout, frame_bytes)
            try:
                raw = read_exact(process.stdout, frame_bytes)
            except PartialFrameError as exc:
                if decoder is not None:
                    raise HardwareDecodePathError(f"{decoder} 输出半帧：{exc}") from exc
                raise
            if raw is None:
                break
            candidate = detector.add(_gray_feature(raw, scan_width, scan_height, samples))
            if candidate is not None:
                classifier.consume(candidate)
                stable_candidates += 1
                classification = classifier.consume(candidate)
                classifications[classification] = classifications.get(classification, 0) + 1
            samples += 1
            if samples % 300 == 0:
                if info.duration and info.duration > 0:
@@ -493,26 +729,32 @@
                    print(f"扫描进度：{samples} 秒", flush=True)
        candidate = detector.flush()
        if candidate is not None:
            classifier.consume(candidate)
        process.stdout.close()
            stable_candidates += 1
            classification = classifier.consume(candidate)
            classifications[classification] = classifications.get(classification, 0) + 1
        try:
            process.stdout.close()
        except OSError as exc:
            raise SlideExtractionError(f"FFmpeg 稳定帧扫描关闭 stdout 失败：{exc}") from exc
        return_code = process.wait()
        if return_code != 0:
            stderr_handle.flush()
            detail = _tail(stderr_path)
            suffix = f"：{detail}" if detail else ""
            raise SlideExtractionError(f"FFmpeg 稳定帧扫描失败（exit={return_code}）{suffix}")
            message = f"FFmpeg 稳定帧扫描失败（exit={return_code}）{suffix}"
            if decoder is not None:
                raise HardwareDecodePathError(f"{decoder} {message}")
            raise SlideExtractionError(message)
    except BaseException as exc:
        active_error = exc
        cleanup_error = _attempt_reap(process) if process is not None else None
        if not isinstance(exc, Exception):
            if cleanup_error is not None:
                _attach_cleanup_note(exc, f"子进程回收失败：{cleanup_error}")
            raise
        if cleanup_error is not None:
            raise SlideExtractionError(
                f"FFmpeg 稳定帧扫描失败：{exc}；子进程清理失败：{cleanup_error}"
            ) from exc
        if isinstance(exc, SlideExtractionError):
            raise
        raise SlideExtractionError(f"FFmpeg 稳定帧扫描失败：{exc}") from exc
            _attach_cleanup_note(exc, f"子进程回收失败：{cleanup_error}")
        raise
    finally:
        if stderr_handle is not None:
            try:
@@ -522,11 +764,53 @@
                    raise SlideExtractionError(
                        f"FFmpeg 稳定帧扫描关闭 stderr 失败：{close_error}"
                    ) from close_error
                _attach_cleanup_note(active_error, f"关闭 stderr 失败：{close_error}")
    if samples == 0:
        if decoder is not None:
            raise HardwareDecodePathError(f"{decoder} 未输出任何完整扫描帧")
        raise SlideExtractionError("第一条视频流没有可扫描画面")
    if not classifier.pages:
    return ScanResult(
        pages=tuple(classifier.pages),
        sample_count=samples,
        stable_candidate_count=stable_candidates,
        classification_counts=tuple(sorted(classifications.items())),
        decoder=decoder,
    )


def scan_video(source: Path, staging: Path, info: VideoInfo) -> list[LogicalPage]:
    decoder = _decoder_for_codec(info.codec_name, info.pixel_format)
    if decoder is None:
        print(
            "警告：视频编码没有 H.264/H.265 NVDEC 映射，"
            "使用 4 解码线程/2 滤镜线程的受限 CPU。",
            file=sys.stderr,
            flush=True,
        )
        result = _scan_video_once(source, staging, info, decoder=None)
        path_label = "受限 CPU"
    else:
        try:
            result = _scan_video_once(source, staging, info, decoder=decoder)
            path_label = f"NVDEC/CUDA ({decoder})"
        except HardwareDecodePathError as exc:
            if getattr(exc, "_media_cleanup_failed", False):
                raise
            print(
                f"警告：NVDEC 不可用（{decoder}）：{exc}；改用 4 解码线程/2 滤镜线程的受限 CPU。",
                file=sys.stderr,
                flush=True,
            )
            result = _scan_video_once(source, staging, info, decoder=None)
            path_label = "受限 CPU（NVDEC 一次回退）"
    print(
        f"扫描解码路径：{path_label}；解码线程={MEDIA_DECODE_THREADS}；"
        f"滤镜线程={MEDIA_FILTER_THREADS}",
        flush=True,
    )
    if not result.pages:
        raise SlideExtractionError("未发现持续至少 3 个采样点的稳定页面")
    return classifier.pages
    return list(result.pages)


def _extract_command(ffmpeg: str, source: Path, timestamp: float, destination: Path) -> list[str]:
@@ -537,6 +821,8 @@
        "error",
        "-ss",
        f"{timestamp:.3f}",
        "-threads",
        str(MEDIA_DECODE_THREADS),
        "-i",
        str(source),
        "-map",
@@ -716,8 +1002,12 @@
            flush=True,
        )
        stage = "扫描稳定页面"
        pages = scan_video(source, staging, info)
        print(f"稳定页面：{len(pages)}", flush=True)
        candidates = scan_video(source, staging, info)
        pages = select_main_sequence(candidates)
        print(
            f"稳定候选：{len(candidates)}；主课件页面：{len(pages)}",
            flush=True,
        )
        stage = "提取原分辨率 PNG"
        png_paths = extract_original_pages(source, staging, info, pages)
        stage = "生成合并 PDF"
@@ -759,12 +1049,13 @@
def main(argv: Sequence[str] | None = None) -> int:
    args = build_parser().parse_args(argv)
    try:
        extract_slides(args.video, args.output)
        with media_host_guard():
            extract_slides(args.video, args.output)
        return 0
    except KeyboardInterrupt:
        print("错误：用户中断，暂存已清理。", file=sys.stderr)
        return 130
    except SlideExtractionError as exc:
    except (SlideExtractionError, MediaHostGuardError) as exc:
        print(f"错误：{exc}", file=sys.stderr)
        return 2