from pathlib import Path import os import json import tempfile import shutil from typing import Dict, List, Optional, Tuple import requests import numpy as np from moviepy.editor import ( ImageClip, AudioFileClip, VideoFileClip, CompositeVideoClip, concatenate_videoclips, ) from moviepy.video.VideoClip import ColorClip from moviepy.audio.AudioClip import AudioArrayClip from Video_generation.helpers import * from dotenv import load_dotenv from celery_app import celery def generate_lip_sync_via_api(api_url, video_path, audio_path, output_path=None): """ Sends video and audio to your Flask Wav2Lip API. Downloads the generated output video to output_path (if provided). Args: api_url (str): Base URL of the running Flask API (e.g. "http://127.0.0.1:5001") video_path (str): Path to input face video file. audio_path (str): Path to input audio file. output_path (str, optional): Path to save returned lip-synced video. Returns: str: Local path to saved output video or None if failed. """ try: # Ensure files exist if not os.path.exists(video_path): raise FileNotFoundError(f"Video file not found: {video_path}") if not os.path.exists(audio_path): raise FileNotFoundError(f"Audio file not found: {audio_path}") files = { "video": open(video_path, "rb"), "audio": open(audio_path, "rb") } print("[INFO] Sending request to Flask API...") response = requests.post(f"{api_url}", files=files) if response.status_code == 200: # Save output video if provided if output_path: with open(output_path, "wb") as f: f.write(response.content) print(f"[DONE] Lip-synced video saved to: {output_path}") return output_path else: print("[DONE] Lip-synced video received (no save path provided).") return "Success" else: print(f"[ERROR] API returned status {response.status_code}: {response.text}") return None except Exception as e: print(f"[ERROR] API call failed: {e}") return None load_dotenv() poppler_path = os.getenv("POPPLER_PATH", None) @celery.task(name="Video_generation.video_sync") def pdf_audio_map_to_video( data: Dict, avatar_path: str, dpi: int = 150, poppler_path: Optional[str] = None, resolution: Tuple[int, int] = (1280, 720), avatar_size: Tuple[int, int] = (180, 180), avatar_margin: int = 16, fps: int = 24, render_segments_to_disk: bool = True, tmp_dir: Optional[str] = "./public/tmp", title_slide_duration: float = 3.0, ) -> str: """ Generate final video with slides + audio mapping + lip-synced avatar. """ api_url = os.getenv("LIPSYNC_API_URL", "https://eduruby.in/lip-sync/api/lipsync") if tmp_dir: tmpdir = Path(tmp_dir) / str(data["id"]) tmpdir.mkdir(parents=True, exist_ok=True) tmp_created = True else: tmpdir = Path(tempfile.mkdtemp()) tmp_created = True pdf_path = data.get("pdf_path") audio_map_path = data.get("audio_mapping_file") with open(audio_map_path, "r", encoding="utf-8") as f: audio_map_source = json.load(f) output_path = Path(f"./public/Video/{data['id']}_final_video.mp4") output_path.parent.mkdir(parents=True, exist_ok=True) audio_clips, seg_files = [], [] try: print("[STEP] Converting PDF pages to images...") page_images = pdf_to_images(pdf_path, str(tmpdir), dpi=dpi, poppler_path=poppler_path) if not page_images: raise RuntimeError("No images produced from PDF conversion.") print(f"[INFO] Produced {len(page_images)} pages") audio_map = load_audio_map(audio_map_source) if not isinstance(audio_map, dict): raise ValueError("audio_map must be a dict mapping page numbers to audio paths") if not Path(avatar_path).exists(): raise FileNotFoundError("Avatar video not found: " + avatar_path) W, H = resolution aw, ah = avatar_size num_pages = len(page_images) for page_no in range(1, num_pages + 1): img_path = page_images[page_no - 1] # --- AUDIO HANDLING --- if page_no == 1: sr = 44100 duration = title_slide_duration silent_arr = np.zeros((int(sr * duration), 1), dtype=np.float32) audio_clip = AudioArrayClip(silent_arr, fps=sr) print(f"[INFO] Page 1: silent title slide") else: audio_path = audio_map.get(page_no - 1) if audio_path and Path(audio_path).exists(): audio_clip = AudioFileClip(audio_path) print(f"[INFO] Page {page_no}: mapped audio found → {Path(audio_path).name}") else: sr = 44100 duration = title_slide_duration silent_arr = np.zeros((int(sr * duration), 1), dtype=np.float32) audio_clip = AudioArrayClip(silent_arr, fps=sr) print(f"[WARN] Page {page_no}: no audio mapping, inserted silence") audio_clips.append(audio_clip) duration = audio_clip.duration # --- SLIDE BASE --- slide_img_clip = ImageClip(str(img_path)).set_duration(duration) slide_img_clip = ( slide_img_clip.resize(width=W) if (slide_img_clip.w / slide_img_clip.h) > (W / H) else slide_img_clip.resize(height=H) ) img_w, img_h = slide_img_clip.w, slide_img_clip.h x_pos, y_pos = (W - img_w) // 2, (H - img_h) // 2 bg = ColorClip(size=(W, H), color=(255, 255, 255)).set_duration(duration) slide_segment = CompositeVideoClip( [bg, slide_img_clip.set_pos((x_pos, y_pos))] ).set_audio(audio_clip) # --- AVATAR OVERLAY --- if page_no == 1: # Title slide: no avatar composed = slide_segment.set_duration(duration).set_fps(fps) else: # Generate lip-synced avatar first synced_avatar_out = str(tmpdir / f"lip_avatar_page_{page_no}.mp4") sync_success = generate_lip_sync_via_api( api_url=api_url, video_path=avatar_path, audio_path=audio_path, output_path=synced_avatar_out, ) if sync_success: print(f"[INFO] Lip-sync successful for page {page_no}") else: print(f"[WARN] Lip-sync failed for page {page_no}, using original avatar") avatar_final_path = ( synced_avatar_out if sync_success else avatar_path ) avatar_clip = make_avatar_clip_for_duration( avatar_final_path, duration=duration, avatar_size=avatar_size ) mask_arr = create_circle_mask_array((aw, ah)) if mask_arr.ndim == 3: mask_arr = mask_arr[..., 0] mask_arr = (mask_arr.astype("float32") / 255.0) mask_clip = ( ImageClip(mask_arr, ismask=True) .set_duration(duration) .resize(newsize=avatar_size) ) avatar_clip = avatar_clip.set_mask(mask_clip) pos_x, pos_y = W - aw - avatar_margin, H - ah - avatar_margin avatar_clip = avatar_clip.set_pos((pos_x, pos_y)) composed = CompositeVideoClip( [slide_segment, avatar_clip], size=(W, H) ).set_duration(duration).set_fps(fps) # --- EXPORT EACH SEGMENT --- seg_dir = tmpdir / "segments" seg_dir.mkdir(parents=True, exist_ok=True) seg_out = str(seg_dir / f"page_{page_no:03d}.mp4") print(f"[INFO] Rendering segment to disk: {seg_out}") composed.write_videofile( seg_out, codec="libx264", audio_codec="aac", fps=fps, verbose=False, threads=min(4, (os.cpu_count() or 1)), ) seg_files.append(seg_out) composed.close() # --- CONCATENATION --- print("[STEP] Concatenating segments...") final = concatenate_videoclips( [VideoFileClip(p) for p in seg_files], method="compose" ) print(f"[STEP] Writing final video: {output_path}") final.write_videofile( str(output_path), codec="libx264", audio_codec="aac", fps=fps, threads=min(4, (os.cpu_count() or 1)), ) final.close() print("[DONE] Final video:", output_path) return str(output_path) finally: if tmp_created and tmpdir.exists(): try: shutil.rmtree(tmpdir) print("[CLEANUP] Removed temporary dir:", tmpdir) except Exception as e: print("[WARN] Failed to remove tmp dir:", e)