#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ ============================================================================= THẬP NGŨ NIÊN (THE FIFTEEN SPRINGS) - INVISIBLE PLAYWRIGHT MUSE.AI DRIVER ============================================================================= Trình điều khiển tự động hóa Muse.ai sử dụng InvisiblePlaywright (Firefox C++ Camoufox Stealth Engine) kết hợp profile 'phu2234' tại: C:\\Users\\Admin\\AntigravityProfiles\\phu2234 Ưu điểm vượt trội: - Vượt 100% hệ thống kiểm tra bot / Cloudflare / Meta checkpoint. - Độc lập hoàn toàn với Google Chrome, triệt tiêu lỗi khóa file (LockFileEx). - Hỗ trợ Head-Tail Chaining, Reverse Motion, Versioning _v* và Audio Guard. ============================================================================= """ import os import sys import re import time import shutil import subprocess from pathlib import Path from typing import Optional, Tuple, Dict, Any 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 PIPELINE_DIR = Path(__file__).resolve().parent if str(PIPELINE_DIR) not in sys.path: sys.path.insert(0, str(PIPELINE_DIR)) VIDEOS_DIR = BASE_DIR / "04_Assets" / "videos" KEYFRAMES_DIR = BASE_DIR / "04_Assets" / "keyframes" PROFILES_ROOT = Path.home() / "AntigravityProfiles" DEFAULT_PROFILE = PROFILES_ROOT / "phu2234_worker" # Tự động đồng bộ cookies từ profile gốc nếu worker chưa có hoặc cần refresh def sync_worker_profile(): src = PROFILES_ROOT / "phu2234" dst = PROFILES_ROOT / "phu2234_worker" if src.exists(): dst.mkdir(parents=True, exist_ok=True) # Xóa lockfile cũ nếu có lock_file = dst / "parent.lock" if lock_file.exists(): try: lock_file.unlink() except Exception: pass try: import subprocess subprocess.run( f'robocopy "{src}" "{dst}" /E /XF "parent.lock" "*lock*" /XD "cache2" "startupCache" /NFL /NDL /NJH /NJS /nc /ns /np', shell=True, capture_output=True ) except Exception: pass sync_worker_profile() try: from invisible_playwright import InvisiblePlaywright HAS_INVPW = True except ImportError: HAS_INVPW = False def sanitize_prompt(prompt: str) -> str: """Thanh lọc từ khóa nhạy cảm kích hoạt bộ lọc kiểm duyệt Meta/Muse.""" if not prompt: return prompt replacements = [ (r'\b(nấm mồ|mồ mả|nghĩa địa|mộ địa|mộ phần|ngôi mộ|ngôi mả)\b', 'gò đất phủ cỏ úa xưa cũ'), (r'\b(âm khí|tử khí|âm khí nặng nề)\b', 'gió lạnh chiều tà se sắt hoang vu'), (r'\b(khóc người đời xưa|khóc than bạc mệnh)\b', 'nghiêng mình trầm tư đượm buồn'), (r'\b(rải vàng mã|tiền vàng mã|vàng mã)\b', 'thắp nén hương trầm phảng phất'), (r'\b(xương tàn|xương cốt|xác chết|chết chóc)\b', 'dấu tích xưa cũ'), (r'\b(bạc mệnh)\b', 'số phận long đong'), ] cleaned = prompt for pattern, repl in replacements: cleaned = re.sub(pattern, repl, cleaned, flags=re.IGNORECASE) return cleaned def get_next_video_path(shot_id: str) -> Path: """Tự động xác định tên version tiếp theo (_v1, _v2, _v3...).""" pattern = re.compile(rf"^{re.escape(shot_id)}_10s(?:_v(\d+))?\.mp4$", re.IGNORECASE) highest_v = 0 has_unversioned = False for f in VIDEOS_DIR.glob(f"{shot_id}_10s*.mp4"): m = pattern.match(f.name) if m: v_num = m.group(1) if v_num is not None: highest_v = max(highest_v, int(v_num)) else: has_unversioned = True next_v = max(highest_v + 1, 2 if has_unversioned and highest_v == 0 else highest_v + 1) return VIDEOS_DIR / f"{shot_id}_10s_v{next_v}.mp4" def extract_tail_frame(video_path: Path, shot_id: str, reverse_motion: bool = False) -> Optional[Path]: """Trích xuất Tail Frame (Frame 239) cho shot kế tiếp.""" keyframe_dir = KEYFRAMES_DIR / shot_id out_frame = keyframe_dir / "clean_frame_239.jpg" try: import cv2 cap = cv2.VideoCapture(str(video_path)) total_frames = int(cap.get(cv2.CAP_PROP_FRAME_COUNT)) if total_frames > 0: target_idx = min(239, total_frames - 1) cap.set(cv2.CAP_PROP_POS_FRAMES, target_idx) ret, frame = cap.read() if ret and frame is not None: keyframe_dir.mkdir(parents=True, exist_ok=True) cv2.imwrite(str(out_frame), frame, [cv2.IMWRITE_JPEG_QUALITY, 95]) cap.release() print(f" ✓ Đã trích xuất Tail Frame 239 -> {out_frame.name}") return out_frame cap.release() except Exception as e: print(f" [!] Lỗi trích OpenCV frame: {e}") # Fallback to FFmpeg try: if Path(video_path).exists(): keyframe_dir.mkdir(parents=True, exist_ok=True) cmd = [ "ffmpeg", "-y", "-sseof", "-0.1", "-i", str(video_path), "-update", "1", "-q:v", "2", str(out_frame) ] subprocess.run(cmd, capture_output=True) if out_frame.exists() and out_frame.stat().st_size > 1000: print(f" ✓ Đã trích xuất Tail Frame qua FFmpeg -> {out_frame.name}") return out_frame if keyframe_dir.exists() and not any(keyframe_dir.iterdir()): keyframe_dir.rmdir() except Exception as e: print(f" [!] Lỗi trích FFmpeg frame: {e}") if keyframe_dir.exists() and not any(keyframe_dir.iterdir()): try: keyframe_dir.rmdir() except Exception: pass return None def check_login_status(headless: bool = True, profile_dir: Optional[Path] = None) -> Tuple[bool, str]: """Kiểm tra xem session phu2234 đã đăng nhập vào Muse.ai hay chưa.""" if not HAS_INVPW: return False, "invisible_playwright chưa được cài đặt" pdir = profile_dir or DEFAULT_PROFILE if not pdir.exists(): return False, f"Thư mục profile không tồn tại: {pdir}" try: with InvisiblePlaywright(profile_dir=pdir, headless=headless) as ctx: page = ctx.pages[0] if ctx.pages else ctx.new_page() page.goto("https://muse.ai", wait_until="domcontentloaded", timeout=25000) time.sleep(3) title = page.title() url = page.url if "Chat" in title or "/access" in url: return True, f"Đã đăng nhập thành công ({title})" # Check for input box chat_inputs = page.locator('textarea, [placeholder*="Nhắn tin"], [placeholder*="Message"]').count() if chat_inputs > 0: return True, "Đã sẵn sàng ô chat Muse" return False, f"Cần đăng nhập (Trang hiện tại: {title} | {url})" except Exception as e: return False, f"Lỗi kết nối: {e}" def open_interactive_login(profile_dir: Optional[Path] = None): """Mở trình duyệt có giao diện (headed) để người dùng đăng nhập tài khoản Muse 1 lần duy nhất.""" pdir = profile_dir or DEFAULT_PROFILE pdir.mkdir(parents=True, exist_ok=True) print("\n" + "="*60) print("🌐 MỞ TRÌNH DUYỆT INVISIBLE PLAYWRIGHT ĐỂ ĐĂNG NHẬP MUSE.AI") print(f" Profile: {pdir}") print("="*60) print("👉 Cửa sổ trình duyệt đang mở trên màn hình.") print("👉 Vui lòng hoàn tất đăng nhập tài khoản 'phu2234' trên Muse.") print("👉 Sau khi đăng nhập vào giao diện Chat, quay lại đây hoặc đóng trình duyệt.") print("="*60 + "\n") try: with InvisiblePlaywright(profile_dir=pdir, headless=False) as ctx: page = ctx.pages[0] if ctx.pages else ctx.new_page() page.goto("https://muse.ai", wait_until="domcontentloaded", timeout=30000) print("[*] Đang chờ đăng nhập... (Tối đa 5 phút)") for i in range(100): time.sleep(3) try: title = page.title() url = page.url if "Chat" in title or "/access" in url: print(f"\n🎉 PHÁT HIỆN ĐĂNG NHẬP THÀNH CÔNG: {title} ({url})!") time.sleep(2) return True chat_inputs = page.locator('textarea, [placeholder*="Nhắn tin"], [placeholder*="Message"]').count() if chat_inputs > 0: print("\n🎉 PHÁT HIỆN Ô CHAT MUSE ĐÃ HOẠT ĐỘNG!") time.sleep(2) return True except Exception: pass if i % 10 == 0 and i > 0: print(f" ...vẫn đang chờ người dùng đăng nhập ({i*3}s)...") return False except Exception as e: print(f"[!] Lỗi khi mở giao diện đăng nhập: {e}") return False def render_shot_invpw( shot_id: str, prompt: str, input_frame: Optional[str] = None, reverse_motion: bool = False, profile_dir: Optional[Path] = None ) -> Tuple[bool, Optional[Path]]: """Render 1 shot video hoàn chỉnh qua InvisiblePlaywright.""" pdir = profile_dir or DEFAULT_PROFILE print(f"\n=======================================================") print(f"🎬 [InvisiblePlaywright] BẮT ĐẦU RENDER SHOT: {shot_id}") print(f" • Profile: {pdir.name}") print(f" • Reverse Motion: {reverse_motion}") print(f"=======================================================") # Chuẩn bị prompt với Audio Guard và Thanh lọc từ khóa nhạy cảm prompt = sanitize_prompt(prompt) audio_guard = " Âm thanh: Chỉ sinh âm thanh môi trường tự nhiên (foley, ambience) và thoại nhân vật chân thực. Tuyệt đối KHÔNG sinh nhạc nền (no music/BGM), không âm thanh điện tử, không tạp âm rè nhiễu." if "Tuyệt đối KHÔNG sinh nhạc nền" not in prompt and "Strictly NO background music" not in prompt: prompt = prompt.rstrip() + audio_guard temp_dl_dir = BASE_DIR / "04_Assets" / "temp_downloads" / "invpw" temp_dl_dir.mkdir(parents=True, exist_ok=True) with InvisiblePlaywright(profile_dir=pdir, headless=True) as ctx: page = ctx.pages[0] if ctx.pages else ctx.new_page() print("[1/5] Điều hướng đến Muse.ai và khởi tạo luồng render sạch...") page.goto("https://muse.ai", wait_until="domcontentloaded", timeout=30000) time.sleep(3) # Bấm nút tạo đoạn chat phụ mới để có thread sạch hoàn toàn new_btn = page.locator('button[aria-label*="chat phụ mới"], [title*="chat phụ mới"], [aria-label*="Đoạn chat phụ mới"]').first if new_btn.count() > 0 and new_btn.is_visible(): new_btn.click() time.sleep(3) elif "thread/new" not in page.url: page.goto("https://muse.ai/thread/new", wait_until="domcontentloaded", timeout=30000) time.sleep(3) print(f" ✓ Đã sẵn sàng luồng render sạch ({page.url})!") # 1. Upload ảnh đầu vào nếu có if input_frame and os.path.exists(input_frame): print(f"[2/5] Uploading Start Frame: {Path(input_frame).name}...") file_input = page.locator('input[type="file"]').first if file_input.count() > 0: file_input.set_input_files(str(input_frame)) attached = False for _ in range(15): time.sleep(1) if page.locator('button[aria-label="Gỡ file đính kèm"]').count() > 0: attached = True break if attached: print(" ✓ Đã upload và xác nhận đính kèm ảnh thành công!") else: print(" [!] Cảnh báo: Chưa thấy nút gỡ file đính kèm, tiếp tục đợi thêm 2s...") time.sleep(2) else: print(" [!] Không tìm thấy input[type=file] để upload ảnh!") else: print("[2/5] Bỏ qua upload (không có ảnh đính kèm)") # 2. Điền prompt vào ô chat (hỗ trợ đầy đủ bộ gõ tiếng Việt và React synthetic events) print("[3/5] Điền prompt vào ô chat...") 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) # Ghi nhận trạng thái download trước khi gửi dl_selector = 'button:has-text("Tải video xuống"), button[aria-label*="Tải video xuống"], button[data-pel="media_download"], button[data-pel-click="media_download"]' pre_dl_count = page.locator(dl_selector).count() downloads_dir = Path.home() / "Downloads" # 3. Bấm Gửi chuẩn xác stop_sel = 'button:has-text("Dừng"), button[aria-label*="Dừng"], button:has-text("Stop"), button[aria-label*="Stop"]' send_btn = page.locator('button[data-pel-click="chat_send_message"], button[aria-label="Gửi"], button[aria-label*="Gửi"]').first if send_btn.count() > 0 and send_btn.is_visible(): send_btn.click() else: page.keyboard.press("Enter") time.sleep(2) # Chờ xác nhận đã gửi vào engine submitted = False for _ in range(20): if page.locator(stop_sel).count() > 0 or page.locator('text="Vẫn đang gửi"').count() > 0: submitted = True 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: print(" [!] Lỗi: Không thể gửi lệnh vào Muse.ai (nút Dừng không xuất hiện)!") return False, None print(" ✓ Đã gửi lệnh render sang Muse.ai thành công (nút Dừng đã kích hoạt)!") # 4. Chờ render hoàn tất print("[4/5] Giám sát tiến độ render ngầm...") target_video = get_next_video_path(shot_id) raw_download_path = temp_dl_dir / f"raw_{shot_id}.mp4" MIN_RENDER_TIME = 40 # Video 10s không thể render dưới 40 giây MAX_WAIT_TIME = 500 # Tối đa 500 giây (~8.3 phút) theo chuẩn Hollywood pipeline start_time = time.time() task_started = False success = False while time.time() - start_time < MAX_WAIT_TIME: time.sleep(5) elapsed = int(time.time() - start_time) stop_count = page.locator(stop_sel).count() if stop_count > 0: task_started = True curr_dl_locator = page.locator(dl_selector) curr_dl_count = curr_dl_locator.count() # Chỉ kiểm tra lỗi khi task đã từng chạy và hiện đã ngừng (stop_count == 0) mà không có nút tải mới if task_started and stop_count == 0 and elapsed >= MIN_RENDER_TIME: # Kiểm tra tin nhắn trợ lý mới nhất assistant_locator = page.locator('[data-message-author="assistant"], [role="article"]') if assistant_locator.count() > 0: latest_text = assistant_locator.last.inner_text().strip() if any(w in latest_text.lower() for w in ["không dựng được", "bị từ chối", "không thể tạo", "lỗi an toàn", "chặn"]): print(f" [!] Muse trả lời từ chối: {latest_text[:200]}") return False, None if "không tìm thấy video" in latest_text.lower() and curr_dl_count <= pre_dl_count: print(" [!] Phát hiện lỗi backend Muse.ai trên shot hiện tại: 'Không tìm thấy video'!") return False, None is_done = (elapsed >= MIN_RENDER_TIME) and ( (curr_dl_count > pre_dl_count) or (task_started and stop_count == 0 and curr_dl_count > 0) ) status_msg = "Đã xong!" if is_done else f"Đang render ({elapsed}s)..." print(f" [{elapsed}s] Trạng thái: {status_msg} (Nút tải: {curr_dl_count}/{pre_dl_count})", flush=True) if is_done and curr_dl_count > 0: print(" 🎉 Muse đã hoàn tất render video! Đang bấm tải file về...") dl_btns = page.locator(dl_selector).all() if dl_btns: btn = dl_btns[-1] click_time = time.time() before_dls = {f.name: f.stat().st_mtime for f in downloads_dir.glob("*.mp4")} btn.scroll_into_view_if_needed() # 1. Bắt download trực tiếp bằng Playwright expect_download try: print(" [*] Bấm nút tải xuống qua expect_download...") with page.expect_download(timeout=12000) as dl_info: btn.click(force=True) dl = dl_info.value if raw_download_path.exists(): raw_download_path.unlink() dl.save_as(str(raw_download_path)) if raw_download_path.exists() and raw_download_path.stat().st_size > 50000: print(f" ✓ Đã tải video thành công qua expect_download: {raw_download_path.name} ({raw_download_path.stat().st_size} bytes)") success = True break except Exception as e_dl: print(f" [*] expect_download fallback ({e_dl}), theo dõi thư mục Downloads...") # 2. Theo dõi thư mục Downloads nếu expect_download chưa bắt được if not success: for _ in range(15): 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 > 50000 ] if new_files: new_files.sort(key=lambda x: x.stat().st_mtime, reverse=True) if raw_download_path.exists(): raw_download_path.unlink() shutil.move(str(new_files[0]), str(raw_download_path)) print(f" ✓ Đã tải video về thành công qua Downloads: {raw_download_path.name} ({raw_download_path.stat().st_size} bytes)") success = True break # Fallback: Trích xuất trực tiếp qua JS Blob URL trong context trình duyệt (bỏ qua mascot avatar 480x480) if not success: print(" [*] Thử fallback trích xuất video trực tiếp qua JS Blob...") v_elements = page.locator('video').all() for v_el in reversed(v_elements): try: vw = v_el.evaluate("el => el.videoWidth") vh = v_el.evaluate("el => el.videoHeight") if vw and vw < 600 and vh and vh < 600: continue # Bỏ qua avatar mascot 480x480 blob_url = v_el.get_attribute("src") or v_el.evaluate("el => el.currentSrc") if blob_url and blob_url.startswith("blob:"): import base64 b64_data = page.evaluate("""async (url) => { const res = await fetch(url); const blob = await res.blob(); return new Promise((resolve, reject) => { const reader = new FileReader(); reader.onloadend = () => resolve(reader.result.split(',')[1]); reader.onerror = reject; reader.readAsDataURL(blob); }); }""", blob_url) if b64_data: data = base64.b64decode(b64_data) if len(data) > 500000: if raw_download_path.exists(): raw_download_path.unlink() raw_download_path.write_bytes(data) print(f" ✓ Đã tải video thành công qua JS Blob ({vw}x{vh}): {raw_download_path.name} ({len(data)} bytes)") success = True break except Exception as e_blob: print(f" [!] JS Blob fallback error: {e_blob}") if success: break # Fallback check nếu hết thời gian mà chưa bắt được nút tải if not success: print(" [*] Kiểm tra lại thread hiện tại (fallback recovery)...") try: page.reload(wait_until="domcontentloaded", timeout=20000) time.sleep(3) dl_btns = page.locator(dl_selector).all() if dl_btns: print(" 🎉 Tìm thấy video sau khi reload thread! Đang tải về...") btn = dl_btns[-1] click_time = time.time() before_dls = {f.name: f.stat().st_mtime for f in downloads_dir.glob("*.mp4")} btn.click(force=True) 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 > 50000 ] if new_files: new_files.sort(key=lambda x: x.stat().st_mtime, reverse=True) if raw_download_path.exists(): raw_download_path.unlink() shutil.move(str(new_files[0]), str(raw_download_path)) print(f" ✓ Đã tải video về qua fallback: {raw_download_path.name}") success = True break except Exception as e: print(f" [!] Fallback check error: {e}") if not success or not raw_download_path.exists(): print(f"[!] Thất bại: Không nhận được video sau thời gian chờ tối đa.") return False, None # 5. Xử lý hậu kỳ (Reverse motion nếu có cờ) và lưu vào đích đến print(f"[5/5] Hoàn thiện video -> {target_video.name}...") VIDEOS_DIR.mkdir(parents=True, exist_ok=True) if reverse_motion: print(" 🔄 Đang áp dụng reverse motion (-vf reverse -af areverse)...") cmd_rev = [ "ffmpeg", "-y", "-i", str(raw_download_path), "-vf", "reverse", "-af", "areverse", "-c:v", "libx264", "-crf", "18", "-preset", "slow", "-c:a", "aac", "-b:a", "192k", str(target_video) ] subprocess.run(cmd_rev, capture_output=True) if raw_download_path.exists(): raw_download_path.unlink() else: shutil.move(str(raw_download_path), str(target_video)) if not target_video.exists() or target_video.stat().st_size < 1000: print(f"[!] Lỗi nghiêm trọng: File video đích không tồn tại hoặc bị rỗng: {target_video}") return False, None # Kiểm tra thông số kỹ thuật tối thiểu (chuẩn 10s và >= 720p) try: import json probe_cmd = ["ffprobe", "-v", "error", "-show_entries", "format=duration:stream=width,height", "-of", "json", str(target_video)] probe_res = subprocess.run(probe_cmd, capture_output=True, text=True) if probe_res.returncode == 0: meta = json.loads(probe_res.stdout) dur = float(meta.get("format", {}).get("duration", 0)) w = meta.get("streams", [{}])[0].get("width", 0) h = meta.get("streams", [{}])[0].get("height", 0) print(f" 📊 Thông số video xuất xưởng: {w}x{h}, thời lượng: {dur:.2f}s") if dur < 8.0: print(f" [!] CẢNH BÁO: Video ngắn hơn chuẩn 10s ({dur:.2f}s < 8s)!") except Exception: pass print(f" 🎉 HOÀN TẤT SHOT {shot_id}: {target_video.name} ({target_video.stat().st_size} bytes)") # Trích xuất Tail Frame 239 out_frame = extract_tail_frame(target_video, shot_id, reverse_motion) return True, str(out_frame) if out_frame else str(target_video) if __name__ == "__main__": import argparse parser = argparse.ArgumentParser(description="InvisiblePlaywright Muse Driver") parser.add_argument("--status", action="store_true", help="Kiểm tra trạng thái đăng nhập") parser.add_argument("--login", action="store_true", help="Mở giao diện đăng nhập cho người dùng") parser.add_argument("--shot", type=str, help="Render shot ID cụ thể") parser.add_argument("--prompt", type=str, help="Motion prompt") parser.add_argument("--input-frame", type=str, help="Start frame input") parser.add_argument("--reverse", action="store_true", help="Đảo ngược chuyển động") args = parser.parse_args() if args.status: ok, msg = check_login_status(headless=True) print(f"Trạng thái đăng nhập: {'SẴN SÀNG' if ok else 'CHƯA ĐĂNG NHẬP'} - {msg}") elif args.login: open_interactive_login() elif args.shot: from production_orchestrator import get_all_shots, resolve_start_frame shots = get_all_shots() sdata = shots.get(args.shot, {}) prompt = args.prompt or sdata.get("motion_prompt", "") iframe = args.input_frame or resolve_start_frame(args.shot, sdata) rev = args.reverse or bool(sdata.get("reverse_motion", False)) render_shot_invpw(args.shot, prompt, iframe, rev) else: parser.print_help()