#!/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())
|
