254 lines
9.9 KiB
Python
Executable File
254 lines
9.9 KiB
Python
Executable File
#!/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()
|