Uyanide/transcode
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor: en

Browse Source
This commit is contained in:
Uyanide
2026-03-18 23:13:09 +01:00
parent 13f98a538d
commit 35dccefc38
3 changed files with 66 additions and 66 deletions
Download Patch File Download Diff File Expand all files Collapse all files
core.py
+17 -17
View File
@@ -11,7 +11,7 @@ class BaseTranscoder:
def __init__(self, args, video_defaults, audio_defaults): def __init__(self, args, video_defaults, audio_defaults):
self.input_path = Path(args.input).resolve() self.input_path = Path(args.input).resolve()
if not self.input_path.exists(): if not self.input_path.exists():
raise FileNotFoundError(f"输入文件不存在: {self.input_path}") raise FileNotFoundError(f"Input file not found: {self.input_path}")
self.vencoder = args.vencoder self.vencoder = args.vencoder
self.aencoder = args.aencoder self.aencoder = args.aencoder
@@ -95,7 +95,7 @@ class BaseTranscoder:
fps = self.orig_info.get("fps_str", "0/1") if self.orig_info else "0/1" fps = self.orig_info.get("fps_str", "0/1") if self.orig_info else "0/1"
print( print(
f"\n[Metrics] 启动 VMAF/PSNR 测试 (采样间隔: {n_subsample}, 线程数: {threads}, 强制帧率: {fps})..." f"\n[Metrics] Starting VMAF/PSNR test (Subsample interval: {n_subsample}, Threads: {threads}, Forced fps: {fps})..."
) )
out_w = new_info.get("width") if new_info else None out_w = new_info.get("width") if new_info else None
@@ -105,7 +105,7 @@ class BaseTranscoder:
orig_fps = self.orig_info.get("fps_str", "0/1") if self.orig_info else "0/1" orig_fps = self.orig_info.get("fps_str", "0/1") if self.orig_info else "0/1"
out_fps = new_info.get("fps_str", "0/1") if new_info else "0/1" out_fps = new_info.get("fps_str", "0/1") if new_info else "0/1"
# 使用滤镜链对齐帧率、分辨率和首帧时间戳 # Use filter chain to align framerate, resolution, and first frame PTS
filters_0 = [] filters_0 = []
if out_fps != orig_fps: if out_fps != orig_fps:
filters_0.append(f"fps={orig_fps}") filters_0.append(f"fps={orig_fps}")
@@ -151,7 +151,7 @@ class BaseTranscoder:
) )
return vmaf, psnr return vmaf, psnr
except Exception as e: except Exception as e:
print(f"[Metrics Error] 无法解析 VMAF 日志: {e}") print(f"[Metrics Error] Failed to parse VMAF log: {e}")
return "N/A", "N/A" return "N/A", "N/A"
def _generate_log_content(self, encode_time, new_info, vmaf, psnr, extra_info=""): def _generate_log_content(self, encode_time, new_info, vmaf, psnr, extra_info=""):
@@ -159,18 +159,18 @@ class BaseTranscoder:
self.output_path.stat().st_size / self.input_path.stat().st_size self.output_path.stat().st_size / self.input_path.stat().st_size
) * 100 ) * 100
content = ( content = (
f"时间: {time.strftime('%Y-%m-%d %H:%M:%S')}\n" f"Time: {time.strftime('%Y-%m-%d %H:%M:%S')}\n"
f"输入: {self.input_path}\n" f"Input: {self.input_path}\n"
f"输出: {self.output_path}\n" f"Output: {self.output_path}\n"
f"{extra_info}" f"{extra_info}"
f"视频编码器: {self.vencoder} ({' '.join(self.v_params)})\n" f"Video encoder: {self.vencoder} ({' '.join(self.v_params)})\n"
f"音频编码器: {self.aencoder} ({' '.join(self.a_params)})\n" f"Audio encoder: {self.aencoder} ({' '.join(self.a_params)})\n"
f"耗时: {encode_time:.2f} s\n" f"Time elapsed: {encode_time:.2f} s\n"
f"[元数据] {self.orig_info.get('codec','') if self.orig_info else ''} -> {new_info.get('codec','') if new_info else ''} | {self.orig_info.get('bitrate_kbps',0) if self.orig_info else 0:.0f} kbps -> {new_info.get('bitrate_kbps',0) if new_info else 0:.0f} kbps\n" f"[Metadata] {self.orig_info.get('codec','') if self.orig_info else ''} -> {new_info.get('codec','') if new_info else ''} | {self.orig_info.get('bitrate_kbps',0) if self.orig_info else 0:.0f} kbps -> {new_info.get('bitrate_kbps',0) if new_info else 0:.0f} kbps\n"
f"[体积比] {size_ratio:.1f}%\n" f"[Size ratio] {size_ratio:.1f}%\n"
) )
if self.run_metrics: if self.run_metrics:
content += f"[质量] VMAF: {vmaf} | PSNR(Y): {psnr}\n" content += f"[Quality] VMAF: {vmaf} | PSNR(Y): {psnr}\n"
content += "-" * 50 + "\n" content += "-" * 50 + "\n"
return content return content
@@ -180,7 +180,7 @@ class BaseTranscoder:
if self.notify_mode == "none": if self.notify_mode == "none":
return return
subject = f"{extra_title}任务{'失败' if error else '完成'}: {self.input_path.name}" subject = f"{extra_title}Task {'failed' if error else 'completed'}: {self.input_path.name}"
if self.notify_mode == "mail": if self.notify_mode == "mail":
body = error if error else full_text body = error if error else full_text
@@ -191,19 +191,19 @@ class BaseTranscoder:
check=True, check=True,
) )
except Exception as e: except Exception as e:
print(f"[Notify Error] 邮件发送失败: {e}") print(f"[Notify Error] Failed to send email: {e}")
elif self.notify_mode == "notify": elif self.notify_mode == "notify":
if error: if error:
body = error body = error
else: else:
body = f"编码器: {self.vencoder}\n耗时: {encode_time:.1f}s" body = f"Encoder: {self.vencoder}\nTime elapsed: {encode_time:.1f}s"
if self.run_metrics: if self.run_metrics:
body += f"\nVMAF: {vmaf}" body += f"\nVMAF: {vmaf}"
try: try:
subprocess.run(["notify-send", subject, body], check=True) subprocess.run(["notify-send", subject, body], check=True)
except Exception as e: except Exception as e:
print(f"[Notify Error] DBus 通知发送失败: {e}") print(f"[Notify Error] DBus Failed to send notification: {e}")
def _cleanup(self): def _cleanup(self):
for temp_file in self.temp_files: for temp_file in self.temp_files:
tc-2pass.py
+26 -26
View File
@@ -26,10 +26,10 @@ class MediaTranscoder2Pass(BaseTranscoder):
self.passlog_prefix = self.out_dir / f"passlog_{self.basename}_{self.vencoder}_{int(time.time())}" self.passlog_prefix = self.out_dir / f"passlog_{self.basename}_{self.vencoder}_{int(time.time())}"
def execute(self): def execute(self):
print(f"--- 初始化 2-Pass 转码任务 ---") print(f"--- Initializing 2-Pass Transcode Task ---")
self.orig_info = self.get_media_info(self.input_path) self.orig_info = self.get_media_info(self.input_path)
if not self.orig_info: if not self.orig_info:
raise ValueError("无法读取输入文件元数据。") raise ValueError("Failed to read input file metadata.")
start_time = time.time() start_time = time.time()
try: try:
@@ -40,8 +40,8 @@ class MediaTranscoder2Pass(BaseTranscoder):
self._run_pass_2_standard() self._run_pass_2_standard()
except subprocess.CalledProcessError as e: except subprocess.CalledProcessError as e:
self._cleanup() self._cleanup()
self._send_notification(error=f"转码失败，退出码: {e.returncode}", extra_title="2-Pass转码") self._send_notification(error=f"Transcode failed, exit code: {e.returncode}", extra_title="2-Pass Transcode ")
sys.exit(f"\n转码失败，退出码: {e.returncode}") sys.exit(f"\nTranscode failed, exit code: {e.returncode}")
self._cleanup() self._cleanup()
encode_time = time.time() - start_time encode_time = time.time() - start_time
@@ -53,17 +53,17 @@ class MediaTranscoder2Pass(BaseTranscoder):
vmaf_score, psnr_score = self._run_vmaf_psnr(new_info) vmaf_score, psnr_score = self._run_vmaf_psnr(new_info)
log_content = self._generate_log_content( log_content = self._generate_log_content(
encode_time, new_info, vmaf_score, psnr_score, extra_info=f"设定码率: {self.target_bitrate}\n" encode_time, new_info, vmaf_score, psnr_score, extra_info=f"Target bitrate: {self.target_bitrate}\n"
) )
with open(self.log_path, "a", encoding="utf-8") as f: with open(self.log_path, "a", encoding="utf-8") as f:
f.write(log_content) f.write(log_content)
self._send_notification(encode_time=encode_time, new_info=new_info, vmaf=vmaf_score, self._send_notification(encode_time=encode_time, new_info=new_info, vmaf=vmaf_score,
full_text=log_content, extra_title="2-Pass转码") full_text=log_content, extra_title="2-Pass Transcode ")
print(f"\n转码完成！耗时: {encode_time:.2f} 秒") print(f"\nTranscode completed! Time elapsed: {encode_time:.2f} s")
print(f"输出文件: {self.output_path}") print(f"Output file: {self.output_path}")
def _get_base_video_cmd(self, pass_num): def _get_base_video_cmd(self, pass_num):
cmd = ["-c:v", self.vencoder] cmd = ["-c:v", self.vencoder]
@@ -78,12 +78,12 @@ class MediaTranscoder2Pass(BaseTranscoder):
return cmd return cmd
def _run_pass_1(self): def _run_pass_1(self):
print("\n[Pass 1] 正在进行第一阶段视频分析...") print("\n[Pass 1] Running first pass video analysis...")
cmd = ["ffmpeg", "-y", "-i", str(self.input_path), *self._get_base_video_cmd(1), "-an", "-f", "null", "-"] cmd = ["ffmpeg", "-y", "-i", str(self.input_path), *self._get_base_video_cmd(1), "-an", "-f", "null", "-"]
subprocess.run(cmd, check=True) subprocess.run(cmd, check=True)
def _run_pass_2_external_audio(self): def _run_pass_2_external_audio(self):
print("\n[Pass 2 - Audio] 提取音频流并使用外部 fdkaac 编码...") print("\n[Pass 2 - Audio] Extracting audio stream and encoding with external fdkaac...")
temp_wav = self.out_dir / f"temp_{self.input_path.stem}.wav" temp_wav = self.out_dir / f"temp_{self.input_path.stem}.wav"
temp_m4a = self.out_dir / f"temp_{self.input_path.stem}.m4a" temp_m4a = self.out_dir / f"temp_{self.input_path.stem}.m4a"
self.temp_files.extend([temp_wav, temp_m4a]) self.temp_files.extend([temp_wav, temp_m4a])
@@ -91,7 +91,7 @@ class MediaTranscoder2Pass(BaseTranscoder):
subprocess.run(["ffmpeg", "-y", "-i", str(self.input_path), "-vn", "-c:a", "pcm_s16le", str(temp_wav)], check=True) subprocess.run(["ffmpeg", "-y", "-i", str(self.input_path), "-vn", "-c:a", "pcm_s16le", str(temp_wav)], check=True)
subprocess.run(["fdkaac"] + self.a_params + [str(temp_wav), "-o", str(temp_m4a)], check=True) subprocess.run(["fdkaac"] + self.a_params + [str(temp_wav), "-o", str(temp_m4a)], check=True)
print("\n[Pass 2 - Video] 正在进行第二阶段最终编码并混流...") print("\n[Pass 2 - Video] Running second pass final encode and muxing...")
cmd = [ cmd = [
"ffmpeg", "-y", "-i", str(self.input_path), "-i", str(temp_m4a), "ffmpeg", "-y", "-i", str(self.input_path), "-i", str(temp_m4a),
"-map", "0:v:0", "-map", "1:a:0", "-map", "0:v:0", "-map", "1:a:0",
@@ -100,7 +100,7 @@ class MediaTranscoder2Pass(BaseTranscoder):
subprocess.run(cmd, check=True) subprocess.run(cmd, check=True)
def _run_pass_2_standard(self): def _run_pass_2_standard(self):
print("\n[Pass 2] 正在进行第二阶段最终编码 (包含音频)...") print("\n[Pass 2] Running second pass final encode (including audio)...")
cmd = ["ffmpeg", "-y", "-i", str(self.input_path), *self._get_base_video_cmd(2), *self.a_params, str(self.output_path)] cmd = ["ffmpeg", "-y", "-i", str(self.input_path), *self._get_base_video_cmd(2), *self.a_params, str(self.output_path)]
subprocess.run(cmd, check=True) subprocess.run(cmd, check=True)
@@ -110,28 +110,28 @@ class MediaTranscoder2Pass(BaseTranscoder):
try: try:
log_file.unlink() log_file.unlink()
except Exception as e: except Exception as e:
print(f"[Cleanup Warning] 无法删除临时日志 {log_file.name}: {e}") print(f"[Cleanup Warning] Failed to delete temporary log {log_file.name}: {e}")
def main(): def main():
parser = argparse.ArgumentParser(description="2-Pass 可配置视频转码脚本") parser = argparse.ArgumentParser(description="2-Pass Configurable video transcoding script")
parser.add_argument("-i", "--input", required=True, help="输入视频文件") parser.add_argument("-i", "--input", required=True, help="Input video file")
parser.add_argument("-b", "--bitrate", required=True, help="目标视频码率 (例如: 8M, 3000K)") parser.add_argument("-b", "--bitrate", required=True, help="Target video bitrate (e.g., 8M, 3000K)")
parser.add_argument("-d", "--outdir", help="指定输出目录 (默认与输入文件同目录)") parser.add_argument("-d", "--outdir", help="Output directory (default: same as input)")
parser.add_argument("-cv", "--vencoder", choices=VIDEO_DEFAULTS.keys(), default="libx265", help="视频编码器 (默认 libx265)") parser.add_argument("-cv", "--vencoder", choices=VIDEO_DEFAULTS.keys(), default="libx265", help="Video encoder (default: libx265)")
parser.add_argument("-ca", "--aencoder", choices=AUDIO_DEFAULTS.keys(), default="opus", help="音频编码器 (默认 opus)") parser.add_argument("-ca", "--aencoder", choices=AUDIO_DEFAULTS.keys(), default="opus", help="Audio encoder (default: opus)")
parser.add_argument("--vargs", type=str, default="", help="覆盖默认视频编码参数") parser.add_argument("--vargs", type=str, default="", help="Override default video encoding parameters")
parser.add_argument("--aargs", type=str, default="", help="覆盖默认音频编码参数") parser.add_argument("--aargs", type=str, default="", help="Override default audio encoding parameters")
parser.add_argument("--fps", type=str, default="", help="指定帧率 (例如 60, 30000/1001)") parser.add_argument("--fps", type=str, default="", help="Specify framerate (e.g., 60, 30000/1001)")
parser.add_argument("--res", type=str, default="", help="指定分辨率，格式与scale滤镜相同 (例如 1920:1080)") parser.add_argument("--res", type=str, default="", help="Specify resolution, same format as scale filter (e.g., 1920:1080)")
parser.add_argument("--metrics", action=argparse.BooleanOptionalAction, default=True, help="结束后运行 VMAF/PSNR") parser.add_argument("--metrics", action=argparse.BooleanOptionalAction, default=True, help="Run VMAF/PSNR after completion")
parser.add_argument("--notify", choices=["none", "mail", "notify"], default="mail", help="完成后的通知方式 (默认 mail)") parser.add_argument("--notify", choices=["none", "mail", "notify"], default="mail", help="Notification method after completion (default: mail)")
try: try:
transcoder = MediaTranscoder2Pass(parser.parse_args()) transcoder = MediaTranscoder2Pass(parser.parse_args())
transcoder.execute() transcoder.execute()
except Exception as e: except Exception as e:
sys.exit(f"发生错误: {e}") sys.exit(f"Error occurred: {e}")
if __name__ == "__main__": if __name__ == "__main__":
tc.py
+23 -23
View File
@@ -7,7 +7,7 @@ from pathlib import Path
from core import BaseTranscoder from core import BaseTranscoder
# 将当前目录加入 path 以引入 core # Add current directory to path to import core
sys.path.append(str(Path(__file__).resolve().parent)) sys.path.append(str(Path(__file__).resolve().parent))
VIDEO_DEFAULTS = { VIDEO_DEFAULTS = {
@@ -29,10 +29,10 @@ class MediaTranscoder(BaseTranscoder):
super().__init__(args, VIDEO_DEFAULTS, AUDIO_DEFAULTS) super().__init__(args, VIDEO_DEFAULTS, AUDIO_DEFAULTS)
def execute(self): def execute(self):
print(f"--- 初始化转码任务 ---") print(f"--- Initializing Transcode Task ---")
self.orig_info = self.get_media_info(self.input_path) self.orig_info = self.get_media_info(self.input_path)
if not self.orig_info: if not self.orig_info:
raise ValueError("无法读取输入文件元数据。") raise ValueError("Failed to read input file metadata.")
start_time = time.time() start_time = time.time()
try: try:
@@ -42,8 +42,8 @@ class MediaTranscoder(BaseTranscoder):
self._encode_standard() self._encode_standard()
except subprocess.CalledProcessError as e: except subprocess.CalledProcessError as e:
self._cleanup() self._cleanup()
self._send_notification(error=f"转码失败，退出码: {e.returncode}", extra_title="转码") self._send_notification(error=f"Transcode failed, exit code: {e.returncode}", extra_title="Transcode ")
sys.exit(f"\n转码失败，退出码: {e.returncode}") sys.exit(f"\nTranscode failed, exit code: {e.returncode}")
self._cleanup() self._cleanup()
encode_time = time.time() - start_time encode_time = time.time() - start_time
@@ -62,13 +62,13 @@ class MediaTranscoder(BaseTranscoder):
f.write(log_content) f.write(log_content)
self._send_notification(encode_time=encode_time, new_info=new_info, self._send_notification(encode_time=encode_time, new_info=new_info,
vmaf=vmaf_score, full_text=log_content, extra_title="转码") vmaf=vmaf_score, full_text=log_content, extra_title="Transcode ")
print(f"\n转码完成！耗时: {encode_time:.2f} 秒") print(f"\nTranscode completed! Time elapsed: {encode_time:.2f} s")
print(f"输出文件: {self.output_path}") print(f"Output file: {self.output_path}")
def _encode_with_external_audio(self): def _encode_with_external_audio(self):
print("\n[Audio] 提取音频流并使用外部 fdkaac 编码...") print("\n[Audio] Extracting audio stream and encoding with external fdkaac...")
temp_wav = self.out_dir / f"temp_{self.input_path.stem}.wav" temp_wav = self.out_dir / f"temp_{self.input_path.stem}.wav"
temp_m4a = self.out_dir / f"temp_{self.input_path.stem}.m4a" temp_m4a = self.out_dir / f"temp_{self.input_path.stem}.m4a"
self.temp_files.extend([temp_wav, temp_m4a]) self.temp_files.extend([temp_wav, temp_m4a])
@@ -76,7 +76,7 @@ class MediaTranscoder(BaseTranscoder):
subprocess.run(["ffmpeg", "-y", "-i", str(self.input_path), "-vn", "-c:a", "pcm_s16le", str(temp_wav)], check=True) subprocess.run(["ffmpeg", "-y", "-i", str(self.input_path), "-vn", "-c:a", "pcm_s16le", str(temp_wav)], check=True)
subprocess.run(["fdkaac"] + self.a_params + [str(temp_wav), "-o", str(temp_m4a)], check=True) subprocess.run(["fdkaac"] + self.a_params + [str(temp_wav), "-o", str(temp_m4a)], check=True)
print("\n[Video] 编码视频并混流...") print("\n[Video] Encoding video and muxing...")
cmd = [ cmd = [
"ffmpeg", "-y", "-i", str(self.input_path), "-i", str(temp_m4a), "ffmpeg", "-y", "-i", str(self.input_path), "-i", str(temp_m4a),
"-map", "0:v:0", "-map", "1:a:0", "-c:v", self.vencoder, "-map", "0:v:0", "-map", "1:a:0", "-c:v", self.vencoder,
@@ -88,7 +88,7 @@ class MediaTranscoder(BaseTranscoder):
subprocess.run(cmd, check=True) subprocess.run(cmd, check=True)
def _encode_standard(self): def _encode_standard(self):
print("\n[Transcode] 使用 ffmpeg 同时处理视音频...") print("\n[Transcode] Processing video and audio simultaneously with ffmpeg...")
cmd = ["ffmpeg", "-y", "-i", str(self.input_path), "-c:v", self.vencoder] cmd = ["ffmpeg", "-y", "-i", str(self.input_path), "-c:v", self.vencoder]
if self.vf_string: if self.vf_string:
cmd.extend(["-vf", self.vf_string]) cmd.extend(["-vf", self.vf_string])
@@ -99,23 +99,23 @@ class MediaTranscoder(BaseTranscoder):
def main(): def main():
parser = argparse.ArgumentParser(description="可配置视频转码脚本") parser = argparse.ArgumentParser(description="Configurable video transcoding script")
parser.add_argument("-i", "--input", required=True, help="输入视频文件") parser.add_argument("-i", "--input", required=True, help="Input video file")
parser.add_argument("-d", "--outdir", help="指定输出目录 (默认与输入文件同目录)") parser.add_argument("-d", "--outdir", help="Output directory (default: same as input)")
parser.add_argument("-cv", "--vencoder", choices=VIDEO_DEFAULTS.keys(), default="libsvtav1", help="视频编码器 (默认 libsvtav1)") parser.add_argument("-cv", "--vencoder", choices=VIDEO_DEFAULTS.keys(), default="libsvtav1", help="Video encoder (default: libsvtav1)")
parser.add_argument("-ca", "--aencoder", choices=AUDIO_DEFAULTS.keys(), default="opus", help="音频编码器 (默认 opus)") parser.add_argument("-ca", "--aencoder", choices=AUDIO_DEFAULTS.keys(), default="opus", help="Audio encoder (default: opus)")
parser.add_argument("--vargs", type=str, default="", help="覆盖默认视频编码参数") parser.add_argument("--vargs", type=str, default="", help="Override default video encoding parameters")
parser.add_argument("--aargs", type=str, default="", help="覆盖默认音频编码参数") parser.add_argument("--aargs", type=str, default="", help="Override default audio encoding parameters")
parser.add_argument("--fps", type=str, default="", help="指定帧率 (例如 60, 30000/1001)") parser.add_argument("--fps", type=str, default="", help="Specify framerate (e.g., 60, 30000/1001)")
parser.add_argument("--res", type=str, default="", help="指定分辨率，格式与scale滤镜相同 (例如 1920:1080)") parser.add_argument("--res", type=str, default="", help="Specify resolution, same format as scale filter (e.g., 1920:1080)")
parser.add_argument("--metrics", action=argparse.BooleanOptionalAction, default=True, help="结束后运行 VMAF/PSNR") parser.add_argument("--metrics", action=argparse.BooleanOptionalAction, default=True, help="Run VMAF/PSNR after completion")
parser.add_argument("--notify", choices=["none", "mail", "notify"], default="mail", help="完成后的通知方式 (默认 mail)") parser.add_argument("--notify", choices=["none", "mail", "notify"], default="mail", help="Notification method after completion (default: mail)")
try: try:
transcoder = MediaTranscoder(parser.parse_args()) transcoder = MediaTranscoder(parser.parse_args())
transcoder.execute() transcoder.execute()
except Exception as e: except Exception as e:
sys.exit(f"发生错误: {e}") sys.exit(f"Error occurred: {e}")
if __name__ == "__main__": if __name__ == "__main__":