193 lines
10 KiB
Python
193 lines
10 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
import shutil
|
|
import subprocess
|
|
from pathlib import Path
|
|
from host.common import VERSION
|
|
from host.bundle import executable, information
|
|
|
|
ROOT = Path(__file__).resolve().parent.parent
|
|
NO_WINDOW = subprocess.CREATE_NO_WINDOW if os.name == "nt" else 0
|
|
|
|
|
|
def config():
|
|
path = ROOT / ".runtime" / "config.json"
|
|
return json.loads(path.read_text(encoding="utf-8-sig")) if path.is_file() else {}
|
|
|
|
|
|
def tool(name):
|
|
bundled = executable(name)
|
|
if bundled:
|
|
return bundled
|
|
configured = config().get(name)
|
|
result = configured if configured and Path(configured).is_file() else shutil.which(name)
|
|
if not result or not Path(result).is_file():
|
|
raise ValueError(f"Не найден {name}. Запустите install.ps1.")
|
|
return str(Path(result).resolve())
|
|
|
|
|
|
def run_small(command, timeout=15):
|
|
return subprocess.run(command, capture_output=True, text=True, encoding="utf-8", errors="replace",
|
|
timeout=timeout, creationflags=NO_WINDOW, stdin=subprocess.DEVNULL)
|
|
|
|
|
|
def capabilities():
|
|
from host.dependencies import status
|
|
dependencies = status()
|
|
ready = all(item["status"] == "installed" for item in dependencies)
|
|
ffmpeg = next((item for item in dependencies if item["id"] == "ffmpeg" and item["status"] == "installed"), None)
|
|
checked = {"nvenc": [], "amf": [], "amf10bit": []}
|
|
for engine in ("nvenc", "amf"):
|
|
for codec in ("h264", "hevc", "av1"):
|
|
if not ffmpeg:
|
|
break
|
|
base = [tool("ffmpeg"), "-hide_banner", "-loglevel", "error", "-f", "lavfi", "-i"]
|
|
preset = ["-preset", "p4"] if engine == "nvenc" else ["-quality", "balanced", "-rc", "cqp", "-qp_i", "20", "-qp_p", "20"]
|
|
try:
|
|
result = run_small([*base, "color=c=black:s=640x360:r=30", "-frames:v", "2", "-c:v", f"{codec}_{engine}", *preset, "-f", "null", "-"], timeout=20)
|
|
if result.returncode == 0:
|
|
checked[engine].append(codec)
|
|
if engine == "amf" and codec in {"hevc", "av1"}:
|
|
extra = ["-profile:v", "main10"] if codec == "hevc" else []
|
|
tenbit = run_small([*base, "color=c=black:s=640x360:r=30,format=p010le", "-frames:v", "2", "-c:v", f"{codec}_amf",
|
|
*preset, "-bitdepth", "10", *extra, "-pix_fmt", "+p010le", "-f", "null", "-"], timeout=20)
|
|
if tenbit.returncode == 0:
|
|
checked["amf10bit"].append(codec)
|
|
except (OSError, subprocess.TimeoutExpired):
|
|
continue
|
|
gpu = ""
|
|
smi = shutil.which("nvidia-smi")
|
|
if smi:
|
|
try:
|
|
result = run_small([smi, "--query-gpu=name", "--format=csv,noheader"], timeout=8)
|
|
gpu = result.stdout.strip()[:200] if result.returncode == 0 else ""
|
|
except (OSError, subprocess.TimeoutExpired):
|
|
pass
|
|
if os.name == "nt":
|
|
try:
|
|
detected = run_small(["powershell.exe", "-NoProfile", "-Command",
|
|
"[Console]::OutputEncoding=[Text.UTF8Encoding]::new(); "
|
|
"Get-CimInstance Win32_VideoController | Select-Object -ExpandProperty Name"], timeout=10)
|
|
if detected.returncode == 0:
|
|
gpu = detected.stdout.strip()[:200] or gpu
|
|
except (OSError, subprocess.TimeoutExpired):
|
|
pass
|
|
return {"version": VERSION, "minimumExtensionVersion": "1.4.5", "supportsWorkFolder": True, "ytDlp": next(item["version"] for item in dependencies if item["id"] == "yt_dlp"), "ffmpeg": ffmpeg["version"] if ffmpeg else "",
|
|
"gpu": gpu, **checked, "defaultFolder": str(Path.home() / "Downloads"),
|
|
"dependencies": dependencies, "ready": ready, "bundle": information()}
|
|
|
|
|
|
def probe(path):
|
|
result = run_small([tool("ffprobe"), "-v", "error", "-show_streams", "-show_format", "-of", "json", str(path)], timeout=30)
|
|
if result.returncode:
|
|
raise ValueError("FFprobe не смог прочитать загруженный файл.")
|
|
return json.loads(result.stdout)
|
|
|
|
|
|
def encoding_plan(options, media, encoders):
|
|
# Older development callers supply the NVENC codec list directly.
|
|
encoders = {"nvenc": encoders} if isinstance(encoders, (list, tuple)) else encoders
|
|
streams = media.get("streams", [])
|
|
video = next((s for s in streams if s.get("codec_type") == "video"), None)
|
|
audio = next((s for s in streams if s.get("codec_type") == "audio"), None)
|
|
mode, container = options["mode"], options["container"]
|
|
if mode != "audio" and not video:
|
|
raise ValueError("В загруженном файле нет видео.")
|
|
if mode in {"audio", "video"} and not audio:
|
|
raise ValueError("В загруженном файле нет звуковой дорожки.")
|
|
if mode == "audio":
|
|
target = {"m4a": "aac", "mp3": "libmp3lame", "opus": "libopus", "flac": "flac", "wav": "pcm_s16le"}[container]
|
|
source_audio = audio.get("codec_name")
|
|
if (container, source_audio) in {("m4a", "aac"), ("mp3", "mp3"), ("opus", "opus"), ("flac", "flac")}:
|
|
target = "copy"
|
|
return {"video": None, "audio": target, "engine": "copy" if target == "copy" else "cpu", "hdr": False}
|
|
source = video.get("codec_name", "")
|
|
compatible = container == "mkv" or (container == "mp4" and source in {"h264", "hevc", "av1"}) or (container == "webm" and source in {"vp8", "vp9", "av1"})
|
|
codec = options["codec"]
|
|
hdr = video.get("color_transfer") in {"smpte2084", "arib-std-b67"}
|
|
ten_bit = "10" in video.get("pix_fmt", "") or video.get("pix_fmt", "").startswith("p010")
|
|
if codec == "copy" and not compatible:
|
|
raise ValueError("Исходный видеокодек несовместим с контейнером. Выберите «Авто» или MKV.")
|
|
if codec == "auto":
|
|
codec = "copy" if compatible else ("av1" if container == "webm" else ("hevc" if hdr else "h264"))
|
|
if container == "webm" and codec not in {"copy", "vp9", "av1"}:
|
|
raise ValueError("WebM поддерживает AV1 и VP9. Для H.264/HEVC выберите MP4 или MKV.")
|
|
if hdr and codec == "h264":
|
|
raise ValueError("Для HDR выберите HEVC, AV1 или сохранение исходного кодека.")
|
|
engine = "copy" if codec == "copy" else "cpu"
|
|
if codec != "copy" and options["engine"] != "cpu":
|
|
for candidate in ("nvenc", "amf"):
|
|
supported = encoders.get("amf10bit" if candidate == "amf" and ten_bit and codec in {"hevc", "av1"} else candidate, [])
|
|
if options["engine"] in {"auto", candidate} and codec in supported:
|
|
engine = candidate
|
|
break
|
|
if options["engine"] in {"nvenc", "amf"} and engine != options["engine"]:
|
|
name = "AMD AMF" if options["engine"] == "amf" else "NVIDIA NVENC"
|
|
raise ValueError(f"Выбранный кодек или глубина цвета недоступны в {name}. Выберите поддерживаемый кодек или «Авто».")
|
|
audio_codec = None
|
|
if mode == "video":
|
|
ac = audio.get("codec_name", "")
|
|
if container == "mkv" or (container == "mp4" and ac in {"aac", "mp3", "alac"}) or (container == "webm" and ac in {"opus", "vorbis"}):
|
|
audio_codec = "copy"
|
|
else:
|
|
audio_codec = "libopus" if container == "webm" else "aac"
|
|
if engine == "copy" and audio_codec not in (None, "copy"):
|
|
engine = "cpu"
|
|
return {"video": codec, "audio": audio_codec, "engine": engine, "hdr": hdr, "tenBit": ten_bit}
|
|
|
|
|
|
def ffmpeg_command(source, target, options, plan, cuda_decode=True):
|
|
command = [tool("ffmpeg"), "-hide_banner", "-nostdin", "-loglevel", "info", "-y", "-progress", "pipe:1", "-stats_period", "0.4"]
|
|
if plan["engine"] == "nvenc" and cuda_decode:
|
|
command += ["-hwaccel", "cuda"]
|
|
elif plan["engine"] == "amf" and cuda_decode:
|
|
command += ["-hwaccel", "d3d11va"]
|
|
command += ["-i", str(source)]
|
|
if options["mode"] == "audio":
|
|
command += ["-map", "0:a:0", "-vn"]
|
|
else:
|
|
command += ["-map", "0:v:0"]
|
|
if options["mode"] == "video":
|
|
command += ["-map", "0:a:0"]
|
|
else:
|
|
command += ["-an"]
|
|
codec = plan["video"]
|
|
if codec:
|
|
if codec == "copy":
|
|
command += ["-c:v", "copy"]
|
|
elif plan["engine"] == "nvenc":
|
|
preset = {"fast": "p2", "balanced": "p4", "quality": "p6"}[options["preset"]]
|
|
command += ["-c:v", codec + "_nvenc", "-preset", preset, "-rc", "vbr", "-cq", "20", "-b:v", "0"]
|
|
command += ["-pix_fmt", "p010le" if plan.get("tenBit") and codec in {"hevc", "av1"} else "yuv420p"]
|
|
elif plan["engine"] == "amf":
|
|
quality = {"fast": "speed", "balanced": "balanced", "quality": "quality"}[options["preset"]]
|
|
command += ["-c:v", codec + "_amf", "-quality", quality, "-rc", "cqp", "-qp_i", "20", "-qp_p", "20"]
|
|
if plan.get("tenBit") and codec in {"hevc", "av1"}:
|
|
command += ["-bitdepth", "10", "-pix_fmt", "p010le"]
|
|
if codec == "hevc": command += ["-profile:v", "main10"]
|
|
else:
|
|
command += ["-pix_fmt", "yuv420p"]
|
|
else:
|
|
encoder = {"h264": "libx264", "hevc": "libx265", "av1": "libsvtav1", "vp9": "libvpx-vp9"}[codec]
|
|
command += ["-c:v", encoder]
|
|
if codec in {"h264", "hevc"}:
|
|
command += ["-preset", {"fast": "veryfast", "balanced": "fast", "quality": "medium"}[options["preset"]], "-crf", "20"]
|
|
elif codec == "av1":
|
|
command += ["-preset", "8", "-crf", "24"]
|
|
else:
|
|
command += ["-crf", "28", "-b:v", "0", "-row-mt", "1", "-cpu-used", "4"]
|
|
command += ["-pix_fmt", "yuv420p10le" if plan.get("tenBit") and codec != "h264" else "yuv420p"]
|
|
if plan["audio"]:
|
|
command += ["-c:a", plan["audio"]]
|
|
if plan["audio"] in {"aac", "libmp3lame", "libopus"}:
|
|
command += ["-b:a", f"{options['bitrate']}k"]
|
|
command += ["-map_metadata", "0"]
|
|
if options["container"] in {"mp4", "m4a"}:
|
|
command += ["-movflags", "+faststart"]
|
|
if codec == "hevc" and options["container"] == "mp4":
|
|
command += ["-tag:v", "hvc1"]
|
|
command += [str(target)]
|
|
return command
|