"""GIF to WebM Video Sticker converter using ffmpeg.""" import io import logging import subprocess import tempfile from typing import Optional, Tuple, Dict, Any from PIL import Image logger = logging.getLogger(__name__) # Telegram Video Sticker requirements MAX_DURATION_SECONDS = 3 MAX_FILE_SIZE_KB = 256 MAX_FILE_SIZE_BYTES = MAX_FILE_SIZE_KB * 1024 STICKER_SIZE = 512 TARGET_FPS = 30 async def get_gif_info(gif_bytes: bytes) -> Dict[str, Any]: """ Get GIF metadata using ffprobe. Returns dict with duration, width, height, fps """ try: with tempfile.NamedTemporaryFile(suffix='.gif', delete=False) as tmp_in: tmp_in.write(gif_bytes) tmp_in.flush() # Use ffprobe to get duration result = subprocess.run( [ 'ffprobe', '-v', 'error', '-show_entries', 'format=duration', '-show_entries', 'stream=width,height,r_frame_rate', '-of', 'default=noprint_wrappers=1', tmp_in.name ], capture_output=True, text=True ) info = {'duration': 0, 'width': 0, 'height': 0, 'fps': 0} for line in result.stdout.strip().split('\n'): if '=' in line: key, value = line.split('=', 1) if key == 'duration': info['duration'] = float(value) elif key == 'width': info['width'] = int(value) elif key == 'height': info['height'] = int(value) elif key == 'r_frame_rate': # Parse fraction like "30/1" if '/' in value: num, den = value.split('/') info['fps'] = int(num) / int(den) if int(den) != 0 else 0 return info except Exception as e: logger.error(f"Error getting GIF info: {e}") return {'duration': 0, 'width': 0, 'height': 0, 'fps': 0} async def validate_gif_duration(duration: float) -> Tuple[bool, str]: """ Check if GIF duration is within limits. Returns: Tuple of (is_valid, message) """ if duration > MAX_DURATION_SECONDS: return False, f"GIF ist {duration:.1f}s lang. Maximal erlaubt: {MAX_DURATION_SECONDS}s" return True, "" async def convert_gif_to_webm( gif_bytes: bytes, max_duration: Optional[float] = None ) -> Optional[bytes]: """ Convert GIF to WebM Video Sticker format using ffmpeg. Args: gif_bytes: Raw GIF file bytes max_duration: Maximum duration in seconds (None = no limit, will be truncated) Returns: WebM video bytes or None if conversion fails """ try: # Get GIF info gif_info = await get_gif_info(gif_bytes) duration = gif_info.get('duration', 0) width = gif_info.get('width', 512) height = gif_info.get('height', 512) logger.info(f"Converting GIF: {width}x{height}, {duration}s, " f"{gif_info.get('fps', 0)}fps") # Calculate dimensions (one side must be exactly 512px) if width >= height: new_width = STICKER_SIZE new_height = int((height / width) * STICKER_SIZE) else: new_height = STICKER_SIZE new_width = int((width / height) * STICKER_SIZE) # Ensure neither exceeds 512 new_width = min(new_width, STICKER_SIZE) new_height = min(new_height, STICKER_SIZE) # Determine duration limit duration_limit = min(duration, max_duration or duration) if duration > MAX_DURATION_SECONDS: duration_limit = MAX_DURATION_SECONDS logger.info(f"GIF too long ({duration}s), truncating to {duration_limit}s") # Create temporary files with tempfile.NamedTemporaryFile(suffix='.gif', delete=False) as tmp_in: tmp_in.write(gif_bytes) tmp_in_path = tmp_in.name with tempfile.NamedTemporaryFile(suffix='.webm', delete=False) as tmp_out: tmp_out_path = tmp_out.name # Build ffmpeg command # Video sticker requirements: WebM, VP9, 512px, 30fps, no audio, max 3s, max 256KB cmd = [ 'ffmpeg', '-i', tmp_in_path, '-t', str(duration_limit), # Duration limit '-vf', f'scale={new_width}:{new_height}:flags=lanczos,fps={TARGET_FPS}', # Resize + FPS '-c:v', 'libvpx-vp9', # VP9 codec '-pix_fmt', 'yuva420p', # With alpha channel (transparency) '-b:v', '0', # Use constant quality '-crf', '40', # Quality (lower = better, 0-63) '-deadline', 'good', # Encoding speed/quality tradeoff '-cpu-used', '2', # Encoding speed '-row-mt', '1', # Multi-threading '-an', # No audio '-f', 'webm', # Output format '-y', # Overwrite output tmp_out_path ] # Run ffmpeg result = subprocess.run( cmd, capture_output=True, text=True ) if result.returncode != 0: logger.error(f"ffmpeg failed: {result.stderr}") return None # Read output file with open(tmp_out_path, 'rb') as f: webm_bytes = f.read() file_size_kb = len(webm_bytes) / 1024 logger.info(f"Initial conversion: {file_size_kb:.1f}KB") # If too large, try with lower quality if len(webm_bytes) > MAX_FILE_SIZE_BYTES: webm_bytes = await _compress_webm(tmp_in_path, tmp_out_path, duration_limit, new_width, new_height) if webm_bytes and len(webm_bytes) <= MAX_FILE_SIZE_BYTES: final_size_kb = len(webm_bytes) / 1024 logger.info(f"Final WebM: {final_size_kb:.1f}KB, " f"{new_width}x{new_height}") return webm_bytes else: logger.error("Could not compress to required size") return None except Exception as e: logger.error(f"Error converting GIF to WebM: {e}") return None async def _compress_webm( input_path: str, output_path: str, duration: float, width: int, height: int ) -> Optional[bytes]: """ Try to compress WebM to meet size requirements. Tries progressively lower quality settings. """ # Try different CRF values (lower = better quality, higher = smaller file) crf_values = [45, 50, 55, 60] for crf in crf_values: try: logger.info(f"Trying compression with CRF={crf}") cmd = [ 'ffmpeg', '-i', input_path, '-t', str(duration), '-vf', f'scale={width}:{height}:flags=lanczos,fps={TARGET_FPS}', '-c:v', 'libvpx-vp9', '-pix_fmt', 'yuva420p', '-b:v', '0', '-crf', str(crf), '-deadline', 'good', '-cpu-used', '3', # Faster encoding for smaller files '-row-mt', '1', '-an', '-f', 'webm', '-y', output_path ] result = subprocess.run(cmd, capture_output=True, text=True) if result.returncode == 0: with open(output_path, 'rb') as f: webm_bytes = f.read() file_size_kb = len(webm_bytes) / 1024 logger.info(f"Compression CRF={crf}: {file_size_kb:.1f}KB") if len(webm_bytes) <= MAX_FILE_SIZE_BYTES: return webm_bytes except Exception as e: logger.error(f"Compression attempt failed: {e}") continue return None async def is_animated_image(file_bytes: bytes) -> bool: """ Check if image file is animated (GIF with multiple frames). Args: file_bytes: Raw file bytes Returns: True if animated, False otherwise """ try: image = Image.open(io.BytesIO(file_bytes)) # Check if it's a GIF if image.format != 'GIF': return False # Check if it has multiple frames try: image.seek(1) # Try to go to second frame return True except EOFError: return False except Exception as e: logger.error(f"Error checking if image is animated: {e}") return False