import sys, os, time, shutil, re from pathlib import Path sys.path.append(str(Path("05_Production_Pipeline").resolve())) from invisible_playwright import InvisiblePlaywright from muse_invpw_driver import DEFAULT_PROFILE, sanitize_prompt, extract_tail_frame from run_shot import ensure_audio_guard import cv2 def run_test_shot01(): shot_id = "ep01_scene01_shot01" start_frame = Path("04_Assets/keyframes/ep01_scene01_shot01/start_frame_720p.png").resolve() target_video = Path("04_Assets/videos/ep01_scene01_shot01_10s.mp4").resolve() target_video.parent.mkdir(parents=True, exist_ok=True) temp_dl = Path("04_Assets/temp_downloads/shot01_muse_test.mp4").resolve() temp_dl.parent.mkdir(parents=True, exist_ok=True) downloads_dir = Path.home() / "Downloads" # Prompt from ep01 muse_prompts.json raw_prompt = ( "Cú máy bay lướt êm ả dọc theo dòng sông biếc uốn lượn ngoại ô kinh thành Đại Minh Gia Tĩnh. " "Mặt nước trong veo như ngọc phản chiếu rặng liễu tơ mềm và làn sương sớm mai lảng bảng. " "Cảnh sắc mùa xuân thanh bình, xa xa thoáng bóng thuyền nan lướt nhẹ, 24fps cinematic drama. " "Âm thanh: Tiếng nước sông vỗ nhẹ vào bờ đất, tiếng mái chèo khua nước êm đềm từ xa, tiếng gió xuân mát rượi lướt qua rặng liễu." ) prompt = ensure_audio_guard(raw_prompt) print(f"[*] Prompt ({len(prompt)} chars):\n{prompt}\n") print("[1/5] Khởi động trình duyệt Camoufox Stealth và mở thread/new...") with InvisiblePlaywright(profile_dir=DEFAULT_PROFILE, headless=True) as ctx: page = ctx.pages[0] if ctx.pages else ctx.new_page() # Listen for falco/chat network events page.on("response", lambda r: print(f" [NET RESP {r.status}] {r.url[:80]}") if any(k in r.url for k in ["falco", "vm", "chat", "message"]) else None) page.goto("https://muse.ai/thread/new", wait_until="domcontentloaded", timeout=30000) time.sleep(4) print(f" ✓ Đang ở URL: {page.url}") # 1. Upload ảnh if start_frame.exists(): print(f"[2/5] Uploading Start Frame: {start_frame.name} ({start_frame.stat().st_size} bytes)...") file_input = page.locator('input[type="file"]').first if file_input.count() > 0: file_input.set_input_files(str(start_frame)) time.sleep(3) print(" ✓ Đã set input file thành công!") else: print(" [!] Cảnh báo: Không tìm thấy input file!") # 2. Gõ prompt print("[3/5] Gõ prompt bằng keyboard.insert_text...") chat_box = page.locator('textarea, [placeholder*="Nhắn tin"], [placeholder*="Message"], [role="textbox"]').first chat_box.click() time.sleep(1) page.keyboard.insert_text(prompt) time.sleep(1) page.keyboard.press("Space") time.sleep(1) # 3. Gửi prompt stop_sel = 'button:has-text("Dừng"), button[aria-label*="Dừng"], button:has-text("Stop"), button[aria-label*="Stop"]' dl_sel = 'button:has-text("Tải video xuống"), button[aria-label*="Tải video xuống"], button[data-pel="media_download"]' initial_dl_count = page.locator(dl_sel).count() send_btn = page.locator('button[data-pel-click="chat_send_message"], button[aria-label="Gửi"]').first print(" Bấm nút Gửi tin nhắn...") if send_btn.count() > 0 and send_btn.is_visible(): send_btn.click() else: page.keyboard.press("Enter") time.sleep(3) # Chờ xác nhận đã gửi submitted = False for s in range(25): stop_cnt = page.locator(stop_sel).count() sending_cnt = page.locator('text="Vẫn đang gửi"').count() if stop_cnt > 0 or sending_cnt > 0: submitted = True print(f" ✓ Đã gửi lệnh (stop_btn={stop_cnt}, sending={sending_cnt}) sau {s+1}s!") break if send_btn.count() > 0 and send_btn.is_visible(): send_btn.click() else: page.keyboard.press("Enter") time.sleep(1) if not submitted: page.screenshot(path="04_Assets/send_failed.png") print(" [!] Lỗi: Không thấy nút Dừng hoặc trạng thái gửi!") return False # 4. Giám sát render print("[4/5] Theo dõi tiến trình render (tối đa 240s)...") start_time = time.time() task_started = False video_downloaded = False for i in range(48): # 48 * 5s = 240s time.sleep(5) elapsed = int(time.time() - start_time) stop_cnt = page.locator(stop_sel).count() dl_cnt = page.locator(dl_sel).count() if stop_cnt > 0: task_started = True print(f" [{elapsed:3d}s] stop_btn={stop_cnt}, dl_btn={dl_cnt}, task_started={task_started}") # Lưu ảnh chụp tiến độ mỗi 30s if elapsed % 30 == 0: page.screenshot(path=f"04_Assets/render_progress_{elapsed}s.png") # Kiểm tra khi task đã chạy và stop_btn biến mất sau ít nhất 40s is_ready = (task_started and stop_cnt == 0 and elapsed >= 35) or (dl_cnt > initial_dl_count and elapsed >= 35) if is_ready: print(f"\n🎉 Render hoàn tất sau {elapsed}s! Đang thu hoạch video...") time.sleep(3) page.screenshot(path="04_Assets/render_completed.png") # Phương thức 1: Click download button với expect_download new_dl_btns = page.locator(dl_sel) if new_dl_btns.count() > 0: btn = new_dl_btns.last btn.scroll_into_view_if_needed() click_time = time.time() before_dls = {f.name: f.stat().st_mtime for f in downloads_dir.glob("*.mp4")} print(" [*] Thử Phương thức 1: expect_download...") try: with page.expect_download(timeout=10000) as dl_info: btn.click(force=True) dl = dl_info.value if temp_dl.exists(): temp_dl.unlink() dl.save_as(str(temp_dl)) if temp_dl.exists() and temp_dl.stat().st_size > 500000: print(f" ✓ Tải thành công qua expect_download: {temp_dl.stat().st_size} bytes!") video_downloaded = True except Exception as e_dl: print(f" [-] expect_download không bắt được ({e_dl})") # Phương thức 2: Quét thư mục Downloads if not video_downloaded: print(" [*] Thử Phương thức 2: Quét thư mục Downloads...") for _ in range(12): time.sleep(1) new_files = [ f for f in downloads_dir.glob("*.mp4") if (f.name not in before_dls or f.stat().st_mtime > click_time - 1.0) and f.stat().st_size > 500000 ] if new_files: new_files.sort(key=lambda x: x.stat().st_mtime, reverse=True) if temp_dl.exists(): temp_dl.unlink() shutil.move(str(new_files[0]), str(temp_dl)) print(f" ✓ Bắt được file từ Downloads: {temp_dl.stat().st_size} bytes!") video_downloaded = True break # Phương thức 3: Trích xuất trực tiếp qua JS Blob URL if not video_downloaded: print(" [*] Thử Phương thức 3: Trích xuất JS Blob từ DOM...") videos = page.locator('video').all() for idx, v in enumerate(reversed(videos)): try: dur = v.evaluate("el => el.duration") vw = v.evaluate("el => el.videoWidth") vh = v.evaluate("el => el.videoHeight") print(f" Video element {idx}: dur={dur}, size={vw}x{vh}") if dur and dur < 6.0 and vw and vw < 500: continue # Bỏ qua mascot avatar 480x480 blob_url = v.get_attribute("src") or v.evaluate("el => el.currentSrc") if blob_url and blob_url.startswith("blob:"): import base64 b64 = page.evaluate("""async (url) => { const res = await fetch(url); const b = await res.blob(); return new Promise((resolve, reject) => { const r = new FileReader(); r.onloadend = () => resolve(r.result.split(',')[1]); r.onerror = reject; r.readAsDataURL(b); }); }""", blob_url) if b64: raw_bytes = base64.b64decode(b64) if len(raw_bytes) > 500000: temp_dl.write_bytes(raw_bytes) print(f" ✓ Trích xuất JS Blob thành công ({vw}x{vh}): {len(raw_bytes)} bytes!") video_downloaded = True break except Exception as e_blob: print(f" [-] Blob extraction error: {e_blob}") if video_downloaded: break if not video_downloaded or not temp_dl.exists(): print("[!] Thất bại: Không tải được video sau khi render.") page.screenshot(path="04_Assets/render_failed.png") return False # Sao chép vào target video shutil.copy2(str(temp_dl), str(target_video)) print(f"\n=======================================================") print(f"🎉 XUẤT XƯỞNG MASTER SHOT THÀNH CÔNG: {target_video.name}") print(f" Đường dẫn: {target_video.resolve()}") print(f" Kích thước: {target_video.stat().st_size:,} bytes") print(f"=======================================================\n") # 5. Trích xuất Tail Frame 239 print("[5/5] Trích xuất Tail Frame 239 bằng OpenCV...") cap = cv2.VideoCapture(str(target_video)) total_f = int(cap.get(cv2.CAP_PROP_FRAME_COUNT)) fps = cap.get(cv2.CAP_PROP_FPS) w = int(cap.get(cv2.CAP_PROP_FRAME_WIDTH)) h = int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT)) print(f" Video specs: {w}x{h}, {total_f} frames, {fps:.1f} fps") cap.set(cv2.CAP_PROP_POS_FRAMES, max(0, total_f - 1)) ret, frame = cap.read() cap.release() tail_frame_path = Path("04_Assets/keyframes/ep01_scene01_shot01/clean_frame_239.jpg") tail_frame_path.parent.mkdir(parents=True, exist_ok=True) if ret and frame is not None: cv2.imwrite(str(tail_frame_path), frame) print(f" ✓ Đã lưu Tail Frame: {tail_frame_path} ({tail_frame_path.stat().st_size:,} bytes)") return True if __name__ == '__main__': ok = run_test_shot01() sys.exit(0 if ok else 1)