124 lines
3.4 KiB
Python
124 lines
3.4 KiB
Python
# @File : ffwrapper.py
|
|
# @Time : 2026/04/21 17:34:20
|
|
# @Author : SilverAg.L
|
|
|
|
import subprocess
|
|
from pathlib import Path
|
|
|
|
from logger import get_logger
|
|
from progress import get_progress_bar, is_tty
|
|
from utils import shorten_name
|
|
|
|
logger = get_logger('cmps_hevc_crf18')
|
|
|
|
|
|
def check_ffmpeg():
|
|
try:
|
|
subprocess.run(['ffmpeg', '-version'],
|
|
capture_output=True, check=True)
|
|
subprocess.run(['ffprobe', '-version'],
|
|
capture_output=True, check=True)
|
|
except (subprocess.CalledProcessError, FileNotFoundError):
|
|
logger.error("ffmpeg not found or not working. Please install ffmpeg.")
|
|
raise SystemExit(1)
|
|
|
|
|
|
def get_duration(file_path: Path):
|
|
cmd = [
|
|
'ffprobe', '-v', 'error', '-show_entries', 'format=duration',
|
|
'-of', 'default=noprint_wrappers=1:nokey=1', str(file_path)
|
|
]
|
|
result = subprocess.run(
|
|
cmd, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True)
|
|
try:
|
|
dur = float(result.stdout.strip())
|
|
if dur <= 0:
|
|
raise ValueError(f"Invalid duration: {dur}")
|
|
return dur
|
|
except (ValueError, TypeError) as e:
|
|
logger.error(f"Cannot determine duration of {file_path}: {e}")
|
|
raise
|
|
|
|
|
|
COMPRESS_OPTIONS = [
|
|
"-c:v", "libx265", "-crf", "18", "-preset", "medium",
|
|
"-c:a", "copy",
|
|
"-tag:v", "hvc1",
|
|
]
|
|
|
|
|
|
def start_ffmpeg(infile: Path, outfile: Path):
|
|
cmd = [
|
|
"ffmpeg", "-y", "-i", shorten_name(infile.name),
|
|
"-progress", "pipe:1", "-nostats", "-loglevel", "error",
|
|
*COMPRESS_OPTIONS, shorten_name(outfile.name, flag=2)
|
|
]
|
|
logger.debug(f"Command: {' '.join(cmd)}")
|
|
cmd[3] = str(infile)
|
|
cmd[-1] = str(outfile)
|
|
return subprocess.Popen(
|
|
cmd,
|
|
stdout=subprocess.PIPE,
|
|
stderr=subprocess.STDOUT,
|
|
bufsize=1,
|
|
text=True,
|
|
encoding='utf-8')
|
|
|
|
|
|
def set_progress(line: str, dur, pbar):
|
|
if not line.startswith("out_time_ms="):
|
|
return
|
|
try:
|
|
us = int(line.split('=')[1])
|
|
if us < 0:
|
|
return
|
|
current_sec = us / 1_000_000
|
|
pbar.n = min(current_sec, dur)
|
|
pbar.refresh()
|
|
except (ValueError, IndexError, OverflowError):
|
|
return
|
|
|
|
|
|
def clean_on_failure(proc: subprocess.Popen, outfile: Path):
|
|
if proc.poll() is None:
|
|
proc.kill()
|
|
proc.wait()
|
|
if outfile.exists():
|
|
try:
|
|
outfile.unlink()
|
|
except OSError:
|
|
pass
|
|
|
|
|
|
def hevc_encode(infile: Path, outfile: Path, progress: bool = True):
|
|
dur = get_duration(infile)
|
|
logger.debug(f"Duration: {dur:.2f} seconds")
|
|
|
|
proc = start_ffmpeg(infile, outfile)
|
|
pbar = get_progress_bar(
|
|
total=dur, unit='s',
|
|
bar_format='{l_bar}{bar}| {n:.2f}/{total:.2f}({unit})'
|
|
) if progress and is_tty() else None
|
|
|
|
try:
|
|
for line in iter(proc.stdout.readline, ''):
|
|
set_progress(line.strip(), dur, pbar)
|
|
if proc.poll() is not None:
|
|
break
|
|
|
|
proc.wait()
|
|
if proc.returncode != 0:
|
|
raise subprocess.CalledProcessError(proc.returncode, "ffmpeg")
|
|
except (
|
|
KeyboardInterrupt,
|
|
subprocess.CalledProcessError
|
|
):
|
|
# Issue: cleaning may also be interrupted by CtrlC
|
|
# (especially in Windows)
|
|
clean_on_failure(proc, outfile)
|
|
raise
|
|
finally:
|
|
if pbar:
|
|
pbar.close()
|
|
proc.stdout.close()
|