adds vcmp
This commit is contained in:
@@ -0,0 +1,253 @@
|
||||
#!/usr/bin/env python3
|
||||
import os
|
||||
import pathlib
|
||||
import subprocess
|
||||
import shutil
|
||||
import sys
|
||||
import argparse
|
||||
import tempfile
|
||||
import logging
|
||||
import time
|
||||
|
||||
video_extensions: list[str] = ["mp4", "mkv", "avi", "wmv", "mov", "m4v", "ts", "flv", "mpg"]
|
||||
|
||||
total_files: int = 0
|
||||
|
||||
def setup_logging(verbose):
|
||||
"""Configure logging to both file and stdout."""
|
||||
log_level = logging.DEBUG if verbose else logging.INFO
|
||||
|
||||
# Use local log directory
|
||||
xdg_state_home = os.environ.get('XDG_STATE_HOME', str(pathlib.Path.home() / '.local' / 'state'))
|
||||
log_dir = pathlib.Path(xdg_state_home) / 'vcmp'
|
||||
log_dir.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
log_file = log_dir / 'vcmp.log'
|
||||
|
||||
# Create console handler
|
||||
console_handler = logging.StreamHandler(sys.stdout)
|
||||
console_handler.setLevel(log_level)
|
||||
console_handler.setFormatter(logging.Formatter('%(asctime)s - %(levelname)s - %(message)s'))
|
||||
|
||||
# Create file handler
|
||||
file_handler = logging.FileHandler(log_file)
|
||||
file_handler.setLevel(logging.DEBUG) # Always log everything to file
|
||||
file_handler.setFormatter(logging.Formatter('%(asctime)s - %(levelname)s - %(message)s'))
|
||||
|
||||
# Set up root logger
|
||||
root_logger = logging.getLogger()
|
||||
root_logger.setLevel(logging.DEBUG)
|
||||
root_logger.addHandler(console_handler)
|
||||
root_logger.addHandler(file_handler)
|
||||
|
||||
def ffmpeg_command(input, output, nice=20):
|
||||
return [
|
||||
"nice", "-n", str(nice),
|
||||
"ffmpeg",
|
||||
"-progress", "pipe:1", # Send progress info to stdout
|
||||
"-i", str(input),
|
||||
"-vf", "fps='if(gte(source_fps,100),source_fps/4,if(gte(source_fps,50),source_fps/2,source_fps))',scale=-2:'min(1080,ih)'",
|
||||
"-c:v", "libsvtav1",
|
||||
"-crf", "40",
|
||||
"-c:a", "aac",
|
||||
"-b:a", "64k",
|
||||
str(output)
|
||||
]
|
||||
|
||||
def clean_directory(directory_path):
|
||||
"""Yeet all files and directories from a directory."""
|
||||
for item in directory_path.iterdir():
|
||||
if item.is_dir():
|
||||
shutil.rmtree(item) # Yeet the directory
|
||||
else:
|
||||
item.unlink() # Yeet the file
|
||||
|
||||
def is_video(path):
|
||||
return path.suffix.lower()[1:] in video_extensions
|
||||
|
||||
def get_video_duration(input_file):
|
||||
cmd = [
|
||||
"ffprobe",
|
||||
"-v", "error",
|
||||
"-show_entries", "format=duration",
|
||||
"-of", "default=noprint_wrappers=1:nokey=1",
|
||||
str(input_file)
|
||||
]
|
||||
result = subprocess.run(cmd, capture_output=True, text=True)
|
||||
return float(result.stdout)
|
||||
|
||||
def run_progress(process, duration, file_index=0):
|
||||
current_time = None
|
||||
last_update = 0
|
||||
progress = 0
|
||||
UPDATE_INTERVAL = 1
|
||||
start_time = time.time()
|
||||
while True:
|
||||
line = process.stdout.readline() if process.stdout else None
|
||||
|
||||
# Check if we've reached the end of output (no more lines to read)
|
||||
# and if the process has finished (poll() returns exit code instead of None)
|
||||
if not line and process.poll() is not None:
|
||||
break
|
||||
|
||||
if line and line.startswith('out_time_ms='):
|
||||
try:
|
||||
time_str = line.split('=')[1].strip()
|
||||
current_time = float(time_str) / 1000000
|
||||
current_time = min(current_time, duration)
|
||||
progress = (current_time / duration) * 100
|
||||
elapsed = time.time() - start_time
|
||||
speed = current_time / elapsed if elapsed > 0 else 0
|
||||
if time.time() - last_update > UPDATE_INTERVAL:
|
||||
total_progress = (file_index / total_files) * 100
|
||||
sys.stdout.write(
|
||||
f"\rProcessing ({file_index}/{total_files} [{total_progress:.1f}%]) :"
|
||||
f"{progress:5.1f}% "
|
||||
f"({current_time:.1f} / {duration:.1f} seconds) "
|
||||
f"[{speed:.2f}x]"
|
||||
f"{'.' * int(progress / 5)}{' ' * (20 - int(progress / 5))}"
|
||||
)
|
||||
sys.stdout.flush()
|
||||
last_update = time.time()
|
||||
except:
|
||||
pass
|
||||
|
||||
def run(filepath, cmd, temp_path, file_index):
|
||||
duration = get_video_duration(filepath)
|
||||
xdg_state_home = os.environ.get('XDG_STATE_HOME', str(pathlib.Path.home() / '.local' / 'state'))
|
||||
ffmpeg_log_path = pathlib.Path(xdg_state_home) / 'vcmp' / 'ffmpeg.log'
|
||||
ffmpeg_log_path.parent.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
with open(ffmpeg_log_path, 'a') as log_file:
|
||||
process = subprocess.Popen(
|
||||
cmd,
|
||||
stdout=subprocess.PIPE,
|
||||
stderr=log_file,
|
||||
universal_newlines=True
|
||||
)
|
||||
run_progress(process, duration, file_index)
|
||||
sys.stdout.write("\n")
|
||||
sys.stdout.flush()
|
||||
|
||||
if process.returncode != 0:
|
||||
logging.error(f"FFmpeg exited with code {process.returncode}")
|
||||
if temp_path.exists():
|
||||
temp_path.unlink()
|
||||
sys.exit(1)
|
||||
|
||||
def process_single_file(filepath, target_path, temp_dir, remove_source, nice, file_index):
|
||||
temp_path = temp_dir / filepath.with_suffix(".mp4").name
|
||||
clean_directory(temp_dir) # Should be empty if mktemp was used, but clear it regardless.
|
||||
logging.info(f"Processing {filepath} to {target_path}")
|
||||
logging.debug(f"Using temporary file: {temp_path}")
|
||||
|
||||
cmd = ffmpeg_command(filepath, temp_path, nice)
|
||||
logging.debug(f"Executing: {' '.join(cmd)}")
|
||||
try:
|
||||
run(filepath, cmd, temp_path, file_index)
|
||||
except KeyboardInterrupt:
|
||||
logging.warning("Process interrupted by user")
|
||||
if temp_path.exists():
|
||||
temp_path.unlink()
|
||||
sys.exit(1)
|
||||
except subprocess.CalledProcessError as e:
|
||||
logging.error(f"FFmpeg error processing {filepath}: {e}")
|
||||
if temp_path.exists():
|
||||
temp_path.unlink()
|
||||
sys.exit(1)
|
||||
else:
|
||||
# File succesfully converted to av1, move to destination
|
||||
# Use shutil move because of inter-fs issues
|
||||
target_path.parent.mkdir(parents=True, exist_ok=True)
|
||||
shutil.move(str(temp_path), str(target_path))
|
||||
logging.info(f"Successfully compressed {filepath}")
|
||||
if remove_source:
|
||||
logging.info(f"Removing source file {filepath}")
|
||||
filepath.unlink()
|
||||
|
||||
def process_directory(*, source_dir, destination_dir, temp_dir=None, remove_source=False, nice=20, count_only=False):
|
||||
"""Process all video files in source directory recursively."""
|
||||
if not count_only:
|
||||
if not temp_dir:
|
||||
raise ValueError("Temporary directory is required for processing files")
|
||||
count = 0
|
||||
for dirpath, _dirnames, filenames in os.walk(source_dir):
|
||||
for filename in filenames:
|
||||
filepath = pathlib.Path(dirpath) / filename
|
||||
|
||||
# File might have been deleted since starting script
|
||||
if not filepath.exists() or not is_video(filepath):
|
||||
logging.debug(f"Skipping {filepath}: not a video file or doesn't exist")
|
||||
continue
|
||||
if ".cmp." in filepath.name:
|
||||
logging.debug(f"Skipping {filepath}: already compressed")
|
||||
continue
|
||||
|
||||
# Determine destination path, remove source if already exists and requested
|
||||
target_path = destination_dir / filepath.relative_to(source_dir)
|
||||
target_path = target_path.with_suffix(".cmp" + target_path.suffix)
|
||||
if target_path.exists():
|
||||
if remove_source:
|
||||
logging.info(f"Removing source file {filepath} as compressed version exists")
|
||||
filepath.unlink()
|
||||
logging.debug(f"Skipping {filepath}: target already exists")
|
||||
continue
|
||||
|
||||
count += 1
|
||||
if count_only:
|
||||
continue
|
||||
|
||||
process_single_file(filepath, target_path, temp_dir, remove_source, nice, count)
|
||||
|
||||
logging.info(f"Found {count} files to process")
|
||||
return count
|
||||
|
||||
|
||||
def main():
|
||||
parser = argparse.ArgumentParser(
|
||||
description="""
|
||||
Video compression utility that converts videos to AV1 format with reduced quality/size.
|
||||
Converts to 1080p max resolution, reduces high framerates, and uses low bitrate audio.
|
||||
The compressed output files will have '.cmp' added to the filename.
|
||||
"""
|
||||
)
|
||||
parser.add_argument('-s', '--source-dir', type=pathlib.Path, default=pathlib.Path.cwd(),
|
||||
help='Source directory containing videos to compress (default: current directory)')
|
||||
parser.add_argument('-d', '--destination-dir', type=pathlib.Path,
|
||||
help='Destination directory for compressed videos (default: same as source)')
|
||||
parser.add_argument('-t', '--temp-dir', type=pathlib.Path, default=pathlib.Path(tempfile.mkdtemp()),
|
||||
help='Temporary directory for processing (default: system temp directory)')
|
||||
parser.add_argument('-r', '--remove-source', action='store_true',
|
||||
help='Remove source files after successful compression')
|
||||
parser.add_argument('-v', '--verbose', action='store_true',
|
||||
help='Enable verbose logging output')
|
||||
parser.add_argument('-n', '--nice', type=lambda x: int(x) if 0 <= int(x) <= 39 else exec('raise ValueError("Nice value must be between 0 and 39")'),
|
||||
default=20,
|
||||
help='Nice value for process priority (default: 20). Valid values are 0 (highest priority) to 39 (lowest priority). Lower values use more system resources.')
|
||||
args = parser.parse_args()
|
||||
|
||||
setup_logging(args.verbose)
|
||||
|
||||
source_dir = args.source_dir
|
||||
destination_dir = args.destination_dir if args.destination_dir else args.source_dir
|
||||
temp_dir = args.temp_dir
|
||||
|
||||
logging.info(f"Starting video compression from {source_dir} to {destination_dir}")
|
||||
|
||||
global total_files
|
||||
total_files = process_directory(
|
||||
source_dir=source_dir,
|
||||
destination_dir=destination_dir,
|
||||
count_only=True
|
||||
)
|
||||
|
||||
process_directory(
|
||||
source_dir=source_dir,
|
||||
destination_dir=destination_dir,
|
||||
temp_dir=temp_dir,
|
||||
remove_source=args.remove_source,
|
||||
nice=args.nice
|
||||
)
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Reference in New Issue
Block a user