import os import sys import time import shutil import subprocess from pathlib import Path from typing import Optional, Dict, Tuple, Any # Đảm bảo UTF-8 Console Windows chống UnicodeEncodeError cp1252 if sys.platform == "win32": try: sys.stdout.reconfigure(encoding="utf-8") sys.stderr.reconfigure(encoding="utf-8") except Exception: pass try: from gradio_client import Client, handle_file except ImportError: raise ImportError("Vui lòng cài đặt gradio_client: pip install gradio_client") from .hf_config import ( BASE_DIR, OUTPUTS_DIR, ASSETS_VIDEOS_DIR, SUPPORTED_SPACES, DEFAULT_MODEL, HF_TOKEN, NEGATIVE_PROMPT_DEFAULT, ) class HuggingFaceVideoClient: """Client kết nối tới các ZeroGPU Spaces trên Hugging Face để sinh Image-to-Video hoàn toàn miễn phí ($0).""" def __init__(self, token: Optional[str] = None): self.token = token or HF_TOKEN self._clients: Dict[str, Client] = {} OUTPUTS_DIR.mkdir(parents=True, exist_ok=True) def get_client(self, model_key: str) -> Client: """Khởi tạo hoặc tái sử dụng Gradio Client kết nối tới Space.""" if model_key not in SUPPORTED_SPACES: raise ValueError( f"Mô hình '{model_key}' không hợp lệ. Chọn một trong: {list(SUPPORTED_SPACES.keys())}" ) if model_key not in self._clients: space_id = SUPPORTED_SPACES[model_key]["space_id"] print(f"[HF Client] Đang kết nối tới Space: {space_id}...") self._clients[model_key] = Client(space_id, token=self.token) print(f"[HF Client] ✓ Kết nối thành công tới {space_id}!") return self._clients[model_key] def generate_i2v( self, image_path: str, prompt: str, model_key: str = DEFAULT_MODEL, negative_prompt: Optional[str] = None, duration: float = 4.0, width: Optional[int] = None, height: Optional[int] = None, seed: Optional[int] = None, output_filename: Optional[str] = None, ) -> Tuple[bool, Optional[str]]: """Sinh video từ ảnh Start Frame và Prompt chuyển động. Trả về: (thành công: bool, đường dẫn file video mp4: str) """ img_p = Path(image_path).resolve() if not img_p.exists(): print(f"[!] Không tìm thấy ảnh đầu vào: {img_p}") return False, None cfg = SUPPORTED_SPACES.get(model_key, SUPPORTED_SPACES[DEFAULT_MODEL]) neg_prompt = negative_prompt or NEGATIVE_PROMPT_DEFAULT w = width or cfg.get("default_width", 704) h = height or cfg.get("default_height", 512) print("\n" + "=" * 60) print(f"🎬 HUGGING FACE ZERO-GPU I2V GENERATION") print(f" • Model: {cfg['name']} ({cfg['space_id']})") print(f" • Input Image: {img_p.name}") print(f" • Prompt: {prompt[:80]}...") print(f" • Chi phí: {cfg['cost']}") print("=" * 60) start_time = time.time() client = self.get_client(model_key) try: if model_key == "ltx": # Lightricks LTX-Video Distilled Endpoint print(f"[...] Đang gửi tác vụ tới LTX-Video Space...") res = client.predict( prompt=prompt, negative_prompt=neg_prompt, input_image_filepath=handle_file(str(img_p)), input_video_filepath=None, height_ui=float(h), width_ui=float(w), mode="image-to-video", duration_ui=float(duration), ui_frames_to_use=9, seed_ui=int(seed) if seed is not None else 42, randomize_seed=True if seed is None else False, ui_guidance_scale=1.0, improve_texture_flag=True, api_name="/image_to_video", ) # res format: (generated_video_dict, seed) video_data = res[0] temp_video_path = ( video_data.get("video") if isinstance(video_data, dict) else video_data ) elif model_key == "cogvideox": # CogVideoX-5B Space Endpoint print(f"[...] Đang gửi tác vụ tới CogVideoX Space...") res = client.predict( prompt=prompt, image_input=handle_file(str(img_p)), video_input=None, video_strength=0.8, seed_value=float(seed) if seed is not None else -1, scale_status=False, rife_status=False, api_name="/generate", ) # res format: (cogvideox_generate_video, _download_video, _download_gif, seed) temp_video_path = res[1] or ( res[0].get("video") if isinstance(res[0], dict) else res[0] ) elif model_key == "wan21": # Alibaba Wan 2.1 Space (Async submission + Polling) print( f"[...] Đang gửi tác vụ bất đồng bộ tới Wan 2.1 Space..." ) client.predict( prompt=prompt, image=handle_file(str(img_p)), watermark_wan=False, seed=float(seed) if seed is not None else -1, api_name="/i2v_generation_async", ) print( "[...] Đang theo dõi tiến trình sinh video Wan 2.1 (Polling status)..." ) temp_video_path = None max_polls = 60 # Đợi tối đa 5 phút for poll_i in range(max_polls): time.sleep(5) poll_res = client.predict(api_name="/status_refresh") # poll_res format: [VideoData, cost_time, waiting_time, progress] if poll_res and isinstance(poll_res, (list, tuple)): vid_candidate = poll_res[0] if isinstance(vid_candidate, dict) and vid_candidate.get( "video" ): temp_video_path = vid_candidate["video"] break elif ( isinstance(vid_candidate, str) and vid_candidate.endswith(".mp4") ): temp_video_path = vid_candidate break print( f" [Wan 2.1 Polling] Lần {poll_i+1}/{max_polls}... Đang xử lý" ) if not temp_video_path: print( "[!] Quá thời gian chờ (Timeout) khi đợi Wan 2.1 hoàn thành." ) return False, None elif model_key == "wan22": # Alibaba Wan 2.2 FP8 AoTI Faster Space print(f"[...] Đang gửi tác vụ tới Wan 2.2 Space ({cfg['space_id']})...") res = client.predict( input_image=handle_file(str(img_p)), prompt=prompt, steps=6.0, negative_prompt=neg_prompt, duration_seconds=float(duration), guidance_scale=1.0, guidance_scale_2=1.0, seed=float(seed) if seed is not None else 42.0, randomize_seed=True if seed is None else False, api_name="/generate_video", ) # res format: (generated_video, seed) video_data = res[0] temp_video_path = ( video_data.get("video") if isinstance(video_data, dict) else video_data ) else: print(f"[!] Không hỗ trợ phương thức xử lý cho: {model_key}") return False, None elapsed = time.time() - start_time print( f"[✓] Space đã tạo xong video trong {elapsed:.1f}s: {temp_video_path}" ) # Lưu video về outputs if not output_filename: timestamp = int(time.time()) output_filename = f"{model_key}_{img_p.stem}_{timestamp}.mp4" final_dest = OUTPUTS_DIR / output_filename shutil.copy2(temp_video_path, str(final_dest)) print(f"[✓] Đã lưu video hoàn tất tại: {final_dest}") return True, str(final_dest) except Exception as e: print(f"[!] Lỗi trong quá trình gọi Space API: {e}") return False, None @staticmethod def extract_tail_frame( video_path: str, output_tail_frame: Optional[str] = None ) -> Optional[str]: """Trích xuất Tail Frame (khung hình cuối cùng) bằng FFmpeg để phục vụ Head-Tail Chaining.""" vid_p = Path(video_path) if not vid_p.exists(): return None if not output_tail_frame: output_tail_frame = str(vid_p.with_name(f"{vid_p.stem}_tail.jpg")) cmd = [ "ffmpeg", "-y", "-sseof", "-1", "-i", str(vid_p), "-update", "1", "-q:v", "2", str(output_tail_frame), ] res = subprocess.run(cmd, capture_output=True) if res.returncode == 0 and os.path.exists(output_tail_frame): print(f"[✓] Đã trích xuất Tail Frame: {output_tail_frame}") return output_tail_frame return None