122 lines
3.6 KiB
Python
122 lines
3.6 KiB
Python
"""
|
|
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
|
|
|