#!/usr/bin/env python3 """Extract stable, deduplicated slide pages from one local recording.""" from __future__ import annotations import argparse import json import math import os from pathlib import Path import shutil import subprocess import sys import time from dataclasses import dataclass from typing import BinaryIO, Callable, Iterable, Sequence import uuid import zlib 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 MIN_STABLE_SAMPLES = 3 PIXEL_CHANGE_THRESHOLD = 12 STABLE_PREVIOUS_MAD = 3.0 STABLE_PREVIOUS_CHANGED_RATIO = 0.045 STABLE_ANCHOR_MAD = 4.5 STABLE_ANCHOR_CHANGED_RATIO = 0.075 STABLE_BEST_UPDATE_MAD = 0.75 ANIMATION_CURRENT_MAD = 24.0 ANIMATION_CURRENT_CHANGED_RATIO = 0.42 ANIMATION_ANCHOR_MAD = 34.0 ANIMATION_ANCHOR_CHANGED_RATIO = 0.58 ANIMATION_HASH_DISTANCE = 24 DEDUPE_MAD = 5.5 DEDUPE_CHANGED_RATIO = 0.085 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 class SlideExtractionError(Exception): """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) class FrameFeature: sample_index: int relative_seconds: float gray: np.ndarray palette_histogram: np.ndarray perceptual_hash: int content_score: float sharpness: float @dataclass(frozen=True) class PageCandidate: timestamp: float feature: np.ndarray palette_histogram: np.ndarray perceptual_hash: int content_score: float sharpness: float @dataclass class LogicalPage: logical_anchor_feature: np.ndarray current_feature: np.ndarray current_palette: np.ndarray current_hash: int current_timestamp: float current_score: float first_seen_order: int def read_exact(stream: BinaryIO, size: int) -> bytes | None: """Read exactly one raw frame, distinguish clean EOF from a partial frame.""" if size <= 0: raise ValueError("size must be positive") chunks: list[bytes] = [] remaining = size while remaining: chunk = stream.read(remaining) if not chunk: if not chunks: return None actual = size - remaining 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: try: with path.open("rb") as handle: handle.seek(0, os.SEEK_END) length = handle.tell() handle.seek(max(0, length - limit)) return handle.read(limit).decode("utf-8", errors="replace").strip() except OSError: return "" def _reap(process: subprocess.Popen[bytes]) -> None: stdout = getattr(process, "stdout", None) if stdout is not None: try: stdout.close() except OSError: pass if process.poll() is not None: process.wait() return process.terminate() try: process.wait(timeout=PROCESS_WAIT_SECONDS) except subprocess.TimeoutExpired: process.kill() process.wait() def _attempt_reap(process: subprocess.Popen[bytes]) -> Exception | None: try: _reap(process) return None except Exception as exc: # cleanup must not replace KeyboardInterrupt/SystemExit return exc def _run_file_command(command: Sequence[str], stderr_path: Path, stage: str) -> None: stderr_handle: BinaryIO | None = None process: subprocess.Popen[bytes] | None = None active_error: BaseException | None = None try: stderr_handle = stderr_path.open("wb") process = subprocess.Popen( list(command), stdin=subprocess.DEVNULL, stdout=subprocess.DEVNULL, stderr=stderr_handle, shell=False, creationflags=_subprocess_creation_flags(), ) return_code = process.wait() if return_code != 0: stderr_handle.flush() detail = _tail(stderr_path) suffix = f"：{detail}" if detail else "" raise SlideExtractionError(f"{stage}失败（exit={return_code}）{suffix}") except BaseException as exc: active_error = exc cleanup_error = _attempt_reap(process) if process is not None else None if not isinstance(exc, Exception): raise if cleanup_error is not None: raise SlideExtractionError( f"{stage}失败：{exc}；子进程清理失败：{cleanup_error}" ) from exc if isinstance(exc, SlideExtractionError): raise raise SlideExtractionError(f"{stage}失败：{exc}") from exc finally: if stderr_handle is not None: try: stderr_handle.close() except Exception as close_error: if active_error is None: raise SlideExtractionError( f"{stage}关闭 stderr 失败：{close_error}" ) from close_error def _which_or_error(name: str) -> str: resolved = shutil.which(name) if not resolved: raise SlideExtractionError(f"未找到 {name}，请先安装并加入 PATH") return resolved def probe_video(source: Path, staging: Path) -> VideoInfo: probe_path = staging / ".probe.json" stderr_path = staging / ".ffprobe.stderr.log" command = [ _which_or_error("ffprobe"), "-v", "error", "-select_streams", "v:0", "-show_entries", "stream=index,width,height,codec_name,pix_fmt,codec_type:format=duration", "-of", "json", "-o", str(probe_path), str(source), ] _run_file_command(command, stderr_path, "FFprobe 探测") try: payload = json.loads(probe_path.read_text(encoding="utf-8")) streams = payload.get("streams") or [] 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, codec_name=codec_name, pixel_format=pixel_format, ) def _scan_size(info: VideoInfo) -> tuple[int, int]: width = min(info.width, SCAN_MAX_WIDTH) height = max(1, round(info.height * width / info.width)) 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( (FEATURE_WIDTH, feature_height), Image.Resampling.BILINEAR ) gray = np.asarray(gray_image, dtype=np.uint8).copy() gx = np.abs(np.diff(gray.astype(np.int16), axis=1)) gy = np.abs(np.diff(gray.astype(np.int16), axis=0)) gradient_count = gx.size + gy.size edge_density = ( (int(np.count_nonzero(gx >= EDGE_THRESHOLD)) + int(np.count_nonzero(gy >= EDGE_THRESHOLD))) / gradient_count if gradient_count else 0.0 ) contrast = float(np.std(gray, dtype=np.float64)) / 255.0 sharpness = float(np.var(gx, dtype=np.float64) + np.var(gy, dtype=np.float64)) with Image.fromarray(gray, mode="L") as feature_image: hash_pixels = np.asarray( feature_image.resize((8, 8), Image.Resampling.BILINEAR), dtype=np.uint8 ).reshape(-1) mean = float(np.mean(hash_pixels)) perceptual_hash = 0 for value in hash_pixels: perceptual_hash = (perceptual_hash << 1) | int(value >= mean) return FrameFeature( 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, ) def difference(left: np.ndarray, right: np.ndarray) -> tuple[float, float]: if left.shape != right.shape: return math.inf, 1.0 delta = np.abs(left.astype(np.int16) - right.astype(np.int16)) return float(np.mean(delta)), float(np.count_nonzero(delta > PIXEL_CHANGE_THRESHOLD) / delta.size) def hash_distance(left: int, right: int) -> int: return (left ^ right).bit_count() def _within( left: np.ndarray, right: np.ndarray, mad_limit: float, ratio_limit: float, ) -> bool: mad, ratio = difference(left, right) return mad <= mad_limit and ratio <= ratio_limit class StableSegmentDetector: """Streaming stable-run detector retaining no frame list.""" def __init__(self) -> None: self.anchor: FrameFeature | None = None self.previous: FrameFeature | None = None self.best: FrameFeature | None = None self.committed_best: FrameFeature | None = None self.count = 0 self.started_from_anchor_drift = False def _candidate(self) -> PageCandidate | None: if self.count < MIN_STABLE_SAMPLES or self.committed_best is None: return None best = self.committed_best 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, ) def _start(self, frame: FrameFeature, *, from_anchor_drift: bool = False) -> None: self.anchor = frame self.previous = frame self.best = frame self.committed_best = None self.count = 1 self.started_from_anchor_drift = from_anchor_drift def add(self, frame: FrameFeature) -> PageCandidate | None: if self.anchor is None or self.previous is None: self._start(frame) return None adjacent = _within( self.previous.gray, frame.gray, STABLE_PREVIOUS_MAD, STABLE_PREVIOUS_CHANGED_RATIO, ) anchored = _within( self.anchor.gray, frame.gray, STABLE_ANCHOR_MAD, STABLE_ANCHOR_CHANGED_RATIO, ) if adjacent and anchored: self.count += 1 self.previous = frame if self.best is None or frame.sharpness >= self.best.sharpness: self.best = frame if self.count == MIN_STABLE_SAMPLES: self.committed_best = self.best elif self.committed_best is not None and self.anchor is not None: # Improve sharpness only while the frame remains very close to the # immutable page anchor; do not let a slow transition win. mad, ratio = difference(self.anchor.gray, frame.gray) if ( mad <= STABLE_BEST_UPDATE_MAD and ratio <= STABLE_PREVIOUS_CHANGED_RATIO and frame.sharpness >= self.committed_best.sharpness ): self.committed_best = frame return None anchor_drift = adjacent and not anchored candidate = None if (anchor_drift and self.started_from_anchor_drift) else self._candidate() self._start(frame, from_anchor_drift=anchor_drift) return candidate def flush(self) -> PageCandidate | None: candidate = self._candidate() self.anchor = None self.previous = None self.best = None self.committed_best = None self.count = 0 self.started_from_anchor_drift = False return candidate class PageClassifier: """Classify candidates as recent-page animation, historical return, or new page.""" def __init__(self) -> None: self.pages: list[LogicalPage] = [] @staticmethod def _animation(candidate: PageCandidate, page: LogicalPage) -> bool: return ( _within( page.current_feature, candidate.feature, ANIMATION_CURRENT_MAD, ANIMATION_CURRENT_CHANGED_RATIO, ) and _within( page.logical_anchor_feature, candidate.feature, ANIMATION_ANCHOR_MAD, ANIMATION_ANCHOR_CHANGED_RATIO, ) and hash_distance(page.current_hash, candidate.perceptual_hash) <= ANIMATION_HASH_DISTANCE ) @staticmethod def _duplicate(candidate: PageCandidate, page: LogicalPage) -> bool: return ( _within( page.current_feature, candidate.feature, DEDUPE_MAD, DEDUPE_CHANGED_RATIO, ) and hash_distance(page.current_hash, candidate.perceptual_hash) <= DEDUPE_HASH_DISTANCE ) def consume(self, candidate: PageCandidate) -> str: if self.pages: recent = self.pages[-1] 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 return "animation_updated" return "animation_retained" for historical in self.pages[:-1]: if self._duplicate(candidate, historical): return "historical_duplicate" self.pages.append( 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, first_seen_order=len(self.pages), ) ) return "new_page" 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", ) ) command = [ ffmpeg, "-hide_banner", "-loglevel", "error", ] 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", ",".join(filters), "-pix_fmt", "rgb24", "-f", "rawvideo", "-", ) ) return command 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 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: 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: 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: 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: percent = min(100.0, samples / info.duration * 100) print(f"扫描进度：{samples} 秒（{percent:.1f}%）", flush=True) else: print(f"扫描进度：{samples} 秒", flush=True) candidate = detector.flush() if candidate is not None: 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 "" 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: _attach_cleanup_note(exc, f"子进程回收失败：{cleanup_error}") raise finally: if stderr_handle is not None: try: stderr_handle.close() except Exception as close_error: if active_error is None: 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("第一条视频流没有可扫描画面") 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 list(result.pages) def _extract_command(ffmpeg: str, source: Path, timestamp: float, destination: Path) -> list[str]: return [ ffmpeg, "-hide_banner", "-loglevel", "error", "-ss", f"{timestamp:.3f}", "-threads", str(MEDIA_DECODE_THREADS), "-i", str(source), "-map", "0:v:0", "-frames:v", "1", "-c:v", "png", "-compression_level", "6", "-n", str(destination), ] def extract_original_pages( source: Path, staging: Path, info: VideoInfo, pages: Sequence[LogicalPage], ) -> list[Path]: ffmpeg = _which_or_error("ffmpeg") digits = max(3, len(str(len(pages)))) pages_directory = staging / "pages" pages_directory.mkdir() outputs: list[Path] = [] for index, page in enumerate(pages, 1): destination = pages_directory / f"slide_{index:0{digits}d}.png" stderr_path = staging / f".extract-{index:0{digits}d}.stderr.log" _run_file_command( _extract_command(ffmpeg, source, page.current_timestamp, destination), stderr_path, f"提取第 {index} 页原分辨率 PNG", ) try: with Image.open(destination) as image: if image.format != "PNG": raise SlideExtractionError(f"第 {index} 页不是 PNG") if image.size != (info.width, info.height): raise SlideExtractionError( f"第 {index} 页尺寸 {image.size}，预期 {(info.width, info.height)}" ) image.verify() except OSError as exc: raise SlideExtractionError(f"第 {index} 页 PNG 无法读取：{exc}") from exc outputs.append(destination) print(f"原图进度：{index}/{len(pages)}", flush=True) return outputs def write_image_pdf( png_paths: Sequence[Path], destination: Path, image_opener: Callable[[Path], Image.Image] = Image.open, ) -> None: if not png_paths: raise SlideExtractionError("没有 PNG 页面可写入 PDF") page_count = len(png_paths) max_object = 2 + page_count * 3 offsets = [0] * (max_object + 1) def page_object(index: int) -> int: return 3 + index * 3 def image_object(index: int) -> int: return 4 + index * 3 def content_object(index: int) -> int: return 5 + index * 3 with destination.open("xb") as output: output.write(b"%PDF-1.4\n%\xe2\xe3\xcf\xd3\n") def write_object(number: int, body: bytes) -> None: offsets[number] = output.tell() output.write(f"{number} 0 obj\n".encode("ascii")) output.write(body) output.write(b"\nendobj\n") write_object(1, b"<< /Type /Catalog /Pages 2 0 R >>") kids = " ".join(f"{page_object(i)} 0 R" for i in range(page_count)) write_object(2, f"<< /Type /Pages /Count {page_count} /Kids [{kids}] >>".encode("ascii")) for index, path in enumerate(png_paths): with image_opener(path) as opened: rgb_image = opened.convert("RGB") try: width, height = rgb_image.size raw_rgb = rgb_image.tobytes() compressed = zlib.compress(raw_rgb, level=6) del raw_rgb page_body = ( f"<< /Type /Page /Parent 2 0 R /MediaBox [0 0 {width} {height}] " f"/Resources << /XObject << /Im0 {image_object(index)} 0 R >> >> " f"/Contents {content_object(index)} 0 R >>" ).encode("ascii") write_object(page_object(index), page_body) image_header = ( f"<< /Type /XObject /Subtype /Image /Width {width} /Height {height} " f"/ColorSpace /DeviceRGB /BitsPerComponent 8 /Filter /FlateDecode " f"/Length {len(compressed)} >>\nstream\n" ).encode("ascii") image_number = image_object(index) offsets[image_number] = output.tell() output.write(f"{image_number} 0 obj\n".encode("ascii")) output.write(image_header) output.write(compressed) output.write(b"\nendstream\nendobj\n") content = f"q\n{width} 0 0 {height} 0 0 cm\n/Im0 Do\nQ\n".encode("ascii") content_body = f"<< /Length {len(content)} >>\nstream\n".encode("ascii") + content + b"endstream" write_object(content_object(index), content_body) finally: rgb_image.close() if "compressed" in locals(): del compressed xref_offset = output.tell() output.write(f"xref\n0 {max_object + 1}\n".encode("ascii")) output.write(b"0000000000 65535 f \n") for number in range(1, max_object + 1): output.write(f"{offsets[number]:010d} 00000 n \n".encode("ascii")) output.write( f"trailer\n<< /Size {max_object + 1} /Root 1 0 R >>\nstartxref\n{xref_offset}\n%%EOF\n".encode( "ascii" ) ) output.flush() os.fsync(output.fileno()) def _lexists(path: Path) -> bool: return os.path.lexists(path) def _is_symlink(path: Path) -> bool: return path.is_symlink() def _absolute_lexical(value: str | os.PathLike[str]) -> Path: """Make an absolute normalized path without following symlinks.""" expanded = Path(value).expanduser() return Path(os.path.abspath(os.fspath(expanded))) def _validate_paths(source_arg: str, output_arg: str | None) -> tuple[Path, Path]: source = _absolute_lexical(source_arg) if _is_symlink(source): raise SlideExtractionError(f"输入视频必须是普通文件，拒绝符号链接：{source}") if not _lexists(source) or not source.is_file(): raise SlideExtractionError(f"输入视频不存在或不是文件：{source}") output = ( _absolute_lexical(output_arg) if output_arg else source.with_name(f"{source.stem}.slides") ) if _lexists(output): raise SlideExtractionError(f"输出目录已存在，默认不覆盖：{output}") if not output.parent.exists() or not output.parent.is_dir(): raise SlideExtractionError(f"输出目录的父目录不存在：{output.parent}") return source, output def extract_slides(source_arg: str, output_arg: str | None = None) -> Path: source, output = _validate_paths(source_arg, output_arg) staging = output.parent / f".{output.name}.staging-{uuid.uuid4().hex[:12]}" started = time.monotonic() stage = "创建暂存目录" staging_created = False try: staging.mkdir() staging_created = True stage = "探测第一条视频流" info = probe_video(source, staging) print( f"视频：{info.width}x{info.height}" + (f"，{info.duration:.3f} 秒" if info.duration is not None else ""), flush=True, ) stage = "扫描稳定页面" 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" pdf_path = staging / f"{source.stem}.slides.pdf" write_image_pdf(png_paths, pdf_path) for temporary in staging.glob(".*.log"): temporary.unlink(missing_ok=True) (staging / ".probe.json").unlink(missing_ok=True) if _lexists(output): raise SlideExtractionError(f"提交前发现输出目录已存在，拒绝覆盖：{output}") stage = "提交正式输出目录" os.rename(staging, output) elapsed = time.monotonic() - started print(f"完成：{output}", flush=True) print(f"页面：{len(png_paths)}；耗时：{elapsed:.3f} 秒", flush=True) return output except BaseException as exc: if staging_created and _lexists(staging): shutil.rmtree(staging, ignore_errors=True) if not isinstance(exc, Exception): raise if isinstance(exc, SlideExtractionError): raise raise SlideExtractionError(f"{stage}失败：{exc}") from exc def build_parser() -> argparse.ArgumentParser: parser = argparse.ArgumentParser( description="从单个本地会议录屏提取稳定、动画合并且去重的原分辨率 PPT 页面。" ) parser.add_argument("video", help="单个本地视频路径") parser.add_argument( "--output", help="输出目录；默认是视频同目录下的 <视频名>.slides，已存在时拒绝覆盖", ) return parser def main(argv: Sequence[str] | None = None) -> int: args = build_parser().parse_args(argv) try: with media_host_guard(): extract_slides(args.video, args.output) return 0 except KeyboardInterrupt: print("错误：用户中断，暂存已清理。", file=sys.stderr) return 130 except (SlideExtractionError, MediaHostGuardError) as exc: print(f"错误：{exc}", file=sys.stderr) return 2 if __name__ == "__main__": raise SystemExit(main())