f634b884ec
Features: - Create and manage sticker packs - Add existing sticker sets to database - Automatic image optimization (512px, WebP conversion) - GIF to animated video sticker conversion (WebM, VP9) - Background removal (local rembg + optional Remove.bg API) - Hybrid API with retry logic and usage tracking - Premium status detection - SQLite database for persistence - Docker containerization - Complete OPENCODE.md documentation Technical Stack: - python-telegram-bot 22.7 - Pillow for image processing - FFmpeg for video conversion - rembg for local background removal - aiosqlite for async database operations
268 lines
8.7 KiB
Python
268 lines
8.7 KiB
Python
"""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
|