feat(pipeline): v3-v4 dedup + panorama stitching + 1080p support

- HSV-aware _trim_to_content (white ratio 30-97%) - pHash cluster dedup: dHash 32x32(1024bit), max_hamming=20 - Panoramic stitching: template matching scroll offset detection - 4-stage pipeline: MSE -> Panorama -> pHash - 1080p download priority + MAX_FRAME_WIDTH=1280 cap - test_pipeline.py with YouTube URLs and --download mode - 3 new known-issues documented - devlog + STATUS.md updated
2026-03-25 21:58:48 +09:00
parent 3d3f74b082
commit 98381d2893
13 changed files with 836 additions and 421 deletions
--- a/test_pipeline.py
+++ b/test_pipeline.py
@@ -0,0 +1,110 @@
+#!/usr/bin/env python3
+"""로컬 캐시된 mp4 파일로 파이프라인 테스트 (다운로드 스킵)
+1080p 다운로드 모드: python test_pipeline.py --download
+"""
+import sys
+import os
+from pathlib import Path
+import importlib.util
+import argparse
+import gc
+
+# youtube_tab_to_pdf 모듈 임포트
+spec = importlib.util.spec_from_file_location(
+    "pipeline", str(Path(__file__).parent / "youtube_tab_to_pdf.py"))
+pipeline = importlib.util.module_from_spec(spec)
+spec.loader.exec_module(pipeline)
+
+# 테스트용 YouTube URLs
+TEST_URLS = {
+    "video_1": "https://www.youtube.com/watch?v=x76IMSvWR0o",  # 晴る
+    "video_2": "https://www.youtube.com/watch?v=90BWvJY6KbE",  # 新宝島
+    "video_3": "https://www.youtube.com/watch?v=Ri9g4lwnrJQ",  # 空奏列車
+}
+
+
+def test_video(mp4_path: Path, label: str):
+    """단일 영상 테스트 — 다운로드 없이 로컬 파일 직접 사용"""
+    print(f"\n{'='*60}")
+    print(f"테스트: {label}")
+    print(f"파일: {mp4_path.name}")
+    print(f"{'='*60}")
+
+    output_dir = Path("output")
+    debug_dir = output_dir / "debug_frames" / label
+    debug_dir.mkdir(parents=True, exist_ok=True)
+
+    # Step 2: 프레임 추출
+    frames = pipeline.extract_frames(mp4_path)
+
+    # Step 3: 패턴 감지
+    pattern = pipeline.detect_pattern(frames)
+
+    # Step 4: 고유 프레임 추출
+    if pattern == "scroll":
+        unique = pipeline.extract_unique_scroll(frames)
+    elif pattern == "split":
+        unique = pipeline.extract_unique_split(frames)
+    else:
+        unique = pipeline.extract_unique_overlay(frames)
+
+    # Step 5: PDF 생성
+    pdf_path = output_dir / f"test_{label}.pdf"
+    pipeline.generate_pdf(unique, pdf_path, debug_dir=debug_dir)
+
+    print(f"\n결과: {pattern} / {len(unique)}개 고유 프레임")
+    return pattern, len(unique)
+
+
+def download_test_videos():
+    """1080p로 테스트 영상 다운로드"""
+    output_dir = Path("output")
+    output_dir.mkdir(exist_ok=True)
+
+    for label, url in TEST_URLS.items():
+        print(f"\n--- {label} 다운로드 ---")
+        try:
+            video_path, title = pipeline.download_video(url, output_dir)
+            print(f"  → 완료: {video_path.name}")
+        except Exception as e:
+            print(f"  → 실패: {e}")
+
+
+def main():
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--download", action="store_true",
+                        help="1080p로 테스트 영상 다운로드")
+    args = parser.parse_args()
+
+    if args.download:
+        download_test_videos()
+        return
+
+    output_dir = Path("output")
+    mp4_files = sorted(output_dir.glob("*.mp4"))
+
+    if not mp4_files:
+        print("output/ 폴더에 mp4 파일이 없습니다!")
+        print("  → python test_pipeline.py --download 로 영상 다운로드")
+        sys.exit(1)
+
+    print(f"캐시된 영상 {len(mp4_files)}개 발견:")
+    for f in mp4_files:
+        print(f"  - {f.name} ({f.stat().st_size / 1024 / 1024:.1f} MB)")
+
+    results = {}
+    for i, mp4 in enumerate(mp4_files):
+        label = f"video_{i+1}"
+        pattern, count = test_video(mp4, label)
+        results[label] = (mp4.name, pattern, count)
+        gc.collect()  # 1080p 프레임 메모리 해제
+
+    print(f"\n{'='*60}")
+    print("전체 결과 요약:")
+    print(f"{'='*60}")
+    for label, (name, pattern, count) in results.items():
+        print(f"  {label}: {pattern:8s} → {count:4d}개 프레임 | {name[:40]}")
+
+
+if __name__ == "__main__":
+    main()