#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ Prototype verification for Critic Gate Remediation. Tests all 5 remediated components before proposing final patches. """ import os import sys import json import tempfile import subprocess from pathlib import Path from typing import List, Optional, Tuple import cv2 import numpy as np PROJECT_ROOT = Path("c:/Projects/KieuStory") CHARACTERS_DIR = PROJECT_ROOT / "04_Assets" / "characters" sys.path.insert(0, str(PROJECT_ROOT / "05_Production_Pipeline")) from antigravity_critic_gate import ( VideoCriticVerdict, ShotEvaluation, SceneEvaluation, _compute_color_histogram, ) # --------------------------------------------------------- # REMEDIATION 1 & 5: Media helpers # --------------------------------------------------------- def _extract_keyframe_bytes(video_path: Path, max_frames: int = 4, max_dimension: int = 720) -> List[bytes]: """Trích xuất JPEG keyframes bytes cho MLLM multimodal.""" if not video_path.exists() or video_path.stat().st_size == 0: return [] cap = cv2.VideoCapture(str(video_path)) if not cap.isOpened(): return [] fc = int(cap.get(cv2.CAP_PROP_FRAME_COUNT)) if fc <= 0: cap.release() return [] if fc <= max_frames: indices = list(range(fc)) else: indices = [int(fc * r) for r in [0.1, 0.35, 0.65, 0.9][:max_frames]] keyframe_bytes = [] for idx in indices: cap.set(cv2.CAP_PROP_POS_FRAMES, idx) ret, frame = cap.read() if not ret or frame is None: continue h, w = frame.shape[:2] if max(h, w) > max_dimension: scale = max_dimension / max(h, w) frame = cv2.resize(frame, (int(w * scale), int(h * scale)), interpolation=cv2.INTER_AREA) success, buf = cv2.imencode(".jpg", frame, [int(cv2.IMWRITE_JPEG_QUALITY), 80]) if success: keyframe_bytes.append(buf.tobytes()) cap.release() return keyframe_bytes def _extract_junction_frame_bytes(shot_video_paths: List[str], max_dimension: int = 720) -> List[bytes]: """Trích xuất cặp tail/head frames tại điểm tiếp biên giữa các shot cho MLLM.""" junction_bytes = [] for i in range(len(shot_video_paths) - 1): p1 = Path(shot_video_paths[i]) p2 = Path(shot_video_paths[i + 1]) if not p1.exists() or not p2.exists(): continue cap1 = cv2.VideoCapture(str(p1)) cap2 = cv2.VideoCapture(str(p2)) if not cap1.isOpened() or not cap2.isOpened(): cap1.release() cap2.release() continue fc1 = int(cap1.get(cv2.CAP_PROP_FRAME_COUNT)) cap1.set(cv2.CAP_PROP_POS_FRAMES, max(0, fc1 - 1)) ret1, f1 = cap1.read() cap2.set(cv2.CAP_PROP_POS_FRAMES, 0) ret2, f2 = cap2.read() cap1.release() cap2.release() for f in (f1, f2): if f is not None: h, w = f.shape[:2] if max(h, w) > max_dimension: scale = max_dimension / max(h, w) f = cv2.resize(f, (int(w * scale), int(h * scale)), interpolation=cv2.INTER_AREA) s, b = cv2.imencode(".jpg", f, [int(cv2.IMWRITE_JPEG_QUALITY), 80]) if s: junction_bytes.append(b.tobytes()) return junction_bytes def _check_audio_stream_remediated(video_path: Path) -> bool: """Kiểm tra luồng audio: parse JSON từ stdout, xác thực AAC/PCM/MP3 và channels > 0.""" if not video_path.exists() or video_path.stat().st_size == 0: return False try: cmd = [ "ffprobe", "-v", "error", "-select_streams", "a", "-show_entries", "stream=codec_name,channels,sample_rate", "-of", "json", str(video_path) ] res = subprocess.run(cmd, capture_output=True, text=True, timeout=5) if res.returncode != 0: return False if not res.stdout or not res.stdout.strip(): return False data = json.loads(res.stdout) streams = data.get("streams", []) if not streams: # Silent video -> tuân thủ Audio Guard (zero audio stream) return True for st in streams: codec = st.get("codec_name", "").lower() channels = st.get("channels", 0) if not codec or codec not in ("aac", "pcm_s16le", "pcm_s24le", "mp3", "opus", "flac"): return False if channels is None or channels <= 0: return False return True except Exception: return False def _find_character_portrait_remediated(character_name: str) -> Optional[Path]: """Tìm đường dẫn ảnh chân dung chuẩn trong 04_Assets/characters/.""" if not character_name or character_name.lower() in ("none", "scenery", ""): return None char_clean = character_name.lower().replace(" ", "_").strip() mapping = { "kim_trong": "01_Main_Protagonists/kim_trong_18yo_720p.png", "thuy_kieu": "01_Main_Protagonists/thuy_kieu_maiden_16yo_720p.png", "thuy_kieu_maiden": "01_Main_Protagonists/thuy_kieu_maiden_16yo_720p.png", "thuy_van": "01_Main_Protagonists/thuy_van_maiden_16yo_720p.png", "vuong_ong": "02_Vuong_Family_And_Fate/vuong_ong_55yo_720p.png", "vuong_ba": "02_Vuong_Family_And_Fate/vuong_ba_50yo_720p.png", "vuong_quan": "02_Vuong_Family_And_Fate/vuong_quan_16yo_720p.png", "dam_tien": "02_Vuong_Family_And_Fate/dam_tien_720p.png", "sai_nha": "05_Imperial_Court_And_Officials/sai_nha_720p.png", "thang_ban_to": "05_Imperial_Court_And_Officials/thang_ban_to_720p.png", "tieu_dong_kim_trong": "02_Vuong_Family_And_Fate/tieu_dong_kim_trong_720p.png", "tieu_dong": "02_Vuong_Family_And_Fate/tieu_dong_kim_trong_720p.png", "ma_giam_sinh": "04_Brokers_And_Brothels/ma_giam_sinh_720p.png", "tu_ba": "04_Brokers_And_Brothels/tu_ba_720p.png", "so_khanh": "04_Brokers_And_Brothels/so_khanh_720p.png", } for key, rel in mapping.items(): if key in char_clean: cand = CHARACTERS_DIR / rel if cand.exists(): return cand for p in CHARACTERS_DIR.glob("**/*.png"): stem = p.stem.lower() if any(tok in stem for tok in char_clean.split("_") if len(tok) >= 4): return p return None # --------------------------------------------------------- # REMEDIATION 3 & 4: Heuristic Shot Evaluation # --------------------------------------------------------- def evaluate_shot_remediated( shot_id: str, video_path_str: str, expected_character: str = "", prompt: str = "" ) -> VideoCriticVerdict: video_path = Path(video_path_str) if not video_path.exists(): shot_eval = ShotEvaluation( character_match=False, character_confidence=0.0, visual_defects=[f"Video file does not exist: {video_path.name}"], squint_or_extra_limbs=False, audio_guard_ok=False, score=0.0 ) return VideoCriticVerdict( overall_score=0.0, approved=False, shot_eval=shot_eval, suggested_action="RETAKE_SHOT", critique_notes=f"Lỗi: Không tìm thấy file video {video_path}" ) if video_path.stat().st_size == 0: shot_eval = ShotEvaluation( character_match=False, character_confidence=0.0, visual_defects=["Zero-byte corrupted file"], score=0.0 ) return VideoCriticVerdict( overall_score=0.0, approved=False, shot_eval=shot_eval, suggested_action="RETAKE_SHOT", critique_notes="Lỗi: File video rỗng 0-byte." ) cap = cv2.VideoCapture(str(video_path)) if not cap.isOpened(): shot_eval = ShotEvaluation( character_match=False, character_confidence=0.0, visual_defects=["OpenCV could not open or decode video stream"], score=0.0 ) return VideoCriticVerdict( overall_score=0.0, approved=False, shot_eval=shot_eval, suggested_action="RETAKE_SHOT", critique_notes="Lỗi: File video bị hỏng, không thể giải mã." ) fps = cap.get(cv2.CAP_PROP_FPS) or 24.0 frame_count = int(cap.get(cv2.CAP_PROP_FRAME_COUNT)) width = int(cap.get(cv2.CAP_PROP_FRAME_WIDTH)) height = int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT)) duration = frame_count / fps if fps > 0 else 0.0 defects: List[str] = [] squint_extra_limbs = False # 1. Thời lượng if duration < 1.0: defects.append(f"Duration too short ({duration:.1f}s, expected ~10s)") # 2. Độ phân giải if width == 0 or height == 0: defects.append("Zero resolution detected") elif width < 640 or height < 360: defects.append(f"Low resolution: {width}x{height}") # 3. Lấy mẫu khung hình sampled_frames = [] step = max(1, frame_count // 10) idx = 0 while cap.isOpened() and len(sampled_frames) < 10: cap.set(cv2.CAP_PROP_POS_FRAMES, idx) ret, frame = cap.read() if not ret or frame is None: break sampled_frames.append(frame) idx += step if idx >= frame_count: break cap.release() # Black frame for i, frame in enumerate(sampled_frames): mean_val = float(np.mean(frame)) if mean_val < 4.0: defects.append(f"Black frame detected at sample index {i} (luminance: {mean_val:.1f})") break # Frozen frame if len(sampled_frames) >= 4: diffs = [] for i in range(len(sampled_frames) - 1): diff = float(np.mean(np.abs(sampled_frames[i].astype(float) - sampled_frames[i+1].astype(float)))) diffs.append(diff) max_diff = max(diffs) if diffs else 0.0 if max_diff < 0.8: defects.append("Frozen video detected (virtually zero micro-motion across frames)") # 4. Audio Guard audio_ok = _check_audio_stream_remediated(video_path) if not audio_ok: defects.append("Audio stream missing or violated Audio Guard") # 5. Đối soát nhân vật (Defect 4 remediation) char_match = True char_conf = 1.0 exp_char_clean = expected_character.lower().strip() if exp_char_clean and exp_char_clean not in ("none", "scenery", ""): is_kieu_expected = ("thuy_kieu" in exp_char_clean) kieu_portrait = _find_character_portrait_remediated("thuy_kieu") exp_portrait = _find_character_portrait_remediated(exp_char_clean) if sampled_frames and kieu_portrait and kieu_portrait.exists(): sample_frame = sampled_frames[len(sampled_frames) // 2] sample_hist = _compute_color_histogram(sample_frame) kieu_img = cv2.imread(str(kieu_portrait)) h_kieu = _compute_color_histogram(kieu_img) if kieu_img is not None else None corr_kieu = float(cv2.compareHist(sample_hist, h_kieu, cv2.HISTCMP_CORREL)) if h_kieu is not None else 0.0 if not is_kieu_expected: corr_exp = 0.0 if exp_portrait and exp_portrait.exists(): exp_img = cv2.imread(str(exp_portrait)) if exp_img is not None: h_exp = _compute_color_histogram(exp_img) corr_exp = float(cv2.compareHist(sample_hist, h_exp, cv2.HISTCMP_CORREL)) # Nhiễm hình Thúy Kiều: tương đồng Kiều cao (>0.85) và tương đồng nhân vật kỳ vọng thấp (<0.60) if corr_kieu > 0.85 and (corr_exp < 0.60 or corr_kieu > corr_exp + 0.30): char_match = False char_conf = max(0.0, round(1.0 - corr_kieu, 2)) squint_extra_limbs = True defects.append( f"Character cross-contamination: Thúy Kiều pattern detected (corr={corr_kieu:.2f}) " f"in '{expected_character}' shot (expected character corr={corr_exp:.2f})" ) elif corr_exp > 0.50: char_conf = max(0.60, min(1.0, float(corr_exp))) else: char_conf = max(0.0, min(1.0, float(corr_kieu))) # 6. Tính điểm Shot (Defect 3 remediation: strict penalty & action override) score = 1.0 is_frozen = any("Frozen" in d for d in defects) has_black_frame = any("Black frame" in d for d in defects) is_too_short = any("Duration too short" in d for d in defects) is_low_res = any("resolution" in d.lower() for d in defects) has_contamination = any("cross-contamination" in d.lower() for d in defects) if not char_match or has_contamination: score -= 0.50 if is_frozen: score -= 0.35 if has_black_frame: score -= 0.35 if is_too_short: score -= 0.35 if is_low_res: score -= 0.35 if not audio_ok: score -= 0.25 if squint_extra_limbs: score -= 0.20 # Phạt thêm các lỗi nhỏ khác nếu có other_defects = [d for d in defects if not any(k in d for k in ("Frozen", "Black frame", "Duration too short", "resolution", "cross-contamination", "Audio stream"))] if other_defects: score -= min(0.30, len(other_defects) * 0.10) score = max(0.0, min(1.0, round(score, 3))) # Action override: bất kỳ lỗi nghiêm trọng nào cũng từ chối và bắt retake has_severe_defect = is_frozen or has_black_frame or is_too_short or is_low_res or has_contamination or not char_match or squint_extra_limbs or not audio_ok approved = (score >= 0.8) and (not has_severe_defect) action = "APPROVE" if approved else "RETAKE_SHOT" notes = ( f"Shot {shot_id}: Duration {duration:.1f}s, Res {width}x{height}, " f"CharMatch={char_match} (conf={char_conf:.2f}), Defects={len(defects)}, Score={score:.2f}" ) shot_eval = ShotEvaluation( character_match=char_match, character_confidence=char_conf, visual_defects=defects, squint_or_extra_limbs=squint_extra_limbs, audio_guard_ok=audio_ok, score=score ) return VideoCriticVerdict( overall_score=score, approved=approved, shot_eval=shot_eval, suggested_action=action, critique_notes=notes ) # --------------------------------------------------------- # REMEDIATION 2: Upfront Scene Evaluation # --------------------------------------------------------- def evaluate_scene_remediated(scene_id: str, shot_video_paths: List[str]) -> VideoCriticVerdict: if not shot_video_paths or len(shot_video_paths) < 1: scene_eval = SceneEvaluation( junction_smoothness=0.0, axis_180_ok=False, eyeline_ok=False, color_continuity=0.0, trigger_color_match=False, tempo_ok=False, trigger_trim_static=False, score=0.0 ) return VideoCriticVerdict( overall_score=0.0, approved=False, scene_eval=scene_eval, suggested_action="RETAKE_SHOT", critique_notes=f"Lỗi: Không có video nào được cung cấp cho cảnh {scene_id}." ) # Upfront validation of all video paths invalid_files = [] for p_str in shot_video_paths: p = Path(p_str) if not p.exists(): invalid_files.append(f"Missing file: {p.name}") continue if p.stat().st_size == 0: invalid_files.append(f"Zero-byte file: {p.name}") continue cap = cv2.VideoCapture(str(p)) if not cap.isOpened(): invalid_files.append(f"Unreadable file: {p.name}") cap.release() continue fc = int(cap.get(cv2.CAP_PROP_FRAME_COUNT)) if fc <= 0: invalid_files.append(f"Zero frames: {p.name}") cap.release() if invalid_files: scene_eval = SceneEvaluation( junction_smoothness=0.0, axis_180_ok=False, eyeline_ok=False, color_continuity=0.0, trigger_color_match=False, tempo_ok=False, trigger_trim_static=False, score=0.0 ) return VideoCriticVerdict( overall_score=0.0, approved=False, scene_eval=scene_eval, suggested_action="RETAKE_SHOT", critique_notes=f"Lỗi: Phát hiện {len(invalid_files)} video không hợp lệ trong cảnh {scene_id}: {'; '.join(invalid_files)}" ) # Frame extraction head_frames = [] tail_frames = [] static_frames_found = False axis_180_ok = True eyeline_ok = True for p_str in shot_video_paths: p = Path(p_str) cap = cv2.VideoCapture(str(p)) fc = int(cap.get(cv2.CAP_PROP_FRAME_COUNT)) cap.set(cv2.CAP_PROP_POS_FRAMES, 0) ret_h, head = cap.read() cap.set(cv2.CAP_PROP_POS_FRAMES, max(0, fc - 1)) ret_t, tail = cap.read() if fc > 60: cap.set(cv2.CAP_PROP_POS_FRAMES, fc // 2) r1, f1 = cap.read() cap.set(cv2.CAP_PROP_POS_FRAMES, min(fc - 1, fc // 2 + 48)) r2, f2 = cap.read() if r1 and r2 and f1 is not None and f2 is not None: diff = float(np.mean(np.abs(f1.astype(float) - f2.astype(float)))) if diff < 0.5: static_frames_found = True cap.release() if ret_h and head is not None: head_frames.append(head) if ret_t and tail is not None: tail_frames.append(tail) if len(head_frames) != len(shot_video_paths) or len(tail_frames) != len(shot_video_paths): scene_eval = SceneEvaluation( junction_smoothness=0.0, axis_180_ok=False, eyeline_ok=False, color_continuity=0.0, trigger_color_match=False, tempo_ok=False, trigger_trim_static=False, score=0.0 ) return VideoCriticVerdict( overall_score=0.0, approved=False, scene_eval=scene_eval, suggested_action="RETAKE_SHOT", critique_notes=f"Lỗi: Không thể trích xuất đủ frame từ tất cả shot trong cảnh {scene_id}." ) junction_scores = [] color_diffs = [] pair_count = min(len(tail_frames) - 1, len(head_frames) - 1) for i in range(max(0, pair_count)): f_tail = tail_frames[i] f_head = head_frames[i + 1] lum_t = float(np.mean(cv2.cvtColor(f_tail, cv2.COLOR_BGR2GRAY))) lum_h = float(np.mean(cv2.cvtColor(f_head, cv2.COLOR_BGR2GRAY))) lum_continuity = 1.0 - min(1.0, abs(lum_t - lum_h) / 255.0) mean_t = np.mean(f_tail, axis=(0, 1)) mean_h = np.mean(f_head, axis=(0, 1)) color_delta = float(np.linalg.norm(mean_t - mean_h) / 441.67) col_cont = max(0.0, 1.0 - color_delta) color_diffs.append(col_cont) corrs = [] for ch in range(3): ht = cv2.calcHist([f_tail], [ch], None, [32], [0, 256]) hh = cv2.calcHist([f_head], [ch], None, [32], [0, 256]) cv2.normalize(ht, ht, 0, 1, cv2.NORM_MINMAX) cv2.normalize(hh, hh, 0, 1, cv2.NORM_MINMAX) c = float(cv2.compareHist(ht, hh, cv2.HISTCMP_CORREL)) corrs.append(max(0.0, c)) hist_corr = float(np.mean(corrs)) if corrs else 1.0 j_smooth = round(0.5 * hist_corr + 0.5 * lum_continuity, 3) junction_scores.append(j_smooth) if len(shot_video_paths) == 1: avg_junction = 1.0 avg_color = 1.0 else: avg_junction = float(np.mean(junction_scores)) if junction_scores else 0.0 avg_color = float(np.mean(color_diffs)) if color_diffs else 0.0 trigger_color_match = avg_color < 0.70 trigger_trim_static = static_frames_found score = (avg_junction * 0.5 + avg_color * 0.5) if trigger_color_match: score -= 0.15 if trigger_trim_static: score -= 0.15 if not axis_180_ok: score -= 0.30 score = max(0.0, min(1.0, round(score, 3))) approved = score >= 0.8 if not approved and not axis_180_ok: action = "RETAKE_SHOT" elif trigger_color_match: action = "APPLY_COLOR_MATCH" elif trigger_trim_static: action = "TRIM_STATIC" else: action = "APPROVE" if approved else "RETAKE_SHOT" notes = ( f"Scene {scene_id} ({len(shot_video_paths)} shots): " f"Junction={avg_junction:.2f}, ColorCont={avg_color:.2f}, " f"ColorMatchTrigger={trigger_color_match}, TrimStaticTrigger={trigger_trim_static}, Score={score:.2f}" ) scene_eval = SceneEvaluation( junction_smoothness=avg_junction, axis_180_ok=axis_180_ok, eyeline_ok=eyeline_ok, color_continuity=avg_color, trigger_color_match=trigger_color_match, tempo_ok=not trigger_trim_static, trigger_trim_static=trigger_trim_static, score=score ) return VideoCriticVerdict( overall_score=score, approved=approved, scene_eval=scene_eval, suggested_action=action, critique_notes=notes ) # --------------------------------------------------------- # EXECUTE ADVERSARIAL VERIFICATION SUITE # --------------------------------------------------------- if __name__ == "__main__": print("=== RUNNING REMEDIATION PROTOTYPE TESTS ===") # Test 1: Missing Scene Files v_missing_scene = evaluate_scene_remediated("sc_missing", ["nonexistent_1.mp4", "nonexistent_2.mp4"]) print("Test 1 (Missing Scene Files):", v_missing_scene.overall_score, v_missing_scene.approved, v_missing_scene.suggested_action) assert v_missing_scene.overall_score == 0.0 assert v_missing_scene.approved is False assert v_missing_scene.suggested_action == "RETAKE_SHOT" # Test 2: Corrupt Scene Files with tempfile.NamedTemporaryFile(suffix=".mp4", delete=False) as f_c1, tempfile.NamedTemporaryFile(suffix=".mp4", delete=False) as f_c2: f_c1.write(b"CORRUPT1") f_c2.write(b"CORRUPT2") f_c1.flush() f_c2.flush() v_corrupt_scene = evaluate_scene_remediated("sc_corrupt", [f_c1.name, f_c2.name]) print("Test 2 (Corrupt Scene Files):", v_corrupt_scene.overall_score, v_corrupt_scene.approved, v_corrupt_scene.suggested_action) assert v_corrupt_scene.overall_score == 0.0 assert v_corrupt_scene.approved is False assert v_corrupt_scene.suggested_action == "RETAKE_SHOT" # Test 3: Frozen Video with tempfile.NamedTemporaryFile(suffix=".mp4", delete=False) as f_fz: out = cv2.VideoWriter(f_fz.name, cv2.VideoWriter_fourcc(*"mp4v"), 24, (1280, 720)) for _ in range(120): out.write(np.full((720, 1280, 3), 100, dtype=np.uint8)) out.release() v_frozen = evaluate_shot_remediated("shot_frozen", f_fz.name) print("Test 3 (Frozen Video):", v_frozen.overall_score, v_frozen.approved, v_frozen.suggested_action, v_frozen.shot_eval.visual_defects) assert v_frozen.overall_score <= 0.65 assert v_frozen.approved is False assert v_frozen.suggested_action == "RETAKE_SHOT" # Test 4: Cross Contamination (Thuy Kieu frame for Vuong Ong) kieu_p = CHARACTERS_DIR / "01_Main_Protagonists" / "thuy_kieu_maiden_16yo_720p.png" kieu_img = cv2.imread(str(kieu_p)) with tempfile.NamedTemporaryFile(suffix=".mp4", delete=False) as f_kieu: out = cv2.VideoWriter(f_kieu.name, cv2.VideoWriter_fourcc(*"mp4v"), 24, (kieu_img.shape[1], kieu_img.shape[0])) for _ in range(48): out.write(kieu_img) out.release() v_kieu_for_vuong = evaluate_shot_remediated("shot_vo", f_kieu.name, expected_character="vuong_ong") print("Test 4 (Thúy Kiều for Vương Ông):", v_kieu_for_vuong.overall_score, v_kieu_for_vuong.approved, v_kieu_for_vuong.shot_eval.character_match, v_kieu_for_vuong.suggested_action) assert v_kieu_for_vuong.overall_score <= 0.50 assert v_kieu_for_vuong.approved is False assert v_kieu_for_vuong.shot_eval.character_match is False assert v_kieu_for_vuong.suggested_action == "RETAKE_SHOT" v_kieu_for_kim = evaluate_shot_remediated("shot_kt", f_kieu.name, expected_character="kim_trong") print("Test 4b (Thúy Kiều for Kim Trọng):", v_kieu_for_kim.overall_score, v_kieu_for_kim.approved, v_kieu_for_kim.shot_eval.character_match, v_kieu_for_kim.suggested_action) assert v_kieu_for_kim.overall_score <= 0.50 assert v_kieu_for_kim.approved is False assert v_kieu_for_kim.shot_eval.character_match is False assert v_kieu_for_kim.suggested_action == "RETAKE_SHOT" # Test 5: Audio Stream check on corrupt/empty vs valid with tempfile.NamedTemporaryFile(suffix=".mp4", delete=False) as f_empty: f_empty.write(b"") f_empty.flush() print("Test 5a (0-byte Audio Check):", _check_audio_stream_remediated(Path(f_empty.name))) assert _check_audio_stream_remediated(Path(f_empty.name)) is False print("=== ALL PROTOTYPE TESTS PASSED PERFECTLY! ===")