#!/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
|
|
|
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
|
PROCESS_WAIT_SECONDS = 5.0
|
STDERR_TAIL_BYTES = 8192
|
|
|
class SlideExtractionError(Exception):
|
"""Expected, user-facing extraction failure."""
|
|
|
@dataclass(frozen=True)
|
class VideoInfo:
|
width: int
|
height: int
|
duration: float | None
|
|
|
@dataclass(frozen=True)
|
class FrameFeature:
|
sample_index: int
|
relative_seconds: float
|
gray: np.ndarray
|
perceptual_hash: int
|
content_score: float
|
sharpness: float
|
|
|
@dataclass(frozen=True)
|
class PageCandidate:
|
timestamp: float
|
feature: np.ndarray
|
perceptual_hash: int
|
content_score: float
|
sharpness: float
|
|
|
@dataclass
|
class LogicalPage:
|
logical_anchor_feature: np.ndarray
|
current_feature: 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 SlideExtractionError(
|
f"rawvideo 半帧 EOF：期望 {size} bytes，实际 {actual} bytes"
|
)
|
chunks.append(chunk)
|
remaining -= len(chunk)
|
return b"".join(chunks)
|
|
|
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,
|
)
|
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_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"])
|
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)
|
|
|
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 _gray_feature(rgb: bytes, width: int, height: int, index: int) -> FrameFeature:
|
array = np.frombuffer(rgb, dtype=np.uint8).reshape((height, width, 3))
|
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,
|
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,
|
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_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_hash=candidate.perceptual_hash,
|
current_timestamp=candidate.timestamp,
|
current_score=candidate.content_score,
|
first_seen_order=len(self.pages),
|
)
|
)
|
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"
|
)
|
return [
|
ffmpeg,
|
"-hide_banner",
|
"-loglevel",
|
"error",
|
"-i",
|
str(source),
|
"-map",
|
"0:v:0",
|
"-vf",
|
filter_graph,
|
"-pix_fmt",
|
"rgb24",
|
"-f",
|
"rawvideo",
|
"-",
|
]
|
|
|
def scan_video(source: Path, staging: Path, info: VideoInfo) -> list[LogicalPage]:
|
scan_width, scan_height = _scan_size(info)
|
frame_bytes = scan_width * scan_height * 3
|
stderr_path = staging / ".scan.stderr.log"
|
stderr_handle: BinaryIO | None = None
|
process: subprocess.Popen[bytes] | None = None
|
active_error: BaseException | None = None
|
detector = StableSegmentDetector()
|
classifier = PageClassifier()
|
samples = 0
|
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,
|
)
|
assert process.stdout is not None
|
while True:
|
raw = read_exact(process.stdout, frame_bytes)
|
if raw is None:
|
break
|
candidate = detector.add(_gray_feature(raw, scan_width, scan_height, samples))
|
if candidate is not None:
|
classifier.consume(candidate)
|
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:
|
classifier.consume(candidate)
|
process.stdout.close()
|
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}")
|
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"FFmpeg 稳定帧扫描失败：{exc}；子进程清理失败：{cleanup_error}"
|
) from exc
|
if isinstance(exc, SlideExtractionError):
|
raise
|
raise SlideExtractionError(f"FFmpeg 稳定帧扫描失败：{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"FFmpeg 稳定帧扫描关闭 stderr 失败：{close_error}"
|
) from close_error
|
if samples == 0:
|
raise SlideExtractionError("第一条视频流没有可扫描画面")
|
if not classifier.pages:
|
raise SlideExtractionError("未发现持续至少 3 个采样点的稳定页面")
|
return classifier.pages
|
|
|
def _extract_command(ffmpeg: str, source: Path, timestamp: float, destination: Path) -> list[str]:
|
return [
|
ffmpeg,
|
"-hide_banner",
|
"-loglevel",
|
"error",
|
"-ss",
|
f"{timestamp:.3f}",
|
"-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 = "扫描稳定页面"
|
pages = scan_video(source, staging, info)
|
print(f"稳定页面：{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:
|
extract_slides(args.video, args.output)
|
return 0
|
except KeyboardInterrupt:
|
print("错误：用户中断，暂存已清理。", file=sys.stderr)
|
return 130
|
except SlideExtractionError as exc:
|
print(f"错误：{exc}", file=sys.stderr)
|
return 2
|
|
|
if __name__ == "__main__":
|
raise SystemExit(main())
|
