comi-logic/mxivideo
7
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

first commit

Browse Source
This commit is contained in:
root
2025-07-10 09:41:40 +08:00
commit 7d72e07639
58 changed files with 7492 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

23
python_core/utils/__init__.py Normal file
View File

@@ -0,0 +1,23 @@
"""
Utilities Module
工具模块
Common utilities and helper functions used across the application.
"""
from .logger import setup_logger
from .validators import validate_video_file, validate_audio_file, validate_image_file
from .helpers import format_duration, get_file_info, ensure_directory, get_unique_filename, format_file_size, clean_temp_files
__all__ = [
"setup_logger",
"validate_video_file",
"validate_audio_file",
"validate_image_file",
"format_duration",
"get_file_info",
"ensure_directory",
"get_unique_filename",
"format_file_size",
"clean_temp_files"
]

195
python_core/utils/helpers.py Normal file
View File

@@ -0,0 +1,195 @@
"""
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

48
python_core/utils/logger.py Normal file
View File

@@ -0,0 +1,48 @@
"""
Logging utilities for MixVideo V2
"""
import sys
from pathlib import Path
from loguru import logger
import sys
import os
sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
from config import settings
def setup_logger(name: str = None):
"""
Set up logger with appropriate configuration.
Args:
name: Logger name (optional)
Returns:
Configured logger instance
"""
# Remove default handler
logger.remove()
# Console handler
logger.add(
sys.stderr,
level=settings.log_level,
format="<green>{time:YYYY-MM-DD HH:mm:ss}</green> | <level>{level: <8}</level> | <cyan>{name}</cyan>:<cyan>{function}</cyan>:<cyan>{line}</cyan> - <level>{message}</level>",
colorize=True
)
# File handler
log_file = settings.temp_dir / settings.log_file
logger.add(
log_file,
level=settings.log_level,
format="{time:YYYY-MM-DD HH:mm:ss} | {level: <8} | {name}:{function}:{line} - {message}",
rotation="10 MB",
retention="7 days",
compression="zip"
)
return logger

117
python_core/utils/validators.py Normal file
View File

@@ -0,0 +1,117 @@
"""
File validation utilities for MixVideo V2
"""
from pathlib import Path
from typing import Union
import sys
import os
sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
from config import settings
def validate_video_file(file_path: Union[str, Path]) -> bool:
"""
Validate if file is a supported video format.
Args:
file_path: Path to video file
Returns:
True if valid video file, False otherwise
"""
try:
path = Path(file_path)
# Check if file exists
if not path.exists():
return False
# Check if it's a file (not directory)
if not path.is_file():
return False
# Check file extension
if path.suffix.lower() not in settings.supported_video_formats:
return False
# Check file size (not empty)
if path.stat().st_size == 0:
return False
return True
except Exception:
return False
def validate_audio_file(file_path: Union[str, Path]) -> bool:
"""
Validate if file is a supported audio format.
Args:
file_path: Path to audio file
Returns:
True if valid audio file, False otherwise
"""
try:
path = Path(file_path)
# Check if file exists
if not path.exists():
return False
# Check if it's a file (not directory)
if not path.is_file():
return False
# Check file extension
if path.suffix.lower() not in settings.supported_audio_formats:
return False
# Check file size (not empty)
if path.stat().st_size == 0:
return False
return True
except Exception:
return False
def validate_image_file(file_path: Union[str, Path]) -> bool:
"""
Validate if file is a supported image format.
Args:
file_path: Path to image file
Returns:
True if valid image file, False otherwise
"""
try:
path = Path(file_path)
# Check if file exists
if not path.exists():
return False
# Check if it's a file (not directory)
if not path.is_file():
return False
# Check file extension
if path.suffix.lower() not in settings.supported_image_formats:
return False
# Check file size (not empty)
if path.stat().st_size == 0:
return False
return True
except Exception:
return False