import os import uuid import io import anyio import numpy as np from PIL import Image, ImageDraw from moviepy import AudioFileClip, VideoClip from fastapi.responses import FileResponse from pydub import AudioSegment from dicebear import DAvatar, DStyle, DFormat import requests from services.tts_service import tts_service from utils import remove_file async def generate_avatar_image(prompt: str, output_path: str) -> bool: try: av = DAvatar( style=DStyle.avataaars, seed=prompt ) def _fetch_and_save(): # نستخدم mouth=serious لضمان فم مغلق جاهز للتحريك، و size=1024 للجودة العالية url = av.url_svg.replace("/svg?", "/png?") + "&mouth=serious&size=1024&backgroundColor=transparent" response = requests.get(url) if response.status_code == 200: with open(output_path, "wb") as f: f.write(response.content) return True return False success = await anyio.to_thread.run_sync(_fetch_and_save) if success and os.path.exists(output_path): print(f"SUCCESS: Avatar image generated using DiceBear (seed: {prompt})") return True return False except Exception as e: print(f"DiceBear Generation Error: {e}") return False async def avatar_service(background_tasks, text, voice, rate, pitch, volume, tashkeel, avatar_prompt=None): file_id = str(uuid.uuid4()) video_path = f"avatar_{file_id}.mp4" audio_path = None temp_files = [] try: # 1) Generate Audio audio_path, extra_temps = await tts_service( background_tasks, text, voice, rate, pitch, volume, tashkeel, return_path=True ) temp_files.extend(extra_temps) temp_files.append(audio_path) # 2) Load or Generate Base Image generated_img_path = f"temp_base_{file_id}.png" temp_files.append(generated_img_path) # تحديد الـ prompt: إذا لم يتوفر، نستخدم أفاتار افتراضي أنيق effective_prompt = avatar_prompt.strip() if (avatar_prompt and avatar_prompt.strip()) else "charismatic young businessman with modern hairstyle and suit" success = await generate_avatar_image(effective_prompt, generated_img_path) if success: base_img_path = generated_img_path else: base_img_path = "avatar_base.png" if not os.path.exists(base_img_path): # إنشاء صورة شفافة كخيار أخير جداً img = Image.new("RGBA", (1024, 1024), color=(0, 0, 0, 0)) img.save(base_img_path) base_img = Image.open(base_img_path).convert("RGBA") w, h = base_img.size face_mask = base_img.getchannel('A') # 3) Animation Settings m_center_x = int(w * 0.5) m_center_y = int(h * 0.55) # ارتفاع الفم m_width = int(w * 0.12) try: skin_color = base_img.getpixel((m_center_x, m_center_y - 40)) except: skin_color = (235, 195, 165, 255) if len(skin_color) < 4: skin_color = (*skin_color[:3], 255) lip_color = (max(0, skin_color[0] - 40), max(0, skin_color[1] - 60), max(0, skin_color[2] - 50), 255) inner_mouth_color = (40, 10, 10, 255) def create_pro_frame(open_ratio): frame = base_img.copy().convert("RGBA") overlay = Image.new("RGBA", frame.size, (0, 0, 0, 0)) draw = ImageDraw.Draw(overlay) m_width_dynamic = int(m_width * (0.9 + (open_ratio * 0.2))) current_open_h = int(35 * open_ratio) mouth_rect = [ m_center_x - m_width_dynamic // 2, m_center_y, m_center_x + m_width_dynamic // 2, m_center_y + current_open_h ] if open_ratio > 0.05: draw.pieslice(mouth_rect, 0, 180, fill=inner_mouth_color) if open_ratio > 0.4: draw.line([ m_center_x - m_width_dynamic // 3, m_center_y + 5, m_center_x + m_width_dynamic // 3, m_center_y + 5 ], fill=(255, 255, 255, 200), width=2) draw.line([ (m_center_x - m_width_dynamic // 2, m_center_y), (m_center_x + m_width_dynamic // 2, m_center_y) ], fill=lip_color, width=4) if open_ratio > 0: draw.arc(mouth_rect, 0, 180, fill=lip_color, width=4) combined_mask = Image.new("L", frame.size, 0) overlay_alpha = overlay.getchannel('A') combined_mask.paste(overlay_alpha, mask=face_mask) overlay.putalpha(combined_mask) frame = Image.alpha_composite(frame, overlay) return np.array(frame.convert("RGB")) # 4) Generate Animation Frames frames = [] for i in range(5): frames.append(create_pro_frame(i / 4.0)) # 5) Analyze Audio audio_seg = AudioSegment.from_file(audio_path) duration = audio_seg.duration_seconds fps = 15 frame_count = max(1, int(duration * fps)) chunk_size = max(1, int(len(audio_seg) / frame_count)) volumes = [] for i in range(frame_count): start = i * chunk_size end = (i + 1) * chunk_size chunk = audio_seg[start:end] volumes.append(chunk.rms) max_vol = max(volumes) if volumes and max(volumes) > 0 else 1 # 6) Create Video def make_frame(t): frame_idx = int(t * fps) if frame_idx >= len(volumes): return frames[0] vol_ratio = volumes[frame_idx] / max_vol if vol_ratio > 0.8: return frames[4] if vol_ratio > 0.6: return frames[3] if vol_ratio > 0.4: return frames[2] if vol_ratio > 0.1: return frames[1] return frames[0] clip = VideoClip(make_frame, duration=duration) audio_clip = AudioFileClip(audio_path) clip = clip.with_audio(audio_clip) clip.write_videofile(video_path, fps=fps, codec="libx264", audio_codec="aac", logger=None) clip.close() audio_clip.close() # 7) Cleanup and Return background_tasks.add_task(remove_file, video_path) for f in temp_files: background_tasks.add_task(remove_file, f) return FileResponse(path=video_path, media_type="video/mp4", filename="avatar.mp4") except Exception as e: print(f"Avatar Service Error: {e}") if os.path.exists(video_path): os.remove(video_path) for f in temp_files: if os.path.exists(f): os.remove(f) return {"error": str(e)}