""" Helper utilities for MixVideo V2 """ import os from pathlib import Path from typing import Dict, Any, Union import ffmpeg def format_duration(seconds: float) -> str: """ Format duration in seconds to human-readable string. Args: seconds: Duration in seconds Returns: Formatted duration string (e.g., "1:23:45") """ hours = int(seconds // 3600) minutes = int((seconds % 3600) // 60) secs = int(seconds % 60) if hours > 0: return f"{hours}:{minutes:02d}:{secs:02d}" else: return f"{minutes}:{secs:02d}" def get_file_info(file_path: Union[str, Path]) -> Dict[str, Any]: """ Get basic file information. Args: file_path: Path to file Returns: Dictionary with file information """ try: path = Path(file_path) if not path.exists(): return {"error": "File not found"} stat = path.stat() info = { "name": path.name, "size": stat.st_size, "size_mb": round(stat.st_size / (1024 * 1024), 2), "modified": stat.st_mtime, "extension": path.suffix.lower(), "is_file": path.is_file(), "is_directory": path.is_dir() } # Try to get media info if it's a media file if path.suffix.lower() in ['.mp4', '.avi', '.mov', '.mkv', '.wmv', '.flv', '.mp3', '.wav', '.aac', '.flac', '.ogg']: try: probe = ffmpeg.probe(str(path)) format_info = probe.get('format', {}) info.update({ "duration": float(format_info.get('duration', 0)), "duration_formatted": format_duration(float(format_info.get('duration', 0))), "bitrate": int(format_info.get('bit_rate', 0)), "format_name": format_info.get('format_name', ''), }) # Video specific info video_streams = [s for s in probe.get('streams', []) if s.get('codec_type') == 'video'] if video_streams: video_stream = video_streams[0] info.update({ "width": int(video_stream.get('width', 0)), "height": int(video_stream.get('height', 0)), "fps": eval(video_stream.get('r_frame_rate', '0/1')), "video_codec": video_stream.get('codec_name', ''), }) # Audio specific info audio_streams = [s for s in probe.get('streams', []) if s.get('codec_type') == 'audio'] if audio_streams: audio_stream = audio_streams[0] info.update({ "sample_rate": int(audio_stream.get('sample_rate', 0)), "channels": int(audio_stream.get('channels', 0)), "audio_codec": audio_stream.get('codec_name', ''), }) except Exception as e: info["media_info_error"] = str(e) return info except Exception as e: return {"error": str(e)} def ensure_directory(directory: Union[str, Path]) -> Path: """ Ensure directory exists, create if it doesn't. Args: directory: Directory path Returns: Path object of the directory """ path = Path(directory) path.mkdir(parents=True, exist_ok=True) return path def get_unique_filename(file_path: Union[str, Path]) -> Path: """ Get a unique filename by appending a number if file already exists. Args: file_path: Original file path Returns: Unique file path """ path = Path(file_path) if not path.exists(): return path counter = 1 while True: stem = path.stem suffix = path.suffix parent = path.parent new_name = f"{stem}_{counter}{suffix}" new_path = parent / new_name if not new_path.exists(): return new_path counter += 1 def format_file_size(size_bytes: int) -> str: """ Format file size in bytes to human-readable string. Args: size_bytes: Size in bytes Returns: Formatted size string (e.g., "1.5 MB") """ if size_bytes == 0: return "0 B" size_names = ["B", "KB", "MB", "GB", "TB"] i = 0 size = float(size_bytes) while size >= 1024.0 and i < len(size_names) - 1: size /= 1024.0 i += 1 return f"{size:.1f} {size_names[i]}" def clean_temp_files(temp_dir: Union[str, Path], max_age_hours: int = 24): """ Clean old temporary files. Args: temp_dir: Temporary directory path max_age_hours: Maximum age of files to keep in hours """ import time temp_path = Path(temp_dir) if not temp_path.exists(): return current_time = time.time() max_age_seconds = max_age_hours * 3600 for file_path in temp_path.rglob('*'): if file_path.is_file(): file_age = current_time - file_path.stat().st_mtime if file_age > max_age_seconds: try: file_path.unlink() except Exception: pass # Ignore errors when deleting files