feat: rewrite application as modular fcco-Converter V2 with new GUI, metadata handling, and updater logic
This commit is contained in:
joel
@@ -0,0 +1,299 @@
|
||||
"""
|
||||
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
|
||||
|
||||
|
||||
def _ensure_ffmpeg() -> tuple[str, str]:
|
||||
"""Stellt sicher, dass ffmpeg und ffprobe verfügbar sind.
|
||||
|
||||
Returns:
|
||||
(ffmpeg_path, ffprobe_path)
|
||||
"""
|
||||
ffmpeg_path, ffprobe_path = static_ffmpeg.run.get_or_fetch_platform_executables_else_raise()
|
||||
return ffmpeg_path, ffprobe_path
|
||||
|
||||
|
||||
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")
|
||||
|
||||
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,
|
||||
) -> 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"])
|
||||
title = 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":
|
||||
cover_path = Path(meta["cover"]) if meta.get("cover") else None
|
||||
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_path=cover_path,
|
||||
)
|
||||
|
||||
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 = []
|
||||
|
||||
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,
|
||||
): 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)
|
||||
Reference in New Issue
Block a user