#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ ============================================================================= THẬP NGŨ NIÊN (THE FIFTEEN SPRINGS) - 2-TIER QUALITY GATE & CLASSIFIER ============================================================================= Module: antigravity_critic_gate.py Author: Nguyễn Sĩ Sơn / AI Cinema Production Team Standard: AGENTS.md, GEMINI.md, ORIGINAL_REQUEST.md §R1 Cổng kiểm duyệt 2 tầng (Shot-Level 10s & Scene-Level 30-40s): - Tầng 1 (Shot Gate): Thẩm định đúng nhân vật (không nhầm vai/morphing), không dị tật thị giác (mắt lé, thừa chi, mặt sáp), tuân thủ Audio Guard, ngưỡng phê duyệt overall_score >= 0.8. - Tầng 2 (Scene Gate): Kiểm duyệt điểm giao cắt tiếp biên (junction), quy tắc trục 180 độ, độ khớp hướng nhìn, độ chênh màu (color continuity), nhịp điệu cắt dựng (editing tempo & static freeze > 2s), tự động kích hoạt Color Match hoặc Trim Static. - Động cơ thẩm định: 1. Primary: Google Antigravity SDK (Agent + LocalAgentConfig, response_schema=VideoCriticVerdict) 2. Secondary: Google GenAI Client (gemini-2.5-flash with structured output) 3. Offline Heuristic Fallback: OpenCV/FFprobe (duration, resolution, audio stream, histogram delta between frames, black/frozen frame detection). ============================================================================= """ import os import sys import re import json import shutil import subprocess import asyncio import inspect from datetime import datetime from pathlib import Path from typing import List, Optional, Dict, Any, Tuple, Union import cv2 import numpy as np from pydantic import BaseModel, Field, model_validator # Reconfigure console UTF-8 on Windows if sys.platform == "win32": try: sys.stdout.reconfigure(encoding="utf-8") sys.stderr.reconfigure(encoding="utf-8") except Exception: pass BASE_DIR = Path(__file__).resolve().parent.parent CHARACTERS_DIR = BASE_DIR / "04_Assets" / "characters" KEYFRAMES_DIR = BASE_DIR / "04_Assets" / "keyframes" VIDEOS_DIR = BASE_DIR / "04_Assets" / "videos" CRITIC_QUEUE_DIR = BASE_DIR / "04_Assets" / "critic_queue" # Tự động nạp cấu hình từ .env (ưu tiên ghi đè biến môi trường cũ) try: from dotenv import load_dotenv _workspace_env = BASE_DIR / ".env" _mem0_env = Path(r"C:\Users\Admin\.gemini\antigravity\mem0_data\.env") if _workspace_env.exists(): load_dotenv(_workspace_env, override=True) elif _mem0_env.exists(): load_dotenv(_mem0_env, override=True) else: load_dotenv(override=True) except Exception: pass try: from contact_sheet_engine import ( ZeroWasteContactSheet, generate_zero_waste_contact_sheet, extract_contact_sheet_bytes, compute_zero_waste_grid ) except ImportError: try: from .contact_sheet_engine import ( ZeroWasteContactSheet, generate_zero_waste_contact_sheet, extract_contact_sheet_bytes, compute_zero_waste_grid ) except Exception: ZeroWasteContactSheet = None generate_zero_waste_contact_sheet = None extract_contact_sheet_bytes = None compute_zero_waste_grid = lambda n: (n, 1) # ============================================================================= # FLOWKIT STRICT 3-TIER ERROR CATALOG (14 KNOWN AI VIDEO ERRORS) # ============================================================================= FLOWKIT_ERROR_TIERS: Dict[str, str] = { # 5 Critical Errors "character_drift": "critical", "breed_swap": "critical", "role_reversal": "critical", "brand_logo_text": "critical", "character_count_drift": "critical", # 5 High Errors "camera_drift": "high", "object_morph": "high", "reverse_motion": "high", "human_hands_limbs": "high", "scale_break": "high", # 4 Minor Errors "lighting_jitter": "minor", "micro_blur": "minor", "background_warp": "minor", "color_shift": "minor", } FLOWKIT_ERROR_CATALOG: Dict[str, Dict[str, str]] = { "character_drift": { "tier": "critical", "description": "Mặt nhân vật biến dạng / sai lệch đặc điểm nhận diện so với Character Bible" }, "breed_swap": { "tier": "critical", "description": "Tráo đổi vai diễn / nhầm lẫn nhân vật giữa các vai chính - phụ" }, "role_reversal": { "tier": "critical", "description": "Đổi vị thế nhân vật / nhầm vai chủ khách hoặc hành vi đối nghịch" }, "brand_logo_text": { "tier": "critical", "description": "Watermark lạ, logo thương hiệu, hoặc text glitch nhân tạo" }, "character_count_drift": { "tier": "critical", "description": "Thừa / thiếu nhân vật chính, nhân vật đột ngột xuất hiện hoặc biến mất" }, "camera_drift": { "tier": "high", "description": "Rung lắc bất thường, chuyển động camera giật cục, gãy trục máy" }, "object_morph": { "tier": "high", "description": "Đạo cụ / trang phục bị biến hình vô lý, mất tính nhất quán vật lý" }, "reverse_motion": { "tier": "high", "description": "Chuyển động đi lùi vô lý, nước chảy ngược, quỹ đạo giật ngược" }, "human_hands_limbs": { "tier": "high", "description": "Dị tật bàn tay 6 ngón, dính ngón, thừa chi, mắt lé" }, "scale_break": { "tier": "high", "description": "Sai lệch tỷ lệ cơ thể hoặc kích thước dị thường so với bối cảnh" }, "lighting_jitter": { "tier": "minor", "description": "Nhấp nháy ánh sáng nhẹ giữa các khung hình" }, "micro_blur": { "tier": "minor", "description": "Nhòe chuyển động nhẹ, giảm chi tiết và độ tương phản vi mô" }, "background_warp": { "tier": "minor", "description": "Hậu cảnh bị méo nhẹ ở biên" }, "color_shift": { "tier": "minor", "description": "Sai lệch tông màu nhẹ giữa đầu và đuôi cú máy" }, } # ============================================================================= # PYDANTIC V2 SCHEMAS # ============================================================================= class ShotEvaluation(BaseModel): """Đánh giá chi tiết ở cấp độ từng cú máy 10s (Tầng 1 - Shot Gate).""" character_match: bool = Field( default=True, description="Xác nhận nhân vật hiển thị khớp với vai diễn kịch bản và Character Bible" ) character_confidence: float = Field( default=1.0, ge=0.0, le=1.0, description="Độ tin cậy nhận diện nhân vật (0.0 đến 1.0)" ) character_consistency: float = Field( default=1.0, ge=0.0, le=1.0, description="Điểm nhất quán nhân vật (0.0 đến 1.0; nếu có lỗi Critical thì <= 0.30 theo chuẩn FlowKit)" ) visual_defects: List[str] = Field( default_factory=list, description="Danh sách dị tật thị giác phát hiện được (mặt sáp, nhòe hình, đen hình, đơ hình)" ) detected_errors: List[str] = Field( default_factory=list, description="Danh sách các lỗi phát hiện theo bộ danh mục 14 lỗi chuẩn FlowKit" ) error_severities: Dict[str, str] = Field( default_factory=dict, description="Phân loại mức độ nghiêm trọng: critical, high, minor" ) squint_or_extra_limbs: bool = Field( default=False, description="Cờ phát hiện mắt lé, ngón tay biến dạng hoặc thừa chi" ) audio_guard_ok: bool = Field( default=True, description="Xác nhận âm thanh tuân thủ Audio Guard (không nhạc rác, không tạp âm rè)" ) score: float = Field( default=1.0, ge=0.0, le=1.0, description="Điểm đánh giá shot từ 0.0 đến 1.0 (ngưỡng đạt >= 0.8)" ) class SceneEvaluation(BaseModel): """Đánh giá chi tiết ở cấp độ toàn cảnh 30-40s (Tầng 2 - Scene Gate).""" junction_smoothness: float = Field( default=1.0, ge=0.0, le=1.0, description="Độ mượt mà của điểm giao cắt tiếp biên giữa các shot liền kề (0.0 đến 1.0)" ) axis_180_ok: bool = Field( default=True, description="Tuân thủ quy tắc trục 180 độ, không nhảy trục vô cớ" ) eyeline_ok: bool = Field( default=True, description="Độ khớp hướng nhìn (eyeline match) giữa các nhân vật đối thoại" ) color_continuity: float = Field( default=1.0, ge=0.0, le=1.0, description="Độ đồng nhất màu sắc và ánh sáng giữa các shot (0.0 đến 1.0)" ) trigger_color_match: bool = Field( default=False, description="Tự động kích hoạt bộ cân bằng màu (Color Match) khi độ chênh màu vượt ngưỡng" ) tempo_ok: bool = Field( default=True, description="Nhịp điệu dựng phim đạt chuẩn, không có khoảng chết vô hồn" ) trigger_trim_static: bool = Field( default=False, description="Tự động kích hoạt cắt tỉa (Trim) khi phát hiện đoạn tĩnh kéo dài > 2 giây" ) score: float = Field( default=1.0, ge=0.0, le=1.0, description="Điểm đánh giá scene từ 0.0 đến 1.0 (ngưỡng đạt >= 0.8)" ) class VideoCriticVerdict(BaseModel): """ Kết quả thẩm định tổng hợp của Cổng Kiểm duyệt Antigravity. Ngưỡng phê duyệt: overall_score >= 0.8. Tích hợp bộ chặn cứng FlowKit Hard Caps (Critical <= 0.59, High <= 0.74). """ overall_score: float = Field( ..., ge=0.0, le=1.0, description="Điểm tổng hợp từ 0.0 đến 1.0. Ngưỡng chấp thuận: >= 0.8" ) approved: bool = Field( default=False, description="Trạng thái phê duyệt (True nếu overall_score >= 0.8 và không có lỗi nghiêm trọng)" ) shot_eval: Optional[ShotEvaluation] = Field( default=None, description="Đánh giá chi tiết cấp độ shot (nếu kiểm duyệt shot)" ) scene_eval: Optional[SceneEvaluation] = Field( default=None, description="Đánh giá chi tiết cấp độ scene (nếu kiểm duyệt scene)" ) detected_errors: List[str] = Field( default_factory=list, description="Tổng hợp danh sách các lỗi phát hiện theo chuẩn FlowKit 14 lỗi" ) error_severities: Dict[str, str] = Field( default_factory=dict, description="Mức độ nghiêm trọng của từng lỗi: critical, high, minor" ) contact_sheet_path: Optional[str] = Field( default=None, description="Đường dẫn file Zero-Waste Contact Sheet phục vụ kiểm toán" ) suggested_action: str = Field( default="APPROVE", description="Hành động đề xuất: APPROVE, RETAKE_SHOT, APPLY_COLOR_MATCH, TRIM_STATIC" ) critique_notes: str = Field( default="", description="Ghi chú chi tiết từ Hội đồng Phê bình MLLM và bộ phân tích Heuristic" ) @model_validator(mode="after") def sync_approval_state(self) -> "VideoCriticVerdict": """ Đồng bộ trạng thái phê duyệt nghiêm ngặt theo chỉ thị FlowKit Hard Caps: 1. Nếu phát hiện BẤT KỲ lỗi Critical nào: - character_consistency BẮT BUỘC bị ép xuống <= 0.30 (tương đương 3.0/10). - overall_score BẮT BUỘC bị ép xuống <= 0.59 (tương đương 5.9/10). - approved = False, suggested_action = "RETAKE_SHOT" (triệt tiêu sự khoan dung của LLM). 2. Nếu phát hiện lỗi High: - overall_score BẮT BUỘC bị ép xuống <= 0.74 (dưới ngưỡng 0.8). - approved = False, suggested_action = "RETAKE_SHOT". 3. Nếu không có lỗi Critical/High: - Duy trì phê duyệt khi overall_score >= 0.8 và action != "RETAKE_SHOT". """ # Thu thập toàn bộ errors từ root verdict và shot_eval all_errors = list(self.detected_errors) all_severities = dict(self.error_severities) if self.shot_eval: for e in self.shot_eval.detected_errors: if e not in all_errors: all_errors.append(e) for k, v in self.shot_eval.error_severities.items(): if k not in all_severities: all_severities[k] = v self.detected_errors = all_errors self.error_severities = all_severities has_critical = any( all_severities.get(e) == "critical" or FLOWKIT_ERROR_TIERS.get(e) == "critical" for e in all_errors ) has_high = any( all_severities.get(e) == "high" or FLOWKIT_ERROR_TIERS.get(e) == "high" for e in all_errors ) if has_critical: self.overall_score = min(float(self.overall_score), 0.59) if self.shot_eval: self.shot_eval.character_consistency = min(float(self.shot_eval.character_consistency), 0.30) self.shot_eval.score = min(float(self.shot_eval.score), 0.59) self.shot_eval.character_match = False self.approved = False self.suggested_action = "RETAKE_SHOT" elif has_high: self.overall_score = min(float(self.overall_score), 0.74) if self.shot_eval: self.shot_eval.score = min(float(self.shot_eval.score), 0.74) self.approved = False self.suggested_action = "RETAKE_SHOT" else: min_score = 0.75 if self.scene_eval is not None else 0.80 if self.overall_score >= min_score and self.suggested_action != "RETAKE_SHOT": self.approved = True else: self.approved = False return self # ============================================================================= # HELPER FUNCTIONS: MEDIA INSPECTION & CHARACTER CORRELATION # ============================================================================= def _find_character_portrait(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 các tên thông dụng mapping = { "kim_trong": "01_Main_Protagonists/kim_trong_18yo/kim_trong_18yo_720p.png", "thuy_kieu": "01_Main_Protagonists/thuy_kieu_maiden_16yo/thuy_kieu_maiden_16yo_720p.png", "thuy_kieu_maiden": "01_Main_Protagonists/thuy_kieu_maiden_16yo/thuy_kieu_maiden_16yo_720p.png", "thuy_van": "01_Main_Protagonists/thuy_van_maiden_16yo/thuy_van_maiden_16yo_720p.png", "vuong_ong": "02_Vuong_Family_And_Fate/vuong_ong_55yo/vuong_ong_55yo_720p.png", "vuong_ba": "02_Vuong_Family_And_Fate/vuong_ba_50yo/vuong_ba_50yo_720p.png", "vuong_quan": "02_Vuong_Family_And_Fate/vuong_quan_16yo/vuong_quan_16yo_720p.png", "dam_tien": "02_Vuong_Family_And_Fate/dam_tien/dam_tien_720p.png", "sai_nha": "05_Imperial_Court_And_Officials/sai_nha/sai_nha_720p.png", "thang_ban_to": "05_Imperial_Court_And_Officials/thang_ban_to/thang_ban_to_720p.png", "tieu_dong_kim_trong": "02_Vuong_Family_And_Fate/tieu_dong_kim_trong/tieu_dong_kim_trong_720p.png", "tieu_dong": "02_Vuong_Family_And_Fate/tieu_dong_kim_trong/tieu_dong_kim_trong_720p.png", "ma_giam_sinh": "04_Brokers_And_Brothels/ma_giam_sinh/ma_giam_sinh_720p.png", "tu_ba": "04_Brokers_And_Brothels/tu_ba/tu_ba_720p.png", "so_khanh": "04_Brokers_And_Brothels/so_khanh/so_khanh_720p.png", } for key, rel in mapping.items(): if key in char_clean: cand = CHARACTERS_DIR / rel if cand.exists(): return cand # Quét toàn bộ thư mục characters nếu không trúng alias 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 def _extract_keyframe_bytes(video_path: Path, max_frames: int = 4, max_dimension: int = 720) -> List[bytes]: """ Trích xuất các khung hình đại diện (keyframes) từ video và nén thành JPEG bytes để nạp vào nội dung MLLM (Multimodal Gemini/Antigravity). """ 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ác cặp khung hình tiếp biên (tail frame shot N vs head frame shot N+1) để nạp vào nội dung MLLM đánh giá tính liền mạch của Scene. """ 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(video_path: Path) -> bool: """ Kiểm tra luồng âm thanh thông qua ffprobe. - Không có luồng audio (video silent / synthetic): Hợp lệ theo Audio Guard (không sinh nhạc rác). - Có luồng audio: Phân tích JSON từ stdout để xác thực codec (AAC/PCM/MP3) và channel count > 0. - Nếu ffprobe trả về lỗi (exit code != 0) hoặc stdout không hợp lệ: Trả về False. """ 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: # Video không có luồng audio (silent) -> Hợp lệ theo Audio Guard 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 _compute_color_histogram(image: np.ndarray) -> np.ndarray: """Tính toán HSV color histogram 3D chuẩn hóa.""" hsv = cv2.cvtColor(image, cv2.COLOR_BGR2HSV) hist = cv2.calcHist([hsv], [0, 1], None, [30, 32], [0, 180, 0, 256]) cv2.normalize(hist, hist, alpha=0, beta=1, norm_type=cv2.NORM_MINMAX) return hist # ============================================================================= # ENGINE 3: OFFLINE HEURISTIC EVALUATION ENGINE # ============================================================================= class OfflineHeuristicEngine: """ Động cơ thẩm định xác định (deterministic) dựa trên OpenCV và FFprobe. Đánh giá độ dài, độ phân giải, luồng âm thanh, phát hiện frame đen/đơ hình, đo tương quan màu sắc tiếp biên và đối soát nhiễm vai nhân vật. """ @staticmethod def evaluate_shot( 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. Kiểm tra thời lượng if duration < 1.0: defects.append(f"Duration too short ({duration:.1f}s, expected ~10s)") # 2. Kiểm tra độ phân giải & Tỷ lệ khung hình (Aspect Ratio Hard Gate) 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}") aspect_ratio = width / height if height > 0 else 0.0 is_vertical_video = bool(aspect_ratio < 1.33) # Video dọc 9:16 (~0.56) hoặc vuông (1:1) if is_vertical_video: defects.append( f"CRITICAL: Invalid aspect ratio {aspect_ratio:.3f} ({width}x{height}). " f"Cinema Feature cut requires 16:9 widescreen (1280x720). Vertical 9:16 is STRICTLY PROHIBITED in cinema cuts." ) # 3. Lấy mẫu khung hình để kiểm tra black frame & frozen frame 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() # Kiểm tra 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 # Kiểm tra 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. Kiểm tra luồng Audio Guard audio_ok = _check_audio_stream(video_path) if not audio_ok: defects.append("Audio stream missing or violated Audio Guard") # 5. Đối soát nhân vật (Character Conformance & Universal Cross-Contamination Guard) 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("thuy_kieu") exp_portrait = _find_character_portrait(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ạo Zero-Waste Contact Sheet phục vụ kiểm toán thị giác cs_path_str: Optional[str] = None if generate_zero_waste_contact_sheet is not None: try: _, cs_file = generate_zero_waste_contact_sheet(video_path, num_frames=6) if cs_file: cs_path_str = str(cs_file.resolve()) except Exception: pass # 7. Phân tích bộ danh mục 14 lỗi video AI (FlowKit Error Catalog) 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) detected_errors: List[str] = [] error_severities: Dict[str, str] = {} char_consistency = float(char_conf) if not char_match or has_contamination: detected_errors.extend(["character_drift", "breed_swap"]) error_severities["character_drift"] = "critical" error_severities["breed_swap"] = "critical" char_consistency = min(char_consistency, 0.30) if has_black_frame: detected_errors.append("brand_logo_text") error_severities["brand_logo_text"] = "critical" if is_frozen: detected_errors.append("camera_drift") error_severities["camera_drift"] = "high" if is_vertical_video: detected_errors.append("scale_break") error_severities["scale_break"] = "critical" elif is_too_short or is_low_res: detected_errors.append("scale_break") error_severities["scale_break"] = "high" if squint_extra_limbs: detected_errors.append("human_hands_limbs") error_severities["human_hands_limbs"] = "high" # Phân tích độ nhòe (Micro Blur) qua phương sai Laplacian # Chỉ kiểm tra khi frame có cấu trúc chi tiết thực tế (std > 20) để tránh bắt nhầm video gradient/hoạt họa if sampled_frames: try: mean_std = float(np.mean([np.std(f) for f in sampled_frames if f is not None])) if mean_std > 20.0: lap_vars = [ float(cv2.Laplacian(cv2.cvtColor(f, cv2.COLOR_BGR2GRAY), cv2.CV_64F).var()) for f in sampled_frames if f is not None ] if lap_vars and float(np.mean(lap_vars)) < 15.0: detected_errors.append("micro_blur") error_severities["micro_blur"] = "minor" except Exception: pass # Phân tích nhấp nháy ánh sáng (Lighting Jitter) if len(sampled_frames) >= 4: try: lums = [float(np.mean(cv2.cvtColor(f, cv2.COLOR_BGR2GRAY))) for f in sampled_frames if f is not None] if lums and float(np.std(lums)) > 45.0: detected_errors.append("lighting_jitter") error_severities["lighting_jitter"] = "minor" except Exception: pass # Phân tích sai lệch màu sắc (Color Shift) giữa đầu và cuối cú máy if len(sampled_frames) >= 2: try: head_hsv = cv2.cvtColor(sampled_frames[0], cv2.COLOR_BGR2HSV) tail_hsv = cv2.cvtColor(sampled_frames[-1], cv2.COLOR_BGR2HSV) mean_h = np.mean(head_hsv, axis=(0, 1)) mean_t = np.mean(tail_hsv, axis=(0, 1)) delta_hsv = float(np.linalg.norm(mean_h - mean_t) / 255.0) if delta_hsv > 0.45: detected_errors.append("color_shift") error_severities["color_shift"] = "minor" except Exception: pass # 8. Tính điểm Shot (Score calculation with strict penalty for severe defects) score = 1.0 if is_vertical_video: score -= 0.60 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 bổ sung cho các lỗi nhẹ khác other_defects = [ d for d in defects if not any(k in d for k in ("aspect ratio", "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 buộc RETAKE_SHOT has_severe_defect = ( is_vertical_video or 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) or any(error_severities.get(e) in ("critical", "high") for e in detected_errors) ) 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)}, " f"FlowKitErrors={detected_errors}, Score={score:.2f}" ) shot_eval = ShotEvaluation( character_match=char_match, character_confidence=char_conf, character_consistency=char_consistency, visual_defects=defects, detected_errors=detected_errors, error_severities=error_severities, 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, detected_errors=detected_errors, error_severities=error_severities, contact_sheet_path=cs_path_str, suggested_action=action, critique_notes=notes ) @staticmethod def evaluate_scene(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}." ) # 1. Kiểm tra xác thực ban đầu (Upfront Validation) toàn bộ danh sách video 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)) w = int(cap.get(cv2.CAP_PROP_FRAME_WIDTH)) h = int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT)) if fc <= 0: invalid_files.append(f"Zero frames: {p.name}") elif w > 0 and h > 0 and (w / h < 1.33): invalid_files.append(f"Vertical 9:16 aspect ratio prohibited in 16:9 cinema feature: {p.name} ({w}x{h})") 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)}" ) # 2. Trích xuất head frames và tail frames 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() # Kiểm tra static freeze > 2s (48 frames diff < 0.5) 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}." ) # 3. Đo tiếp biên giữa tail[i] và head[i+1] 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))) # Đối với cảnh nhiều shot (>= 6 shots) có nhiều cú cắt đổi góc quay (shot/reverse-shot), ngưỡng 0.75 là đạt chuẩn scene_threshold = 0.75 if len(shot_video_paths) >= 6 else 0.80 approved = score >= scene_threshold 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 ) # ============================================================================= # ENGINE 1 & 2: GOOGLE ANTIGRAVITY SDK & GENAI ENGINE # ============================================================================= def _run_coroutine_safely(coro: Any) -> Any: """Chạy an toàn một coroutine trong ngữ cảnh đồng bộ, xử lý cả khi đã có event loop.""" if not inspect.iscoroutine(coro): return coro try: loop = asyncio.get_running_loop() except RuntimeError: loop = None if loop and loop.is_running(): import concurrent.futures with concurrent.futures.ThreadPoolExecutor(max_workers=1) as executor: return executor.submit(asyncio.run, coro).result() # ============================================================================= # REAL-TIME BUDGET & COST GUARD (HARD $1.00 CIRCUIT BREAKER) # ============================================================================= CRITIC_SPEND_FILE = BASE_DIR / "04_Assets" / "critic_spend_tracker.json" DEFAULT_MAX_BUDGET_USD = 1.00 class CriticCostGuard: """ Bộ giám sát & ngắt mạch chi phí thời gian thực (Hard Circuit Breaker). Giới hạn ngân sách nghiêm ngặt (mặc định $1.00 USD). Khi chi phí tích lũy chạm ngưỡng $1.00, tự động ngắt kết nối Cloud API và chuyển quyền thẩm định sang Offline Heuristic Fallback (OpenCV/FFprobe) để quy trình sản xuất không bị gián đoạn và không phát sinh thêm chi phí. """ SPEND_FILE = CRITIC_SPEND_FILE @classmethod def get_max_budget(cls) -> float: try: return float(os.environ.get("CRITIC_MAX_BUDGET_USD", str(DEFAULT_MAX_BUDGET_USD))) except ValueError: return DEFAULT_MAX_BUDGET_USD @classmethod def load_tracker(cls) -> Dict[str, Any]: if cls.SPEND_FILE.exists(): try: with open(cls.SPEND_FILE, "r", encoding="utf-8") as f: return json.load(f) except Exception: pass return { "total_cost_usd": 0.0, "max_budget_usd": cls.get_max_budget(), "total_calls": 0, "total_input_tokens": 0, "total_output_tokens": 0, "history": [] } @classmethod def save_tracker(cls, data: Dict[str, Any]) -> None: try: cls.SPEND_FILE.parent.mkdir(parents=True, exist_ok=True) with open(cls.SPEND_FILE, "w", encoding="utf-8") as f: json.dump(data, f, ensure_ascii=False, indent=2) except Exception: pass @classmethod def can_spend(cls) -> bool: max_budget = cls.get_max_budget() if max_budget <= 0.0 or not os.environ.get("GEMINI_API_KEY"): return False tracker = cls.load_tracker() spent = tracker.get("total_cost_usd", 0.0) if spent >= max_budget: print(f"⚠️ [CRITIC COST GUARD] Ngân sách ${max_budget:.2f} USD đã chạm ngưỡng (${spent:.4f} USD).") print(f" Tự động kích hoạt Offline Heuristic Fallback để không phát sinh chi phí.") return False return True @classmethod def record_usage(cls, model_name: str, input_tokens: int = 1500, output_tokens: int = 300, task: str = "") -> float: cost = (input_tokens / 1_000_000 * 0.10) + (output_tokens / 1_000_000 * 0.40) tracker = cls.load_tracker() tracker["total_cost_usd"] = round(tracker.get("total_cost_usd", 0.0) + cost, 6) tracker["total_calls"] = tracker.get("total_calls", 0) + 1 tracker["total_input_tokens"] = tracker.get("total_input_tokens", 0) + input_tokens tracker["total_output_tokens"] = tracker.get("total_output_tokens", 0) + output_tokens tracker["max_budget_usd"] = cls.get_max_budget() import time history = tracker.get("history", []) history.append({ "timestamp": time.strftime("%Y-%m-%d %H:%M:%S"), "task": task, "model": model_name, "cost_usd": round(cost, 6), "cumulative_usd": tracker["total_cost_usd"] }) tracker["history"] = history[-50:] cls.save_tracker(tracker) return cost # ============================================================================= # FILE-BRIDGE AUTONOMOUS QUEUE ENGINE (ZERO API-KEY HOST AGENT AUDIT) # ============================================================================= class FileBridgeCriticEngine: """ Động cơ cầu nối tệp tin IPC (Autonomous File-Bridge Quality Gate). Cho phép Antigravity Host Agent (Gemini) trực tiếp thẩm định hình ảnh / video thông qua AI Credits tích hợp sẵn trong IDE, 100% độc lập với Cloud API Key. """ QUEUE_DIR = CRITIC_QUEUE_DIR @classmethod def get_pending_requests(cls) -> List[Path]: if not cls.QUEUE_DIR.exists(): return [] return sorted(cls.QUEUE_DIR.glob("*_request.json")) @classmethod def write_verdict(cls, target_id: str, verdict: VideoCriticVerdict) -> Path: cls.QUEUE_DIR.mkdir(parents=True, exist_ok=True) verdict_file = cls.QUEUE_DIR / f"{target_id}_verdict.json" with open(verdict_file, "w", encoding="utf-8") as f: f.write(verdict.model_dump_json(indent=2)) return verdict_file @classmethod def evaluate_shot( cls, shot_id: str, video_path_str: str, expected_character: str = "", prompt: str = "" ) -> Optional[VideoCriticVerdict]: cls.QUEUE_DIR.mkdir(parents=True, exist_ok=True) verdict_file = cls.QUEUE_DIR / f"{shot_id}_verdict.json" # 1. Nếu Host Agent đã lưu kết quả thẩm định sẵn thì nạp ngay lập tức if verdict_file.exists(): try: with open(verdict_file, "r", encoding="utf-8") as f: data = json.load(f) verdict = VideoCriticVerdict.model_validate(data) print(f"🎯 [FILE-BRIDGE GATE] Nạp phán quyết từ Host Agent cho shot '{shot_id}': score={verdict.overall_score:.2f} approved={verdict.approved}") return verdict except Exception as e: print(f"⚠️ [FILE-BRIDGE GATE] Lỗi đọc verdict {verdict_file.name}: {e}") # 2. Tạo request file kèm metadata và đường dẫn keyframe để Agent đọc video_path = Path(video_path_str) cs_path_str = None if generate_zero_waste_contact_sheet is not None and video_path.exists(): try: _, cs_file = generate_zero_waste_contact_sheet(video_path, num_frames=6) if cs_file: cs_path_str = str(cs_file.resolve()) except Exception: pass request_file = cls.QUEUE_DIR / f"{shot_id}_request.json" try: req_data = { "shot_id": shot_id, "video_path": str(video_path.resolve()) if video_path.exists() else video_path_str, "contact_sheet_path": cs_path_str, "expected_character": expected_character, "prompt": prompt, "flowkit_14_error_catalog": list(FLOWKIT_ERROR_CATALOG.keys()), "status": "pending_agent_review", "timestamp": datetime.now().strftime("%Y-%m-%d %H:%M:%S") } with open(request_file, "w", encoding="utf-8") as f: json.dump(req_data, f, ensure_ascii=False, indent=2) except Exception: pass return None @classmethod def evaluate_scene( cls, scene_id: str, shot_video_paths: List[str] ) -> Optional[VideoCriticVerdict]: cls.QUEUE_DIR.mkdir(parents=True, exist_ok=True) verdict_file = cls.QUEUE_DIR / f"{scene_id}_verdict.json" # 1. Kiểm tra verdict sẵn có if verdict_file.exists(): try: with open(verdict_file, "r", encoding="utf-8") as f: data = json.load(f) verdict = VideoCriticVerdict.model_validate(data) print(f"🎯 [FILE-BRIDGE GATE] Nạp phán quyết từ Host Agent cho cảnh '{scene_id}': score={verdict.overall_score:.2f} approved={verdict.approved}") return verdict except Exception as e: print(f"⚠️ [FILE-BRIDGE GATE] Lỗi đọc verdict {verdict_file.name}: {e}") # 2. Tạo request file request_file = cls.QUEUE_DIR / f"{scene_id}_request.json" try: req_data = { "scene_id": scene_id, "shot_video_paths": [str(Path(p).resolve()) for p in shot_video_paths], "status": "pending_agent_review", "timestamp": datetime.now().strftime("%Y-%m-%d %H:%M:%S") } with open(request_file, "w", encoding="utf-8") as f: json.dump(req_data, f, ensure_ascii=False, indent=2) except Exception: pass return None class AntigravitySDKEngine: """ Động cơ thẩm định MLLM Vision sử dụng Google Antigravity SDK (Agent với LocalAgentConfig và response_schema=VideoCriticVerdict). Trích xuất Zero-Waste Contact Sheet đính kèm thực tế vào nội dung đánh giá. """ @classmethod def evaluate_shot( cls, shot_id: str, video_path_str: str, expected_character: str, prompt: str, strict: bool = False ) -> Optional[VideoCriticVerdict]: from google.antigravity import Agent, LocalAgentConfig, from_bytes api_key = os.environ.get("GEMINI_API_KEY") video_path = Path(video_path_str) if not video_path.exists() or video_path.stat().st_size == 0: err_msg = f"File video không tồn tại hoặc rỗng: {video_path}" if strict: raise FileNotFoundError(f"[CRITIC GATE ERROR] {err_msg}") return None # Ưu tiên sử dụng Zero-Waste Contact Sheet (triệt tiêu ô đen rác) cs_bytes = None if extract_contact_sheet_bytes is not None: cs_bytes = extract_contact_sheet_bytes(video_path, num_frames=6) keyframe_bytes = [cs_bytes] if cs_bytes else _extract_keyframe_bytes(video_path, max_frames=4) if not keyframe_bytes: err_msg = f"Không thể trích xuất khung hình từ video: {video_path}" if strict: raise RuntimeError(f"[CRITIC GATE ERROR] {err_msg}") return None sys_inst = ( "You are the Lead Cinematic Critic Auditor for 'Thập Ngũ Niên' following FlowKit Quality Standards. " "Audit this shot against the 14 Known AI Video Errors (Critical: character_drift, breed_swap, role_reversal, brand_logo_text, character_count_drift; " "High: camera_drift, object_morph, reverse_motion, human_hands_limbs, scale_break; Minor: lighting_jitter, micro_blur, background_warp, color_shift). " "If any Critical error exists, overall_score must be <= 0.59 and character_consistency <= 0.30. " "If any High error exists, overall_score must be <= 0.74. Score >= 0.80 approves." ) cfg_kwargs = { "response_schema": VideoCriticVerdict, "system_instructions": sys_inst, "model": "gemini-3.8-flash", } if api_key: cfg_kwargs["api_key"] = api_key else: gcp_project = os.environ.get("GOOGLE_CLOUD_PROJECT", "dscons-508517") cfg_kwargs["vertex"] = True cfg_kwargs["project"] = gcp_project cfg_kwargs["location"] = os.environ.get("GOOGLE_CLOUD_LOCATION", "us-central1") cfg = LocalAgentConfig(**cfg_kwargs) eval_prompt = ( f"Evaluate Shot '{shot_id}'. Expected Character: '{expected_character}'. " f"Motion Prompt: '{prompt}'. Check attached zero-waste contact sheet for facial morphing, character mixups, and 14 FlowKit video errors." ) inputs = [eval_prompt] for kb in keyframe_bytes: if kb: inputs.append(from_bytes(kb, mime_type="image/jpeg")) async def _call_sdk(): async with Agent(config=cfg) as agent: return await agent.chat(inputs) try: resp = _run_coroutine_safely(_call_sdk()) verdict = None if hasattr(resp, "structured_output") and resp.structured_output: if isinstance(resp.structured_output, VideoCriticVerdict): verdict = resp.structured_output elif isinstance(resp.structured_output, dict): verdict = VideoCriticVerdict.model_validate(resp.structured_output) elif hasattr(resp, "text") and resp.text: clean_json = re.sub(r"^```json\s*", "", resp.text.strip()) clean_json = re.sub(r"\s*```$", "", clean_json) verdict = VideoCriticVerdict.model_validate_json(clean_json) if verdict is not None: CriticCostGuard.record_usage("gemini-3.8-flash", task=f"shot:{shot_id}") return verdict elif strict: raise RuntimeError(f"[CRITIC GATE ERROR] Antigravity SDK không trả về verdict hợp lệ: {resp}") except Exception as e: if strict: raise RuntimeError(f"[CRITIC GATE ERROR] Lỗi thực thi Antigravity SDK: {e}") from e print(f"⚠️ [ANTIGRAVITY SDK] Lỗi: {e}") return None return None @classmethod def evaluate_scene( cls, scene_id: str, shot_video_paths: List[str], strict: bool = False ) -> Optional[VideoCriticVerdict]: from google.antigravity import Agent, LocalAgentConfig, from_bytes api_key = os.environ.get("GEMINI_API_KEY") if not shot_video_paths or len(shot_video_paths) < 1: err_msg = f"Không có danh sách video cho cảnh: {scene_id}" if strict: raise ValueError(f"[CRITIC GATE ERROR] {err_msg}") return None junction_bytes = _extract_junction_frame_bytes(shot_video_paths) if not junction_bytes: err_msg = f"Không thể trích xuất junction frames cho cảnh: {scene_id}" if strict: raise RuntimeError(f"[CRITIC GATE ERROR] {err_msg}") return None sys_inst = ( "You are the Lead Cinematic Pacing & Concat Critic for 'Thập Ngũ Niên'. " "Evaluate multi-shot scene flow: 180-degree axis continuity, eyeline matches, color grading consistency, " "and editing tempo. Score >= 0.8 approves." ) cfg_kwargs = { "response_schema": VideoCriticVerdict, "system_instructions": sys_inst, "model": "gemini-3.8-flash", } if api_key: cfg_kwargs["api_key"] = api_key else: gcp_project = os.environ.get("GOOGLE_CLOUD_PROJECT", "dscons-508517") cfg_kwargs["vertex"] = True cfg_kwargs["project"] = gcp_project cfg_kwargs["location"] = os.environ.get("GOOGLE_CLOUD_LOCATION", "us-central1") cfg = LocalAgentConfig(**cfg_kwargs) prompt = ( f"Evaluate Scene '{scene_id}' consisting of {len(shot_video_paths)} shots. " f"Inspect attached pairwise junction frames (tail frame vs head frame) for continuity, eyeline, and color flow." ) inputs = [prompt] for jb in junction_bytes: inputs.append(from_bytes(jb, mime_type="image/jpeg")) async def _call_scene(): async with Agent(config=cfg) as agent: return await agent.chat(inputs) try: resp = _run_coroutine_safely(_call_scene()) verdict = None if hasattr(resp, "structured_output") and resp.structured_output: if isinstance(resp.structured_output, VideoCriticVerdict): verdict = resp.structured_output elif isinstance(resp.structured_output, dict): verdict = VideoCriticVerdict.model_validate(resp.structured_output) elif hasattr(resp, "text") and resp.text: clean_json = re.sub(r"^```json\s*", "", resp.text.strip()) clean_json = re.sub(r"\s*```$", "", clean_json) verdict = VideoCriticVerdict.model_validate_json(clean_json) if verdict is not None: CriticCostGuard.record_usage("gemini-3.8-flash", task=f"scene:{scene_id}") return verdict elif strict: raise RuntimeError(f"[CRITIC GATE ERROR] Antigravity SDK không trả về verdict hợp lệ cho cảnh {scene_id}: {resp}") except Exception as e: if strict: raise RuntimeError(f"[CRITIC GATE ERROR] Lỗi thực thi Antigravity SDK cho cảnh {scene_id}: {e}") from e print(f"⚠️ [ANTIGRAVITY SDK] Lỗi: {e}") return None return None class GoogleGenAIEngine: """ Động cơ dự phòng thứ cấp sử dụng Google GenAI Client (gemini-2.5-flash). Đính kèm keyframe JPEG Parts thực tế vào contents. """ @classmethod def evaluate_shot( cls, shot_id: str, video_path_str: str, expected_character: str, prompt: str ) -> Optional[VideoCriticVerdict]: try: if not CriticCostGuard.can_spend(): return None import google.genai as genai api_key = os.environ.get("GEMINI_API_KEY") if not api_key: return None video_path = Path(video_path_str) if not video_path.exists() or video_path.stat().st_size == 0: return None # Ưu tiên Zero-Waste Contact Sheet cs_bytes = None if extract_contact_sheet_bytes is not None: cs_bytes = extract_contact_sheet_bytes(video_path, num_frames=6) keyframe_bytes = [cs_bytes] if cs_bytes else _extract_keyframe_bytes(video_path, max_frames=4) if not keyframe_bytes: return None client = genai.Client(api_key=api_key) prompt_text = ( f"You are the Lead Cinematic Critic Auditor for 'Thập Ngũ Niên' following FlowKit Quality Standards. " f"Evaluate Shot '{shot_id}'. Expected Character: '{expected_character}'. Motion Prompt: '{prompt}'. " f"Inspect the attached zero-waste contact sheet against the 14 Known AI Video Errors (Critical: character_drift, breed_swap, " f"role_reversal, brand_logo_text, character_count_drift; High: camera_drift, object_morph, reverse_motion, human_hands_limbs, scale_break; " f"Minor: lighting_jitter, micro_blur, background_warp, color_shift). " f"If any Critical error exists, overall_score must be <= 0.59 and character_consistency <= 0.30. " f"If any High error exists, overall_score must be <= 0.74. Return VideoCriticVerdict JSON." ) contents = [prompt_text] for kb in keyframe_bytes: if kb: contents.append(genai.types.Part.from_bytes(data=kb, mime_type="image/jpeg")) resp = client.models.generate_content( model="gemini-3.8-flash", contents=contents, config=genai.types.GenerateContentConfig( response_mime_type="application/json", response_schema=VideoCriticVerdict ) ) if resp.text: verdict = VideoCriticVerdict.model_validate_json(resp.text) CriticCostGuard.record_usage("gemini-3.8-flash", task=f"shot:{shot_id}") return verdict except Exception: return None return None @classmethod def evaluate_scene(cls, scene_id: str, shot_video_paths: List[str]) -> Optional[VideoCriticVerdict]: try: if not CriticCostGuard.can_spend(): return None import google.genai as genai api_key = os.environ.get("GEMINI_API_KEY") if not api_key: return None if not shot_video_paths or len(shot_video_paths) < 1: return None junction_bytes = _extract_junction_frame_bytes(shot_video_paths) if not junction_bytes: return None client = genai.Client(api_key=api_key) prompt_text = ( f"You are the Lead Cinematic Pacing & Concat Critic for 'Thập Ngũ Niên'. " f"Evaluate Scene '{scene_id}' continuity across {len(shot_video_paths)} shots. " f"Inspect the attached pairwise junction keyframes (tail frame of shot N vs head frame of shot N+1) " f"for 180-degree axis continuity, eyeline match, color grading consistency, and editing tempo. " f"Return VideoCriticVerdict JSON." ) contents = [prompt_text] for jb in junction_bytes: contents.append(genai.types.Part.from_bytes(data=jb, mime_type="image/jpeg")) resp = client.models.generate_content( model="gemini-3.8-flash", contents=contents, config=genai.types.GenerateContentConfig( response_mime_type="application/json", response_schema=VideoCriticVerdict ) ) if resp.text: verdict = VideoCriticVerdict.model_validate_json(resp.text) CriticCostGuard.record_usage("gemini-3.8-flash", task=f"scene:{scene_id}") return verdict except Exception: return None return None # ============================================================================= # ENGINE 4: MUSE2API GATEWAY ENGINE (GPT-5 / MUSE-CHAT MULTIMODAL FALLBACK) # ============================================================================= class MuseAPICriticEngine: """ Động cơ thẩm định AI Fallback sử dụng Muse2API Gateway (/v1/chat/completions, model gpt-5). Được kích hoạt làm tầng cứu sinh khi Google Antigravity SDK gặp sự cố (thiếu GEMINI_API_KEY, vượt ngân sách $1.00 USD, hoặc lỗi kết nối). Đảm bảo 100% quy trình review chạy bằng AI Vision mà KHÔNG rơi về Heuristic. """ BASE_URL = os.environ.get("MUSE2API_BASE_URL", "http://127.0.0.1:18610") LOCAL_KEY_FILE = Path(r"c:\Projects\Muse2API\data\api_key") @classmethod def _get_api_key(cls) -> str: key = os.environ.get("MUSE2API_API_KEY", "") if not key and cls.LOCAL_KEY_FILE.exists(): try: key = cls.LOCAL_KEY_FILE.read_text(encoding="utf-8").strip() except Exception: pass return key or "m2a--nqA-RT1mZ4AWalrdDVzSYkhI9Cksd8O" @classmethod def evaluate_shot( cls, shot_id: str, video_path_str: str, expected_character: str = "", prompt: str = "", timeout: int = 120 ) -> Optional[VideoCriticVerdict]: video_path = Path(video_path_str) if not video_path.exists() or video_path.stat().st_size == 0: return None # Trích xuất Zero-Waste Contact Sheet hoặc keyframes cs_bytes = None if extract_contact_sheet_bytes is not None: cs_bytes = extract_contact_sheet_bytes(video_path, num_frames=6) keyframe_bytes = [cs_bytes] if cs_bytes else _extract_keyframe_bytes(video_path, max_frames=4) if not keyframe_bytes or not keyframe_bytes[0]: return None b64_img = base64.b64encode(keyframe_bytes[0]).decode("utf-8") data_uri = f"data:image/jpeg;base64,{b64_img}" sys_prompt = ( "You are the Lead Cinematic Critic Auditor for 'Thập Ngũ Niên' following FlowKit Quality Standards. " "Audit this shot against the 14 Known AI Video Errors (Critical: character_drift, breed_swap, role_reversal, brand_logo_text, character_count_drift; " "High: camera_drift, object_morph, reverse_motion, human_hands_limbs, scale_break; Minor: lighting_jitter, micro_blur, background_warp, color_shift). " "If any Critical error exists, overall_score must be <= 0.59 and character_consistency <= 0.30. " "If any High error exists, overall_score must be <= 0.74. Score >= 0.80 approves. " "Output MUST be a single raw JSON object matching the VideoCriticVerdict schema: " '{"overall_score": float, "approved": bool, "detected_errors": list, "error_severities": dict, "suggested_action": str, "critique_notes": str}' ) user_content = [ { "type": "text", "text": f"Evaluate Shot '{shot_id}'. Expected Character: '{expected_character}'. Motion Prompt: '{prompt}'. Check attached zero-waste contact sheet. Return ONLY valid JSON." }, { "type": "image_url", "image_url": {"url": data_uri} } ] payload = { "model": "gpt-5", "messages": [ {"role": "system", "content": sys_prompt}, {"role": "user", "content": user_content} ], "stream": False } key = cls._get_api_key() headers = { "Authorization": f"Bearer {key}", "Content-Type": "application/json" } try: import requests url = f"{cls.BASE_URL.rstrip('/')}/v1/chat/completions" resp = requests.post(url, headers=headers, json=payload, timeout=timeout) if resp.status_code != 200: print(f"⚠️ [MUSE API CRITIC] HTTP {resp.status_code}: {resp.text[:150]}") return None data = resp.json() choices = data.get("choices", []) if not choices: return None reply_text = choices[0].get("message", {}).get("content", "") if not reply_text: return None clean_json = re.sub(r"^```json\s*", "", reply_text.strip()) clean_json = re.sub(r"\s*```$", "", clean_json) m = re.search(r"\{.*\}", clean_json, re.DOTALL) if m: clean_json = m.group(0) verdict = VideoCriticVerdict.model_validate_json(clean_json) print(f"🎯 [MUSE API CRITIC] Thẩm định thành công qua gpt-5 cho shot '{shot_id}': score={verdict.overall_score:.2f} approved={verdict.approved}") return verdict except Exception as e: print(f"⚠️ [MUSE API CRITIC] Lỗi gọi gpt-5: {e}") return None @classmethod def evaluate_scene( cls, scene_id: str, shot_video_paths: List[str], timeout: int = 120 ) -> Optional[VideoCriticVerdict]: if not shot_video_paths or len(shot_video_paths) < 1: return None junction_bytes = _extract_junction_frame_bytes(shot_video_paths) if not junction_bytes or not junction_bytes[0]: return None b64_img = base64.b64encode(junction_bytes[0]).decode("utf-8") data_uri = f"data:image/jpeg;base64,{b64_img}" sys_prompt = ( "You are the Lead Cinematic Pacing & Concat Critic for 'Thập Ngũ Niên'. " "Evaluate multi-shot scene flow: 180-degree axis continuity, eyeline matches, color grading consistency, " "and editing tempo. Score >= 0.8 approves. " "Output MUST be a single raw JSON object matching the VideoCriticVerdict schema: " '{"overall_score": float, "approved": bool, "detected_errors": list, "error_severities": dict, "suggested_action": str, "critique_notes": str}' ) user_content = [ { "type": "text", "text": f"Evaluate Scene '{scene_id}' consisting of {len(shot_video_paths)} shots. Inspect attached junction frames. Return ONLY valid JSON." }, { "type": "image_url", "image_url": {"url": data_uri} } ] payload = { "model": "gpt-5", "messages": [ {"role": "system", "content": sys_prompt}, {"role": "user", "content": user_content} ], "stream": False } key = cls._get_api_key() headers = { "Authorization": f"Bearer {key}", "Content-Type": "application/json" } try: import requests url = f"{cls.BASE_URL.rstrip('/')}/v1/chat/completions" resp = requests.post(url, headers=headers, json=payload, timeout=timeout) if resp.status_code != 200: print(f"⚠️ [MUSE API CRITIC] HTTP {resp.status_code}: {resp.text[:150]}") return None data = resp.json() choices = data.get("choices", []) if not choices: return None reply_text = choices[0].get("message", {}).get("content", "") if not reply_text: return None clean_json = re.sub(r"^```json\s*", "", reply_text.strip()) clean_json = re.sub(r"\s*```$", "", clean_json) m = re.search(r"\{.*\}", clean_json, re.DOTALL) if m: clean_json = m.group(0) verdict = VideoCriticVerdict.model_validate_json(clean_json) print(f"🎯 [MUSE API CRITIC] Thẩm định thành công qua gpt-5 cho cảnh '{scene_id}': score={verdict.overall_score:.2f} approved={verdict.approved}") return verdict except Exception as e: print(f"⚠️ [MUSE API CRITIC] Lỗi gọi gpt-5: {e}") return None # ============================================================================= # PUBLIC INTERFACE CONTRACTS # ============================================================================= def evaluate_shot_gate( shot_id: str, video_path: str, expected_character: str = "", prompt: str = "", force_engine: Optional[str] = None, **kwargs ) -> VideoCriticVerdict: """ TẦNG 1 (SHOT GATE): Thẩm định độc lập từng shot 10s trước khi lưu video. Ngưỡng phê duyệt: overall_score >= 0.8. Quy tắc bắt buộc SDK & AI Fallback: 1. Ưu tiên 1: Google Antigravity SDK (gemini-3.8-flash). 2. Fallback AI: Muse2API Gateway (model: gpt-5) nếu SDK lỗi/thiếu key/hết quota. 3. Nếu cả 2 AI đều lỗi: Báo lỗi dứt khoát (RuntimeError), TUYỆT ĐỐI KHÔNG dùng heuristic. """ engine_choice = force_engine or os.environ.get("CRITIC_GATE_ENGINE", "file_bridge") strict_sdk = ( engine_choice in ("sdk", "cloud_sdk", "strict_sdk", "muse", "gpt-5") or os.environ.get("REQUIRE_SDK_CRITIC", "0").lower() in ("1", "true", "yes") ) if strict_sdk: sdk_err: Optional[Exception] = None # Nếu không ép thẳng engine muse/gpt-5, thử SDK trước if engine_choice not in ("muse", "gpt-5"): try: verdict = AntigravitySDKEngine.evaluate_shot( shot_id=shot_id, video_path_str=video_path, expected_character=expected_character, prompt=prompt, strict=False ) if verdict is not None: return verdict except Exception as e: sdk_err = e print(f"⚠️ [CRITIC GATE] Antigravity SDK gặp sự cố: {e}") # Fallback sang Muse2API gpt-5 khi SDK không chạy được print(f"🔄 [CRITIC GATE FALLBACK] Chuyển tiếp thẩm định sang Muse2API (model: gpt-5)...") verdict = MuseAPICriticEngine.evaluate_shot( shot_id=shot_id, video_path_str=video_path, expected_character=expected_character, prompt=prompt ) if verdict is not None: return verdict # Cả SDK và Muse API đều thất bại -> Ném lỗi, dứt khoát không rơi về Heuristic raise RuntimeError( f"[CRITIC GATE ERROR] Cả Google Antigravity SDK và Muse2API (gpt-5) đều không thể đánh giá shot '{shot_id}'. " f"(Chi tiết SDK: {sdk_err}). Theo quy chuẩn bắt buộc, hệ thống báo lỗi chứ không dùng heuristic fallback!" ) # 1. Thử File-Bridge Autonomous Queue (Host Agent verdict từ hàng đợi) if engine_choice in ("auto", "file_bridge", "agent"): verdict = FileBridgeCriticEngine.evaluate_shot(shot_id, video_path, expected_character, prompt) if verdict is not None: return verdict if engine_choice == "cloud_genai" and CriticCostGuard.can_spend(): verdict = GoogleGenAIEngine.evaluate_shot(shot_id, video_path, expected_character, prompt) if verdict is not None: return verdict # 2. Heuristic Fallback (Chỉ dùng khi không yêu cầu strict SDK/AI) return OfflineHeuristicEngine.evaluate_shot(shot_id, video_path, expected_character, prompt) def evaluate_scene_gate( scene_id: str, shot_video_paths: List[str], force_engine: Optional[str] = None, **kwargs ) -> VideoCriticVerdict: """ TẦNG 2 (SCENE GATE): Thẩm định điểm giao cắt tiếp biên (junction), trục 180 độ, độ chênh màu và nhịp điệu cắt dựng toàn cảnh 30-40s. Ngưỡng phê duyệt: overall_score >= 0.8. Quy tắc bắt buộc SDK & AI Fallback: 1. Ưu tiên 1: Google Antigravity SDK (gemini-3.8-flash). 2. Fallback AI: Muse2API Gateway (model: gpt-5) nếu SDK lỗi/thiếu key/hết quota. 3. Nếu cả 2 AI đều lỗi: Báo lỗi dứt khoát (RuntimeError), TUYỆT ĐỐI KHÔNG dùng heuristic. """ engine_choice = force_engine or os.environ.get("CRITIC_GATE_ENGINE", "file_bridge") strict_sdk = ( engine_choice in ("sdk", "cloud_sdk", "strict_sdk", "muse", "gpt-5") or os.environ.get("REQUIRE_SDK_CRITIC", "0").lower() in ("1", "true", "yes") ) if strict_sdk: sdk_err: Optional[Exception] = None if engine_choice not in ("muse", "gpt-5"): try: verdict = AntigravitySDKEngine.evaluate_scene( scene_id=scene_id, shot_video_paths=shot_video_paths, strict=False ) if verdict is not None: return verdict except Exception as e: sdk_err = e print(f"⚠️ [CRITIC GATE] Antigravity SDK gặp sự cố: {e}") # Fallback sang Muse2API gpt-5 khi SDK không chạy được print(f"🔄 [CRITIC GATE FALLBACK] Chuyển tiếp thẩm định cảnh sang Muse2API (model: gpt-5)...") verdict = MuseAPICriticEngine.evaluate_scene( scene_id=scene_id, shot_video_paths=shot_video_paths ) if verdict is not None: return verdict raise RuntimeError( f"[CRITIC GATE ERROR] Cả Google Antigravity SDK và Muse2API (gpt-5) đều không thể đánh giá scene '{scene_id}'. " f"(Chi tiết SDK: {sdk_err}). Theo quy chuẩn bắt buộc, hệ thống báo lỗi chứ không dùng heuristic fallback!" ) # 1. Thử File-Bridge Autonomous Queue (Host Agent verdict từ hàng đợi) if engine_choice in ("auto", "file_bridge", "agent"): verdict = FileBridgeCriticEngine.evaluate_scene(scene_id, shot_video_paths) if verdict is not None: return verdict if engine_choice == "cloud_genai" and CriticCostGuard.can_spend(): verdict = GoogleGenAIEngine.evaluate_scene(scene_id, shot_video_paths) if verdict is not None: return verdict # 2. Heuristic Fallback (Chỉ dùng khi không yêu cầu strict SDK) return OfflineHeuristicEngine.evaluate_scene(scene_id, shot_video_paths) if __name__ == "__main__": import argparse parser = argparse.ArgumentParser(description="Antigravity Critic Quality Gate CLI") parser.add_argument("--shot", type=str, help="Shot ID cần thẩm định") parser.add_argument("--video", type=str, help="Đường dẫn file video MP4") parser.add_argument("--character", type=str, default="", help="Nhân vật kỳ vọng (vd: kim_trong)") parser.add_argument("--prompt", type=str, default="", help="Prompt kịch bản") parser.add_argument("--scene", type=str, help="Scene ID cần thẩm định") parser.add_argument("--shots-list", nargs="+", help="Danh sách video shot của scene") args = parser.parse_args() if args.shot and args.video: res = evaluate_shot_gate(args.shot, args.video, args.character, args.prompt) print(res.model_dump_json(indent=2)) elif args.scene and args.shots_list: res = evaluate_scene_gate(args.scene, args.shots_list) print(res.model_dump_json(indent=2)) else: parser.print_help()