From 2c699fc08247494faf054f9c1f5b90a47a105ec1 Mon Sep 17 00:00:00 2001
From: Cai <cai@nbcai.cc>
Date: Fri, 28 Aug 2026 18:13:13 +0800
Subject: [PATCH] chore: 更新2026-08-28股票估值每日台账
---
dev/project-dev/extract_ppt_slides.py | 367 +++++++++++++++++++++++++++++++++++++++++++++++-----
1 files changed, 329 insertions(+), 38 deletions(-)
diff --git a/dev/project-dev/extract_ppt_slides.py b/dev/project-dev/extract_ppt_slides.py
index 1d0ced7..0e7a20e 100644
--- a/dev/project-dev/extract_ppt_slides.py
+++ b/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
--
Gitblit v1.9.3