343 lines
10 KiB
Python
343 lines
10 KiB
Python
"""
|
||
fcco-Converter V2 – Konvertierungsmodul
|
||
|
||
Enthält die FFmpeg-basierte Audio-Konvertierung mit static-ffmpeg.
|
||
Unterstützt erweiterte Einstellungen (Bitrate, Sample-Rate, Kanäle, VBR/CBR).
|
||
"""
|
||
|
||
import os
|
||
import subprocess
|
||
from pathlib import Path
|
||
from concurrent.futures import ThreadPoolExecutor, as_completed
|
||
from typing import Callable
|
||
|
||
import static_ffmpeg
|
||
|
||
from .metadata import apply_id3_tags
|
||
|
||
|
||
import sys
|
||
import io
|
||
import shutil
|
||
|
||
def _ensure_ffmpeg() -> tuple[str, str]:
|
||
"""Stellt sicher, dass ffmpeg und ffprobe verfügbar sind.
|
||
|
||
Returns:
|
||
(ffmpeg_path, ffprobe_path)
|
||
"""
|
||
# 1. Behebe NoneType 'write' Fehler, wenn App im windowed-Modus gestartet wurde
|
||
if sys.stdout is None:
|
||
sys.stdout = io.StringIO()
|
||
if sys.stderr is None:
|
||
sys.stderr = io.StringIO()
|
||
|
||
# 2. Versuch 1: static_ffmpeg (jetzt direkt in der App gebündelt)
|
||
try:
|
||
ffmpeg_path, ffprobe_path = static_ffmpeg.run.get_or_fetch_platform_executables_else_raise()
|
||
return ffmpeg_path, ffprobe_path
|
||
except Exception:
|
||
pass
|
||
|
||
# 3. Versuch 2: System-FFmpeg (z.B. über winget, apt, brew) als Fallback
|
||
sys_ffmpeg = shutil.which("ffmpeg")
|
||
sys_ffprobe = shutil.which("ffprobe")
|
||
if sys_ffmpeg and sys_ffprobe:
|
||
return sys_ffmpeg, sys_ffprobe
|
||
|
||
raise RuntimeError("FFmpeg konnte weder im Bundle noch im System gefunden werden.")
|
||
|
||
|
||
def get_duration(file_path: Path, ffprobe_path: str) -> float:
|
||
"""Ermittelt die Dauer einer Audiodatei in Sekunden."""
|
||
cmd = [
|
||
ffprobe_path,
|
||
"-v", "error",
|
||
"-show_entries", "format=duration",
|
||
"-of", "default=noprint_wrappers=1:nokey=1",
|
||
str(file_path),
|
||
]
|
||
try:
|
||
result = subprocess.check_output(
|
||
cmd, universal_newlines=True, stderr=subprocess.DEVNULL
|
||
).strip()
|
||
return float(result)
|
||
except (subprocess.CalledProcessError, ValueError):
|
||
return 0.0
|
||
|
||
|
||
def _build_ffmpeg_cmd(
|
||
ffmpeg_path: str,
|
||
input_path: Path,
|
||
output_path: Path,
|
||
audio_settings: dict,
|
||
threads: int = 0,
|
||
) -> list[str]:
|
||
"""Baut den ffmpeg-Befehl basierend auf den Audio-Einstellungen.
|
||
|
||
Args:
|
||
ffmpeg_path: Pfad zur ffmpeg-Binary.
|
||
input_path: Quell-Audiodatei.
|
||
output_path: Ziel-Ausgabedatei.
|
||
audio_settings: Dict mit Audio-Konfiguration.
|
||
threads: Thread-Anzahl (0 = auto).
|
||
|
||
Returns:
|
||
Liste der Befehlsargumente.
|
||
"""
|
||
output_format = audio_settings.get("output_format", "mp3")
|
||
encoding_mode = audio_settings.get("encoding_mode", "CBR")
|
||
bitrate = audio_settings.get("bitrate", "192")
|
||
vbr_quality = audio_settings.get("vbr_quality", "2")
|
||
sample_rate = audio_settings.get("sample_rate", "Original")
|
||
channels = audio_settings.get("channels", "Original")
|
||
overwrite = audio_settings.get("overwrite_existing", True)
|
||
|
||
cmd = [ffmpeg_path, "-i", str(input_path)]
|
||
|
||
# Threads
|
||
cmd.extend(["-threads", str(threads) if threads > 0 else "0"])
|
||
|
||
# Format-spezifische Codec-Einstellungen
|
||
if output_format == "mp3":
|
||
cmd.extend(["-c:a", "libmp3lame"])
|
||
if encoding_mode == "VBR":
|
||
cmd.extend(["-q:a", vbr_quality])
|
||
else: # CBR
|
||
cmd.extend(["-b:a", f"{bitrate}k"])
|
||
|
||
elif output_format == "flac":
|
||
cmd.extend(["-c:a", "flac"])
|
||
|
||
elif output_format == "ogg":
|
||
cmd.extend(["-c:a", "libvorbis"])
|
||
if encoding_mode == "VBR":
|
||
# OGG VBR: Qualität 0–10 (wir mappen 0–9 auf 1–10)
|
||
q = min(10, max(0, 10 - int(vbr_quality)))
|
||
cmd.extend(["-q:a", str(q)])
|
||
else:
|
||
cmd.extend(["-b:a", f"{bitrate}k"])
|
||
|
||
elif output_format == "wav":
|
||
cmd.extend(["-c:a", "pcm_s16le"])
|
||
|
||
# Sample-Rate
|
||
if sample_rate != "Original":
|
||
cmd.extend(["-ar", sample_rate])
|
||
|
||
# Kanäle
|
||
if channels == "Mono":
|
||
cmd.extend(["-ac", "1"])
|
||
elif channels == "Stereo":
|
||
cmd.extend(["-ac", "2"])
|
||
|
||
# Überschreiben
|
||
if overwrite:
|
||
cmd.append("-y")
|
||
else:
|
||
cmd.append("-n")
|
||
|
||
# Keine Video/Bilder-Streams übernehmen (verhindert Konflikte mit eigenen Covern)
|
||
cmd.append("-vn")
|
||
|
||
cmd.append(str(output_path))
|
||
return cmd
|
||
|
||
|
||
def convert_single_file(
|
||
input_path: Path,
|
||
output_path: Path,
|
||
ffmpeg_path: str,
|
||
ffprobe_path: str,
|
||
audio_settings: dict | None = None,
|
||
threads: int = 0,
|
||
progress_callback: Callable[[int], None] | None = None,
|
||
) -> None:
|
||
"""Konvertiert eine einzelne Audiodatei.
|
||
|
||
Args:
|
||
input_path: Pfad zur Quelldatei.
|
||
output_path: Pfad für die Zieldatei.
|
||
ffmpeg_path: Pfad zur ffmpeg-Binary.
|
||
ffprobe_path: Pfad zur ffprobe-Binary.
|
||
audio_settings: Dict mit Audio-Konfiguration (aus config).
|
||
threads: Anzahl der Threads (0 = auto).
|
||
progress_callback: Wird mit Prozent (0–100) aufgerufen.
|
||
|
||
Raises:
|
||
RuntimeError: Bei Fehlern in der Konvertierung.
|
||
"""
|
||
if audio_settings is None:
|
||
audio_settings = {"output_format": "mp3", "bitrate": "192", "encoding_mode": "CBR"}
|
||
|
||
output_path.parent.mkdir(parents=True, exist_ok=True)
|
||
|
||
duration = get_duration(input_path, ffprobe_path)
|
||
if duration <= 0:
|
||
raise RuntimeError(f"Dauer der Datei konnte nicht ermittelt werden: {input_path.name}")
|
||
|
||
cmd = _build_ffmpeg_cmd(ffmpeg_path, input_path, output_path, audio_settings, threads)
|
||
|
||
process = subprocess.Popen(
|
||
cmd,
|
||
stderr=subprocess.PIPE,
|
||
stdout=subprocess.DEVNULL,
|
||
universal_newlines=True,
|
||
)
|
||
|
||
while True:
|
||
line = process.stderr.readline()
|
||
if not line:
|
||
break
|
||
if "time=" in line and progress_callback:
|
||
try:
|
||
time_str = line.split("time=")[1].split(" ")[0]
|
||
h, m, s = map(float, time_str.split(":"))
|
||
current_time = h * 3600 + m * 60 + s
|
||
percentage = min(int((current_time / duration) * 100), 100)
|
||
progress_callback(percentage)
|
||
except (ValueError, IndexError):
|
||
pass
|
||
|
||
process.wait()
|
||
if process.returncode != 0:
|
||
raise RuntimeError(f"FFmpeg-Fehler (Code {process.returncode}) bei: {input_path.name}")
|
||
|
||
|
||
def convert_file_entry(
|
||
entry: dict,
|
||
index: int,
|
||
export_dir: Path,
|
||
meta: dict,
|
||
use_subdirs: bool,
|
||
ffmpeg_path: str,
|
||
ffprobe_path: str,
|
||
audio_settings: dict | None = None,
|
||
threads: int = 0,
|
||
progress_callback: Callable[[int, int, str], None] | None = None,
|
||
cover_data: bytes | None = None,
|
||
cover_mime: str = "image/jpeg",
|
||
) -> tuple[int, str, str | None]:
|
||
"""Konvertiert eine Datei und schreibt die Metadaten.
|
||
|
||
Returns:
|
||
(index, filename, error_or_None)
|
||
"""
|
||
if audio_settings is None:
|
||
audio_settings = {"output_format": "mp3", "bitrate": "192", "encoding_mode": "CBR"}
|
||
|
||
file_path = Path(entry["full_path"])
|
||
from .metadata import extract_title_from_filename
|
||
title = extract_title_from_filename(file_path.stem)
|
||
output_format = audio_settings.get("output_format", "mp3")
|
||
|
||
try:
|
||
if use_subdirs and entry["rel_dir"]:
|
||
output_dir = export_dir / entry["rel_dir"]
|
||
else:
|
||
output_dir = export_dir
|
||
|
||
output_file = output_dir / f"{title}.{output_format}"
|
||
|
||
def _on_progress(pct: int):
|
||
if progress_callback:
|
||
progress_callback(index, pct, title)
|
||
|
||
convert_single_file(
|
||
input_path=file_path,
|
||
output_path=output_file,
|
||
ffmpeg_path=ffmpeg_path,
|
||
ffprobe_path=ffprobe_path,
|
||
audio_settings=audio_settings,
|
||
threads=threads,
|
||
progress_callback=_on_progress,
|
||
)
|
||
|
||
# ID3-Metadaten nur für MP3 schreiben
|
||
if output_format == "mp3":
|
||
apply_id3_tags(
|
||
mp3_path=output_file,
|
||
title=title,
|
||
artist=meta.get("artist", ""),
|
||
album=meta.get("album", ""),
|
||
track_number=index + 1,
|
||
year=meta.get("year", ""),
|
||
cover_data=cover_data,
|
||
cover_mime=cover_mime,
|
||
)
|
||
|
||
return (index, f"{title}.{output_format}", None)
|
||
|
||
except Exception as e:
|
||
return (index, file_path.name, str(e))
|
||
|
||
|
||
def run_batch_conversion(
|
||
entries: list[dict],
|
||
export_dir: Path,
|
||
meta: dict,
|
||
use_subdirs: bool,
|
||
audio_settings: dict | None = None,
|
||
max_workers: int = 0,
|
||
progress_callback: Callable[[int, int, str], None] | None = None,
|
||
completion_callback: Callable[[list[str], list[tuple[str, str]]], None] | None = None,
|
||
) -> None:
|
||
"""Führt die Batch-Konvertierung aller Dateien durch."""
|
||
if audio_settings is None:
|
||
audio_settings = {"output_format": "mp3", "bitrate": "192", "encoding_mode": "CBR"}
|
||
|
||
ffmpeg_path, ffprobe_path = _ensure_ffmpeg()
|
||
|
||
if max_workers <= 0:
|
||
max_workers = os.cpu_count() or 1
|
||
|
||
successful_files = []
|
||
errors = []
|
||
|
||
# Cover-Bild einmalig in den Arbeitsspeicher laden, um Dateisperren (Locking)
|
||
# bei parallelem Zugriff durch Windows Defender/Antivirus zu vermeiden
|
||
cover_data = None
|
||
cover_mime = "image/jpeg"
|
||
cover_path_str = meta.get("cover")
|
||
if cover_path_str:
|
||
cp = Path(cover_path_str)
|
||
if cp.exists():
|
||
if cp.suffix.lower() == ".png":
|
||
cover_mime = "image/png"
|
||
with open(cp, "rb") as f:
|
||
cover_data = f.read()
|
||
|
||
with ThreadPoolExecutor(max_workers=max_workers) as executor:
|
||
futures = {
|
||
executor.submit(
|
||
convert_file_entry,
|
||
entry=entry,
|
||
index=idx,
|
||
export_dir=export_dir,
|
||
meta=meta,
|
||
use_subdirs=use_subdirs,
|
||
ffmpeg_path=ffmpeg_path,
|
||
ffprobe_path=ffprobe_path,
|
||
audio_settings=audio_settings,
|
||
threads=0,
|
||
progress_callback=progress_callback,
|
||
cover_data=cover_data,
|
||
cover_mime=cover_mime,
|
||
): idx
|
||
for idx, entry in enumerate(entries)
|
||
}
|
||
|
||
for future in as_completed(futures):
|
||
index, filename, error = future.result()
|
||
if error:
|
||
errors.append((filename, error))
|
||
if progress_callback:
|
||
progress_callback(index, 100, filename)
|
||
else:
|
||
successful_files.append(filename)
|
||
if progress_callback:
|
||
progress_callback(index, 100, filename)
|
||
|
||
if completion_callback:
|
||
completion_callback(successful_files, errors)
|