pat-e/encoding-scripts-old
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

renamed file

Browse Source
This commit is contained in:
pat-e
2025-07-16 22:25:14 +02:00
parent f133ad12bf
commit 5271fd3d7d
1 changed files with 0 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

510
anime_audio_encoder.py Normal file
View File

@@ -0,0 +1,510 @@
#!/usr/bin/env python3
import os
import sys
import subprocess
import shutil
import tempfile
import json
import re # Added for VFR frame rate parsing
from datetime import datetime
from pathlib import Path
REQUIRED_TOOLS = [
"ffmpeg", "ffprobe", "mkvmerge", "mkvpropedit",
"sox", "opusenc", "mediainfo", "av1an", "HandBrakeCLI" # Added HandBrakeCLI
]
DIR_COMPLETED = Path("completed")
DIR_ORIGINAL = Path("original")
DIR_CONV_LOGS = Path("conv_logs") # Directory for conversion logs
REMUX_CODECS = {"aac", "opus"} # Using a set for efficient lookups
def check_tools():
for tool in REQUIRED_TOOLS:
if shutil.which(tool) is None:
print(f"Required tool '{tool}' not found in PATH.")
sys.exit(1)
def run_cmd(cmd, capture_output=False, check=True):
if capture_output:
result = subprocess.run(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, check=check, text=True)
return result.stdout
else:
subprocess.run(cmd, check=check)
def convert_audio_track(index, ch, lang, audio_temp_dir, source_file, should_downmix):
audio_temp_path = Path(audio_temp_dir)
temp_extracted = audio_temp_path / f"track_{index}_extracted.flac"
temp_normalized = audio_temp_path / f"track_{index}_normalized.flac"
final_opus = audio_temp_path / f"track_{index}_final.opus"
print(f" - Extracting Audio Track #{index} to FLAC...")
ffmpeg_args = [
"ffmpeg", "-v", "quiet", "-stats", "-y", "-i", str(source_file), "-map", f"0:{index}"
]
if should_downmix and ch >= 6:
if ch == 6:
ffmpeg_args += ["-af", "pan=stereo|c0=c2+0.30*c0+0.30*c4|c1=c2+0.30*c1+0.30*c5"]
elif ch == 8:
ffmpeg_args += ["-af", "pan=stereo|c0=c2+0.30*c0+0.30*c4+0.30*c6|c1=c2+0.30*c1+0.30*c5+0.30*c7"]
else: # Other multi-channel (e.g. 7ch, 10ch)
ffmpeg_args += ["-ac", "2"]
ffmpeg_args += ["-c:a", "flac", str(temp_extracted)]
run_cmd(ffmpeg_args)
print(f" - Normalizing Audio Track #{index} with SoX...")
run_cmd([
"sox", str(temp_extracted), str(temp_normalized), "-S", "--temp", str(audio_temp_path), "--guard", "gain", "-n"
])
# Set bitrate based on the final channel count of the Opus file.
# If we are downmixing, the result is stereo.
# If not, the result has the original channel count.
is_being_downmixed = should_downmix and ch >= 6
if is_being_downmixed:
# Downmixing from 5.1 or 7.1 results in a stereo track.
bitrate = "128k"
else:
# Not downmixing (or source is already stereo or less).
# Base bitrate on the source channel count.
if ch == 2: # Stereo
bitrate = "128k"
elif ch == 6: # 5.1 Surround
bitrate = "256k"
elif ch == 8: # 7.1 Surround
bitrate = "384k"
else: # Mono or other layouts
bitrate = "96k" # A sensible default for mono.
print(f" - Encoding Audio Track #{index} to Opus at {bitrate}...")
run_cmd([
"opusenc", "--vbr", "--bitrate", bitrate, str(temp_normalized), str(final_opus)
])
return final_opus
def convert_video(source_file_base, source_file_full, is_vfr, target_cfr_fps_for_handbrake):
print(" --- Starting Video Processing ---")
# source_file_base is file_path.stem (e.g., "my.anime.episode.01")
scene_file = Path(f"{source_file_base}.txt")
vpy_file = Path(f"{source_file_base}.vpy")
ut_video_file = Path(f"{source_file_base}.ut.mkv")
encoded_video_file = Path(f"temp-{source_file_base}.mkv")
handbrake_cfr_intermediate_file = None # To store path of HandBrake output if created
current_input_for_utvideo = Path(source_file_full)
if is_vfr and target_cfr_fps_for_handbrake:
print(f" - Source is VFR. Converting to CFR ({target_cfr_fps_for_handbrake}) with HandBrakeCLI...")
handbrake_cfr_intermediate_file = Path(f"{source_file_base}.cfr_temp.mkv")
handbrake_args = [
"HandBrakeCLI",
"--input", str(source_file_full),
"--output", str(handbrake_cfr_intermediate_file),
"--cfr",
"--rate", str(target_cfr_fps_for_handbrake),
"--encoder", "x264_10bit", # Changed to x264_10bit for 10-bit CFR intermediate
"--quality", "0", # CRF 0 for x264 is often considered visually lossless, or near-lossless
"--encoder-preset", "superfast", # Use a fast preset for quicker processing
"--encoder-tune", "fastdecode", # Added tune for faster decoding
"--audio", "none",
"--subtitle", "none",
"--crop-mode", "none" # Disable auto-cropping
]
print(f" - Running HandBrakeCLI: {' '.join(handbrake_args)}")
try:
run_cmd(handbrake_args)
if handbrake_cfr_intermediate_file.exists() and handbrake_cfr_intermediate_file.stat().st_size > 0:
print(f" - HandBrake VFR to CFR conversion successful: {handbrake_cfr_intermediate_file}")
current_input_for_utvideo = handbrake_cfr_intermediate_file
else:
print(f" - Warning: HandBrakeCLI VFR-to-CFR conversion failed or produced an empty file. Proceeding with original source for UTVideo.")
handbrake_cfr_intermediate_file = None # Ensure it's None if failed
except subprocess.CalledProcessError as e:
print(f" - Error during HandBrakeCLI execution: {e}")
print(f" - Proceeding with original source for UTVideo.")
handbrake_cfr_intermediate_file = None # Ensure it's None if failed
print(" - Creating UTVideo intermediate file (overwriting if exists)...")
# Check if source is already UTVideo
ffprobe_cmd = [
"ffprobe", "-v", "error", "-select_streams", "v:0",
"-show_entries", "stream=codec_name", "-of", "default=noprint_wrappers=1:nokey=1",
str(current_input_for_utvideo) # Use current input, which might be HandBrake output
]
source_codec = run_cmd(ffprobe_cmd, capture_output=True, check=True).strip()
video_codec_args = ["-c:v", "utvideo"]
if source_codec == "utvideo" and current_input_for_utvideo == Path(source_file_full): # Only copy if original was UTVideo
print(" - Source is already UTVideo. Copying video stream...")
video_codec_args = ["-c:v", "copy"]
ffmpeg_args = [
"ffmpeg", "-hide_banner", "-v", "quiet", "-stats", "-y", "-i", str(current_input_for_utvideo),
"-map", "0:v:0", "-map_metadata", "-1", "-map_chapters", "-1", "-an", "-sn", "-dn",
] + video_codec_args + [str(ut_video_file)]
run_cmd(ffmpeg_args)
ut_video_full_path = os.path.abspath(ut_video_file)
vpy_script_content = f"""import vapoursynth as vs
core = vs.core
core.num_threads = 4
clip = core.lsmas.LWLibavSource(source=r'''{ut_video_full_path}''')
clip = core.resize.Point(clip, format=vs.YUV420P10, matrix_in_s="709") # type: ignore
clip.set_output()
"""
with vpy_file.open("w", encoding="utf-8") as f:
f.write(vpy_script_content)
if not scene_file.exists():
print(" - Performing scene detection with av1an...")
av1an_sc_args = [
"av1an", "-i", str(vpy_file), "-s", str(scene_file), "--sc-only", "--verbose"
]
run_cmd(av1an_sc_args)
else:
print(" - Found existing scene file, skipping detection.")
print(" - Starting AV1 encode with av1an (this will take a long time)...")
total_cores = os.cpu_count() or 4 # Fallback if cpu_count is None
workers = max(total_cores - 2, 1) # Ensure at least 1 worker
print(f" - Using {workers} workers for av1an (Total Cores: {total_cores}).")
svt_av1_params = {
"preset": 2,
"crf": 27,
"film-grain": 6,
"lp": 1,
"tune": 1,
"keyint": -1,
"color-primaries": 1,
"transfer-characteristics": 1,
"matrix-coefficients": 1,
}
# Create the parameter string for av1an's -v option, which expects a single string.
av1an_video_params_str = " ".join([f"--{key} {value}" for key, value in svt_av1_params.items()])
print(f" - Using SVT-AV1 parameters: {av1an_video_params_str}")
av1an_enc_args = [
"av1an", "-i", str(vpy_file), "-o", str(encoded_video_file), "-s", str(scene_file), "-n",
"-e", "svt-av1", "--resume", "--sc-pix-format", "yuv420p", "-c", "mkvmerge",
"--set-thread-affinity", "1", "--pix-format", "yuv420p10le", "--force",
"-w", str(workers),
"-v", av1an_video_params_str
]
run_cmd(av1an_enc_args)
print(" --- Finished Video Processing ---")
return encoded_video_file, handbrake_cfr_intermediate_file
def is_ffmpeg_decodable(file_path):
"""Quickly check if ffmpeg can decode the input file."""
try:
# Try to decode a short segment of the first audio stream
subprocess.run([
"ffmpeg", "-v", "error", "-i", str(file_path), "-map", "0:a:0", "-t", "1", "-f", "null", "-"
], check=True)
return True
except subprocess.CalledProcessError:
return False
def main(no_downmix=False):
check_tools()
DIR_COMPLETED.mkdir(exist_ok=True, parents=True)
DIR_ORIGINAL.mkdir(exist_ok=True, parents=True)
DIR_CONV_LOGS.mkdir(exist_ok=True, parents=True) # Create conv_logs directory
current_dir = Path(".")
while True:
files_to_process = sorted(
f for f in current_dir.glob("*.mkv")
if not (f.name.endswith(".ut.mkv") or f.name.startswith("temp-") or f.name.startswith("output-") or f.name.endswith(".cfr_temp.mkv"))
)
if not files_to_process:
print("No more .mkv files found to process in the current directory. The script will now exit.")
break
# Process the first file in the list. The list is requeried in the next iteration.
file_path = files_to_process[0]
# --- Add ffmpeg decodability check here ---
if not is_ffmpeg_decodable(file_path):
print(f"ERROR: ffmpeg cannot decode '{file_path.name}'. Skipping this file.", file=sys.stderr)
shutil.move(str(file_path), DIR_ORIGINAL / file_path.name)
continue
print("-" * shutil.get_terminal_size(fallback=(80, 24)).columns)
# This print remains on the console, indicating which file is starting.
# The detailed "Starting full processing for..." will be in the log.
log_file_name = f"{file_path.stem}.log" # Use stem to avoid .mkv.log
log_file_path = DIR_CONV_LOGS / log_file_name
original_stdout_console = sys.stdout
original_stderr_console = sys.stderr
# Announce to console (original stdout)
print(f"Processing: {file_path.name}", file=original_stdout_console)
print(f"Logging output to: {log_file_path}", file=original_stdout_console)
log_file_handle = None
processing_error_occurred = False
date_for_runtime_calc = datetime.now() # For runtime calculation
try: # Outer try for log redirection and file handling
log_file_handle = open(log_file_path, 'w', encoding='utf-8')
sys.stdout = log_file_handle
sys.stderr = log_file_handle
# --- Start of log-specific messages ---
print(f"STARTING LOG FOR: {file_path.name}")
print(f"Processing started at: {date_for_runtime_calc}")
print(f"Full input file path: {file_path.resolve()}")
print("-" * shutil.get_terminal_size(fallback=(80, 24)).columns)
input_file_abs = file_path.resolve() # Used by original logic
intermediate_output_file = current_dir / f"output-{file_path.name}" # Used by original logic
audio_temp_dir = None # Initialize before inner try
handbrake_intermediate_for_cleanup = None # Initialize before inner try
# This is the original try...except...finally block for processing a single file.
# All its print statements will now go to the log file.
try:
audio_temp_dir = tempfile.mkdtemp(prefix="anime_audio_")
print(f"Audio temporary directory created at: {audio_temp_dir}")
print(f"Analyzing file: {input_file_abs}")
ffprobe_info_json = run_cmd([
"ffprobe", "-v", "quiet", "-print_format", "json", "-show_streams", "-show_format", str(input_file_abs)
], capture_output=True)
ffprobe_info = json.loads(ffprobe_info_json)
mkvmerge_info_json = run_cmd([
"mkvmerge", "-J", str(input_file_abs)
], capture_output=True)
mkv_info = json.loads(mkvmerge_info_json)
mediainfo_json = run_cmd([
"mediainfo", "--Output=JSON", "-f", str(input_file_abs)
], capture_output=True)
media_info = json.loads(mediainfo_json)
is_vfr = False
target_cfr_fps_for_handbrake = None
video_track_info = None
if media_info.get("media") and media_info["media"].get("track"):
for track in media_info["media"]["track"]:
if track.get("@type") == "Video":
video_track_info = track
break
if video_track_info:
frame_rate_mode = video_track_info.get("FrameRate_Mode")
if frame_rate_mode and frame_rate_mode.upper() in ["VFR", "VARIABLE"]:
is_vfr = True
print(f" - Detected VFR based on MediaInfo FrameRate_Mode: {frame_rate_mode}")
original_fps_str = video_track_info.get("FrameRate_Original_String")
if original_fps_str:
match = re.search(r'\((\d+/\d+)\)', original_fps_str)
if match:
target_cfr_fps_for_handbrake = match.group(1)
else: # Fallback to decimal part if fraction not in parentheses
target_cfr_fps_for_handbrake = video_track_info.get("FrameRate_Original")
if not target_cfr_fps_for_handbrake: # Fallback if Original_String didn't yield
target_cfr_fps_for_handbrake = video_track_info.get("FrameRate_Original")
if not target_cfr_fps_for_handbrake: # Further fallback to current FrameRate
target_cfr_fps_for_handbrake = video_track_info.get("FrameRate")
if target_cfr_fps_for_handbrake:
print(f" - Using MediaInfo FrameRate ({target_cfr_fps_for_handbrake}) as fallback for HandBrake target FPS.")
if target_cfr_fps_for_handbrake:
print(f" - Target CFR for HandBrake: {target_cfr_fps_for_handbrake}")
# Convert fractional FPS to decimal for HandBrakeCLI if needed
if isinstance(target_cfr_fps_for_handbrake, str) and "/" in target_cfr_fps_for_handbrake:
try:
num, den = map(float, target_cfr_fps_for_handbrake.split('/'))
target_cfr_fps_for_handbrake = f"{num / den:.3f}"
print(f" - Converted fractional FPS to decimal for HandBrake: {target_cfr_fps_for_handbrake}")
except ValueError:
print(f" - Warning: Could not parse fractional FPS '{target_cfr_fps_for_handbrake}'. HandBrakeCLI might fail.")
is_vfr = False # Revert if conversion fails
else:
print(" - Warning: VFR detected, but could not determine target CFR from MediaInfo. Will attempt standard UTVideo conversion without HandBrake.")
is_vfr = False # Revert to non-HandBrake path
else:
print(f" - Video appears to be CFR or FrameRate_Mode not specified as VFR/Variable by MediaInfo.")
encoded_video_file, handbrake_intermediate_for_cleanup = convert_video(
file_path.stem, str(input_file_abs), is_vfr, target_cfr_fps_for_handbrake
)
print("--- Starting Audio Processing ---")
processed_audio_files = []
audio_tracks_to_remux = []
audio_streams = [s for s in ffprobe_info.get("streams", []) if s.get("codec_type") == "audio"]
# Build mkvmerge track mapping by track ID
mkv_audio_tracks = {t["id"]: t for t in mkv_info.get("tracks", []) if t.get("type") == "audio"}
# Build mediainfo track mapping by StreamOrder
media_tracks_data = media_info.get("media", {}).get("track", [])
mediainfo_audio_tracks = {int(t.get("StreamOrder", -1)): t for t in media_tracks_data if t.get("@type") == "Audio"}
for stream in audio_streams:
stream_index = stream["index"]
codec = stream.get("codec_name")
channels = stream.get("channels", 2)
language = stream.get("tags", {}).get("language", "und")
# Find mkvmerge track by matching ffprobe stream index to mkvmerge track's 'properties'->'stream_id'
mkv_track = None
for t in mkv_info.get("tracks", []):
if t.get("type") == "audio" and t.get("properties", {}).get("stream_id") == stream_index:
mkv_track = t
break
if not mkv_track:
# Fallback: try by position
mkv_track = mkv_info.get("tracks", [])[stream_index] if stream_index < len(mkv_info.get("tracks", [])) else {}
track_id = mkv_track.get("id", -1)
track_title = mkv_track.get("properties", {}).get("track_name", "")
# Find mediainfo track by StreamOrder
audio_track_info = mediainfo_audio_tracks.get(stream_index)
track_delay = 0
delay_in_seconds = audio_track_info.get("Video_Delay") if audio_track_info else None
if delay_in_seconds is not None:
try:
track_delay = round(float(delay_in_seconds) * 1000)
except Exception:
track_delay = 0
print(f"Processing Audio Stream #{stream_index} (TID: {track_id}, Codec: {codec}, Channels: {channels})")
if codec in REMUX_CODECS:
audio_tracks_to_remux.append(str(track_id))
else:
# Convert any codec that is not in REMUX_CODECS
opus_file = convert_audio_track(
stream_index, channels, language, audio_temp_dir, str(input_file_abs), not no_downmix
)
processed_audio_files.append({
"Path": opus_file,
"Language": language,
"Title": track_title,
"Delay": track_delay
})
print("--- Finished Audio Processing ---")
# Final mux
print("Assembling final file with mkvmerge...")
mkvmerge_args = ["mkvmerge", "-o", str(intermediate_output_file), str(encoded_video_file)]
for file_info in processed_audio_files:
sync_switch = ["--sync", f"0:{file_info['Delay']}"] if file_info["Delay"] else []
mkvmerge_args += [
"--language", f"0:{file_info['Language']}",
"--track-name", f"0:{file_info['Title']}"
] + sync_switch + [str(file_info["Path"])]
source_copy_args = ["--no-video"]
if audio_tracks_to_remux:
source_copy_args += ["--audio-tracks", ",".join(audio_tracks_to_remux)]
else:
source_copy_args += ["--no-audio"]
mkvmerge_args += source_copy_args + [str(input_file_abs)]
run_cmd(mkvmerge_args)
# Move files
print("Moving files to final destinations...")
shutil.move(str(file_path), DIR_ORIGINAL / file_path.name)
shutil.move(str(intermediate_output_file), DIR_COMPLETED / file_path.name)
print("Cleaning up persistent video temporary files (after successful processing)...")
video_temp_files_on_success = [
current_dir / f"{file_path.stem}.txt",
current_dir / f"{file_path.stem}.vpy",
current_dir / f"{file_path.stem}.ut.mkv",
current_dir / f"temp-{file_path.stem}.mkv", # This is encoded_video_file
current_dir / f"{file_path.stem}.ut.mkv.lwi",
]
if handbrake_intermediate_for_cleanup and handbrake_intermediate_for_cleanup.exists():
video_temp_files_on_success.append(handbrake_intermediate_for_cleanup)
for temp_vid_file in video_temp_files_on_success:
if temp_vid_file.exists():
print(f" Deleting: {temp_vid_file}")
temp_vid_file.unlink(missing_ok=True)
else:
print(f" Skipping (not found): {temp_vid_file}")
except Exception as e:
print(f"ERROR: An error occurred while processing '{file_path.name}': {e}", file=sys.stderr) # Goes to log
original_stderr_console.write(f"ERROR during processing of '{file_path.name}': {e}\nSee log '{log_file_path}' for details.\n")
processing_error_occurred = True
finally:
# This is the original 'finally' block. Its prints go to the log file.
print("--- Starting Universal Cleanup (for this file) ---")
print(" - Cleaning up disposable audio temporary directory...")
if audio_temp_dir and Path(audio_temp_dir).exists():
shutil.rmtree(audio_temp_dir, ignore_errors=True)
print(f" - Deleted audio temp dir: {audio_temp_dir}")
elif audio_temp_dir: # Was created but now not found
print(f" - Audio temp dir not found or already cleaned: {audio_temp_dir}")
else: # Was never created
print(f" - Audio temp dir was not created.")
print(" - Cleaning up intermediate output file (if it wasn't moved on success)...")
if intermediate_output_file.exists(): # Check if it still exists (e.g. error before move)
if processing_error_occurred:
print(f" - WARNING: Processing error occurred. Intermediate output file '{intermediate_output_file}' is being preserved at its original path for inspection.")
else:
# No processing error, so it should have been moved.
# If it's still here, it's unexpected but we'll clean it up.
print(f" - INFO: Intermediate output file '{intermediate_output_file}' found at original path despite no errors (expected to be moved). Cleaning up.")
intermediate_output_file.unlink(missing_ok=True) # Only unlink if no error and it exists
print(f" - Deleted intermediate output file from original path: {intermediate_output_file}")
else:
# File does not exist at original path
if not processing_error_occurred:
print(f" - Intermediate output file successfully moved (not found at original path, as expected): {intermediate_output_file}")
else:
print(f" - Processing error occurred, and intermediate output file '{intermediate_output_file}' not found at original path (likely not created or cleaned by another step).")
# --- End of original per-file processing block ---
print(f"FINISHED LOG FOR: {file_path.name}")
# --- End of log-specific messages ---
finally: # Outer finally for restoring stdout/stderr and closing log file
runtime = datetime.now() - date_for_runtime_calc
runtime_str = str(runtime).split('.')[0]
# This print goes to the log file, as stdout is not yet restored.
print(f"\nTotal runtime for this file: {runtime_str}")
if sys.stdout != original_stdout_console:
sys.stdout = original_stdout_console
if sys.stderr != original_stderr_console:
sys.stderr = original_stderr_console
if log_file_handle:
log_file_handle.close()
# Announce to console (original stdout/stderr) that this file is done
if processing_error_occurred:
original_stderr_console.write(f"File: {file_path.name}\n")
original_stderr_console.write(f"Log: {log_file_path}\n")
original_stderr_console.write(f"Runtime: {runtime_str}\n")
else:
original_stdout_console.write(f"File: {file_path.name}\n")
original_stdout_console.write(f"Log: {log_file_path}\n")
original_stdout_console.write(f"Runtime: {runtime_str}\n")
if __name__ == "__main__":
import argparse
parser = argparse.ArgumentParser(description="Batch-process MKV files with resumable video encoding, audio downmixing, and per-file logging.")
parser.add_argument("--no-downmix", action="store_true", help="Preserve original audio channel layout.")
args = parser.parse_args()
main(no_downmix=args.no_downmix)