""" File processing (archives, thumbnails) """ import asyncio from pathlib import Path from typing import Optional from PIL import Image import logging logger = logging.getLogger(__name__) async def generate_thumbnail(video_path: str, output_path: str, size: tuple = (320, 240)) -> bool: """ Generate thumbnail for video using ffmpeg Args: video_path: Path to video file output_path: Path to save thumbnail size: Thumbnail size (width, height) Returns: True if successful, False otherwise """ try: import subprocess import asyncio # Check if ffmpeg is available try: result = await asyncio.create_subprocess_exec( 'ffmpeg', '-version', stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE ) await result.wait() if result.returncode != 0: logger.warning("ffmpeg not found, thumbnail generation not possible") return False except FileNotFoundError: logger.warning("ffmpeg not installed, thumbnail generation not possible") return False # Generate thumbnail from middle of video output_file = Path(output_path) output_file.parent.mkdir(parents=True, exist_ok=True) # Get video duration duration_cmd = [ 'ffprobe', '-v', 'error', '-show_entries', 'format=duration', '-of', 'default=noprint_wrappers=1:nokey=1', str(video_path) ] proc = await asyncio.create_subprocess_exec( *duration_cmd, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE ) stdout, _ = await proc.communicate() if proc.returncode != 0: logger.warning("Failed to get video duration") # Use 1 second as default seek_time = 1 else: try: duration = float(stdout.decode().strip()) seek_time = duration / 2 # Middle of video except (ValueError, IndexError): seek_time = 1 # Generate thumbnail cmd = [ 'ffmpeg', '-i', str(video_path), '-ss', str(seek_time), '-vframes', '1', '-vf', f'scale={size[0]}:{size[1]}', '-y', # Overwrite if exists str(output_path) ] proc = await asyncio.create_subprocess_exec( *cmd, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE ) await proc.wait() if proc.returncode == 0 and Path(output_path).exists(): logger.info(f"Thumbnail created: {output_path}") return True else: logger.warning(f"Failed to create thumbnail for {video_path}") return False except Exception as e: logger.error(f"Error generating thumbnail: {e}", exc_info=True) return False async def extract_archive(archive_path: str, output_dir: str, password: Optional[str] = None) -> bool: """ Extract archive Args: archive_path: Path to archive output_dir: Directory for extraction password: Archive password (if required) Returns: True if successful, False otherwise """ try: # TODO: Implement archive extraction # Support zip, rar, 7z logger.warning("Archive extraction not implemented yet") return False except Exception as e: logger.error(f"Error extracting archive: {e}") return False