from __future__ import annotations
|
|
import hashlib
|
import importlib.util
|
import io
|
import os
|
from pathlib import Path
|
import re
|
import shutil
|
import subprocess
|
import sys
|
import tempfile
|
import unittest
|
from unittest import mock
|
import zlib
|
|
import numpy as np
|
from PIL import Image, ImageDraw
|
|
|
MODULE_PATH = Path(__file__).resolve().parents[1] / "extract_ppt_slides.py"
|
SPEC = importlib.util.spec_from_file_location("extract_ppt_slides", MODULE_PATH)
|
assert SPEC and SPEC.loader
|
slides = importlib.util.module_from_spec(SPEC)
|
sys.modules[SPEC.name] = slides
|
SPEC.loader.exec_module(slides)
|
|
|
def feature(index: int, array: np.ndarray, score: float = 0.2, sharpness: float | None = None):
|
value = int(np.mean(array))
|
phash = int.from_bytes(hashlib.sha256(array.tobytes()).digest()[:8], "big")
|
return slides.FrameFeature(
|
sample_index=index,
|
relative_seconds=float(index),
|
gray=array.astype(np.uint8).copy(),
|
perceptual_hash=phash,
|
content_score=score,
|
sharpness=float(value if sharpness is None else sharpness),
|
)
|
|
|
def candidate(timestamp: float, array: np.ndarray, score: float, phash: int = 0):
|
return slides.PageCandidate(timestamp, array.astype(np.uint8).copy(), phash, score, 1.0)
|
|
|
def sha256(path: Path) -> str:
|
digest = hashlib.sha256()
|
with path.open("rb") as handle:
|
for chunk in iter(lambda: handle.read(1024 * 1024), b""):
|
digest.update(chunk)
|
return digest.hexdigest().upper()
|
|
|
def pdf_rgb_pages(path: Path):
|
data = path.read_bytes()
|
objects = {
|
int(match.group(1)): match.group(2)
|
for match in re.finditer(rb"(?m)^(\d+) 0 obj\n(.*?)\nendobj\n", data, re.DOTALL)
|
}
|
pages_body = objects[2]
|
kids_match = re.search(rb"/Kids \[(.*?)\]", pages_body, re.DOTALL)
|
assert kids_match
|
page_ids = [int(value) for value in re.findall(rb"(\d+) 0 R", kids_match.group(1))]
|
result = []
|
for page_id in page_ids:
|
page = objects[page_id]
|
media = re.search(rb"/MediaBox \[0 0 (\d+) (\d+)\]", page)
|
image_ref = re.search(rb"/Im0 (\d+) 0 R", page)
|
assert media and image_ref
|
image_body = objects[int(image_ref.group(1))]
|
length = int(re.search(rb"/Length (\d+)", image_body).group(1))
|
marker = b"stream\n"
|
start = image_body.index(marker) + len(marker)
|
compressed = image_body[start : start + length]
|
result.append(((int(media.group(1)), int(media.group(2))), zlib.decompress(compressed)))
|
return result
|
|
|
class ShortReadStream:
|
def __init__(self, chunks):
|
self.chunks = list(chunks)
|
|
def read(self, size):
|
if not self.chunks:
|
return b""
|
chunk = self.chunks.pop(0)
|
if len(chunk) <= size:
|
return chunk
|
self.chunks.insert(0, chunk[size:])
|
return chunk[:size]
|
|
|
class ReadExactTests(unittest.TestCase):
|
def test_short_reads_clean_eof_and_partial_eof(self):
|
self.assertEqual(slides.read_exact(ShortReadStream([b"a", b"bc", b"def"]), 6), b"abcdef")
|
self.assertIsNone(slides.read_exact(ShortReadStream([]), 6))
|
with self.assertRaisesRegex(slides.SlideExtractionError, "半帧 EOF"):
|
slides.read_exact(ShortReadStream([b"abc"]), 6)
|
|
|
class SegmentAndClassifierTests(unittest.TestCase):
|
def test_anchor_drift_emits_only_old_and_new_plateaus(self):
|
detector = slides.StableSegmentDetector()
|
emitted = []
|
arrays = [np.full((8, 8), 20, np.uint8)] * 3
|
arrays += [np.full((8, 8), value, np.uint8) for value in (22, 24, 26, 28, 30, 32)]
|
arrays += [np.full((8, 8), 80, np.uint8)] * 3
|
for index, array in enumerate(arrays):
|
value = detector.add(feature(index, array, score=index / 100, sharpness=index))
|
if value:
|
emitted.append(value)
|
final = detector.flush()
|
if final:
|
emitted.append(final)
|
self.assertEqual([round(float(np.mean(item.feature))) for item in emitted], [20, 80])
|
|
def test_animation_increase_decrease_and_equal_score(self):
|
classifier = slides.PageClassifier()
|
base = np.zeros((8, 8), np.uint8)
|
added = base.copy()
|
added[:, :2] = 20
|
self.assertEqual(classifier.consume(candidate(1, base, 0.20)), "new_page")
|
self.assertEqual(classifier.consume(candidate(2, added, 0.30)), "animation_updated")
|
self.assertEqual(classifier.pages[0].current_timestamp, 2)
|
reduced = base.copy()
|
self.assertEqual(classifier.consume(candidate(3, reduced, 0.10)), "animation_retained")
|
self.assertEqual(classifier.pages[0].current_timestamp, 2)
|
self.assertEqual(
|
classifier.consume(candidate(4, added, 0.30 + slides.CONTENT_SCORE_TOLERANCE / 2)),
|
"animation_updated",
|
)
|
self.assertEqual(classifier.pages[0].current_timestamp, 4)
|
|
def test_a_b_a_keeps_first_order(self):
|
classifier = slides.PageClassifier()
|
a = np.zeros((8, 8), np.uint8)
|
b = np.full((8, 8), 255, np.uint8)
|
classifier.consume(candidate(1, a, 0.2, 0))
|
classifier.consume(candidate(2, b, 0.2, (1 << 64) - 1))
|
action = classifier.consume(candidate(3, a, 0.2, 0))
|
self.assertEqual(action, "historical_duplicate")
|
self.assertEqual(len(classifier.pages), 2)
|
self.assertEqual([page.first_seen_order for page in classifier.pages], [0, 1])
|
|
def test_recent_animation_has_priority_over_historical_duplicate(self):
|
classifier = slides.PageClassifier()
|
historical = np.zeros((8, 8), np.uint8)
|
recent = np.full((8, 8), 4, np.uint8)
|
classifier.consume(candidate(1, historical, 0.20, 0))
|
classifier.pages.append(
|
slides.LogicalPage(recent.copy(), recent.copy(), 0, 2.0, 0.20, 1)
|
)
|
action = classifier.consume(candidate(3, historical, 0.22, 0))
|
self.assertEqual(action, "animation_updated")
|
self.assertEqual(len(classifier.pages), 2)
|
|
|
class ProcessAndCommandTests(unittest.TestCase):
|
def test_command_freezes_time_mapping(self):
|
scan = slides._scan_command("ffmpeg", Path("x.mp4"), 320, 180)
|
self.assertIn("setpts=PTS-STARTPTS,fps=fps=1:start_time=0,scale=320:180:flags=bilinear", scan)
|
extract = slides._extract_command("ffmpeg", Path("x.mp4"), 12.0, Path("x.png"))
|
self.assertEqual(extract[extract.index("-ss") + 1], "12.000")
|
self.assertEqual(feature(9, np.zeros((2, 2), np.uint8)).relative_seconds, 9.0)
|
|
@unittest.skipUnless(shutil.which("ffmpeg") and shutil.which("ffprobe"), "FFmpeg is required")
|
def test_vfr_nonzero_pts_scan_and_exact_seek_share_relative_time(self):
|
with tempfile.TemporaryDirectory() as raw:
|
root = Path(raw)
|
video = root / "vfr-nonzero.mkv"
|
subprocess.run(
|
[
|
shutil.which("ffmpeg"), "-hide_banner", "-loglevel", "error",
|
"-f", "lavfi", "-i", "color=red:size=64x64:rate=2:duration=2",
|
"-f", "lavfi", "-i", "color=green:size=64x64:rate=3:duration=2",
|
"-f", "lavfi", "-i", "color=blue:size=64x64:rate=4:duration=2",
|
"-filter_complex", "[0:v][1:v][2:v]concat=n=3:v=1:a=0,setpts=PTS+5/TB[v]",
|
"-map", "[v]", "-fps_mode", "vfr", "-c:v", "ffv1", str(video),
|
],
|
check=True,
|
)
|
start_time = float(
|
subprocess.check_output(
|
[
|
shutil.which("ffprobe"), "-v", "error", "-select_streams", "v:0",
|
"-show_entries", "stream=start_time", "-of", "default=nw=1:nk=1", str(video),
|
],
|
text=True,
|
).strip()
|
)
|
self.assertGreaterEqual(start_time, 5.0)
|
|
scan_process = subprocess.Popen(
|
slides._scan_command(shutil.which("ffmpeg"), video, 64, 64),
|
stdin=subprocess.DEVNULL,
|
stdout=subprocess.PIPE,
|
stderr=subprocess.PIPE,
|
)
|
assert scan_process.stdout is not None
|
scanned = []
|
while True:
|
payload = slides.read_exact(scan_process.stdout, 64 * 64 * 3)
|
if payload is None:
|
break
|
scanned.append(np.frombuffer(payload, np.uint8).reshape((64, 64, 3)))
|
scan_process.stdout.close()
|
stderr = scan_process.stderr.read() if scan_process.stderr else b""
|
if scan_process.stderr:
|
scan_process.stderr.close()
|
self.assertEqual(scan_process.wait(), 0, stderr.decode(errors="replace"))
|
self.assertGreaterEqual(len(scanned), 5)
|
|
extracted = []
|
for timestamp in (0.0, 2.0, 4.0):
|
path = root / f"at-{timestamp:.0f}.png"
|
slides._run_file_command(
|
slides._extract_command(shutil.which("ffmpeg"), video, timestamp, path),
|
root / f"at-{timestamp:.0f}.stderr.log",
|
"VFR 定位",
|
)
|
with Image.open(path) as image:
|
extracted.append(np.asarray(image.convert("RGB"), np.uint8))
|
|
for sequence in ((scanned[0], scanned[2], scanned[4]), tuple(extracted)):
|
means = [np.mean(item, axis=(0, 1)) for item in sequence]
|
self.assertEqual([int(np.argmax(value)) for value in means], [0, 1, 2])
|
|
@unittest.skipUnless(shutil.which("ffmpeg"), "FFmpeg is required")
|
def test_large_stderr_does_not_deadlock_and_nonzero_is_reported(self):
|
with tempfile.TemporaryDirectory() as raw:
|
root = Path(raw)
|
ok_log = root / "ok.log"
|
slides._run_file_command(
|
[sys.executable, "-c", "import sys; sys.stderr.write('x'*300000)"],
|
ok_log,
|
"桩命令",
|
)
|
self.assertEqual(ok_log.stat().st_size, 300000)
|
bad_log = root / "bad.log"
|
with self.assertRaisesRegex(slides.SlideExtractionError, "exit=7"):
|
slides._run_file_command(
|
[sys.executable, "-c", "import sys; sys.stderr.write('boom'); raise SystemExit(7)"],
|
bad_log,
|
"桩命令",
|
)
|
|
def test_keyboard_interrupt_reaps_file_process_and_preserves_type(self):
|
events = []
|
handles = []
|
|
class FakeProcess:
|
stdout = None
|
|
def __init__(self):
|
self.wait_calls = 0
|
self.running = True
|
|
def wait(self, timeout=None):
|
self.wait_calls += 1
|
events.append(("wait", timeout))
|
if self.wait_calls == 1:
|
raise KeyboardInterrupt()
|
self.running = False
|
return 0
|
|
def poll(self):
|
return None if self.running else 0
|
|
def terminate(self):
|
events.append(("terminate", None))
|
|
def kill(self):
|
events.append(("kill", None))
|
|
def make_process(*args, **kwargs):
|
handles.append(kwargs["stderr"])
|
return FakeProcess()
|
|
with tempfile.TemporaryDirectory() as raw:
|
root = Path(raw)
|
source = root / "source.mp4"
|
source.write_bytes(b"stub")
|
output = root / "result"
|
page_feature = np.zeros((8, 8), np.uint8)
|
logical_page = slides.LogicalPage(page_feature, page_feature, 0, 1.0, 0.2, 0)
|
with mock.patch.object(slides, "probe_video", return_value=slides.VideoInfo(640, 360, 10)), mock.patch.object(
|
slides, "scan_video", return_value=[logical_page]
|
), mock.patch.object(slides, "_which_or_error", return_value="ffmpeg"), mock.patch.object(
|
slides.subprocess, "Popen", side_effect=make_process
|
):
|
with self.assertRaises(KeyboardInterrupt):
|
slides.extract_slides(str(source), str(output))
|
self.assertFalse(output.exists())
|
self.assertFalse(any(root.glob(".result.staging-*")))
|
self.assertTrue(handles[0].closed)
|
self.assertEqual(events[1][0], "terminate")
|
self.assertEqual(events[2][0], "wait")
|
|
def test_keyboard_interrupt_reaps_scan_process_and_preserves_type(self):
|
events = []
|
|
class Stdout:
|
def read(self, size):
|
raise KeyboardInterrupt()
|
|
def close(self):
|
events.append("close")
|
|
class FakeProcess:
|
def __init__(self):
|
self.stdout = Stdout()
|
self.running = True
|
|
def wait(self, timeout=None):
|
events.append("wait")
|
self.running = False
|
return 0
|
|
def poll(self):
|
return None if self.running else 0
|
|
def terminate(self):
|
events.append("terminate")
|
|
def kill(self):
|
events.append("kill")
|
|
with tempfile.TemporaryDirectory() as raw:
|
root = Path(raw)
|
source = root / "source.mp4"
|
source.write_bytes(b"stub")
|
output = root / "result"
|
with mock.patch.object(
|
slides.subprocess, "Popen", return_value=FakeProcess()
|
), mock.patch.object(slides, "_which_or_error", return_value="ffmpeg"), mock.patch.object(
|
slides, "probe_video", return_value=slides.VideoInfo(640, 360, 10)
|
):
|
with self.assertRaises(KeyboardInterrupt):
|
slides.extract_slides(str(source), str(output))
|
self.assertFalse(output.exists())
|
self.assertFalse(any(root.glob(".result.staging-*")))
|
self.assertEqual(events[:3], ["close", "terminate", "wait"])
|
|
def test_stderr_open_failure_is_wrapped_with_stage(self):
|
with tempfile.TemporaryDirectory() as raw, mock.patch.object(
|
slides.Path, "open", side_effect=OSError("denied")
|
):
|
with self.assertRaisesRegex(slides.SlideExtractionError, "桩阶段失败：denied"):
|
slides._run_file_command(["fake"], Path(raw) / "stderr.log", "桩阶段")
|
|
def test_cleanup_error_does_not_replace_keyboard_interrupt(self):
|
class FakeProcess:
|
stdout = None
|
|
def wait(self, timeout=None):
|
raise KeyboardInterrupt()
|
|
def poll(self):
|
return None
|
|
def terminate(self):
|
raise OSError("cleanup denied")
|
|
with tempfile.TemporaryDirectory() as raw, mock.patch.object(
|
slides.subprocess, "Popen", return_value=FakeProcess()
|
):
|
with self.assertRaises(KeyboardInterrupt):
|
slides._run_file_command(["fake"], Path(raw) / "stderr.log", "桩阶段")
|
|
def test_stderr_close_error_preserves_interrupt_in_both_helpers(self):
|
class CloseFailingHandle:
|
def close(self):
|
raise OSError("stderr close denied")
|
|
def flush(self):
|
return None
|
|
class FileProcess:
|
stdout = None
|
|
def __init__(self):
|
self.calls = 0
|
|
def wait(self, timeout=None):
|
self.calls += 1
|
if self.calls == 1:
|
raise KeyboardInterrupt()
|
return 0
|
|
def poll(self):
|
return None
|
|
def terminate(self):
|
return None
|
|
class ScanStdout:
|
def read(self, size):
|
raise KeyboardInterrupt()
|
|
def close(self):
|
return None
|
|
class ScanProcess:
|
def __init__(self):
|
self.stdout = ScanStdout()
|
|
def wait(self, timeout=None):
|
return 0
|
|
def poll(self):
|
return None
|
|
def terminate(self):
|
return None
|
|
with tempfile.TemporaryDirectory() as raw:
|
root = Path(raw)
|
with mock.patch.object(slides.Path, "open", return_value=CloseFailingHandle()), mock.patch.object(
|
slides.subprocess, "Popen", return_value=FileProcess()
|
):
|
with self.assertRaises(KeyboardInterrupt):
|
slides._run_file_command(["fake"], root / "file.stderr.log", "文件阶段")
|
with mock.patch.object(slides.Path, "open", return_value=CloseFailingHandle()), mock.patch.object(
|
slides.subprocess, "Popen", return_value=ScanProcess()
|
), mock.patch.object(slides, "_which_or_error", return_value="ffmpeg"):
|
with self.assertRaises(KeyboardInterrupt):
|
slides.scan_video(Path("x.mp4"), root, slides.VideoInfo(640, 360, 10))
|
|
def test_stderr_close_error_is_staged_on_normal_and_ordinary_failure(self):
|
class CloseFailingHandle:
|
def close(self):
|
raise OSError("stderr close denied")
|
|
def flush(self):
|
return None
|
|
class FakeProcess:
|
stdout = None
|
|
def __init__(self, code):
|
self.code = code
|
|
def wait(self, timeout=None):
|
return self.code
|
|
def poll(self):
|
return self.code
|
|
with tempfile.TemporaryDirectory() as raw:
|
root = Path(raw)
|
with mock.patch.object(slides.Path, "open", return_value=CloseFailingHandle()), mock.patch.object(
|
slides.subprocess, "Popen", return_value=FakeProcess(0)
|
):
|
with self.assertRaisesRegex(slides.SlideExtractionError, "文件阶段关闭 stderr 失败"):
|
slides._run_file_command(["fake"], root / "normal.log", "文件阶段")
|
with mock.patch.object(slides.Path, "open", return_value=CloseFailingHandle()), mock.patch.object(
|
slides.subprocess, "Popen", return_value=FakeProcess(7)
|
), mock.patch.object(slides, "_tail", return_value="boom"):
|
with self.assertRaisesRegex(slides.SlideExtractionError, "文件阶段失败（exit=7）"):
|
slides._run_file_command(["fake"], root / "failed.log", "文件阶段")
|
|
|
class PdfTests(unittest.TestCase):
|
def test_pdf_rgb_content_and_order(self):
|
with tempfile.TemporaryDirectory() as raw:
|
root = Path(raw)
|
paths = []
|
for index, color in enumerate(((255, 20, 20), (20, 40, 255))):
|
path = root / f"{index}.png"
|
Image.new("RGB", (17 + index, 11 + index), color).save(path)
|
paths.append(path)
|
pdf = root / "slides.pdf"
|
slides.write_image_pdf(paths, pdf)
|
decoded = pdf_rgb_pages(pdf)
|
self.assertEqual([item[0] for item in decoded], [(17, 11), (18, 12)])
|
for path, (_, rgb) in zip(paths, decoded):
|
with Image.open(path) as image:
|
self.assertEqual(hashlib.sha256(rgb).digest(), hashlib.sha256(image.convert("RGB").tobytes()).digest())
|
pdfinfo = shutil.which("pdfinfo")
|
if pdfinfo and Path(pdfinfo).suffix.lower() not in {".cmd", ".bat"}:
|
result = subprocess.run([pdfinfo, str(pdf)], capture_output=True, text=True, check=True)
|
self.assertIn("Pages: 2", result.stdout)
|
|
def test_twenty_pages_keep_one_decoded_image_active(self):
|
with tempfile.TemporaryDirectory() as raw:
|
root = Path(raw)
|
paths = []
|
for index in range(20):
|
path = root / f"{index:02d}.png"
|
Image.new("RGB", (12, 8), (index, index * 2, index * 3)).save(path)
|
paths.append(path)
|
active = 0
|
maximum = 0
|
|
class CountedOpen:
|
def __init__(self, path):
|
self.image = Image.open(path)
|
|
def __enter__(self):
|
nonlocal active, maximum
|
active += 1
|
maximum = max(maximum, active)
|
return self.image
|
|
def __exit__(self, exc_type, exc, traceback):
|
nonlocal active
|
self.image.close()
|
active -= 1
|
|
pdf = root / "twenty.pdf"
|
slides.write_image_pdf(paths, pdf, image_opener=CountedOpen)
|
self.assertEqual(maximum, 1)
|
self.assertEqual(active, 0)
|
self.assertEqual(len(pdf_rgb_pages(pdf)), 20)
|
|
|
class SafetyAndIntegrationTests(unittest.TestCase):
|
def test_existing_output_is_rejected_before_external_process(self):
|
with tempfile.TemporaryDirectory() as raw:
|
root = Path(raw)
|
source = root / "source.mp4"
|
source.write_bytes(b"not a video")
|
for kind in ("directory", "file"):
|
with self.subTest(kind=kind):
|
output = root / f"existing-{kind}"
|
if kind == "directory":
|
output.mkdir()
|
else:
|
output.write_bytes(b"occupied")
|
with mock.patch.object(slides.subprocess, "Popen") as popen:
|
with self.assertRaisesRegex(slides.SlideExtractionError, "默认不覆盖"):
|
slides.extract_slides(str(source), str(output))
|
popen.assert_not_called()
|
|
def test_live_and_broken_output_links_are_rejected_by_lexical_precheck(self):
|
with tempfile.TemporaryDirectory() as raw:
|
root = Path(raw)
|
source = root / "source.mp4"
|
source.write_bytes(b"stub")
|
for link_kind in ("live", "broken"):
|
with self.subTest(link_kind=link_kind):
|
output = slides._absolute_lexical(root / f"{link_kind}-link")
|
original_lexists = slides._lexists
|
|
def controlled_lexists(path):
|
return True if path == output else original_lexists(path)
|
|
with mock.patch.object(slides, "_lexists", side_effect=controlled_lexists), mock.patch.object(
|
slides.subprocess, "Popen"
|
) as popen:
|
with self.assertRaisesRegex(slides.SlideExtractionError, "默认不覆盖"):
|
slides.extract_slides(str(source), str(output))
|
popen.assert_not_called()
|
|
def test_input_symlink_is_rejected_before_external_process(self):
|
with tempfile.TemporaryDirectory() as raw:
|
root = Path(raw)
|
source = slides._absolute_lexical(root / "source-link.mp4")
|
with mock.patch.object(slides, "_is_symlink", side_effect=lambda path: path == source), mock.patch.object(
|
slides.subprocess, "Popen"
|
) as popen:
|
with self.assertRaisesRegex(slides.SlideExtractionError, "拒绝符号链接"):
|
slides.extract_slides(str(source), str(root / "result"))
|
popen.assert_not_called()
|
|
def test_baseexception_removes_staging_and_leaves_no_formal_output(self):
|
with tempfile.TemporaryDirectory() as raw:
|
root = Path(raw)
|
source = root / "source.mp4"
|
source.write_bytes(b"stub")
|
output = root / "result"
|
with mock.patch.object(slides, "probe_video", side_effect=KeyboardInterrupt()):
|
with self.assertRaises(KeyboardInterrupt):
|
slides.extract_slides(str(source), str(output))
|
self.assertFalse(output.exists())
|
self.assertFalse(any(root.glob(".result.staging-*")))
|
|
def test_commit_conflict_is_wrapped_and_staging_is_removed(self):
|
with tempfile.TemporaryDirectory() as raw:
|
root = Path(raw)
|
source = root / "source.mp4"
|
source.write_bytes(b"stub")
|
output = root / "result"
|
with mock.patch.object(slides, "probe_video", return_value=slides.VideoInfo(640, 360, 10)), mock.patch.object(
|
slides, "scan_video", return_value=[]
|
), mock.patch.object(slides, "extract_original_pages", return_value=[]), mock.patch.object(
|
slides, "write_image_pdf"
|
), mock.patch.object(slides.os, "rename", side_effect=FileExistsError("race")):
|
with self.assertRaisesRegex(slides.SlideExtractionError, "提交正式输出目录失败：race"):
|
slides.extract_slides(str(source), str(output))
|
self.assertFalse(output.exists())
|
self.assertFalse(any(root.glob(".result.staging-*")))
|
|
@unittest.skipUnless(shutil.which("ffmpeg") and shutil.which("ffprobe"), "FFmpeg is required")
|
def test_chinese_path_synthetic_video_end_to_end(self):
|
with tempfile.TemporaryDirectory() as raw:
|
root = Path(raw) / "中文路径"
|
root.mkdir()
|
frame_root = root / "frames"
|
frame_root.mkdir()
|
size = (640, 360)
|
|
def page_a(complete: bool):
|
image = Image.new("RGB", size, "white")
|
draw = ImageDraw.Draw(image)
|
draw.rectangle((0, 0, 639, 65), fill=(26, 78, 145))
|
draw.rectangle((55, 100, 585, 125), fill=(30, 30, 30))
|
if complete:
|
draw.rectangle((55, 165, 470, 190), fill=(30, 30, 30))
|
draw.rectangle((55, 230, 540, 255), fill=(30, 30, 30))
|
draw.rectangle((570, 320, 580, 330), fill=(220, 20, 20))
|
return image
|
|
def page_b():
|
image = Image.new("RGB", size, (18, 34, 68))
|
draw = ImageDraw.Draw(image)
|
draw.rectangle((65, 75, 575, 285), fill=(230, 190, 45))
|
draw.rectangle((110, 125, 530, 155), fill=(18, 34, 68))
|
return image
|
|
frames = [page_a(False)] * 3
|
frames += [Image.new("RGB", size, (127, 127, 127))]
|
frames += [page_a(True)] * 3
|
frames += [Image.new("RGB", size, (90, 20, 130))]
|
frames += [page_b()] * 3
|
frames += [Image.new("RGB", size, (20, 150, 90))]
|
frames += [page_a(True)] * 3
|
for index, image in enumerate(frames):
|
image.save(frame_root / f"frame_{index:03d}.png")
|
|
source = root / "合成会议录屏.mp4"
|
subprocess.run(
|
[
|
shutil.which("ffmpeg"), "-hide_banner", "-loglevel", "error", "-framerate", "1",
|
"-i", str(frame_root / "frame_%03d.png"), "-c:v", "libx264rgb", "-crf", "0", "-pix_fmt", "rgb24", str(source),
|
],
|
check=True,
|
)
|
before = (source.stat().st_size, source.stat().st_ctime_ns, source.stat().st_mtime_ns, sha256(source))
|
output = root / "验收输出"
|
result = subprocess.run(
|
[sys.executable, str(MODULE_PATH), str(source), "--output", str(output)],
|
capture_output=True,
|
text=True,
|
timeout=120,
|
)
|
self.assertEqual(result.returncode, 0, result.stderr + result.stdout)
|
self.assertEqual(
|
{entry.name for entry in output.iterdir()},
|
{"pages", "合成会议录屏.slides.pdf"},
|
)
|
pngs = sorted((output / "pages").glob("slide_*.png"))
|
self.assertEqual(len(pngs), 2, result.stdout)
|
for png in pngs:
|
with Image.open(png) as image:
|
self.assertEqual(image.size, size)
|
expected = [page_a(True), page_b()]
|
for actual_path, expected_image in zip(pngs, expected):
|
with Image.open(actual_path) as actual:
|
self.assertEqual(
|
hashlib.sha256(actual.convert("RGB").tobytes()).digest(),
|
hashlib.sha256(expected_image.tobytes()).digest(),
|
)
|
decoded = pdf_rgb_pages(output / "合成会议录屏.slides.pdf")
|
self.assertEqual(len(decoded), 2)
|
for png, (_, rgb) in zip(pngs, decoded):
|
with Image.open(png) as image:
|
self.assertEqual(hashlib.sha256(rgb).digest(), hashlib.sha256(image.convert("RGB").tobytes()).digest())
|
after = (source.stat().st_size, source.stat().st_ctime_ns, source.stat().st_mtime_ns, sha256(source))
|
self.assertEqual(after, before)
|
self.assertFalse(any(root.glob(".验收输出.staging-*")))
|
|
|
if __name__ == "__main__":
|
unittest.main()
|
