116 lines
4.2 KiB
Python
Executable File
116 lines
4.2 KiB
Python
Executable File
#!/usr/bin/python3
|
|
import argparse
|
|
from pathlib import Path
|
|
|
|
from ffwrapper import check_ffmpeg, hevc_encode
|
|
from logger import get_logger
|
|
from utils import format_size
|
|
|
|
logger = get_logger('cmps_hevc_crf18')
|
|
|
|
|
|
def build_file_list(srcdir, extensions):
|
|
exts = {ext.lower().lstrip('.') for ext in extensions}
|
|
return [
|
|
path.absolute() for path in Path(srcdir).rglob('*')
|
|
if path.is_file() and path.suffix.lower().lstrip('.') in exts
|
|
]
|
|
|
|
|
|
def make_output_path(infile: Path, outdir: Path = None, relroot: str = None):
|
|
if not outdir:
|
|
return infile.with_suffix('.hevc.mp4')
|
|
outdir = Path(outdir)
|
|
if not relroot:
|
|
return outdir / infile.with_suffix('.hevc.mp4').name
|
|
rel_path = infile.relative_to(
|
|
Path(relroot).absolute()).with_suffix('.hevc.mp4')
|
|
return outdir / rel_path
|
|
|
|
|
|
def resolve_output(infile: Path, outfile: Path, rm_original: bool = False):
|
|
final_out = outfile.with_name(
|
|
outfile.name.replace(".hevc.mp4", ".mp4"))
|
|
is_path_conflict = final_out.resolve() == infile.resolve()
|
|
if is_path_conflict:
|
|
logger.warning("Path conflict!")
|
|
logger.debug(f" original: {infile}")
|
|
logger.debug(f" conflict: {outfile.name} -> {final_out.name}")
|
|
if not rm_original:
|
|
logger.debug("Renaming original to avoid overwriting.")
|
|
infile.rename(infile.with_stem(infile.stem + '.bak'))
|
|
logger.info(f"Resolved output: {final_out}")
|
|
outfile.replace(final_out)
|
|
if rm_original and not is_path_conflict:
|
|
logger.debug(f"Remove original: {infile}")
|
|
infile.unlink()
|
|
|
|
|
|
def parse_args():
|
|
parser = argparse.ArgumentParser(
|
|
description='Encode video files to HEVC CRF 18.',
|
|
formatter_class=argparse.RawDescriptionHelpFormatter,
|
|
)
|
|
parser.add_argument('-d', '--dir', dest='srcdir',
|
|
help='Source directory')
|
|
parser.add_argument('-o', '--outdir', dest='outdir',
|
|
help='Output directory')
|
|
parser.add_argument('-f', '--ext', dest='extensions', action='append',
|
|
default=[],
|
|
help='Extensions to search for. Can be repeated.')
|
|
parser.add_argument('-r', '--rm-original', dest='remove_original',
|
|
action='store_true',
|
|
help='Remove original after encoding')
|
|
parser.add_argument('-s', '--silent', dest='silent',
|
|
action='store_true',
|
|
help='Keep silent (no progress bar)')
|
|
parser.add_argument('files', nargs='*',
|
|
help='Input files (with no -d/--dir)')
|
|
|
|
args = parser.parse_args()
|
|
if args.srcdir and args.files:
|
|
parser.error('cannot mix -d/--dir with input files')
|
|
if not args.srcdir and not args.files:
|
|
parser.error('either -d/--dir or input files must be provided')
|
|
if args.srcdir and not args.extensions:
|
|
args.extensions = ['mp4']
|
|
return args
|
|
|
|
|
|
def main():
|
|
check_ffmpeg()
|
|
args = parse_args()
|
|
|
|
if args.srcdir:
|
|
files = build_file_list(args.srcdir, args.extensions)
|
|
errmsg = (f'No files found in {args.srcdir} '
|
|
f'with extensions: {args.extensions}')
|
|
else:
|
|
files = [Path(f).absolute() for f in args.files if Path(f).is_file()]
|
|
errmsg = 'No input files provided'
|
|
if not files:
|
|
raise SystemExit(errmsg)
|
|
if len(files) > 1:
|
|
files.sort(key=lambda x: x.stat().st_size)
|
|
|
|
for idx, infile in enumerate(files, 1):
|
|
logger.info(f"[{idx}/{len(files)}] {infile.name}")
|
|
|
|
outfile = make_output_path(infile, args.outdir, args.srcdir)
|
|
outfile.parent.mkdir(parents=True, exist_ok=True)
|
|
try:
|
|
outfile.touch(0o644)
|
|
hevc_encode(infile, outfile, not args.silent)
|
|
except (OSError, PermissionError) as e:
|
|
logger.error(f"Job failed: {e}")
|
|
raise
|
|
isize, osize = infile.stat().st_size, outfile.stat().st_size
|
|
delta = (osize - isize) / 1024 ** 2
|
|
logger.info(f"Δ: {delta:+.2f} MB"
|
|
f" ({format_size(isize)} -> {format_size(osize)})")
|
|
resolve_output(infile, outfile, args.remove_original)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
main()
|