diff --git a/combined_files.txt b/combined_files.txt deleted file mode 100644 index 1aa4afb..0000000 --- a/combined_files.txt +++ /dev/null @@ -1,2613 +0,0 @@ -# ========================================= -# Файл: .gitignore -# ========================================= - -__pycache__/ -*.py[cod] -*.egg-info/ -.venv/ -venv/ -.env - -transcribed/ -output/ -*.wav -*.flac -*.mp3 -!tests/fixtures/* - -preset.json -*.log - - -# ========================================= -# Файл: main.py -# ========================================= - -#!/usr/bin/env python3 -"""Song Transcriber — Qt GUI entry point.""" -from __future__ import annotations - -import ctypes -import os -import sys - -# Must be set BEFORE PySide6 is imported. -os.environ.setdefault("QT_LOGGING_RULES", "qt.qpa.wayland.textinput=false") - - -def preload_cudnn() -> None: - try: - import nvidia.cudnn - except ImportError: - return - lib_dir = os.path.join(nvidia.cudnn.__path__[0], "lib") - if not os.path.isdir(lib_dir): - return - for fname in sorted(os.listdir(lib_dir)): - if ".so" in fname: - try: - ctypes.CDLL(os.path.join(lib_dir, fname), mode=ctypes.RTLD_GLOBAL) - except OSError: - pass - - -def enable_tf32() -> None: - """Enable TF32 for faster matmul on Ampere+ GPUs (RTX 30/40 series).""" - try: - import torch - if torch.cuda.is_available(): - torch.backends.cuda.matmul.allow_tf32 = True - torch.backends.cudnn.allow_tf32 = True - except ImportError: - pass - - -def main() -> int: - preload_cudnn() - enable_tf32() - - from PySide6.QtWidgets import QApplication - from gui.theme import DARK_STYLESHEET - from gui.main_window import MainWindow - - app = QApplication(sys.argv) - app.setApplicationName("Audio Transcriber") - app.setStyle("Fusion") - app.setStyleSheet(DARK_STYLESHEET) - - window = MainWindow() - window.show() - return app.exec() - - -if __name__ == "__main__": - sys.exit(main()) - - -# ========================================= -# Файл: LICENSE -# ========================================= - -MIT License - -Copyright (c) 2026 jads - -Permission is hereby granted, free of charge, to any person obtaining a copy of this software and -associated documentation files (the "Software"), to deal in the Software without restriction, including -without limitation the rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell -copies of the Software, and to permit persons to whom the Software is furnished to do so, subject to the -following conditions: - -The above copyright notice and this permission notice shall be included in all copies or substantial -portions of the Software. - -THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT -LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO -EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER -IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE -USE OR OTHER DEALINGS IN THE SOFTWARE. - - -# ========================================= -# Файл: core/separate.py -# ========================================= - -"""Vocal separation via unblend (Mel-Band RoFormer).""" -from __future__ import annotations - -import os -import subprocess -import time -from pathlib import Path -from typing import Callable, Optional - - -def _build_env(custom_kernels: bool) -> dict: - env = os.environ.copy() - env["UNBLEND_CUSTOM_KERNELS"] = "1" if custom_kernels else "0" - return env - - -def _check_result(result: subprocess.CompletedProcess, what: str) -> None: - if result.returncode != 0: - raise RuntimeError( - f"unblend failed ({what}):\n{result.stderr.strip()[:1200]}" - ) - - -def separate_vocals( - input_wav: Path, - work_dir: Path, - model: str = "bs_roformer_anvuew", - isolate_stem: str = "vocals", - unblend_path: str = "unblend", - custom_kernels: bool = False, - device: str = "cuda", -) -> Path: - work_dir.mkdir(parents=True, exist_ok=True) - output_pattern = work_dir / "{track}" / "{stem}.wav" - cmd = [ - unblend_path, "separate", str(input_wav), - "-m", model, - "-o", str(output_pattern), - "--isolate-stem", isolate_stem, - "--device", device, - ] - result = subprocess.run(cmd, capture_output=True, text=True, env=_build_env(custom_kernels)) - _check_result(result, input_wav.name) - - candidates = sorted(work_dir.rglob(f"{isolate_stem}.wav")) - if not candidates: - raise RuntimeError( - f"unblend finished but stem '{isolate_stem}.wav' was not found in {work_dir}" - ) - return candidates[0] - - -def separate_vocals_batch( - input_wavs: list[Path], - work_root: Path, - model: str = "bs_roformer_anvuew", - isolate_stem: str = "vocals", - unblend_path: str = "unblend", - custom_kernels: bool = False, - device: str = "cuda", - progress_callback: Optional[Callable[[int], None]] = None, -) -> dict[Path, Path]: - """Separate many files in one unblend invocation. - - unblend doesn't emit per-file progress, so progress_callback is called - with an incrementing counter roughly every 2 seconds, up to N-1. - The final N is always reported after the process exits. - """ - if not input_wavs: - return {} - - work_root.mkdir(parents=True, exist_ok=True) - output_pattern = work_root / "{parent}" / "{stem}.wav" - - cmd = [ - unblend_path, "separate", - *[str(p) for p in input_wavs], - "-m", model, - "-o", str(output_pattern), - "--isolate-stem", isolate_stem, - "--device", device, - ] - - proc = subprocess.Popen( - cmd, - stdout=subprocess.DEVNULL, - stderr=subprocess.DEVNULL, - env=_build_env(custom_kernels), - ) - - total = len(input_wavs) - processed = 0 - while proc.poll() is None: - if progress_callback and processed < total - 1: - if proc.wait(2.0) is not None: - break - processed += 1 - progress_callback(processed) - - if proc.returncode != 0: - raise RuntimeError(f"unblend batch failed with exit code {proc.returncode}") - - if progress_callback: - progress_callback(total) - - mapping: dict[Path, Path] = {} - missing: list[Path] = [] - for src in input_wavs: - candidate = src.parent / f"{isolate_stem}.wav" - if candidate.exists(): - mapping[src] = candidate - else: - missing.append(src) - - if missing: - found = sorted(work_root.rglob(f"{isolate_stem}.wav")) - if len(found) == len(input_wavs): - for src, dst in zip(input_wavs, found): - mapping[src] = dst - else: - names = ", ".join(p.name for p in missing[:3]) - raise RuntimeError( - f"unblend finished but outputs missing for {len(missing)} file(s): {names}" - ) - - return mapping - - -# ========================================= -# Файл: core/init.py -# ========================================= - -"""Core pipeline: convert, separate, transcribe.""" - - -# ========================================= -# Файл: core/convert.py -# ========================================= - -"""Convert any audio file to a WAV with chosen sample rate and channels via ffmpeg.""" -from __future__ import annotations - -import subprocess -from pathlib import Path - - -def convert_to_wav( - input_path: Path, - output_wav: Path, - sample_rate: int = 44100, - channels: int = 2, - ffmpeg_path: str = "ffmpeg", -) -> None: - """Run ffmpeg to produce a WAV file. Raises RuntimeError on ffmpeg failure.""" - output_wav.parent.mkdir(parents=True, exist_ok=True) - cmd = [ - ffmpeg_path, "-y", "-hide_banner", "-loglevel", "error", - "-i", str(input_path), - "-ar", str(sample_rate), - "-ac", str(channels), - str(output_wav), - ] - result = subprocess.run(cmd, capture_output=True, text=True) - if result.returncode != 0: - raise RuntimeError( - f"ffmpeg failed for {input_path.name}:\n{result.stderr.strip()[:800]}" - ) - - -# ========================================= -# Файл: core/metadata.py -# ========================================= - -"""Read audio file tags via mutagen. Always safe — never raises.""" -from __future__ import annotations - -from dataclasses import dataclass -from pathlib import Path -from typing import Optional - - -@dataclass -class TrackMetadata: - title: Optional[str] = None - artist: Optional[str] = None - album: Optional[str] = None - duration: Optional[float] = None - has_identity: bool = False - - -def read_metadata(path: Path) -> TrackMetadata: - """Extract title/artist/album/duration from tags. Never raises.""" - try: - from mutagen import File - except ImportError: - return TrackMetadata() - - try: - audio = File(str(path), easy=True) - except Exception: - return TrackMetadata() - - if audio is None: - return TrackMetadata() - - def _first(key: str) -> Optional[str]: - v = audio.get(key) - if isinstance(v, list): - return v[0] if v else None - if isinstance(v, str): - return v - return None - - title = _first("title") - artist = _first("artist") - album = _first("album") - - duration: Optional[float] = None - try: - if audio.info is not None: - duration = float(audio.info.length) - except Exception: - pass - - return TrackMetadata( - title=title, - artist=artist, - album=album, - duration=duration, - has_identity=bool(title and artist), - ) - - -# ========================================= -# Файл: core/pipeline.py -# ========================================= - -"""Orchestrate identify -> convert -> separate -> transcribe -> lyrics-match. - -Pipeline order per track (short-circuits as soon as text is found): - 1. metadata tags -> LRCLIB search (if `use_tags_first`) - 2. Shazam fingerprint (if enabled) - 3. Shazam result -> LRCLIB search (if `use_shazam_result`) - 4. full pipeline: convert -> separate -> transcribe - 5. post-ASR: LRCLIB search by recognized text (fuzzy match) - -Two modes: - - sequential: run all of the above per file - - staged: identification pass first, then convert/separate/transcribe - passes only for unresolved files, then post-ASR lyrics pass. -""" -from __future__ import annotations - -import difflib -import json -import shutil -import tempfile -import threading -from concurrent.futures import ThreadPoolExecutor, as_completed -import re -from dataclasses import dataclass, field -from pathlib import Path -from typing import Callable, Optional - -from .config import PipelineConfig -from .convert import convert_to_wav -from .lyrics import LyricsMatch, search_by_identity, search_by_text -from .metadata import TrackMetadata, read_metadata -from .recognize import ShazamResult, recognize -from .separate import separate_vocals, separate_vocals_batch -from .transcribe import transcribe - - -AUDIO_EXTENSIONS = { - ".mp3", ".wav", ".flac", ".ogg", ".m4a", ".aac", ".opus", ".wma", -} - -LogFn = Callable[[str], None] -ProgressFn = Callable[[int, int, str], None] - - -@dataclass -class FileResult: - input_path: Path - text: Optional[str] - output_dir: Optional[Path] - error: Optional[str] = None - source: Optional[str] = None - metadata: Optional[dict] = field(default=None) - - -class Cancelled(Exception): - """Raised when the stop event is set.""" - - -# ---------------------------------------------------------------- helpers - -def collect_audio_files(input_dir: Path, recursive: bool) -> list[Path]: - it = input_dir.rglob("*") if recursive else input_dir.iterdir() - return sorted(p for p in it if p.is_file() and p.suffix.lower() in AUDIO_EXTENSIONS) - - -def _common_parent(paths: list[Path]) -> Path: - """Return the deepest common parent directory of the given paths.""" - if not paths: - raise ValueError("empty paths") - parts = [p.resolve().parts for p in paths] - common = [] - for chunk in zip(*parts): - if all(c == chunk[0] for c in chunk): - common.append(chunk[0]) - else: - break - if not common: - return paths[0].parent - return Path(*common) - - -def _check_cancel(stop_event: Optional[threading.Event]) -> None: - if stop_event is not None and stop_event.is_set(): - raise Cancelled() - - -def _text_head(text: str, n_words: int = 8) -> str: - return " ".join(text.split()[:n_words]) - - -def _build_metadata_dict( - metadata: TrackMetadata, - shazam: Optional[ShazamResult], - source: Optional[str], -) -> Optional[dict]: - d: dict = {} - if metadata.has_identity: - d["tags"] = { - "title": metadata.title, - "artist": metadata.artist, - "album": metadata.album, - "duration": metadata.duration, - } - if shazam: - d["shazam"] = { - "title": shazam.title, - "artist": shazam.artist, - "album": shazam.album, - "year": shazam.year, - } - if source: - d["source"] = source - return d if d else None - - -def _write_metadata_json(path: Path, data: Optional[dict]) -> None: - if not data: - return - path.parent.mkdir(parents=True, exist_ok=True) - path.write_text( - json.dumps(data, indent=2, ensure_ascii=False), - encoding="utf-8", - ) - - -@dataclass -class IdentifyResult: - metadata: TrackMetadata - shazam: Optional[ShazamResult] = None - text: Optional[str] = None - source: Optional[str] = None - needs_asr: bool = True - - -def _identify_file( - input_path: Path, - work_dir: Path, - config: PipelineConfig, - log: LogFn, -) -> IdentifyResult: - """Run cheap identification steps. May populate .text and .source, - and set .needs_asr=False if the track is fully resolved.""" - result = IdentifyResult(metadata=read_metadata(input_path)) - - # --- tags -> LRCLIB --- - if ( - config.lyrics.enabled - and config.lyrics.use_tags_first - and result.metadata.has_identity - ): - log(" LRCLIB по тегам...") - match = search_by_identity( - title=result.metadata.title, - artist=result.metadata.artist, - album=result.metadata.album, - duration=result.metadata.duration, - user_agent=config.lyrics.user_agent, - timeout=config.lyrics.timeout, - ) - if match: - result.text = match.lyrics - result.source = "tag+lrclib" - result.needs_asr = False - log(f" ✓ {match.artist} — {match.title}") - return result - - # --- Shazam --- - if config.shazam.enabled: - log(" Shazam...") - shazam_wav = work_dir / "shazam_16k.wav" - try: - convert_to_wav( - input_path, shazam_wav, - sample_rate=16000, channels=1, - ffmpeg_path=config.convert.ffmpeg_path, - ) - result.shazam = recognize(shazam_wav) - except Exception as e: # noqa: BLE001 - log(f" ⚠ Shazam error: {e}") - result.shazam = None - - if result.shazam: - log(f" ✓ {result.shazam.artist} — {result.shazam.title}") - - if config.lyrics.enabled and config.lyrics.use_shazam_result: - match = search_by_identity( - title=result.shazam.title, - artist=result.shazam.artist, - album=result.shazam.album, - duration=result.metadata.duration, - user_agent=config.lyrics.user_agent, - timeout=config.lyrics.timeout, - ) - if match: - result.text = match.lyrics - result.source = "shazam+lrclib" - result.needs_asr = False - log(" ✓ текст найден через LRCLIB") - return result - - # Identified, but no lyrics - result.source = "shazam" - if not config.lyrics.transcribe_if_identified_no_lyrics: - result.needs_asr = False - log(" ⚠ текста нет") - else: - log(" ⚠ не распознан") - - return result - - -def _try_post_asr_match( - text: str, - config: PipelineConfig, - log: LogFn, -) -> Optional[str]: - if not text: - return None - - # Filter out common ASR watermarks/credits - ignore_phrases = ["субтитры", "редактор", "перевод", "сделал", "озвучка"] - lines = [] - for line in text.splitlines(): - line = line.strip() - if not line: - continue - if any(p in line.lower() for p in ignore_phrases): - continue - lines.append(line) - - if not lines: - return None - - # Take the first line with at least 3 words, strip punctuation - query = "" - for line in lines: - words = line.split() - if len(words) >= 3: - clean_words = [re.sub(r'[^\w\s]', '', w) for w in words] - query = " ".join(clean_words[:6]) - break - - if not query: - return None - - log(f" LRCLIB по тексту: {query!r}...") - - try: - candidates = search_by_text( - query, limit=5, - user_agent=config.lyrics.user_agent, - timeout=config.lyrics.timeout, - ) - except Exception as e: # noqa: BLE001 - log(f" ⚠ LRCLIB error: {e}") - return None - - if not candidates: - log(" ⚠ нет кандидатов") - return None - - best: Optional[LyricsMatch] = None - best_ratio = 0.0 - for c in candidates: - ratio = difflib.SequenceMatcher(None, text.lower(), c.lyrics.lower()).ratio() - if ratio > best_ratio: - best_ratio = ratio - best = c - - if best is not None and best_ratio >= config.lyrics.fuzzy_threshold: - log(f" ✓ совпадение {best_ratio:.2f}: {best.artist} — {best.title}") - return best.lyrics - - log(f" ⚠ лучшее совпадение {best_ratio:.2f} < {config.lyrics.fuzzy_threshold}") - return None - - -def _persist_artifacts( - work_dir: Path, - persist_dir: Path, - config: PipelineConfig, - has_text: bool, - has_metadata: bool, - log: LogFn, -) -> None: - """Move wanted artifacts from work_dir to persist_dir.""" - persist_dir.mkdir(parents=True, exist_ok=True) - - names: list[str] = [] - if has_metadata and (work_dir / "metadata.json").exists(): - names.append("metadata.json") - if has_text and (work_dir / "text.txt").exists(): - names.append("text.txt") - - if config.keep_intermediate: - for n in ("convert.wav", "vocals.wav"): - if (work_dir / n).exists(): - names.append(n) - elif not has_text: - if config.separate.enabled and (work_dir / "vocals.wav").exists(): - names.append("vocals.wav") - elif config.convert.enabled and (work_dir / "convert.wav").exists(): - names.append("convert.wav") - - for name in names: - src = work_dir / name - dst = persist_dir / name - if dst.exists(): - dst.unlink() - shutil.move(str(src), str(dst)) - log(f" [persist] {name}") - - -# ---------------------------------------------------------------- entry - -def _run_identification_parallel( - config: PipelineConfig, - files: list[Path], - log: LogFn, - on_progress: Optional[ProgressFn], - stop_event: Optional[threading.Event], - work_root: Path, -) -> dict[Path, Optional[IdentifyResult]]: - """Parallel identification pass (tags → LRCLIB, Shazam, etc.).""" - if not (config.shazam.enabled or config.lyrics.enabled): - return {} - - concurrency = max(1, int(config.shazam.identify_concurrency)) - total = len(files) - - if concurrency == 1 or total == 1: - results: dict[Path, Optional[IdentifyResult]] = {} - for i, f in enumerate(files, 1): - if stop_event is not None and stop_event.is_set(): - break - if on_progress: - on_progress(i, total, "identify") - try: - results[f] = _identify_file(f, work_root / f.stem, config, log) - except Exception as e: # noqa: BLE001 - log(f" ✗ {f.name} (identify): {e}") - results[f] = None - return results - - log(f" Идентификация: {total} файлов, конкурентность {concurrency}") - - def worker(f: Path) -> tuple[Path, Optional[IdentifyResult]]: - if stop_event is not None and stop_event.is_set(): - return f, None - - def prefixed(msg: str) -> None: - log(f" [{f.name}] {msg.strip()}") - - try: - ident = _identify_file(f, work_root / f.stem, config, prefixed) - return f, ident - except Exception as e: # noqa: BLE001 - log(f" ✗ {f.name} (identify): {e}") - return f, None - - results: dict[Path, Optional[IdentifyResult]] = {} - with ThreadPoolExecutor(max_workers=concurrency) as pool: - futures = {pool.submit(worker, f): f for f in files} - done = 0 - for fut in as_completed(futures): - f = futures[fut] - try: - _, ident = fut.result() - results[f] = ident - except Exception as e: # noqa: BLE001 - log(f" ✗ {f.name} (identify): {e}") - results[f] = None - done += 1 - if on_progress: - on_progress(done, total, "identify") - - return results - - -def _run_convert_parallel( - config: PipelineConfig, - files: list[Path], - state: dict, - log: LogFn, - on_progress: Optional[ProgressFn], - stop_event: Optional[threading.Event], -) -> None: - """Parallel convert stage. Writes state[f]['audio_path'] or state[f]['error'].""" - concurrency = max(1, int(config.convert.convert_concurrency)) - total = len(files) - - def worker(f: Path) -> tuple[Path, Optional[Path], Optional[str]]: - if stop_event is not None and stop_event.is_set(): - return f, None, None - try: - target = Path(tempfile.mkdtemp(prefix=f"at_{f.stem}_")) / "convert.wav" - if config.overwrite or not target.exists(): - convert_to_wav( - f, target, - sample_rate=config.convert.sample_rate, - channels=config.convert.channels, - ffmpeg_path=config.convert.ffmpeg_path, - ) - return f, target, None - except Exception as e: # noqa: BLE001 - return f, None, f"{type(e).__name__}: {e}" - - if concurrency == 1 or total == 1: - for i, f in enumerate(files, 1): - if stop_event is not None and stop_event.is_set(): - break - if on_progress: - on_progress(i, total, "convert") - f2, target, err = worker(f) - if err: - state[f2]["error"] = err - log(f" ✗ {f2.name}: {err}") - else: - state[f2]["audio_path"] = target - log(f" ✓ {f2.name}") - return - - log(f" Конвертация: {total} файлов, конкурентность {concurrency}") - with ThreadPoolExecutor(max_workers=concurrency) as pool: - futures = {pool.submit(worker, f): f for f in files} - done = 0 - for fut in as_completed(futures): - f2, target, err = fut.result() - if err: - state[f2]["error"] = err - log(f" ✗ {f2.name}: {err}") - else: - state[f2]["audio_path"] = target - log(f" ✓ {f2.name}") - done += 1 - if on_progress: - on_progress(done, total, "convert") - -def run_pipeline( - config: PipelineConfig, - log: LogFn = print, - on_progress: Optional[ProgressFn] = None, - stop_event: Optional[threading.Event] = None, -) -> list[FileResult]: - input_dir = Path(config.input_dir) - if not input_dir.is_dir(): - raise FileNotFoundError(f"Input directory not found: {input_dir}") - - output_dir = Path(config.output_dir) - output_dir.mkdir(parents=True, exist_ok=True) - - files = collect_audio_files(input_dir, config.recursive) - if not files: - log("Аудиофайлы не найдены.") - return [] - - mode = (config.mode or "sequential").lower() - log( - f"Найдено треков: {len(files)} | режим: {mode} | " - f"keep_intermediate={config.keep_intermediate}" - ) - if config.shazam.enabled: - log(" Shazam: включён") - if config.lyrics.enabled: - log(f" LRCLIB: включён (порог fuzzy={config.lyrics.fuzzy_threshold:.2f})") - - if mode == "staged": - return _run_staged(config, files, log, on_progress, stop_event) - return _run_sequential(config, files, log, on_progress, stop_event) - - -# ---------------------------------------------------------------- sequential - -def _run_sequential( - config: PipelineConfig, - files: list[Path], - log: LogFn, - on_progress: Optional[ProgressFn], - stop_event: Optional[threading.Event], -) -> list[FileResult]: - total = len(files) - results: list[FileResult] = [] - output_root = Path(config.output_dir) - - for idx, f in enumerate(files, 1): - if stop_event is not None and stop_event.is_set(): - log("⏹ Отменено") - break - if on_progress: - on_progress(idx - 1, total, "file") - log(f"\n▶ [{idx}/{total}] {f.name}") - - tmp_root = Path(tempfile.mkdtemp(prefix=f"at_{f.stem}_")) - track_out = output_root / f.stem - track_out.mkdir(parents=True, exist_ok=True) - - text: Optional[str] = None - source: Optional[str] = None - error: Optional[str] = None - metadata_dict: Optional[dict] = None - - try: - # --- Phase 1: identification --- - ident = _identify_file(f, tmp_root, config, log) - - if not ident.needs_asr: - text = ident.text - source = ident.source - metadata_dict = _build_metadata_dict(ident.metadata, ident.shazam, source) - if text: - (tmp_root / "text.txt").write_text(text, encoding="utf-8") - log(f" ✓ text.txt ({len(text)} символов, {source})") - if metadata_dict: - _write_metadata_json(tmp_root / "metadata.json", metadata_dict) - log(" ✓ metadata.json") - _persist_artifacts( - tmp_root, track_out, config, - has_text=text is not None, - has_metadata=bool(metadata_dict), - log=log, - ) - results.append(FileResult(f, text, track_out, None, source, metadata_dict)) - continue - - # --- Phase 2: full pipeline --- - audio_path: Path = f - if config.convert.enabled: - _check_cancel(stop_event) - log(" [1] Конвертация в WAV...") - target = tmp_root / "convert.wav" - if config.overwrite or not target.exists(): - convert_to_wav( - f, target, - sample_rate=config.convert.sample_rate, - channels=config.convert.channels, - ffmpeg_path=config.convert.ffmpeg_path, - ) - audio_path = target - log(" ✓ convert.wav") - - if config.separate.enabled: - _check_cancel(stop_event) - log(" [2] Отделение вокала...") - target = tmp_root / "vocals.wav" - with tempfile.TemporaryDirectory() as tmp: - raw = separate_vocals( - audio_path, Path(tmp), - model=config.separate.model, - isolate_stem=config.separate.isolate_stem, - unblend_path=config.separate.unblend_path, - custom_kernels=config.separate.custom_kernels, - ) - shutil.move(str(raw), str(target)) - audio_path = target - log(" ✓ vocals.wav") - - if config.transcribe.enabled: - _check_cancel(stop_event) - log(" [3] Транскрибация...") - text = transcribe( - audio_path, - engine=config.transcribe.engine, - model=config.transcribe.model, - language=config.transcribe.language, - device=config.transcribe.device, - compute_type=config.transcribe.compute_type, - batch_size=config.transcribe.batch_size, - gigaam_provider=config.transcribe.gigaam_provider, - local_files_only=config.transcribe.local_files_only, - ) - source = config.transcribe.engine - log(f" ✓ распознано {len(text)} символов") - - if config.lyrics.enabled and config.lyrics.try_after_asr: - matched = _try_post_asr_match(text, config, log) - if matched: - text = matched - source = f"{source}+lrclib" - - (tmp_root / "text.txt").write_text(text, encoding="utf-8") - - metadata_dict = _build_metadata_dict(ident.metadata, ident.shazam, source) - if metadata_dict: - _write_metadata_json(tmp_root / "metadata.json", metadata_dict) - - _persist_artifacts( - tmp_root, track_out, config, - has_text=text is not None, - has_metadata=bool(metadata_dict), - log=log, - ) - - except Cancelled: - log("⏹ Отменено") - break - except Exception as e: # noqa: BLE001 - error = f"{type(e).__name__}: {e}" - log(f" ✗ Ошибка: {error}") - finally: - shutil.rmtree(tmp_root, ignore_errors=True) - if on_progress: - on_progress(idx, total, "file") - results.append( - FileResult(f, text, track_out if not error else None, error, source, metadata_dict) - ) - - _log_summary(results, log) - return results - - -# ---------------------------------------------------------------- staged - -def _run_staged( - config: PipelineConfig, - files: list[Path], - log: LogFn, - on_progress: Optional[ProgressFn], - stop_event: Optional[threading.Event], -) -> list[FileResult]: - output_root = Path(config.output_dir) - total = len(files) - - state: dict[Path, dict] = { - f: { - "audio_path": f, - "ident": None, - "text": None, - "source": None, - "metadata_dict": None, - "error": None, - "resolved": False, - } - for f in files - } - - def active() -> list[Path]: - return [f for f in files if state[f]["error"] is None and not state[f]["resolved"]] - - def cancelled() -> bool: - return stop_event is not None and stop_event.is_set() - - # --- Stage 0: identification --- - if config.shazam.enabled or config.lyrics.enabled: - log(f"\n══ Этап 0 · Идентификация ({total} файлов) ══") - ident_tmp = Path(tempfile.mkdtemp(prefix="at_ident_")) - try: - for i, f in enumerate(files, 1): - if cancelled(): - log("⏹ Отменено") - break - if on_progress: - on_progress(i, total, "identify") - try: - ident = _identify_file(f, ident_tmp / f.stem, config, log) - state[f]["ident"] = ident - - if not ident.needs_asr: - state[f]["resolved"] = True - state[f]["text"] = ident.text - state[f]["source"] = ident.source - state[f]["metadata_dict"] = _build_metadata_dict( - ident.metadata, ident.shazam, ident.source, - ) - # Persist resolved tracks right away - track_out = output_root / f.stem - track_out.mkdir(parents=True, exist_ok=True) - if ident.text: - (track_out / "text.txt").write_text(ident.text, encoding="utf-8") - if state[f]["metadata_dict"]: - _write_metadata_json(track_out / "metadata.json", state[f]["metadata_dict"]) - log(f" ✓ {f.name} [{ident.source}]") - else: - log(f" → {f.name} (нужен полный пайплайн)") - except Exception as e: # noqa: BLE001 - state[f]["error"] = f"{type(e).__name__}: {e}" - log(f" ✗ {f.name}: {e}") - finally: - shutil.rmtree(ident_tmp, ignore_errors=True) - - # --- Stage 1: convert --- - if config.convert.enabled and not cancelled(): - todo = active() - log(f"\n══ Этап 1 · Конвертация ({len(todo)} файлов) ══") - for i, f in enumerate(todo, 1): - if cancelled(): - log("⏹ Отменено") - break - if on_progress: - on_progress(i, len(todo), "convert") - try: - target = Path(tempfile.mkdtemp(prefix=f"at_{f.stem}_")) / "convert.wav" - if config.overwrite or not target.exists(): - convert_to_wav( - f, target, - sample_rate=config.convert.sample_rate, - channels=config.convert.channels, - ffmpeg_path=config.convert.ffmpeg_path, - ) - state[f]["audio_path"] = target - log(f" ✓ {f.name}") - except Exception as e: # noqa: BLE001 - state[f]["error"] = f"{type(e).__name__}: {e}" - log(f" ✗ {f.name}: {e}") - - # ---- Stage 2: separate (batched) ---- - if config.separate.enabled and not cancelled(): - todo = active() - log(f"\n══ Этап 2/3 · Отделение вокала ({len(todo)} файлов) ══") - if not todo: - pass - elif len(todo) == 1: - # Single file — no batching advantage - f = todo[0] - if on_progress: - on_progress(1, 1, "separate") - try: - src = state[f]["audio_path"] - target = src.parent / "vocals.wav" - with tempfile.TemporaryDirectory() as tmp: - raw = separate_vocals( - src, Path(tmp), - model=config.separate.model, - isolate_stem=config.separate.isolate_stem, - unblend_path=config.separate.unblend_path, - custom_kernels=config.separate.custom_kernels, - ) - shutil.move(str(raw), str(target)) - state[f]["audio_path"] = target - log(f" ✓ {f.name}") - except Exception as e: # noqa: BLE001 - state[f]["error"] = f"{type(e).__name__}: {e}" - log(f" ✗ {f.name}: {e}") - else: - # Batch: one unblend invocation, model loaded once - input_wavs = [state[f]["audio_path"] for f in todo] - common_root = _common_parent(input_wavs) - log(f" Батч: {len(todo)} файлов, один вызов unblend (модель загрузится один раз)") - if on_progress: - on_progress(0, len(todo), "separate") - try: - mapping = separate_vocals_batch( - input_wavs, - common_root, - model=config.separate.model, - isolate_stem=config.separate.isolate_stem, - unblend_path=config.separate.unblend_path, - custom_kernels=config.separate.custom_kernels, - progress_callback=lambda p: on_progress(p, len(todo), "separate") if on_progress else None, - ) - for f in todo: - src = state[f]["audio_path"] - vocal = mapping.get(src) - if vocal is None: - state[f]["error"] = "batch produced no output" - log(f" ✗ {f.name}: нет результата в батче") - continue - state[f]["audio_path"] = vocal - log(f" ✓ {f.name}") - if on_progress: - on_progress(len(todo), len(todo), "separate") - except Exception as e: # noqa: BLE001 - log(f" ✗ Батч упал: {e}") - log(" ↻ Fallback: по одному...") - for i, f in enumerate(todo, 1): - if cancelled(): - break - if on_progress: - on_progress(i, len(todo), "separate") - try: - src = state[f]["audio_path"] - target = src.parent / "vocals.wav" - with tempfile.TemporaryDirectory() as tmp: - raw = separate_vocals( - src, Path(tmp), - model=config.separate.model, - isolate_stem=config.separate.isolate_stem, - unblend_path=config.separate.unblend_path, - custom_kernels=config.separate.custom_kernels, - ) - shutil.move(str(raw), str(target)) - state[f]["audio_path"] = target - log(f" ✓ {f.name}") - except Exception as inner: # noqa: BLE001 - state[f]["error"] = f"{type(inner).__name__}: {inner}" - log(f" ✗ {f.name}: {inner}") - - # --- Stage 3: transcribe --- - if config.transcribe.enabled and not cancelled(): - todo = active() - log(f"\n══ Этап 3 · Транскрибация ({len(todo)} файлов) ══") - for i, f in enumerate(todo, 1): - if cancelled(): - log("⏹ Отменено") - break - if on_progress: - on_progress(i, len(todo), "transcribe") - try: - text = transcribe( - state[f]["audio_path"], - engine=config.transcribe.engine, - model=config.transcribe.model, - language=config.transcribe.language, - device=config.transcribe.device, - compute_type=config.transcribe.compute_type, - batch_size=config.transcribe.batch_size, - gigaam_provider=config.transcribe.gigaam_provider, - local_files_only=config.transcribe.local_files_only, - ) - state[f]["text"] = text - state[f]["source"] = config.transcribe.engine - log(f" ✓ {f.name} ({len(text)} символов)") - except Exception as e: # noqa: BLE001 - state[f]["error"] = f"{type(e).__name__}: {e}" - log(f" ✗ {f.name}: {e}") - - # --- Stage 4: post-ASR lyrics match + persist --- - if not cancelled(): - log("\n══ Финализация ══") - for f in files: - s = state[f] - if s["error"] is not None: - continue - if s["resolved"]: - continue # already persisted in identification pass - - text = s["text"] - source = s["source"] - - if text and config.lyrics.enabled and config.lyrics.try_after_asr: - matched = _try_post_asr_match(text, config, log) - if matched: - text = matched - source = f"{source}+lrclib" - s["text"] = text - s["source"] = source - - ident = s["ident"] - metadata_dict = _build_metadata_dict( - ident.metadata, ident.shazam, source, - ) if ident else None - s["metadata_dict"] = metadata_dict - - track_out = output_root / f.stem - track_out.mkdir(parents=True, exist_ok=True) - - audio_work = s["audio_path"] - work_dir = audio_work.parent if isinstance(audio_work, Path) else track_out - if text: - (work_dir / "text.txt").write_text(text, encoding="utf-8") - if metadata_dict: - _write_metadata_json(work_dir / "metadata.json", metadata_dict) - - _persist_artifacts( - work_dir, track_out, config, - has_text=text is not None, - has_metadata=bool(metadata_dict), - log=log, - ) - log(f" ✓ {f.name}") - - # cleanup temp dirs - for f in files: - audio_path = state[f]["audio_path"] - if isinstance(audio_path, Path) and audio_path.parent.name.startswith(f"at_{f.stem}"): - shutil.rmtree(audio_path.parent, ignore_errors=True) - - results = [ - FileResult( - f, - state[f]["text"], - output_root / f.stem if state[f]["error"] is None else None, - state[f]["error"], - state[f]["source"], - state[f]["metadata_dict"], - ) - for f in files - ] - _log_summary(results, log) - return results - - -# ---------------------------------------------------------------- summary - -def _log_summary(results: list[FileResult], log: LogFn) -> None: - from collections import Counter - counts = Counter(r.source or ("error" if r.error else "no_result") for r in results) - log("\n══ Итог ══") - for src, n in sorted(counts.items(), key=lambda kv: -kv[1]): - log(f" {src}: {n}") - - -# ========================================= -# Файл: core/transcribe.py -# ========================================= - -"""Speech recognition via WhisperX or GigaAM (onnx-asr). - -Includes aggressive VRAM cleanup to prevent OOM on long batches, -and automatic model reload if the CUDA context gets corrupted. -""" -from __future__ import annotations - -import gc -import time -from pathlib import Path -from typing import Optional - - -def _resolve_device(device: str) -> str: - if device in ("cuda", "cpu"): - return device - try: - import torch - return "cuda" if torch.cuda.is_available() else "cpu" - except Exception: - return "cpu" - - -def _retry(fn, attempts: int = 3, base_delay: float = 2.0): - last = None - for i in range(attempts): - try: - return fn() - except Exception as e: - last = e - if i < attempts - 1: - time.sleep(base_delay * (i + 1)) - raise last - - -class _WhisperXModels: - _cache: dict = {} - - @classmethod - def get(cls, model_name: str, device: str, compute_type: Optional[str], - local_files_only: bool = False): - if compute_type is None: - compute_type = "int8_float16" if device == "cuda" else "int8" - key = (model_name, device, compute_type, local_files_only) - if key not in cls._cache: - import whisperx - print(f" [ASR] loading WhisperX {model_name} on {device} " - f"({compute_type}, local_files_only={local_files_only})...") - - def load(): - return whisperx.load_model( - model_name, device, - compute_type=compute_type, - local_files_only=local_files_only, - ) - - cls._cache[key] = _retry(load) - print(" [ASR] model loaded.") - return cls._cache[key] - - -class _GigaAMModels: - _cache: dict = {} - - @classmethod - def get(cls, provider: str): - providers = { - "cuda": ["CUDAExecutionProvider"], - "cpu": ["CPUExecutionProvider"], - "auto": ["CUDAExecutionProvider", "CPUExecutionProvider"], - }.get(provider, ["CPUExecutionProvider"]) - key = tuple(providers) - if key not in cls._cache: - import onnx_asr - - def load(): - return onnx_asr.load_model("gigaam-v3-ctc", providers=list(providers)) - - cls._cache[key] = _retry(load) - return cls._cache[key] - - -def transcribe( - audio_path: Path, - engine: str = "whisperx", - model: str = "large-v3", - language: Optional[str] = "ru", - device: str = "auto", - compute_type: Optional[str] = None, - batch_size: int = 8, - gigaam_provider: str = "auto", - local_files_only: bool = False, - vad_filter: bool = True, -) -> str: - device = _resolve_device(device) - - if engine == "whisperx": - try: - return _transcribe_whisperx( - audio_path, model, language, device, compute_type, - batch_size, local_files_only, vad_filter, - ) - except Exception as e: - err = str(e).lower() - if "out of memory" in err or "cuda" in err or "device ordinal" in err: - print(f" [ASR] CUDA error: {e}. Clearing model cache and retrying...") - _WhisperXModels._cache.clear() - gc.collect() - try: - import torch - if torch.cuda.is_available(): - torch.cuda.empty_cache() - except ImportError: - pass - return _transcribe_whisperx( - audio_path, model, language, device, compute_type, - batch_size, local_files_only, vad_filter, - ) - raise - - if engine == "gigaam": - return _transcribe_gigaam(audio_path, gigaam_provider) - raise ValueError(f"Unknown ASR engine: {engine!r}") - - -def _transcribe_whisperx( - audio_path: Path, - model_name: str, - language: Optional[str], - device: str, - compute_type: Optional[str], - batch_size: int, - local_files_only: bool, - vad_filter: bool, -) -> str: - import whisperx - model = _WhisperXModels.get(model_name, device, compute_type, local_files_only) - audio = whisperx.load_audio(str(audio_path)) - result = model.transcribe( - audio, - batch_size=batch_size, - language=language, - ) - - text = " ".join(seg["text"].strip() for seg in result["segments"]).strip() - - # Free VRAM aggressively to prevent fragmentation/OOM on long batches - del audio - del result - gc.collect() - try: - import torch - if torch.cuda.is_available(): - torch.cuda.empty_cache() - except ImportError: - pass - - return text - - -def _transcribe_gigaam(audio_path: Path, provider: str) -> str: - model = _GigaAMModels.get(provider) - return model.recognize(str(audio_path), channel="mean").strip() - - -# ========================================= -# Файл: core/recognize.py -# ========================================= - -"""Audio fingerprinting via Shazam (unofficial, no API key needed). - -Wraps shazamlite. If Shazam can't identify the track — return None. -Never raises on 'not found', only on missing dependency. -""" -from __future__ import annotations - -from dataclasses import dataclass -from pathlib import Path -from typing import Optional - - -@dataclass -class ShazamResult: - title: str - artist: str - album: Optional[str] = None - year: Optional[str] = None - - -def recognize(audio_wav: Path) -> Optional[ShazamResult]: - """Run Shazam fingerprinting on a WAV file. Returns None if not found.""" - try: - from shazamlite import Shazam - except ImportError as e: - raise RuntimeError( - "shazamlite не установлен. Установите: pip install shazamlite numpy soundfile" - ) from e - - shazam = Shazam() - try: - track = shazam.recognize(str(audio_wav)) - except Exception: - return None - - if track is None: - return None - - title = getattr(track, "title", None) or getattr(track, "track", None) - artist = getattr(track, "artist", None) or getattr(track, "subtitle", None) - album = getattr(track, "album", None) - year = getattr(track, "year", None) or getattr(track, "released", None) - - if not title or not artist: - return None - - return ShazamResult( - title=str(title), - artist=str(artist), - album=str(album) if album else None, - year=str(year) if year else None, - ) - - -# ========================================= -# Файл: core/lyrics.py -# ========================================= - -"""Lyrics lookup via LRCLIB (https://lrclib.net) — free, no API key. - -Rate limit: LRCLIB allows ~90 req/min with a proper User-Agent, -~45 without. We throttle to 0.4s between requests (150/min theoretical, -well within limits) and always send a UA. -""" -from __future__ import annotations - -import threading -import time -from dataclasses import dataclass -from typing import Optional - -import requests - - -DEFAULT_USER_AGENT = "audio-transcriber/1.0" -_API_BASE = "https://lrclib.net/api" - -_throttle_lock = threading.Lock() -_last_request_at = 0.0 -_MIN_INTERVAL = 0.4 # seconds - - -@dataclass -class LyricsMatch: - title: str - artist: str - lyrics: str - duration: Optional[float] = None - - -def _throttle() -> None: - global _last_request_at - with _throttle_lock: - now = time.monotonic() - wait = _last_request_at + _MIN_INTERVAL - now - if wait > 0: - time.sleep(wait) - _last_request_at = time.monotonic() - - -def _get(url: str, params: dict, user_agent: str, timeout: float = 10.0) -> Optional[dict]: - _throttle() - try: - resp = requests.get( - url, params=params, - headers={"User-Agent": user_agent}, - timeout=timeout, - ) - if resp.status_code == 404: - return None - resp.raise_for_status() - return resp.json() - except requests.RequestException: - return None - - -def _make_match(item: dict) -> Optional[LyricsMatch]: - lyrics = item.get("plainLyrics") or item.get("syncedLyrics") - if not lyrics: - return None - return LyricsMatch( - title=item.get("trackName", ""), - artist=item.get("artistName", ""), - lyrics=lyrics.strip(), - duration=item.get("duration"), - ) - - -def search_by_identity( - title: Optional[str], - artist: Optional[str], - album: Optional[str] = None, - duration: Optional[float] = None, - user_agent: str = DEFAULT_USER_AGENT, - timeout: float = 10.0, -) -> Optional[LyricsMatch]: - """Search LRCLIB by track metadata. Works even with messy tags.""" - if not title and not artist: - return None - - query = " ".join(filter(None, [artist or "", title or ""])).strip() - if not query: - return None - - data = _get(f"{_API_BASE}/search", {"q": query}, user_agent, timeout) - if not isinstance(data, list) or not data: - return None - - for item in data[:5]: - match = _make_match(item) - if match: - return match - return None - - -def search_by_text( - query: str, - limit: int = 5, - user_agent: str = DEFAULT_USER_AGENT, - timeout: float = 10.0, -) -> list[LyricsMatch]: - """Search LRCLIB by plain text (e.g. first words of recognized speech).""" - if not query or not query.strip(): - return [] - data = _get(f"{_API_BASE}/search", {"q": query.strip()}, user_agent, timeout) - if not isinstance(data, list): - return [] - matches: list[LyricsMatch] = [] - for item in data[:limit]: - match = _make_match(item) - if match: - matches.append(match) - return matches - - -# ========================================= -# Файл: core/config.py -# ========================================= - -"""Pipeline configuration dataclasses with JSON save/load.""" -from __future__ import annotations - -import json -from dataclasses import asdict, dataclass, field, fields -from pathlib import Path -from typing import Optional - - -MODES = ("sequential", "staged") - - -@dataclass -class ConvertConfig: - enabled: bool = True - sample_rate: int = 44100 - channels: int = 2 - ffmpeg_path: str = "ffmpeg" - convert_concurrency: int = 4 - - -@dataclass -class SeparateConfig: - enabled: bool = True - model: str = "bs_roformer_anvuew" - isolate_stem: str = "vocals" - unblend_path: str = "unblend" - custom_kernels: bool = False - - -@dataclass -class TranscribeConfig: - enabled: bool = True - engine: str = "whisperx" - model: str = "large-v3" - language: Optional[str] = "ru" - device: str = "auto" - compute_type: Optional[str] = None - batch_size: int = 8 - gigaam_provider: str = "auto" - local_files_only: bool = False - - -@dataclass -class ShazamConfig: - enabled: bool = False - identify_concurrency: int = 4 - - -@dataclass -class LyricsConfig: - enabled: bool = False - user_agent: str = "audio-transcriber/1.0" - timeout: float = 10.0 - use_tags_first: bool = True - use_shazam_result: bool = True - try_after_asr: bool = True - fuzzy_threshold: float = 0.6 - transcribe_if_identified_no_lyrics: bool = True - - -@dataclass -class PipelineConfig: - input_dir: str = "" - output_dir: str = "transcribed" - recursive: bool = False - keep_intermediate: bool = True - overwrite: bool = True - mode: str = "sequential" - - convert: ConvertConfig = field(default_factory=ConvertConfig) - separate: SeparateConfig = field(default_factory=SeparateConfig) - transcribe: TranscribeConfig = field(default_factory=TranscribeConfig) - shazam: ShazamConfig = field(default_factory=ShazamConfig) - lyrics: LyricsConfig = field(default_factory=LyricsConfig) - - def to_dict(self) -> dict: - return asdict(self) - - def to_json(self) -> str: - return json.dumps(self.to_dict(), indent=2, ensure_ascii=False) - - @classmethod - def from_dict(cls, data: dict) -> "PipelineConfig": - cfg = cls() - for top in fields(cls): - name = top.name - if name not in data: - continue - value = data[name] - current = getattr(cfg, name) - if hasattr(current, "__dataclass_fields__"): - sub = {f.name for f in fields(current)} - for k, v in value.items(): - if k in sub: - setattr(current, k, v) - else: - setattr(cfg, name, value) - return cfg - - @classmethod - def from_json(cls, text: str) -> "PipelineConfig": - return cls.from_dict(json.loads(text)) - - def save(self, path: Path) -> None: - Path(path).write_text(self.to_json(), encoding="utf-8") - - @classmethod - def load(cls, path: Path) -> "PipelineConfig": - return cls.from_json(Path(path).read_text(encoding="utf-8")) - - -# ========================================= -# Файл: README.md -# ========================================= - -# Audio Transcriber Combine - -Локальное десктоп-приложение для извлечения текста из аудио. - -## Этапы - -1. **convert** — любой аудиоформат → WAV через `ffmpeg` (настраиваемый sample rate / каналы). -2. **separate** — отделение вокала через `unblend` (Mel-Band RoFormer). -3. **transcribe** — ASR через **WhisperX** или **GigaAM** (onnx-asr). - -Каждый этап можно включать/выключать отдельно. Промежуточные файлы -можно сохранять (по умолчанию — да) или удалять, оставляя только финальный -результат последнего включённого этапа. - -## Раскладка результатов - - -# ========================================= -# Файл: requirements.txt -# ========================================= - -# GUI -PySide6>=6.6 - -# ASR -whisperx>=3.1 -onnx-asr>=0.4 - -# Recognition & lyrics -shazamlite>=0.1 -mutagen>=1.47 -requests>=2.31 - -# WhisperX pulls torch; for CUDA you must install the matching torch build, e.g.: -# pip install torch torchaudio --index-url https://download.pytorch.org/whl/cu121 -# For cuDNN preload helper used in main.py: -# pip install nvidia-cudnn-cu12 - -# System dependencies (not pip-installable): -# ffmpeg — must be in PATH -# unblend — vocal separation CLI (Mel-Band RoFormer), must be in PATH - - -# ========================================= -# Файл: gui/main_window.py -# ========================================= - -"""Main application window — dark theme, mode selector, staged/sequential pipelines.""" -from __future__ import annotations - -import shutil -from pathlib import Path -from typing import Optional - -from PySide6.QtCore import Qt, QUrl -from PySide6.QtGui import QColor, QDesktopServices -from PySide6.QtWidgets import ( - QCheckBox, QComboBox, QDoubleSpinBox, QFileDialog, QFormLayout, QGroupBox, - QHBoxLayout, QLabel, QLineEdit, QMainWindow, QMessageBox, - QPlainTextEdit, QProgressBar, QPushButton, QSpinBox, QTabWidget, - QVBoxLayout, QWidget, -) - -from core.config import ( - ConvertConfig, LyricsConfig, PipelineConfig, SeparateConfig, - ShazamConfig, TranscribeConfig, -) -from gui.worker import PipelineWorker - - -LANGUAGES = ["auto", "ru", "en", "de", "fr", "es", "it", "uk", "ja", "zh", "ko"] -SAMPLE_RATES = [16000, 22050, 44100, 48000] -ENGINES = ["whisperx", "gigaam"] - -STAGE_LABELS = { - "file": "Файлы", - "identify": "Идентификация", - "convert": "Конвертация", - "separate": "Отделение", - "transcribe": "Транскрибация", -} - -MODE_LABELS = { - "sequential": "Последовательно (файл за файлом)", - "staged": "Пакетно (по этапам)", -} - - -def _detect_device_info() -> tuple[str, str]: - try: - import torch - except ImportError: - return "⚠ PyTorch не установлен", "#f38ba8" - - if torch.cuda.is_available(): - try: - name = torch.cuda.get_device_name(0) - vram = torch.cuda.get_device_properties(0).total_memory / (1024 ** 3) - return f"✓ CUDA: {name} · {vram:.1f} GB VRAM", "#a6e3a1" - except Exception: # noqa: BLE001 - return "✓ CUDA доступна", "#a6e3a1" - - return "⚠ CUDA не обнаружена — расчёт пойдёт на CPU (медленно)", "#f9e2af" - - -class MainWindow(QMainWindow): - def __init__(self) -> None: - super().__init__() - self.setWindowTitle("Song Transcriber") - self.resize(1140, 880) - self._worker: Optional[PipelineWorker] = None - self._build_ui() - self._apply_defaults() - self._sync_tab_indicators() - self._sync_engine_fields() - self._log_device_info() - - # ---------------- UI ---------------- - - def _build_ui(self) -> None: - central = QWidget() - self.setCentralWidget(central) - root = QVBoxLayout(central) - root.setContentsMargins(18, 16, 18, 16) - root.setSpacing(12) - - header = QHBoxLayout() - title = QLabel("🎵 Song Transcriber") - title.setObjectName("appTitle") - header.addWidget(title) - header.addStretch(1) - self.device_label = QLabel("") - self.device_label.setObjectName("deviceInfo") - header.addWidget(self.device_label) - root.addLayout(header) - - top_box = QGroupBox("Общие настройки") - top_layout = QFormLayout(top_box) - top_layout.setSpacing(10) - - self.input_edit = QLineEdit() - top_layout.addRow("Входная папка:", self._with_browse(self.input_edit, self._pick_input_dir)) - - self.output_edit = QLineEdit() - top_layout.addRow("Выходная папка:", self._with_browse(self.output_edit, self._pick_output_dir)) - - self.mode_cb = QComboBox() - for key, label in MODE_LABELS.items(): - self.mode_cb.addItem(label, key) - top_layout.addRow("Режим:", self.mode_cb) - - opts = QHBoxLayout() - self.recursive_cb = QCheckBox("Рекурсивный обход") - self.keep_intermediate_cb = QCheckBox("Сохранять промежуточные") - self.overwrite_cb = QCheckBox("Перезаписывать") - opts.addWidget(self.recursive_cb) - opts.addWidget(self.keep_intermediate_cb) - opts.addWidget(self.overwrite_cb) - opts.addStretch(1) - top_layout.addRow("Опции:", opts) - - root.addWidget(top_box) - - self.tabs = QTabWidget() - self.tabs.addTab(self._build_shazam_tab(), "Shazam") - self.tabs.addTab(self._build_lrclib_tab(), "LRCLIB") - self.tabs.addTab(self._build_convert_tab(), "Конвертация") - self.tabs.addTab(self._build_separate_tab(), "Отделение вокала") - self.tabs.addTab(self._build_transcribe_tab(), "Транскрибация") - root.addWidget(self.tabs) - - self.shazam_enabled_cb.toggled.connect(self._sync_tab_indicators) - self.lyrics_enabled_cb.toggled.connect(self._sync_tab_indicators) - self.convert_enabled_cb.toggled.connect(self._sync_tab_indicators) - self.separate_enabled_cb.toggled.connect(self._sync_tab_indicators) - self.transcribe_enabled_cb.toggled.connect(self._sync_tab_indicators) - self.engine_cb.currentTextChanged.connect(self._sync_engine_fields) - - row = QHBoxLayout() - row.setSpacing(10) - self.start_btn = QPushButton("▶ Запустить") - self.start_btn.setObjectName("primary") - self.start_btn.clicked.connect(self._on_start) - self.stop_btn = QPushButton("⏹ Стоп") - self.stop_btn.setObjectName("danger") - self.stop_btn.setEnabled(False) - self.stop_btn.clicked.connect(self._on_stop) - self.open_out_btn = QPushButton("📂 Открыть результат") - self.open_out_btn.clicked.connect(self._open_output_dir) - self.save_preset_btn = QPushButton("💾 Сохранить пресет") - self.save_preset_btn.clicked.connect(self._save_preset) - self.load_preset_btn = QPushButton("📁 Загрузить пресет") - self.load_preset_btn.clicked.connect(self._load_preset) - - row.addWidget(self.start_btn) - row.addWidget(self.stop_btn) - row.addStretch(1) - row.addWidget(self.open_out_btn) - row.addWidget(self.save_preset_btn) - row.addWidget(self.load_preset_btn) - root.addLayout(row) - - self.progress = QProgressBar() - self.progress.setFormat("Ожидание") - root.addWidget(self.progress) - - self.log_view = QPlainTextEdit() - self.log_view.setObjectName("log") - self.log_view.setReadOnly(True) - self.log_view.setPlaceholderText("Здесь появится лог выполнения…") - root.addWidget(self.log_view, 1) - - def _with_browse(self, edit: QLineEdit, picker) -> QWidget: - w = QWidget() - h = QHBoxLayout(w) - h.setContentsMargins(0, 0, 0, 0) - h.setSpacing(6) - h.addWidget(edit) - btn = QPushButton("…") - btn.setFixedWidth(36) - btn.clicked.connect(picker) - h.addWidget(btn) - return w - - def _build_shazam_tab(self) -> QWidget: - w = QWidget() - form = QFormLayout(w) - form.setContentsMargins(16, 16, 16, 16) - form.setSpacing(10) - - self.shazam_enabled_cb = QCheckBox("Включён") - self.shazam_concurrency = QSpinBox() - self.shazam_concurrency.setRange(1, 16) - self.shazam_concurrency.setToolTip( - "Сколько файлов опознавать параллельно.\n" - "1 — последовательно, 4–8 — оптимум для Shazam." - ) - form.addRow("Параллельных запросов:", self.shazam_concurrency) - form.addRow("", self.shazam_enabled_cb) - - info = QLabel( - "Идентификация трека через Shazam (unofficial API).\n" - "При совпадении сохраняются метаданные и (если возможно) текст из LRCLIB.\n" - "Требует ffmpeg (для конвертации в 16 кГц моно)." - ) - info.setWordWrap(True) - info.setStyleSheet("color: #a6adc8;") - form.addRow(info) - - return w - - def _build_lrclib_tab(self) -> QWidget: - w = QWidget() - form = QFormLayout(w) - form.setContentsMargins(16, 16, 16, 16) - form.setSpacing(10) - - self.lyrics_enabled_cb = QCheckBox("Включён") - form.addRow("", self.lyrics_enabled_cb) - - self.lyrics_user_agent = QLineEdit() - form.addRow("User-Agent:", self.lyrics_user_agent) - - self.lyrics_use_tags_cb = QCheckBox("Искать по тегам (artist + title)") - form.addRow("", self.lyrics_use_tags_cb) - - self.lyrics_use_shazam_cb = QCheckBox("Искать по результату Shazam") - form.addRow("", self.lyrics_use_shazam_cb) - - self.lyrics_after_asr_cb = QCheckBox("Проверять после ASR (fuzzy-match)") - form.addRow("", self.lyrics_after_asr_cb) - - self.lyrics_transcribe_unresolved_cb = QCheckBox( - "Запускать ASR, если Shazam опознал, но текста нет" - ) - form.addRow("", self.lyrics_transcribe_unresolved_cb) - - self.lyrics_threshold = QDoubleSpinBox() - self.lyrics_threshold.setRange(0.0, 1.0) - self.lyrics_threshold.setSingleStep(0.05) - self.lyrics_threshold.setDecimals(2) - self.lyrics_threshold.setToolTip( - "Насколько должен совпасть распознанный ASR-текст с текстом из LRCLIB,\n" - "чтобы считать это тем же треком. 1.0 = точное совпадение, 0.0 = любое." - ) - form.addRow("Порог fuzzy:", self.lyrics_threshold) - - return w - - def _build_convert_tab(self) -> QWidget: - w = QWidget() - form = QFormLayout(w) - form.setContentsMargins(16, 16, 16, 16) - form.setSpacing(10) - - self.convert_enabled_cb = QCheckBox("Включён") - form.addRow("", self.convert_enabled_cb) - - self.convert_sr = QComboBox() - for sr in SAMPLE_RATES: - self.convert_sr.addItem(f"{sr} Hz", sr) - form.addRow("Sample rate:", self.convert_sr) - - self.convert_ch = QComboBox() - self.convert_ch.addItem("Моно", 1) - self.convert_ch.addItem("Стерео", 2) - form.addRow("Каналы:", self.convert_ch) - - self.convert_ffmpeg = QLineEdit() - self.convert_concurrency = QSpinBox() - self.convert_concurrency.setRange(1, 16) - self.convert_concurrency.setToolTip( - "Сколько ffmpeg-процессов запускать одновременно.\n" - "Для SSD 4–8 нормально; для HDD лучше 1–2." - ) - form.addRow("Параллельных конвертаций:", self.convert_concurrency) - form.addRow("ffmpeg:", self.convert_ffmpeg) - return w - - def _build_separate_tab(self) -> QWidget: - w = QWidget() - form = QFormLayout(w) - form.setContentsMargins(16, 16, 16, 16) - form.setSpacing(10) - - self.separate_enabled_cb = QCheckBox("Включён") - form.addRow("", self.separate_enabled_cb) - - self.separate_model = QLineEdit() - form.addRow("Модель:", self.separate_model) - - self.separate_stem = QLineEdit() - form.addRow("Isolate stem:", self.separate_stem) - - self.separate_unblend = QLineEdit() - form.addRow("unblend CLI:", self.separate_unblend) - - self.separate_kernels_cb = QCheckBox("Использовать кастомные ядра") - form.addRow("", self.separate_kernels_cb) - return w - - def _build_transcribe_tab(self) -> QWidget: - w = QWidget() - form = QFormLayout(w) - form.setContentsMargins(16, 16, 16, 16) - form.setSpacing(10) - - self.transcribe_enabled_cb = QCheckBox("Включён") - form.addRow("", self.transcribe_enabled_cb) - - self.engine_cb = QComboBox() - self.engine_cb.addItems(ENGINES) - form.addRow("Движок:", self.engine_cb) - - self.model_edit = QLineEdit() - form.addRow("Модель:", self.model_edit) - - self.lang_cb = QComboBox() - self.lang_cb.setEditable(True) - self.lang_cb.addItems(LANGUAGES) - self.lang_cb.setToolTip( - "'auto' — WhisperX определяет язык на каждом файле.\n" - "Если знаешь язык — выбери его явно, будет быстрее." - ) - form.addRow("Язык:", self.lang_cb) - - self.device_cb = QComboBox() - self.device_cb.addItems(["auto", "cuda", "cpu"]) - form.addRow("Device:", self.device_cb) - - self.compute_type_cb = QComboBox() - self.compute_type_cb.addItems(["auto", "int8_float16", "int8", "float16", "float32"]) - form.addRow("Compute type:", self.compute_type_cb) - - self.batch_spin = QSpinBox() - self.batch_spin.setRange(1, 512) - self.batch_spin.setToolTip( - "Размер батча сегментов ВНУТРИ одного файла (не файлов!).\n" - "8–16 для 8 ГБ VRAM, 32–64 для 24 ГБ." - ) - form.addRow("Batch size:", self.batch_spin) - - self.gigaam_provider_cb = QComboBox() - self.gigaam_provider_cb.addItems(["auto", "cuda", "cpu"]) - form.addRow("GigaAM provider:", self.gigaam_provider_cb) - - self.local_files_only_cb = QCheckBox("Не проверять HuggingFace (offline)") - self.local_files_only_cb.setToolTip( - "Если модель уже скачана — включите, чтобы не ждать сетевую проверку." - ) - form.addRow("", self.local_files_only_cb) - return w - - # ---------------- visual sync ---------------- - - def _log_device_info(self) -> None: - text, color = _detect_device_info() - self.device_label.setText(text) - self.device_label.setStyleSheet(f"color: {color};") - self._append_log(text) - - def _sync_tab_indicators(self) -> None: - active = QColor("#89b4fa") - inactive = QColor("#585b70") - bar = self.tabs.tabBar() - bar.setTabTextColor(0, active if self.shazam_enabled_cb.isChecked() else inactive) - bar.setTabTextColor(1, active if self.lyrics_enabled_cb.isChecked() else inactive) - bar.setTabTextColor(2, active if self.convert_enabled_cb.isChecked() else inactive) - bar.setTabTextColor(3, active if self.separate_enabled_cb.isChecked() else inactive) - bar.setTabTextColor(4, active if self.transcribe_enabled_cb.isChecked() else inactive) - - def _sync_engine_fields(self) -> None: - is_whisperx = self.engine_cb.currentText() == "whisperx" - for w in (self.lang_cb, self.device_cb, self.compute_type_cb, self.batch_spin): - w.setEnabled(is_whisperx) - self.gigaam_provider_cb.setEnabled(not is_whisperx) - - # ---------------- defaults & config <-> UI ---------------- - - def _apply_defaults(self) -> None: - self._apply_config(PipelineConfig()) - - def _apply_config(self, cfg: PipelineConfig) -> None: - self.input_edit.setText(cfg.input_dir) - self.output_edit.setText(str(Path(cfg.output_dir).absolute())) - self.recursive_cb.setChecked(cfg.recursive) - self.keep_intermediate_cb.setChecked(cfg.keep_intermediate) - self.overwrite_cb.setChecked(cfg.overwrite) - - i = self.mode_cb.findData(cfg.mode) - if i >= 0: - self.mode_cb.setCurrentIndex(i) - - self.shazam_enabled_cb.setChecked(cfg.shazam.enabled) - self.shazam_concurrency.setValue(cfg.shazam.identify_concurrency) - - self.lyrics_enabled_cb.setChecked(cfg.lyrics.enabled) - self.lyrics_user_agent.setText(cfg.lyrics.user_agent) - self.lyrics_use_tags_cb.setChecked(cfg.lyrics.use_tags_first) - self.lyrics_use_shazam_cb.setChecked(cfg.lyrics.use_shazam_result) - self.lyrics_after_asr_cb.setChecked(cfg.lyrics.try_after_asr) - self.lyrics_transcribe_unresolved_cb.setChecked( - cfg.lyrics.transcribe_if_identified_no_lyrics - ) - self.lyrics_threshold.setValue(cfg.lyrics.fuzzy_threshold) - - self.convert_enabled_cb.setChecked(cfg.convert.enabled) - i = self.convert_sr.findData(cfg.convert.sample_rate) - if i >= 0: - self.convert_sr.setCurrentIndex(i) - i = self.convert_ch.findData(cfg.convert.channels) - if i >= 0: - self.convert_ch.setCurrentIndex(i) - self.convert_ffmpeg.setText(cfg.convert.ffmpeg_path) - self.convert_concurrency.setValue(cfg.convert.convert_concurrency) - - self.separate_enabled_cb.setChecked(cfg.separate.enabled) - self.separate_model.setText(cfg.separate.model) - self.separate_stem.setText(cfg.separate.isolate_stem) - self.separate_unblend.setText(cfg.separate.unblend_path) - self.separate_kernels_cb.setChecked(cfg.separate.custom_kernels) - - self.transcribe_enabled_cb.setChecked(cfg.transcribe.enabled) - self.engine_cb.setCurrentText(cfg.transcribe.engine) - self.model_edit.setText(cfg.transcribe.model) - self.lang_cb.setCurrentText(cfg.transcribe.language or "auto") - self.device_cb.setCurrentText(cfg.transcribe.device) - self.compute_type_cb.setCurrentText(cfg.transcribe.compute_type or "auto") - self.batch_spin.setValue(cfg.transcribe.batch_size) - self.gigaam_provider_cb.setCurrentText(cfg.transcribe.gigaam_provider) - self.local_files_only_cb.setChecked(cfg.transcribe.local_files_only) - - def _collect_config(self) -> PipelineConfig: - cfg = PipelineConfig() - cfg.input_dir = self.input_edit.text().strip() - cfg.output_dir = self.output_edit.text().strip() or "transcribed" - cfg.recursive = self.recursive_cb.isChecked() - cfg.keep_intermediate = self.keep_intermediate_cb.isChecked() - cfg.overwrite = self.overwrite_cb.isChecked() - cfg.mode = self.mode_cb.currentData() - - cfg.shazam = ShazamConfig( - enabled=self.shazam_enabled_cb.isChecked(), - identify_concurrency=self.shazam_concurrency.value(), - ) - - cfg.lyrics = LyricsConfig( - enabled=self.lyrics_enabled_cb.isChecked(), - user_agent=self.lyrics_user_agent.text().strip() or "audio-transcriber/1.0", - use_tags_first=self.lyrics_use_tags_cb.isChecked(), - use_shazam_result=self.lyrics_use_shazam_cb.isChecked(), - try_after_asr=self.lyrics_after_asr_cb.isChecked(), - transcribe_if_identified_no_lyrics=self.lyrics_transcribe_unresolved_cb.isChecked(), - fuzzy_threshold=float(self.lyrics_threshold.value()), - ) - - cfg.convert = ConvertConfig( - enabled=self.convert_enabled_cb.isChecked(), - sample_rate=int(self.convert_sr.currentData()), - channels=int(self.convert_ch.currentData()), - ffmpeg_path=self.convert_ffmpeg.text().strip() or "ffmpeg", - convert_concurrency=self.convert_concurrency.value(), - ) - cfg.separate = SeparateConfig( - enabled=self.separate_enabled_cb.isChecked(), - model=self.separate_model.text().strip(), - isolate_stem=self.separate_stem.text().strip() or "vocals", - unblend_path=self.separate_unblend.text().strip() or "unblend", - custom_kernels=self.separate_kernels_cb.isChecked(), - ) - - lang = self.lang_cb.currentText().strip() - compute = self.compute_type_cb.currentText().strip() - cfg.transcribe = TranscribeConfig( - enabled=self.transcribe_enabled_cb.isChecked(), - engine=self.engine_cb.currentText(), - model=self.model_edit.text().strip() or "large-v3", - language=None if lang in ("", "auto") else lang, - device=self.device_cb.currentText(), - compute_type=None if compute == "auto" else compute, - batch_size=self.batch_spin.value(), - gigaam_provider=self.gigaam_provider_cb.currentText(), - local_files_only=self.local_files_only_cb.isChecked(), - ) - return cfg - - # ---------------- handlers ---------------- - - def _pick_input_dir(self) -> None: - d = QFileDialog.getExistingDirectory(self, "Выберите входную папку") - if d: - self.input_edit.setText(d) - - def _pick_output_dir(self) -> None: - d = QFileDialog.getExistingDirectory(self, "Выберите выходную папку") - if d: - self.output_edit.setText(d) - - def _open_output_dir(self) -> None: - p = Path(self.output_edit.text().strip() or "transcribed").absolute() - p.mkdir(parents=True, exist_ok=True) - QDesktopServices.openUrl(QUrl.fromLocalFile(str(p))) - - def _save_preset(self) -> None: - path, _ = QFileDialog.getSaveFileName(self, "Сохранить пресет", "preset.json", "JSON (*.json)") - if not path: - return - try: - self._collect_config().save(Path(path)) - except Exception as exc: # noqa: BLE001 - QMessageBox.critical(self, "Ошибка", str(exc)) - - def _load_preset(self) -> None: - path, _ = QFileDialog.getOpenFileName(self, "Загрузить пресет", "", "JSON (*.json)") - if not path: - return - try: - self._apply_config(PipelineConfig.load(Path(path))) - except Exception as exc: # noqa: BLE001 - QMessageBox.critical(self, "Ошибка", str(exc)) - - def _on_start(self) -> None: - cfg = self._collect_config() - if not cfg.input_dir or not Path(cfg.input_dir).is_dir(): - QMessageBox.warning(self, "Нет входной папки", "Выберите существующую входную папку.") - return - has_work = any([ - cfg.shazam.enabled, cfg.lyrics.enabled, - cfg.convert.enabled, cfg.separate.enabled, cfg.transcribe.enabled, - ]) - if not has_work: - QMessageBox.warning(self, "Нет этапов", "Включите хотя бы один этап.") - return - needs_ffmpeg = cfg.convert.enabled or cfg.shazam.enabled - if needs_ffmpeg and not shutil.which(cfg.convert.ffmpeg_path): - QMessageBox.warning(self, "ffmpeg не найден", - f"Команда '{cfg.convert.ffmpeg_path}' не найдена в PATH.") - return - if cfg.separate.enabled and not shutil.which(cfg.separate.unblend_path): - QMessageBox.warning(self, "unblend не найден", - f"Команда '{cfg.separate.unblend_path}' не найдена в PATH.") - return - - self.log_view.clear() - self._log_device_info() - if cfg.transcribe.enabled and cfg.transcribe.language is None: - self._append_log("⚠ Язык = auto: WhisperX определит язык на каждом файле.") - - self.progress.setValue(0) - self.progress.setFormat("Запуск…") - self.start_btn.setEnabled(False) - self.stop_btn.setEnabled(True) - - self._worker = PipelineWorker(cfg) - self._worker.log.connect(self._append_log) - self._worker.progress.connect(self._on_progress) - self._worker.finished_ok.connect(self._on_finished) - self._worker.failed.connect(self._on_failed) - self._worker.start() - - def _on_stop(self) -> None: - if self._worker is not None: - self._worker.stop() - self._append_log("⏹ Запрошена остановка…") - - def _append_log(self, text: str) -> None: - self.log_view.appendPlainText(text) - sb = self.log_view.verticalScrollBar() - sb.setValue(sb.maximum()) - - def _on_progress(self, done: int, total: int, stage: str) -> None: - self.progress.setMaximum(total) - self.progress.setValue(done) - label = STAGE_LABELS.get(stage, stage) - self.progress.setFormat(f"{label} — %v / %m") - - def _on_finished(self, results: list) -> None: - self.start_btn.setEnabled(True) - self.stop_btn.setEnabled(False) - ok = sum(1 for r in results if r.error is None) - fail = len(results) - ok - self.progress.setFormat(f"Готово — успешно {ok}, с ошибкой {fail}") - self._append_log(f"\n✅ Готово. Успешно: {ok}, с ошибкой: {fail}.") - - def _on_failed(self, message: str) -> None: - self.start_btn.setEnabled(True) - self.stop_btn.setEnabled(False) - self.progress.setFormat("Ошибка") - self._append_log(f"\n✗ Ошибка пайплайна: {message}") - QMessageBox.critical(self, "Ошибка", message) - - -# ========================================= -# Файл: gui/worker.py -# ========================================= - -"""QThread wrapper around core.pipeline.run_pipeline.""" -from __future__ import annotations - -import threading - -from PySide6.QtCore import QThread, Signal - -from core.config import PipelineConfig -from core.pipeline import run_pipeline - - -class PipelineWorker(QThread): - log = Signal(str) - progress = Signal(int, int, str) # done, total, stage - finished_ok = Signal(list) - failed = Signal(str) - - def __init__(self, config: PipelineConfig, parent=None): - super().__init__(parent) - self._config = config - self._stop_event = threading.Event() - - def stop(self) -> None: - self._stop_event.set() - - def run(self) -> None: - try: - results = run_pipeline( - self._config, - log=self.log.emit, - on_progress=lambda d, t, s: self.progress.emit(d, t, s), - stop_event=self._stop_event, - ) - self.finished_ok.emit(results) - except Exception as exc: # noqa: BLE001 - self.failed.emit(f"{type(exc).__name__}: {exc}") - - -# ========================================= -# Файл: gui/init.py -# ========================================= - -"""Qt GUI package.""" - - -# ========================================= -# Файл: gui/theme.py -# ========================================= - -"""Dark theme for the whole application (Catppuccin Macchiato-inspired).""" -from __future__ import annotations - -DARK_STYLESHEET = """ -QWidget { - background-color: #1e1e2e; - color: #cdd6f4; - font-family: 'Inter', 'Segoe UI', 'Noto Sans', sans-serif; - font-size: 10pt; -} -QMainWindow { background-color: #181825; } - -QLabel { background: transparent; } -QLabel#appTitle { - font-size: 18pt; - font-weight: 700; - color: #89b4fa; - padding: 4px 0 10px 0; -} -QLabel#appSubtitle { - color: #7f849c; - font-size: 9pt; - padding-bottom: 12px; -} -QLabel#deviceInfo { - color: #a6adc8; - font-size: 9pt; - padding: 4px 8px; - background-color: #181825; - border: 1px solid #313244; - border-radius: 6px; -} - -QGroupBox { - border: 1px solid #313244; - border-radius: 10px; - margin-top: 14px; - padding: 16px 12px 12px 12px; - background-color: #1e1e2e; - font-weight: 600; -} -QGroupBox::title { - subcontrol-origin: margin; - subcontrol-position: top left; - left: 12px; - padding: 0 6px; - color: #89b4fa; -} - -QLineEdit, QSpinBox, QComboBox, QPlainTextEdit { - background-color: #181825; - border: 1px solid #313244; - border-radius: 6px; - padding: 6px 10px; - color: #cdd6f4; - selection-background-color: #89b4fa; - selection-color: #1e1e2e; - min-height: 22px; -} -QLineEdit:focus, QSpinBox:focus, QComboBox:focus, QPlainTextEdit:focus { - border: 1px solid #89b4fa; -} -QLineEdit:disabled, QComboBox:disabled, QSpinBox:disabled { - color: #585b70; - background-color: #11111b; -} - -QComboBox::drop-down { border: none; width: 22px; } -QComboBox::down-arrow { - width: 0; height: 0; - border-left: 4px solid transparent; - border-right: 4px solid transparent; - border-top: 5px solid #89b4fa; - margin-right: 8px; -} -QComboBox QAbstractItemView { - background-color: #181825; - border: 1px solid #313244; - selection-background-color: #313244; - selection-color: #89b4fa; - outline: none; -} - -/* ─── buttons: higher contrast ─────────────────────────────────── */ -QPushButton { - background-color: #3d4159; - border: 1px solid #5c6080; - border-radius: 6px; - padding: 8px 16px; - color: #ffffff; - font-weight: 600; - min-height: 22px; -} -QPushButton:hover { background-color: #4a4e6b; border-color: #7a7fa0; } -QPushButton:pressed { background-color: #585b70; } -QPushButton:focus { border: 1px solid #a6c8ff; outline: none; } -QPushButton:disabled { - background-color: #2a2d42; - color: #6c7086; - border-color: #3d4055; -} - -QPushButton#primary { - background-color: #89b4fa; - color: #11111b; - font-weight: 700; - border: none; - padding: 8px 20px; -} -QPushButton#primary:hover { background-color: #a6c8ff; } -QPushButton#primary:pressed { background-color: #74a0e0; } -QPushButton#primary:disabled { - background-color: #45475a; - color: #6c7086; -} - -QPushButton#danger { - background-color: #f38ba8; - color: #11111b; - font-weight: 700; - border: none; - padding: 8px 20px; -} -QPushButton#danger:hover { background-color: #f5a4b8; } -QPushButton#danger:pressed { background-color: #e07898; } -QPushButton#danger:disabled { - background-color: #45475a; - color: #6c7086; -} - -QTabWidget::pane { - border: 1px solid #313244; - border-radius: 10px; - background-color: #1e1e2e; - top: -1px; -} -QTabBar::tab { - background-color: #181825; - color: #7f849c; - padding: 8px 20px; - border: 1px solid #313244; - border-bottom: none; - border-top-left-radius: 8px; - border-top-right-radius: 8px; - margin-right: 3px; - font-weight: 500; -} -QTabBar::tab:selected { - background-color: #1e1e2e; - color: #89b4fa; - border-color: #313244; -} -QTabBar::tab:hover:!selected { background-color: #252535; } - -QCheckBox { - spacing: 8px; - background: transparent; -} -QCheckBox::indicator { - width: 16px; - height: 16px; - border: 1px solid #5c6080; - border-radius: 4px; - background-color: #181825; -} -QCheckBox::indicator:hover { border-color: #89b4fa; } -QCheckBox::indicator:checked { - background-color: #89b4fa; - border-color: #89b4fa; -} - -QProgressBar { - background-color: #181825; - border: 1px solid #313244; - border-radius: 8px; - text-align: center; - color: #cdd6f4; - min-height: 24px; - max-height: 24px; - font-weight: 500; -} -QProgressBar::chunk { - background-color: #89b4fa; - border-radius: 7px; - margin: 1px; -} - -QScrollBar:vertical { - background: #181825; - width: 12px; - border-radius: 6px; - margin: 0; -} -QScrollBar::handle:vertical { - background: #45475a; - border-radius: 6px; - min-height: 24px; - margin: 2px; -} -QScrollBar::handle:vertical:hover { background: #585b70; } -QScrollBar::add-line:vertical, QScrollBar::sub-line:vertical { height: 0; } -QScrollBar::add-page:vertical, QScrollBar::sub-page:vertical { background: none; } - -QSplitter::handle { background-color: #313244; } - -QPlainTextEdit#log { - background-color: #11111b; - border: 1px solid #313244; - border-radius: 10px; - padding: 10px; - font-family: 'JetBrains Mono', 'Fira Code', 'Consolas', monospace; - font-size: 9.5pt; - color: #cdd6f4; -} -""" - - diff --git a/core/config.py b/core/config.py index c09469c..9a725b3 100644 --- a/core/config.py +++ b/core/config.py @@ -26,6 +26,7 @@ class SeparateConfig: isolate_stem: str = "vocals" unblend_path: str = "unblend" custom_kernels: bool = False + save_vocals_copy: bool = False @dataclass @@ -57,6 +58,19 @@ class LyricsConfig: try_after_asr: bool = True fuzzy_threshold: float = 0.6 transcribe_if_identified_no_lyrics: bool = True + # NEW: + use_embedded_lyrics: bool = True # read lyrics already stored in file tags first + save_unrecognized: bool = False # write unresolved texts to JSONL for LRCLIB submission + unrecognized_path: str = "" # "" = /unrecognized_lyrics.jsonl + + +@dataclass +class LrcConfig: + language: Optional[str] = "ru" + align_model: Optional[str] = None + device: str = "auto" + audio_source: str = "auto" + embed_in_place: bool = False @dataclass @@ -73,6 +87,7 @@ class PipelineConfig: transcribe: TranscribeConfig = field(default_factory=TranscribeConfig) shazam: ShazamConfig = field(default_factory=ShazamConfig) lyrics: LyricsConfig = field(default_factory=LyricsConfig) + lrc: LrcConfig = field(default_factory=LrcConfig) def to_dict(self) -> dict: return asdict(self) diff --git a/core/lrc.py b/core/lrc.py new file mode 100644 index 0000000..9761481 --- /dev/null +++ b/core/lrc.py @@ -0,0 +1,195 @@ +"""LRC generation, parsing, embedding, and reading embedded lyrics. + +Reading embedded lyrics checks all tag locations for a given audio format: + - MP3 -> ID3 USLT / SYLT + - FLAC/OGG -> LYRICS, UNSYNCEDLYRICS (Vorbis comments) + - M4A/MP4 -> `©lyr` atom +""" +from __future__ import annotations + +import re +from dataclasses import dataclass +from pathlib import Path +from typing import Optional + + +_LRC_LINE_RE = re.compile(r"^\s*\[\d{1,3}:\d{2}(?:\.\d{1,3})?\]") +_LRC_PARSE_RE = re.compile(r"\[(\d+):(\d+)\.(\d+)\](.*)$") + + +@dataclass +class LrcLine: + start: float + text: str + + +def format_timestamp(seconds: float) -> str: + if seconds < 0: + seconds = 0.0 + minutes = int(seconds // 60) + secs = seconds - minutes * 60 + return f"[{minutes:02d}:{secs:05.2f}]" + + +def segments_to_lrc(segments: list[dict]) -> str: + lines: list[str] = [] + for seg in segments: + text = (seg.get("text") or "").strip() + if not text: + continue + start = float(seg.get("start", 0.0)) + lines.append(f"{format_timestamp(start)}{text}") + return "\n".join(lines) + "\n" + + +def parse_lrc(content: str) -> list[LrcLine]: + out: list[LrcLine] = [] + for line in content.splitlines(): + m = _LRC_PARSE_RE.match(line.strip()) + if not m: + continue + mm, ss, frac, text = m.groups() + frac_sec = float(f"0.{frac}") if frac else 0.0 + out.append(LrcLine(start=int(mm) * 60 + int(ss) + frac_sec, text=text.strip())) + return out + + +def is_lrc(content: str) -> bool: + """True if content contains at least one LRC-style timestamp line.""" + if not content: + return False + for line in content.splitlines(): + if _LRC_LINE_RE.match(line): + return True + return False + + +# ---------------------------------------------------------------- read embedded + +def read_embedded_lyrics(audio_path: Path) -> Optional[str]: + """Return lyrics stored in the file's tags, or None. + + Order: prefer synced (LRC-like) -> plain. Never raises. + """ + suffix = audio_path.suffix.lower() + try: + if suffix in (".mp3", ".mp2"): + return _read_mp3(audio_path) + if suffix in (".flac", ".ogg", ".opus"): + return _read_vorbis(audio_path) + if suffix in (".m4a", ".mp4", ".m4b", ".aac"): + return _read_mp4(audio_path) + except Exception: # noqa: BLE001 — never break the pipeline on tag issues + return None + return None + + +def _read_mp3(path: Path) -> Optional[str]: + from mutagen.id3 import ID3, ID3NoHeaderError + try: + tags = ID3(str(path)) + except ID3NoHeaderError: + return None + + # SYLT = synced lyrics (rare); USLT = unsynced (common, may contain LRC text) + sylt = tags.getall("SYLT") + for tag in sylt: + if tag.text: + try: + content = "\n".join( + f"{format_timestamp(ms / 1000.0)}{txt}" + for txt, ms in tag.text + ) + if content.strip(): + return content + except Exception: # noqa: BLE001 + continue + + uslt = tags.getall("USLT") + for tag in uslt: + if tag.text and str(tag.text).strip(): + return str(tag.text) + + return None + + +def _read_vorbis(path: Path) -> Optional[str]: + suffix = path.suffix.lower() + if suffix == ".flac": + from mutagen.flac import FLAC as Audio + elif suffix == ".ogg": + from mutagen.oggvorbis import OggVorbis as Audio + else: + from mutagen.oggopus import OggOpus as Audio + audio = Audio(str(path)) + for key in ("LYRICS", "UNSYNCEDLYRICS", "LYRIC", "SYNCEDLYRICS"): + if key in audio: + values = audio[key] + if isinstance(values, list) and values: + text = str(values[0]) + else: + text = str(values) + if text.strip(): + return text + return None + + +def _read_mp4(path: Path) -> Optional[str]: + from mutagen.mp4 import MP4 + audio = MP4(str(path)) + for key in ("\xa9lyr", "©lyr"): + if key in audio: + values = audio[key] + if isinstance(values, list) and values: + text = str(values[0]) + else: + text = str(values) + if text.strip(): + return text + return None + + +# ---------------------------------------------------------------- embedding + +def embed_lrc(audio_path: Path, lrc_content: str) -> None: + suffix = audio_path.suffix.lower() + if suffix in (".mp3", ".mp2"): + _embed_mp3(audio_path, lrc_content) + elif suffix in (".flac", ".ogg", ".opus"): + _embed_vorbis(audio_path, lrc_content) + elif suffix in (".m4a", ".mp4", ".m4b", ".aac"): + _embed_mp4(audio_path, lrc_content) + else: + raise ValueError(f"Unsupported format for LRC embedding: {suffix}") + + +def _embed_mp3(path: Path, lrc_content: str) -> None: + from mutagen.id3 import ID3, USLT, ID3NoHeaderError + try: + tags = ID3(str(path)) + except ID3NoHeaderError: + tags = ID3() + tags.delall("USLT") + tags.add(USLT(encoding=3, lang="rus", desc="", text=lrc_content)) + tags.save(str(path), v2_version=3) + + +def _embed_vorbis(path: Path, lrc_content: str) -> None: + suffix = path.suffix.lower() + if suffix == ".flac": + from mutagen.flac import FLAC as Audio + elif suffix == ".ogg": + from mutagen.oggvorbis import OggVorbis as Audio + else: + from mutagen.oggopus import OggOpus as Audio + audio = Audio(str(path)) + audio["LYRICS"] = lrc_content + audio["UNSYNCEDLYRICS"] = lrc_content + audio.save() + + +def _embed_mp4(path: Path, lrc_content: str) -> None: + from mutagen.mp4 import MP4 + audio = MP4(str(path)) + audio["\xa9lyr"] = [lrc_content] + audio.save() diff --git a/core/lrc_pipeline.py b/core/lrc_pipeline.py new file mode 100644 index 0000000..41b455e --- /dev/null +++ b/core/lrc_pipeline.py @@ -0,0 +1,195 @@ +"""LRC-generation and LRC-embedding pipelines.""" +from __future__ import annotations + +import shutil +import tempfile +import threading +from pathlib import Path +from typing import Callable, Optional + +from .config import PipelineConfig +from .convert import convert_to_wav +from .lrc import embed_lrc, is_lrc, read_embedded_lyrics, segments_to_lrc +from .metadata import read_metadata +from .pipeline import FileResult, _save_unrecognized, collect_audio_files +from .transcribe import transcribe_aligned + + +LogFn = Callable[[str], None] +ProgressFn = Callable[[int, int, str], None] + + +def _pick_audio_source( + f: Path, output_dir: Path, config: PipelineConfig, work_dir: Path, log: LogFn, +) -> Path: + mode = (config.lrc.audio_source or "auto").lower() + vocals = output_dir / f.stem / "vocals.wav" + convert = output_dir / f.stem / "convert.wav" + if mode == "vocals": + return vocals + if mode == "convert": + return convert + if mode == "input": + return f + if vocals.exists(): + return vocals + if convert.exists(): + return convert + target = work_dir / "input_16k.wav" + log(f" [convert] {f.name} -> 16k mono WAV (для align)") + convert_to_wav( + f, target, sample_rate=16000, channels=1, + ffmpeg_path=config.convert.ffmpeg_path, + ) + return target + + +def run_lrc_pipeline( + config: PipelineConfig, + log: LogFn = print, + on_progress: Optional[ProgressFn] = None, + stop_event: Optional[threading.Event] = None, +) -> list[FileResult]: + input_dir = Path(config.input_dir) + if not input_dir.is_dir(): + raise FileNotFoundError(f"Input directory not found: {input_dir}") + output_dir = Path(config.output_dir) + output_dir.mkdir(parents=True, exist_ok=True) + + files = collect_audio_files(input_dir, config.recursive) + if not files: + log("Аудиофайлы не найдены.") + return [] + + lang = config.lrc.language or config.transcribe.language + log(f"LRC: {len(files)} файлов | язык={lang or 'auto'} | " + f"источник={config.lrc.audio_source}") + + results: list[FileResult] = [] + total = len(files) + + for idx, f in enumerate(files, 1): + if stop_event is not None and stop_event.is_set(): + log("⏹ Отменено") + break + if on_progress: + on_progress(idx - 1, total, "lrc") + log(f"\n▶ [{idx}/{total}] {f.name}") + + track_out = output_dir / f.stem + track_out.mkdir(parents=True, exist_ok=True) + lrc_path = track_out / "text.lrc" + error: Optional[str] = None + + tmp = Path(tempfile.mkdtemp(prefix=f"at_lrc_{f.stem}_")) + try: + embedded = read_embedded_lyrics(f) if config.lyrics.use_embedded_lyrics else None + if embedded and is_lrc(embedded): + lrc_path.write_text(embedded, encoding="utf-8") + log(" ✓ text.lrc (взят из тегов аудио)") + results.append(FileResult(f, None, track_out, None, "embedded+lrc")) + continue + + audio_path = _pick_audio_source(f, output_dir, config, tmp, log) + if not audio_path.exists(): + raise FileNotFoundError( + f"Источник аудио не найден: {audio_path}. " + "Включите conversion/separation в основном пайплайне " + "или выберите audio_source=input." + ) + log(f" [align] {audio_path.name}") + + segments, detected = transcribe_aligned( + audio_path, + model=config.transcribe.model, + language=lang, + device=config.lrc.device, + compute_type=config.transcribe.compute_type, + batch_size=config.transcribe.batch_size, + align_model=config.lrc.align_model, + local_files_only=config.transcribe.local_files_only, + ) + + lrc_content = segments_to_lrc(segments) + lrc_path.write_text(lrc_content, encoding="utf-8") + log(f" ✓ text.lrc ({len(segments)} строк, язык: {detected})") + + # NEW: save to JSONL for LRCLIB submission + _save_unrecognized( + config, f, lrc_content, source="lrc", + metadata=read_metadata(f), + ) + results.append(FileResult(f, None, track_out, None, "lrc")) + except Exception as e: # noqa: BLE001 + error = f"{type(e).__name__}: {e}" + log(f" ✗ Ошибка: {error}") + results.append(FileResult(f, None, None, error, "lrc")) + finally: + shutil.rmtree(tmp, ignore_errors=True) + if on_progress: + on_progress(idx, total, "lrc") + + return results + + +def run_embed_pipeline( + config: PipelineConfig, + log: LogFn = print, + on_progress: Optional[ProgressFn] = None, + stop_event: Optional[threading.Event] = None, +) -> list[FileResult]: + input_dir = Path(config.input_dir) + if not input_dir.is_dir(): + raise FileNotFoundError(f"Input directory not found: {input_dir}") + output_dir = Path(config.output_dir) + output_dir.mkdir(parents=True, exist_ok=True) + + files = collect_audio_files(input_dir, config.recursive) + if not files: + log("Аудиофайлы не найдены.") + return [] + + log(f"Embed LRC: {len(files)} файлов | " + f"режим={'in-place' if config.lrc.embed_in_place else 'копия в output'}") + + results: list[FileResult] = [] + total = len(files) + + for idx, f in enumerate(files, 1): + if stop_event is not None and stop_event.is_set(): + log("⏹ Отменено") + break + if on_progress: + on_progress(idx - 1, total, "embed") + + lrc_path = output_dir / f.stem / "text.lrc" + error: Optional[str] = None + + try: + if not lrc_path.exists(): + log(f" ⊘ {f.name}: text.lrc отсутствует, пропуск") + results.append(FileResult(f, None, None, "no LRC", "embed")) + continue + + lrc_content = lrc_path.read_text(encoding="utf-8") + + if config.lrc.embed_in_place: + target = f + else: + target = output_dir / f.stem / f.name + target.parent.mkdir(parents=True, exist_ok=True) + if not target.exists() or config.overwrite: + shutil.copy2(f, target) + + embed_lrc(target, lrc_content) + log(f" ✓ {f.name} → {target if config.lrc.embed_in_place else target.name}") + results.append(FileResult(f, None, target.parent, None, "embed")) + except Exception as e: # noqa: BLE001 + error = f"{type(e).__name__}: {e}" + log(f" ✗ {f.name}: {error}") + results.append(FileResult(f, None, None, error, "embed")) + finally: + if on_progress: + on_progress(idx, total, "embed") + + return results diff --git a/core/pipeline.py b/core/pipeline.py index cc3a3a8..37da09d 100644 --- a/core/pipeline.py +++ b/core/pipeline.py @@ -1,17 +1,4 @@ -"""Orchestrate identify -> convert -> separate -> transcribe -> lyrics-match. - -Pipeline order per track (short-circuits as soon as text is found): - 1. metadata tags -> LRCLIB search (if `use_tags_first`) - 2. Shazam fingerprint (if enabled) - 3. Shazam result -> LRCLIB search (if `use_shazam_result`) - 4. full pipeline: convert -> separate -> transcribe - 5. post-ASR: LRCLIB search by recognized text (fuzzy match) - -Two modes: - - sequential: run all of the above per file - - staged: identification pass first, then convert/separate/transcribe - passes only for unresolved files, then post-ASR lyrics pass. -""" +"""Orchestrate identify -> convert -> separate -> transcribe -> lyrics-match.""" from __future__ import annotations import difflib @@ -32,6 +19,7 @@ from .metadata import TrackMetadata, read_metadata from .recognize import ShazamResult, recognize from .separate import separate_vocals, separate_vocals_batch from .transcribe import transcribe +from .lrc import is_lrc, read_embedded_lyrics AUDIO_EXTENSIONS = { @@ -64,7 +52,6 @@ def collect_audio_files(input_dir: Path, recursive: bool) -> list[Path]: def _common_parent(paths: list[Path]) -> Path: - """Return the deepest common parent directory of the given paths.""" if not paths: raise ValueError("empty paths") parts = [p.resolve().parts for p in paths] @@ -117,10 +104,44 @@ def _write_metadata_json(path: Path, data: Optional[dict]) -> None: if not data: return path.parent.mkdir(parents=True, exist_ok=True) - path.write_text( - json.dumps(data, indent=2, ensure_ascii=False), - encoding="utf-8", - ) + path.write_text(json.dumps(data, indent=2, ensure_ascii=False), encoding="utf-8") + + +def _save_unrecognized( + config: PipelineConfig, + f: Path, + text: Optional[str], + source: Optional[str], + metadata: Optional[TrackMetadata] = None, + lrc_path: Optional[Path] = None, +) -> None: + """Append a record to the JSONL file for texts NOT confirmed by LRCLIB.""" + if not config.lyrics.save_unrecognized or not text: + return + if source and "+lrclib" in source: + return + + if config.lyrics.unrecognized_path: + out_path = Path(config.lyrics.unrecognized_path) + else: + out_path = Path(config.output_dir) / "unrecognized_lyrics.jsonl" + out_path.parent.mkdir(parents=True, exist_ok=True) + + record = { + "file": str(f), + "trackName": (metadata.title if metadata else None) or f.stem, + "artistName": metadata.artist if metadata else None, + "albumName": metadata.album if metadata else None, + "duration": metadata.duration if metadata else None, + "plainLyrics": text if not is_lrc(text) else None, + "syncedLyrics": text if is_lrc(text) else None, + "source": source, + } + try: + with out_path.open("a", encoding="utf-8") as fh: + fh.write(json.dumps(record, ensure_ascii=False) + "\n") + except OSError: + pass @dataclass @@ -138,24 +159,30 @@ def _identify_file( config: PipelineConfig, log: LogFn, ) -> IdentifyResult: - """Run cheap identification steps. May populate .text and .source, - and set .needs_asr=False if the track is fully resolved.""" result = IdentifyResult(metadata=read_metadata(input_path)) - # --- tags -> LRCLIB --- - if ( - config.lyrics.enabled - and config.lyrics.use_tags_first - and result.metadata.has_identity - ): + # --- 0) embedded lyrics в самом файле --- + if config.lyrics.enabled and config.lyrics.use_embedded_lyrics: + embedded = read_embedded_lyrics(input_path) + if embedded and embedded.strip(): + if is_lrc(embedded): + result.text = embedded + result.source = "embedded+lrc" + log(" ✓ встроенный LRC найден") + else: + result.text = embedded + result.source = "embedded" + log(" ✓ встроенный текст найден") + result.needs_asr = False + return result + + # --- 1) LRCLIB по тегам --- + if config.lyrics.enabled and config.lyrics.use_tags_first and result.metadata.has_identity: log(" LRCLIB по тегам...") match = search_by_identity( - title=result.metadata.title, - artist=result.metadata.artist, - album=result.metadata.album, - duration=result.metadata.duration, - user_agent=config.lyrics.user_agent, - timeout=config.lyrics.timeout, + title=result.metadata.title, artist=result.metadata.artist, + album=result.metadata.album, duration=result.metadata.duration, + user_agent=config.lyrics.user_agent, timeout=config.lyrics.timeout, ) if match: result.text = match.lyrics @@ -164,14 +191,13 @@ def _identify_file( log(f" ✓ {match.artist} — {match.title}") return result - # --- Shazam --- + # --- 2) Shazam --- if config.shazam.enabled: log(" Shazam...") shazam_wav = work_dir / "shazam_16k.wav" try: convert_to_wav( - input_path, shazam_wav, - sample_rate=16000, channels=1, + input_path, shazam_wav, sample_rate=16000, channels=1, ffmpeg_path=config.convert.ffmpeg_path, ) result.shazam = recognize(shazam_wav) @@ -181,15 +207,11 @@ def _identify_file( if result.shazam: log(f" ✓ {result.shazam.artist} — {result.shazam.title}") - if config.lyrics.enabled and config.lyrics.use_shazam_result: match = search_by_identity( - title=result.shazam.title, - artist=result.shazam.artist, - album=result.shazam.album, - duration=result.metadata.duration, - user_agent=config.lyrics.user_agent, - timeout=config.lyrics.timeout, + title=result.shazam.title, artist=result.shazam.artist, + album=result.shazam.album, duration=result.metadata.duration, + user_agent=config.lyrics.user_agent, timeout=config.lyrics.timeout, ) if match: result.text = match.lyrics @@ -197,8 +219,6 @@ def _identify_file( result.needs_asr = False log(" ✓ текст найден через LRCLIB") return result - - # Identified, but no lyrics result.source = "shazam" if not config.lyrics.transcribe_if_identified_no_lyrics: result.needs_asr = False @@ -209,29 +229,18 @@ def _identify_file( return result -def _try_post_asr_match( - text: str, - config: PipelineConfig, - log: LogFn, -) -> Optional[str]: +def _try_post_asr_match(text: str, config: PipelineConfig, log: LogFn) -> Optional[str]: if not text: return None - - # Filter out common ASR watermarks/credits ignore_phrases = ["субтитры", "редактор", "перевод", "сделал", "озвучка"] lines = [] for line in text.splitlines(): line = line.strip() - if not line: - continue - if any(p in line.lower() for p in ignore_phrases): + if not line or any(p in line.lower() for p in ignore_phrases): continue lines.append(line) - if not lines: return None - - # Take the first line with at least 3 words, strip punctuation query = "" for line in lines: words = line.split() @@ -239,26 +248,20 @@ def _try_post_asr_match( clean_words = [re.sub(r'[^\w\s]', '', w) for w in words] query = " ".join(clean_words[:6]) break - if not query: return None - log(f" LRCLIB по тексту: {query!r}...") - try: candidates = search_by_text( - query, limit=5, - user_agent=config.lyrics.user_agent, + query, limit=5, user_agent=config.lyrics.user_agent, timeout=config.lyrics.timeout, ) except Exception as e: # noqa: BLE001 log(f" ⚠ LRCLIB error: {e}") return None - if not candidates: log(" ⚠ нет кандидатов") return None - best: Optional[LyricsMatch] = None best_ratio = 0.0 for c in candidates: @@ -266,11 +269,9 @@ def _try_post_asr_match( if ratio > best_ratio: best_ratio = ratio best = c - if best is not None and best_ratio >= config.lyrics.fuzzy_threshold: log(f" ✓ совпадение {best_ratio:.2f}: {best.artist} — {best.title}") return best.lyrics - log(f" ⚠ лучшее совпадение {best_ratio:.2f} < {config.lyrics.fuzzy_threshold}") return None @@ -283,7 +284,9 @@ def _persist_artifacts( has_metadata: bool, log: LogFn, ) -> None: - """Move wanted artifacts from work_dir to persist_dir.""" + if work_dir.resolve() == persist_dir.resolve(): + return # nothing to move — same directory + persist_dir.mkdir(parents=True, exist_ok=True) names: list[str] = [] @@ -291,6 +294,8 @@ def _persist_artifacts( names.append("metadata.json") if has_text and (work_dir / "text.txt").exists(): names.append("text.txt") + if has_text and (work_dir / "text.lrc").exists(): + names.append("text.lrc") if config.keep_intermediate: for n in ("convert.wav", "vocals.wav"): @@ -302,6 +307,10 @@ def _persist_artifacts( elif config.convert.enabled and (work_dir / "convert.wav").exists(): names.append("convert.wav") + if config.separate.save_vocals_copy and (work_dir / "vocals.wav").exists(): + if "vocals.wav" not in names: + names.append("vocals.wav") + for name in names: src = work_dir / name dst = persist_dir / name @@ -314,14 +323,10 @@ def _persist_artifacts( # ---------------------------------------------------------------- entry def _run_identification_parallel( - config: PipelineConfig, - files: list[Path], - log: LogFn, - on_progress: Optional[ProgressFn], - stop_event: Optional[threading.Event], + config: PipelineConfig, files: list[Path], log: LogFn, + on_progress: Optional[ProgressFn], stop_event: Optional[threading.Event], work_root: Path, ) -> dict[Path, Optional[IdentifyResult]]: - """Parallel identification pass (tags → LRCLIB, Shazam, etc.).""" if not (config.shazam.enabled or config.lyrics.enabled): return {} @@ -347,13 +352,10 @@ def _run_identification_parallel( def worker(f: Path) -> tuple[Path, Optional[IdentifyResult]]: if stop_event is not None and stop_event.is_set(): return f, None - def prefixed(msg: str) -> None: log(f" [{f.name}] {msg.strip()}") - try: - ident = _identify_file(f, work_root / f.stem, config, prefixed) - return f, ident + return f, _identify_file(f, work_root / f.stem, config, prefixed) except Exception as e: # noqa: BLE001 log(f" ✗ {f.name} (identify): {e}") return f, None @@ -373,19 +375,13 @@ def _run_identification_parallel( done += 1 if on_progress: on_progress(done, total, "identify") - return results def _run_convert_parallel( - config: PipelineConfig, - files: list[Path], - state: dict, - log: LogFn, - on_progress: Optional[ProgressFn], - stop_event: Optional[threading.Event], + config: PipelineConfig, files: list[Path], state: dict, log: LogFn, + on_progress: Optional[ProgressFn], stop_event: Optional[threading.Event], ) -> None: - """Parallel convert stage. Writes state[f]['audio_path'] or state[f]['error'].""" concurrency = max(1, int(config.convert.convert_concurrency)) total = len(files) @@ -396,8 +392,7 @@ def _run_convert_parallel( target = Path(tempfile.mkdtemp(prefix=f"at_{f.stem}_")) / "convert.wav" if config.overwrite or not target.exists(): convert_to_wav( - f, target, - sample_rate=config.convert.sample_rate, + f, target, sample_rate=config.convert.sample_rate, channels=config.convert.channels, ffmpeg_path=config.convert.ffmpeg_path, ) @@ -436,51 +431,49 @@ def _run_convert_parallel( if on_progress: on_progress(done, total, "convert") + def run_pipeline( - config: PipelineConfig, - log: LogFn = print, + config: PipelineConfig, log: LogFn = print, on_progress: Optional[ProgressFn] = None, stop_event: Optional[threading.Event] = None, ) -> list[FileResult]: input_dir = Path(config.input_dir) if not input_dir.is_dir(): raise FileNotFoundError(f"Input directory not found: {input_dir}") - output_dir = Path(config.output_dir) output_dir.mkdir(parents=True, exist_ok=True) - files = collect_audio_files(input_dir, config.recursive) if not files: log("Аудиофайлы не найдены.") return [] - mode = (config.mode or "sequential").lower() - log( - f"Найдено треков: {len(files)} | режим: {mode} | " - f"keep_intermediate={config.keep_intermediate}" - ) + log(f"Найдено треков: {len(files)} | режим: {mode} | " + f"keep_intermediate={config.keep_intermediate}") + if config.separate.save_vocals_copy: + log(" ✓ Вокал будет сохраняться всегда (save_vocals_copy=True)") if config.shazam.enabled: log(" Shazam: включён") if config.lyrics.enabled: log(f" LRCLIB: включён (порог fuzzy={config.lyrics.fuzzy_threshold:.2f})") - if mode == "staged": return _run_staged(config, files, log, on_progress, stop_event) return _run_sequential(config, files, log, on_progress, stop_event) -# ---------------------------------------------------------------- sequential - def _run_sequential( - config: PipelineConfig, - files: list[Path], - log: LogFn, - on_progress: Optional[ProgressFn], - stop_event: Optional[threading.Event], + config: PipelineConfig, files: list[Path], log: LogFn, + on_progress: Optional[ProgressFn], stop_event: Optional[threading.Event], ) -> list[FileResult]: total = len(files) results: list[FileResult] = [] output_root = Path(config.output_dir) + ident_tmp = Path(tempfile.mkdtemp(prefix="at_ident_")) + try: + ident_map = _run_identification_parallel( + config, files, log, on_progress, stop_event, ident_tmp, + ) + finally: + shutil.rmtree(ident_tmp, ignore_errors=True) for idx, f in enumerate(files, 1): if stop_event is not None and stop_event.is_set(): @@ -500,8 +493,7 @@ def _run_sequential( metadata_dict: Optional[dict] = None try: - # --- Phase 1: identification --- - ident = _identify_file(f, tmp_root, config, log) + ident = ident_map.get(f) or IdentifyResult(metadata=TrackMetadata()) if not ident.needs_asr: text = ident.text @@ -509,20 +501,22 @@ def _run_sequential( metadata_dict = _build_metadata_dict(ident.metadata, ident.shazam, source) if text: (tmp_root / "text.txt").write_text(text, encoding="utf-8") - log(f" ✓ text.txt ({len(text)} символов, {source})") if metadata_dict: _write_metadata_json(tmp_root / "metadata.json", metadata_dict) - log(" ✓ metadata.json") _persist_artifacts( tmp_root, track_out, config, has_text=text is not None, has_metadata=bool(metadata_dict), log=log, ) + _save_unrecognized( + config, f, text, source, + metadata=ident.metadata, + ) + log(f" ✓ {f.name} [{source}]") results.append(FileResult(f, text, track_out, None, source, metadata_dict)) continue - # --- Phase 2: full pipeline --- audio_path: Path = f if config.convert.enabled: _check_cancel(stop_event) @@ -530,8 +524,7 @@ def _run_sequential( target = tmp_root / "convert.wav" if config.overwrite or not target.exists(): convert_to_wav( - f, target, - sample_rate=config.convert.sample_rate, + f, target, sample_rate=config.convert.sample_rate, channels=config.convert.channels, ffmpeg_path=config.convert.ffmpeg_path, ) @@ -558,8 +551,7 @@ def _run_sequential( _check_cancel(stop_event) log(" [3] Транскрибация...") text = transcribe( - audio_path, - engine=config.transcribe.engine, + audio_path, engine=config.transcribe.engine, model=config.transcribe.model, language=config.transcribe.language, device=config.transcribe.device, @@ -570,13 +562,11 @@ def _run_sequential( ) source = config.transcribe.engine log(f" ✓ распознано {len(text)} символов") - if config.lyrics.enabled and config.lyrics.try_after_asr: matched = _try_post_asr_match(text, config, log) if matched: text = matched source = f"{source}+lrclib" - (tmp_root / "text.txt").write_text(text, encoding="utf-8") metadata_dict = _build_metadata_dict(ident.metadata, ident.shazam, source) @@ -589,6 +579,10 @@ def _run_sequential( has_metadata=bool(metadata_dict), log=log, ) + _save_unrecognized( + config, f, text, source, + metadata=ident.metadata, + ) except Cancelled: log("⏹ Отменено") @@ -600,36 +594,24 @@ def _run_sequential( shutil.rmtree(tmp_root, ignore_errors=True) if on_progress: on_progress(idx, total, "file") - results.append( - FileResult(f, text, track_out if not error else None, error, source, metadata_dict) - ) + results.append(FileResult( + f, text, track_out if not error else None, error, source, metadata_dict, + )) _log_summary(results, log) return results -# ---------------------------------------------------------------- staged - def _run_staged( - config: PipelineConfig, - files: list[Path], - log: LogFn, - on_progress: Optional[ProgressFn], - stop_event: Optional[threading.Event], + config: PipelineConfig, files: list[Path], log: LogFn, + on_progress: Optional[ProgressFn], stop_event: Optional[threading.Event], ) -> list[FileResult]: output_root = Path(config.output_dir) total = len(files) state: dict[Path, dict] = { - f: { - "audio_path": f, - "ident": None, - "text": None, - "source": None, - "metadata_dict": None, - "error": None, - "resolved": False, - } + f: {"audio_path": f, "ident": None, "text": None, "source": None, + "metadata_dict": None, "error": None, "resolved": False} for f in files } @@ -639,77 +621,55 @@ def _run_staged( def cancelled() -> bool: return stop_event is not None and stop_event.is_set() - # --- Stage 0: identification --- if config.shazam.enabled or config.lyrics.enabled: log(f"\n══ Этап 0 · Идентификация ({total} файлов) ══") ident_tmp = Path(tempfile.mkdtemp(prefix="at_ident_")) try: - for i, f in enumerate(files, 1): - if cancelled(): - log("⏹ Отменено") - break - if on_progress: - on_progress(i, total, "identify") - try: - ident = _identify_file(f, ident_tmp / f.stem, config, log) - state[f]["ident"] = ident - - if not ident.needs_asr: - state[f]["resolved"] = True - state[f]["text"] = ident.text - state[f]["source"] = ident.source - state[f]["metadata_dict"] = _build_metadata_dict( - ident.metadata, ident.shazam, ident.source, - ) - # Persist resolved tracks right away - track_out = output_root / f.stem - track_out.mkdir(parents=True, exist_ok=True) - if ident.text: - (track_out / "text.txt").write_text(ident.text, encoding="utf-8") - if state[f]["metadata_dict"]: - _write_metadata_json(track_out / "metadata.json", state[f]["metadata_dict"]) - log(f" ✓ {f.name} [{ident.source}]") - else: - log(f" → {f.name} (нужен полный пайплайн)") - except Exception as e: # noqa: BLE001 - state[f]["error"] = f"{type(e).__name__}: {e}" - log(f" ✗ {f.name}: {e}") + ident_map = _run_identification_parallel( + config, files, log, on_progress, stop_event, ident_tmp, + ) finally: shutil.rmtree(ident_tmp, ignore_errors=True) - # --- Stage 1: convert --- + for f in files: + ident = ident_map.get(f) + if ident is None: + state[f]["ident"] = IdentifyResult(metadata=TrackMetadata()) + log(f" → {f.name} (identification failed)") + continue + state[f]["ident"] = ident + if not ident.needs_asr: + state[f]["resolved"] = True + state[f]["text"] = ident.text + state[f]["source"] = ident.source + state[f]["metadata_dict"] = _build_metadata_dict( + ident.metadata, ident.shazam, ident.source, + ) + track_out = output_root / f.stem + track_out.mkdir(parents=True, exist_ok=True) + if ident.text: + (track_out / "text.txt").write_text(ident.text, encoding="utf-8") + if state[f]["metadata_dict"]: + _write_metadata_json(track_out / "metadata.json", state[f]["metadata_dict"]) + _save_unrecognized( + config, f, ident.text, ident.source, + metadata=ident.metadata, + ) + log(f" ✓ {f.name} [{ident.source}]") + else: + log(f" → {f.name} (нужен полный пайплайн)") + if config.convert.enabled and not cancelled(): todo = active() log(f"\n══ Этап 1 · Конвертация ({len(todo)} файлов) ══") - for i, f in enumerate(todo, 1): - if cancelled(): - log("⏹ Отменено") - break - if on_progress: - on_progress(i, len(todo), "convert") - try: - target = Path(tempfile.mkdtemp(prefix=f"at_{f.stem}_")) / "convert.wav" - if config.overwrite or not target.exists(): - convert_to_wav( - f, target, - sample_rate=config.convert.sample_rate, - channels=config.convert.channels, - ffmpeg_path=config.convert.ffmpeg_path, - ) - state[f]["audio_path"] = target - log(f" ✓ {f.name}") - except Exception as e: # noqa: BLE001 - state[f]["error"] = f"{type(e).__name__}: {e}" - log(f" ✗ {f.name}: {e}") + _run_convert_parallel(config, todo, state, log, on_progress, stop_event) - # ---- Stage 2: separate (batched) ---- if config.separate.enabled and not cancelled(): todo = active() - log(f"\n══ Этап 2/3 · Отделение вокала ({len(todo)} файлов) ══") + log(f"\n══ Этап 2 · Отделение вокала ({len(todo)} файлов) ══") if not todo: pass elif len(todo) == 1: - # Single file — no batching advantage f = todo[0] if on_progress: on_progress(1, 1, "separate") @@ -718,8 +678,7 @@ def _run_staged( target = src.parent / "vocals.wav" with tempfile.TemporaryDirectory() as tmp: raw = separate_vocals( - src, Path(tmp), - model=config.separate.model, + src, Path(tmp), model=config.separate.model, isolate_stem=config.separate.isolate_stem, unblend_path=config.separate.unblend_path, custom_kernels=config.separate.custom_kernels, @@ -731,16 +690,14 @@ def _run_staged( state[f]["error"] = f"{type(e).__name__}: {e}" log(f" ✗ {f.name}: {e}") else: - # Batch: one unblend invocation, model loaded once input_wavs = [state[f]["audio_path"] for f in todo] common_root = _common_parent(input_wavs) - log(f" Батч: {len(todo)} файлов, один вызов unblend (модель загрузится один раз)") + log(f" Батч: {len(todo)} файлов, один вызов unblend") if on_progress: on_progress(0, len(todo), "separate") try: mapping = separate_vocals_batch( - input_wavs, - common_root, + input_wavs, common_root, model=config.separate.model, isolate_stem=config.separate.isolate_stem, unblend_path=config.separate.unblend_path, @@ -771,8 +728,7 @@ def _run_staged( target = src.parent / "vocals.wav" with tempfile.TemporaryDirectory() as tmp: raw = separate_vocals( - src, Path(tmp), - model=config.separate.model, + src, Path(tmp), model=config.separate.model, isolate_stem=config.separate.isolate_stem, unblend_path=config.separate.unblend_path, custom_kernels=config.separate.custom_kernels, @@ -784,7 +740,6 @@ def _run_staged( state[f]["error"] = f"{type(inner).__name__}: {inner}" log(f" ✗ {f.name}: {inner}") - # --- Stage 3: transcribe --- if config.transcribe.enabled and not cancelled(): todo = active() log(f"\n══ Этап 3 · Транскрибация ({len(todo)} файлов) ══") @@ -794,8 +749,7 @@ def _run_staged( break try: text = transcribe( - state[f]["audio_path"], - engine=config.transcribe.engine, + state[f]["audio_path"], engine=config.transcribe.engine, model=config.transcribe.model, language=config.transcribe.language, device=config.transcribe.device, @@ -814,7 +768,6 @@ def _run_staged( if on_progress: on_progress(i, len(todo), "transcribe") - # --- Stage 4: post-ASR lyrics match + persist --- if not cancelled(): log("\n══ Финализация ══") for f in files: @@ -822,11 +775,9 @@ def _run_staged( if s["error"] is not None: continue if s["resolved"]: - continue # already persisted in identification pass - + continue text = s["text"] source = s["source"] - if text and config.lyrics.enabled and config.lyrics.try_after_asr: matched = _try_post_asr_match(text, config, log) if matched: @@ -834,7 +785,6 @@ def _run_staged( source = f"{source}+lrclib" s["text"] = text s["source"] = source - ident = s["ident"] metadata_dict = _build_metadata_dict( ident.metadata, ident.shazam, source, @@ -845,21 +795,31 @@ def _run_staged( track_out.mkdir(parents=True, exist_ok=True) audio_work = s["audio_path"] - work_dir = audio_work.parent if isinstance(audio_work, Path) else track_out - if text: - (work_dir / "text.txt").write_text(text, encoding="utf-8") - if metadata_dict: - _write_metadata_json(work_dir / "metadata.json", metadata_dict) + if isinstance(audio_work, Path) and audio_work != f: + work_dir = audio_work.parent # intermediates exist + if text: + (work_dir / "text.txt").write_text(text, encoding="utf-8") + if metadata_dict: + _write_metadata_json(work_dir / "metadata.json", metadata_dict) + _persist_artifacts( + work_dir, track_out, config, + has_text=text is not None, + has_metadata=bool(metadata_dict), + log=log, + ) + else: + # no intermediates (convert/separate disabled): write directly + if text: + (track_out / "text.txt").write_text(text, encoding="utf-8") + if metadata_dict: + _write_metadata_json(track_out / "metadata.json", metadata_dict) - _persist_artifacts( - work_dir, track_out, config, - has_text=text is not None, - has_metadata=bool(metadata_dict), - log=log, + _save_unrecognized( + config, f, text, source, + metadata=(ident.metadata if ident else None), ) log(f" ✓ {f.name}") - # cleanup temp dirs for f in files: audio_path = state[f]["audio_path"] if isinstance(audio_path, Path) and audio_path.parent.name.startswith(f"at_{f.stem}"): @@ -867,12 +827,9 @@ def _run_staged( results = [ FileResult( - f, - state[f]["text"], + f, state[f]["text"], output_root / f.stem if state[f]["error"] is None else None, - state[f]["error"], - state[f]["source"], - state[f]["metadata_dict"], + state[f]["error"], state[f]["source"], state[f]["metadata_dict"], ) for f in files ] @@ -880,8 +837,6 @@ def _run_staged( return results -# ---------------------------------------------------------------- summary - def _log_summary(results: list[FileResult], log: LogFn) -> None: from collections import Counter counts = Counter(r.source or ("error" if r.error else "no_result") for r in results) diff --git a/core/separate.py b/core/separate.py index 4e181d5..5857edb 100644 --- a/core/separate.py +++ b/core/separate.py @@ -96,6 +96,8 @@ def separate_vocals_batch( break processed += 1 progress_callback(processed) + else: + proc.wait(2.0) if proc.returncode != 0: raise RuntimeError(f"unblend batch failed with exit code {proc.returncode}") diff --git a/core/transcribe.py b/core/transcribe.py index f445f2b..f5b4c7b 100644 --- a/core/transcribe.py +++ b/core/transcribe.py @@ -1,7 +1,7 @@ """Speech recognition via WhisperX or GigaAM (onnx-asr). -Includes aggressive VRAM cleanup to prevent OOM on long batches, -and automatic model reload if the CUDA context gets corrupted. +Also exposes `transcribe_aligned()` for word-level alignment (used by LRC). +Models are cached; aggressive VRAM cleanup happens after each file. """ from __future__ import annotations @@ -33,6 +33,16 @@ def _retry(fn, attempts: int = 3, base_delay: float = 2.0): raise last +def _free_vram() -> None: + gc.collect() + try: + import torch + if torch.cuda.is_available(): + torch.cuda.empty_cache() + except ImportError: + pass + + class _WhisperXModels: _cache: dict = {} @@ -59,6 +69,28 @@ class _WhisperXModels: return cls._cache[key] +class _AlignModels: + _cache: dict = {} + + @classmethod + def get(cls, language_code: str, device: str, model_name: Optional[str] = None): + key = (language_code, device, model_name or "") + if key not in cls._cache: + import whisperx + print(f" [ALIGN] loading align model for {language_code} " + f"({model_name or 'default'}) on {device}...") + + def load(): + kwargs = {"language_code": language_code, "device": device} + if model_name: + kwargs["model_name"] = model_name + return whisperx.load_align_model(**kwargs) + + cls._cache[key] = _retry(load) + print(" [ALIGN] align model loaded.") + return cls._cache[key] + + class _GigaAMModels: _cache: dict = {} @@ -105,13 +137,7 @@ def transcribe( if "out of memory" in err or "cuda" in err or "device ordinal" in err: print(f" [ASR] CUDA error: {e}. Clearing model cache and retrying...") _WhisperXModels._cache.clear() - gc.collect() - try: - import torch - if torch.cuda.is_available(): - torch.cuda.empty_cache() - except ImportError: - pass + _free_vram() return _transcribe_whisperx( audio_path, model, language, device, compute_type, batch_size, local_files_only, vad_filter, @@ -136,28 +162,52 @@ def _transcribe_whisperx( import whisperx model = _WhisperXModels.get(model_name, device, compute_type, local_files_only) audio = whisperx.load_audio(str(audio_path)) - result = model.transcribe( - audio, - batch_size=batch_size, - language=language, - ) - + result = model.transcribe(audio, batch_size=batch_size, language=language) text = " ".join(seg["text"].strip() for seg in result["segments"]).strip() - - # Free VRAM aggressively to prevent fragmentation/OOM on long batches - del audio - del result - gc.collect() - try: - import torch - if torch.cuda.is_available(): - torch.cuda.empty_cache() - except ImportError: - pass - + del audio, result + _free_vram() return text def _transcribe_gigaam(audio_path: Path, provider: str) -> str: model = _GigaAMModels.get(provider) return model.recognize(str(audio_path), channel="mean").strip() + + +def transcribe_aligned( + audio_path: Path, + model: str = "large-v3", + language: Optional[str] = "ru", + device: str = "auto", + compute_type: Optional[str] = None, + batch_size: int = 8, + align_model: Optional[str] = None, + local_files_only: bool = False, +) -> tuple[list[dict], str]: + """Transcribe + word-level align. Returns (aligned_segments, detected_language). + + Each segment has 'start', 'end', 'text', and optionally 'words'. + Requires whisperx engine. + """ + import whisperx + device = _resolve_device(device) + model_obj = _WhisperXModels.get(model, device, compute_type, local_files_only) + audio = whisperx.load_audio(str(audio_path)) + result = model_obj.transcribe(audio, batch_size=batch_size, language=language) + + detected = result.get("language") or language or "en" + + try: + align_model_obj, align_metadata = _AlignModels.get(detected, device, align_model) + aligned = whisperx.align( + result["segments"], align_model_obj, align_metadata, + audio, device, return_char_alignments=False, + ) + segments = aligned["segments"] + except Exception as e: + print(f" [ALIGN] failed: {e}. Falling back to raw WhisperX segments.") + segments = result["segments"] + + del audio, result + _free_vram() + return segments, detected diff --git a/gui/main_window.py b/gui/main_window.py index 7f65dfb..6b04179 100644 --- a/gui/main_window.py +++ b/gui/main_window.py @@ -1,4 +1,4 @@ -"""Main application window — dark theme, mode selector, staged/sequential pipelines.""" +"""Main application window — dark theme, mode selector, LRC tab.""" from __future__ import annotations import shutil @@ -15,7 +15,7 @@ from PySide6.QtWidgets import ( ) from core.config import ( - ConvertConfig, LyricsConfig, PipelineConfig, SeparateConfig, + ConvertConfig, LrcConfig, LyricsConfig, PipelineConfig, SeparateConfig, ShazamConfig, TranscribeConfig, ) from gui.worker import PipelineWorker @@ -31,6 +31,8 @@ STAGE_LABELS = { "convert": "Конвертация", "separate": "Отделение", "transcribe": "Транскрибация", + "lrc": "LRC", + "embed": "Встраивание LRC", } MODE_LABELS = { @@ -44,7 +46,6 @@ def _detect_device_info() -> tuple[str, str]: import torch except ImportError: return "⚠ PyTorch не установлен", "#f38ba8" - if torch.cuda.is_available(): try: name = torch.cuda.get_device_name(0) @@ -52,7 +53,6 @@ def _detect_device_info() -> tuple[str, str]: return f"✓ CUDA: {name} · {vram:.1f} GB VRAM", "#a6e3a1" except Exception: # noqa: BLE001 return "✓ CUDA доступна", "#a6e3a1" - return "⚠ CUDA не обнаружена — расчёт пойдёт на CPU (медленно)", "#f9e2af" @@ -60,7 +60,7 @@ class MainWindow(QMainWindow): def __init__(self) -> None: super().__init__() self.setWindowTitle("Song Transcriber") - self.resize(1140, 880) + self.resize(1140, 920) self._worker: Optional[PipelineWorker] = None self._build_ui() self._apply_defaults() @@ -93,7 +93,6 @@ class MainWindow(QMainWindow): self.input_edit = QLineEdit() top_layout.addRow("Входная папка:", self._with_browse(self.input_edit, self._pick_input_dir)) - self.output_edit = QLineEdit() top_layout.addRow("Выходная папка:", self._with_browse(self.output_edit, self._pick_output_dir)) @@ -111,7 +110,6 @@ class MainWindow(QMainWindow): opts.addWidget(self.overwrite_cb) opts.addStretch(1) top_layout.addRow("Опции:", opts) - root.addWidget(top_box) self.tabs = QTabWidget() @@ -120,6 +118,7 @@ class MainWindow(QMainWindow): self.tabs.addTab(self._build_convert_tab(), "Конвертация") self.tabs.addTab(self._build_separate_tab(), "Отделение вокала") self.tabs.addTab(self._build_transcribe_tab(), "Транскрибация") + self.tabs.addTab(self._build_lrc_tab(), "LRC") root.addWidget(self.tabs) self.shazam_enabled_cb.toggled.connect(self._sync_tab_indicators) @@ -131,9 +130,9 @@ class MainWindow(QMainWindow): row = QHBoxLayout() row.setSpacing(10) - self.start_btn = QPushButton("▶ Запустить") + self.start_btn = QPushButton("▶ Запустить пайплайн") self.start_btn.setObjectName("primary") - self.start_btn.clicked.connect(self._on_start) + self.start_btn.clicked.connect(lambda: self._run_task("main")) self.stop_btn = QPushButton("⏹ Стоп") self.stop_btn.setObjectName("danger") self.stop_btn.setEnabled(False) @@ -180,26 +179,18 @@ class MainWindow(QMainWindow): form = QFormLayout(w) form.setContentsMargins(16, 16, 16, 16) form.setSpacing(10) - self.shazam_enabled_cb = QCheckBox("Включён") self.shazam_concurrency = QSpinBox() self.shazam_concurrency.setRange(1, 16) - self.shazam_concurrency.setToolTip( - "Сколько файлов опознавать параллельно.\n" - "1 — последовательно, 4–8 — оптимум для Shazam." - ) form.addRow("Параллельных запросов:", self.shazam_concurrency) form.addRow("", self.shazam_enabled_cb) - info = QLabel( "Идентификация трека через Shazam (unofficial API).\n" - "При совпадении сохраняются метаданные и (если возможно) текст из LRCLIB.\n" - "Требует ffmpeg (для конвертации в 16 кГц моно)." + "При совпадении сохраняются метаданные и (если возможно) текст из LRCLIB." ) info.setWordWrap(True) info.setStyleSheet("color: #a6adc8;") form.addRow(info) - return w def _build_lrclib_tab(self) -> QWidget: @@ -207,37 +198,47 @@ class MainWindow(QMainWindow): form = QFormLayout(w) form.setContentsMargins(16, 16, 16, 16) form.setSpacing(10) - self.lyrics_enabled_cb = QCheckBox("Включён") form.addRow("", self.lyrics_enabled_cb) - + self.lyrics_use_embedded_cb = QCheckBox("Использовать встроенный текст из тегов (USLT/LYRICS/©lyr)") + self.lyrics_use_embedded_cb.setToolTip( + "Сначала проверяем, есть ли текст (LRC или обычный) уже в тегах аудио.\n" + "Если есть — ASR/Shazam/LRCLIB не запускаются." + ) + form.addRow("", self.lyrics_use_embedded_cb) self.lyrics_user_agent = QLineEdit() form.addRow("User-Agent:", self.lyrics_user_agent) - - self.lyrics_use_tags_cb = QCheckBox("Искать по тегам (artist + title)") + self.lyrics_use_tags_cb = QCheckBox("Искать по тегам") form.addRow("", self.lyrics_use_tags_cb) - - self.lyrics_use_shazam_cb = QCheckBox("Искать по результату Shazam") + self.lyrics_use_shazam_cb = QCheckBox("Искать по Shazam") form.addRow("", self.lyrics_use_shazam_cb) - - self.lyrics_after_asr_cb = QCheckBox("Проверять после ASR (fuzzy-match)") + self.lyrics_after_asr_cb = QCheckBox("Проверять после ASR (fuzzy)") form.addRow("", self.lyrics_after_asr_cb) - self.lyrics_transcribe_unresolved_cb = QCheckBox( - "Запускать ASR, если Shazam опознал, но текста нет" + "ASR, если Shazam опознал без текста" ) form.addRow("", self.lyrics_transcribe_unresolved_cb) - self.lyrics_threshold = QDoubleSpinBox() self.lyrics_threshold.setRange(0.0, 1.0) self.lyrics_threshold.setSingleStep(0.05) self.lyrics_threshold.setDecimals(2) - self.lyrics_threshold.setToolTip( - "Насколько должен совпасть распознанный ASR-текст с текстом из LRCLIB,\n" - "чтобы считать это тем же треком. 1.0 = точное совпадение, 0.0 = любое." - ) form.addRow("Порог fuzzy:", self.lyrics_threshold) + self.lyrics_save_unrecognized_cb = QCheckBox( + "Сохранять тексты без LRCLIB в JSONL (для добавления в LRCLIB)" + ) + self.lyrics_save_unrecognized_cb.setToolTip( + "Если текст получен НЕ из LRCLIB (WhisperX / GigaAM / embedded), " + "записать его в /unrecognized_lyrics.jsonl.\n" + "Формат совместим с LRCLIB API для ручной отправки." + ) + form.addRow("", self.lyrics_save_unrecognized_cb) + + self.lyrics_unrecognized_path_edit = QLineEdit() + self.lyrics_unrecognized_path_edit.setPlaceholderText( + "(пусто — /unrecognized_lyrics.jsonl)" + ) + form.addRow("Файл unrecognized:", self.lyrics_unrecognized_path_edit) return w def _build_convert_tab(self) -> QWidget: @@ -245,27 +246,19 @@ class MainWindow(QMainWindow): form = QFormLayout(w) form.setContentsMargins(16, 16, 16, 16) form.setSpacing(10) - self.convert_enabled_cb = QCheckBox("Включён") form.addRow("", self.convert_enabled_cb) - self.convert_sr = QComboBox() for sr in SAMPLE_RATES: self.convert_sr.addItem(f"{sr} Hz", sr) form.addRow("Sample rate:", self.convert_sr) - self.convert_ch = QComboBox() self.convert_ch.addItem("Моно", 1) self.convert_ch.addItem("Стерео", 2) form.addRow("Каналы:", self.convert_ch) - self.convert_ffmpeg = QLineEdit() self.convert_concurrency = QSpinBox() self.convert_concurrency.setRange(1, 16) - self.convert_concurrency.setToolTip( - "Сколько ffmpeg-процессов запускать одновременно.\n" - "Для SSD 4–8 нормально; для HDD лучше 1–2." - ) form.addRow("Параллельных конвертаций:", self.convert_concurrency) form.addRow("ffmpeg:", self.convert_ffmpeg) return w @@ -275,21 +268,22 @@ class MainWindow(QMainWindow): form = QFormLayout(w) form.setContentsMargins(16, 16, 16, 16) form.setSpacing(10) - self.separate_enabled_cb = QCheckBox("Включён") form.addRow("", self.separate_enabled_cb) - self.separate_model = QLineEdit() form.addRow("Модель:", self.separate_model) - self.separate_stem = QLineEdit() form.addRow("Isolate stem:", self.separate_stem) - self.separate_unblend = QLineEdit() form.addRow("unblend CLI:", self.separate_unblend) - self.separate_kernels_cb = QCheckBox("Использовать кастомные ядра") form.addRow("", self.separate_kernels_cb) + self.separate_save_vocals_cb = QCheckBox("Всегда сохранять vocals.wav в output") + self.separate_save_vocals_cb.setToolTip( + "Независимо от 'Сохранять промежуточные': вокал всегда будет лежать " + "в //vocals.wav — полезно для LRC-этапа." + ) + form.addRow("", self.separate_save_vocals_cb) return w def _build_transcribe_tab(self) -> QWidget: @@ -297,53 +291,90 @@ class MainWindow(QMainWindow): form = QFormLayout(w) form.setContentsMargins(16, 16, 16, 16) form.setSpacing(10) - self.transcribe_enabled_cb = QCheckBox("Включён") form.addRow("", self.transcribe_enabled_cb) - self.engine_cb = QComboBox() self.engine_cb.addItems(ENGINES) form.addRow("Движок:", self.engine_cb) - self.model_edit = QLineEdit() form.addRow("Модель:", self.model_edit) - self.lang_cb = QComboBox() self.lang_cb.setEditable(True) self.lang_cb.addItems(LANGUAGES) - self.lang_cb.setToolTip( - "'auto' — WhisperX определяет язык на каждом файле.\n" - "Если знаешь язык — выбери его явно, будет быстрее." - ) form.addRow("Язык:", self.lang_cb) - self.device_cb = QComboBox() self.device_cb.addItems(["auto", "cuda", "cpu"]) form.addRow("Device:", self.device_cb) - self.compute_type_cb = QComboBox() self.compute_type_cb.addItems(["auto", "int8_float16", "int8", "float16", "float32"]) form.addRow("Compute type:", self.compute_type_cb) - self.batch_spin = QSpinBox() self.batch_spin.setRange(1, 512) - self.batch_spin.setToolTip( - "Размер батча сегментов ВНУТРИ одного файла (не файлов!).\n" - "8–16 для 8 ГБ VRAM, 32–64 для 24 ГБ." - ) form.addRow("Batch size:", self.batch_spin) - self.gigaam_provider_cb = QComboBox() self.gigaam_provider_cb.addItems(["auto", "cuda", "cpu"]) form.addRow("GigaAM provider:", self.gigaam_provider_cb) - self.local_files_only_cb = QCheckBox("Не проверять HuggingFace (offline)") - self.local_files_only_cb.setToolTip( - "Если модель уже скачана — включите, чтобы не ждать сетевую проверку." - ) form.addRow("", self.local_files_only_cb) return w + def _build_lrc_tab(self) -> QWidget: + w = QWidget() + form = QFormLayout(w) + form.setContentsMargins(16, 16, 16, 16) + form.setSpacing(10) + + info = QLabel( + "1) Сгенерировать LRC: WhisperX + align → //text.lrc.\n" + "2) Встроить LRC: записывает LRC в тег lyrics аудио (USLT/LYRICS/©lyr).\n" + "Если 'in-place' выключено — оригинал не трогается, создаётся копия в output." + ) + info.setWordWrap(True) + info.setStyleSheet("color: #a6adc8;") + form.addRow(info) + + self.lrc_audio_source_cb = QComboBox() + self.lrc_audio_source_cb.addItems(["auto", "vocals", "convert", "input"]) + self.lrc_audio_source_cb.setToolTip( + "auto: vocals.wav → convert.wav → 16k WAV из исходника.\n" + "vocals/convert: только соответствующий файл.\n" + "input: оригинальный mp3/flac." + ) + form.addRow("Источник аудио:", self.lrc_audio_source_cb) + + self.lrc_language_cb = QComboBox() + self.lrc_language_cb.setEditable(True) + self.lrc_language_cb.addItems(LANGUAGES) + form.addRow("Язык (для align):", self.lrc_language_cb) + + self.lrc_align_model_edit = QLineEdit() + self.lrc_align_model_edit.setPlaceholderText("(пусто — whisperx по языку)") + form.addRow("Align модель:", self.lrc_align_model_edit) + + self.lrc_device_cb = QComboBox() + self.lrc_device_cb.addItems(["auto", "cuda", "cpu"]) + form.addRow("Device:", self.lrc_device_cb) + + self.lrc_embed_in_place_cb = QCheckBox("Встраивать in-place (перезаписывать оригинал)") + self.lrc_embed_in_place_cb.setToolTip( + "Если выключено (по умолчанию) — создаётся копия в //.\n" + "Если включено — оригинальный файл модифицируется." + ) + form.addRow("", self.lrc_embed_in_place_cb) + + btns = QHBoxLayout() + self.lrc_generate_btn = QPushButton("🎵 Сгенерировать LRC") + self.lrc_generate_btn.setObjectName("primary") + self.lrc_generate_btn.clicked.connect(lambda: self._run_task("lrc")) + self.lrc_embed_btn = QPushButton("📝 Встроить LRC в аудио") + self.lrc_embed_btn.clicked.connect(lambda: self._run_task("embed")) + btns.addWidget(self.lrc_generate_btn) + btns.addWidget(self.lrc_embed_btn) + btns.addStretch(1) + form.addRow("", btns) + + return w + # ---------------- visual sync ---------------- def _log_device_info(self) -> None: @@ -379,7 +410,6 @@ class MainWindow(QMainWindow): self.recursive_cb.setChecked(cfg.recursive) self.keep_intermediate_cb.setChecked(cfg.keep_intermediate) self.overwrite_cb.setChecked(cfg.overwrite) - i = self.mode_cb.findData(cfg.mode) if i >= 0: self.mode_cb.setCurrentIndex(i) @@ -392,10 +422,11 @@ class MainWindow(QMainWindow): self.lyrics_use_tags_cb.setChecked(cfg.lyrics.use_tags_first) self.lyrics_use_shazam_cb.setChecked(cfg.lyrics.use_shazam_result) self.lyrics_after_asr_cb.setChecked(cfg.lyrics.try_after_asr) - self.lyrics_transcribe_unresolved_cb.setChecked( - cfg.lyrics.transcribe_if_identified_no_lyrics - ) + self.lyrics_transcribe_unresolved_cb.setChecked(cfg.lyrics.transcribe_if_identified_no_lyrics) self.lyrics_threshold.setValue(cfg.lyrics.fuzzy_threshold) + self.lyrics_use_embedded_cb.setChecked(cfg.lyrics.use_embedded_lyrics) + self.lyrics_save_unrecognized_cb.setChecked(cfg.lyrics.save_unrecognized) + self.lyrics_unrecognized_path_edit.setText(cfg.lyrics.unrecognized_path) self.convert_enabled_cb.setChecked(cfg.convert.enabled) i = self.convert_sr.findData(cfg.convert.sample_rate) @@ -412,6 +443,7 @@ class MainWindow(QMainWindow): self.separate_stem.setText(cfg.separate.isolate_stem) self.separate_unblend.setText(cfg.separate.unblend_path) self.separate_kernels_cb.setChecked(cfg.separate.custom_kernels) + self.separate_save_vocals_cb.setChecked(cfg.separate.save_vocals_copy) self.transcribe_enabled_cb.setChecked(cfg.transcribe.enabled) self.engine_cb.setCurrentText(cfg.transcribe.engine) @@ -423,6 +455,12 @@ class MainWindow(QMainWindow): self.gigaam_provider_cb.setCurrentText(cfg.transcribe.gigaam_provider) self.local_files_only_cb.setChecked(cfg.transcribe.local_files_only) + self.lrc_audio_source_cb.setCurrentText(cfg.lrc.audio_source) + self.lrc_language_cb.setCurrentText(cfg.lrc.language or "auto") + self.lrc_align_model_edit.setText(cfg.lrc.align_model or "") + self.lrc_device_cb.setCurrentText(cfg.lrc.device) + self.lrc_embed_in_place_cb.setChecked(cfg.lrc.embed_in_place) + def _collect_config(self) -> PipelineConfig: cfg = PipelineConfig() cfg.input_dir = self.input_edit.text().strip() @@ -436,7 +474,6 @@ class MainWindow(QMainWindow): enabled=self.shazam_enabled_cb.isChecked(), identify_concurrency=self.shazam_concurrency.value(), ) - cfg.lyrics = LyricsConfig( enabled=self.lyrics_enabled_cb.isChecked(), user_agent=self.lyrics_user_agent.text().strip() or "audio-transcriber/1.0", @@ -445,8 +482,10 @@ class MainWindow(QMainWindow): try_after_asr=self.lyrics_after_asr_cb.isChecked(), transcribe_if_identified_no_lyrics=self.lyrics_transcribe_unresolved_cb.isChecked(), fuzzy_threshold=float(self.lyrics_threshold.value()), + use_embedded_lyrics=self.lyrics_use_embedded_cb.isChecked(), + save_unrecognized=self.lyrics_save_unrecognized_cb.isChecked(), + unrecognized_path=self.lyrics_unrecognized_path_edit.text().strip(), ) - cfg.convert = ConvertConfig( enabled=self.convert_enabled_cb.isChecked(), sample_rate=int(self.convert_sr.currentData()), @@ -460,8 +499,8 @@ class MainWindow(QMainWindow): isolate_stem=self.separate_stem.text().strip() or "vocals", unblend_path=self.separate_unblend.text().strip() or "unblend", custom_kernels=self.separate_kernels_cb.isChecked(), + save_vocals_copy=self.separate_save_vocals_cb.isChecked(), ) - lang = self.lang_cb.currentText().strip() compute = self.compute_type_cb.currentText().strip() cfg.transcribe = TranscribeConfig( @@ -475,6 +514,14 @@ class MainWindow(QMainWindow): gigaam_provider=self.gigaam_provider_cb.currentText(), local_files_only=self.local_files_only_cb.isChecked(), ) + lrc_lang = self.lrc_language_cb.currentText().strip() + cfg.lrc = LrcConfig( + language=None if lrc_lang in ("", "auto") else lrc_lang, + align_model=self.lrc_align_model_edit.text().strip() or None, + device=self.lrc_device_cb.currentText(), + audio_source=self.lrc_audio_source_cb.currentText(), + embed_in_place=self.lrc_embed_in_place_cb.isChecked(), + ) return cfg # ---------------- handlers ---------------- @@ -512,39 +559,50 @@ class MainWindow(QMainWindow): except Exception as exc: # noqa: BLE001 QMessageBox.critical(self, "Ошибка", str(exc)) - def _on_start(self) -> None: + def _set_busy(self, busy: bool) -> None: + self.start_btn.setEnabled(not busy) + self.lrc_generate_btn.setEnabled(not busy) + self.lrc_embed_btn.setEnabled(not busy) + self.stop_btn.setEnabled(busy) + + def _run_task(self, task: str) -> None: cfg = self._collect_config() if not cfg.input_dir or not Path(cfg.input_dir).is_dir(): QMessageBox.warning(self, "Нет входной папки", "Выберите существующую входную папку.") return - has_work = any([ - cfg.shazam.enabled, cfg.lyrics.enabled, - cfg.convert.enabled, cfg.separate.enabled, cfg.transcribe.enabled, - ]) - if not has_work: - QMessageBox.warning(self, "Нет этапов", "Включите хотя бы один этап.") - return - needs_ffmpeg = cfg.convert.enabled or cfg.shazam.enabled - if needs_ffmpeg and not shutil.which(cfg.convert.ffmpeg_path): - QMessageBox.warning(self, "ffmpeg не найден", - f"Команда '{cfg.convert.ffmpeg_path}' не найдена в PATH.") - return - if cfg.separate.enabled and not shutil.which(cfg.separate.unblend_path): - QMessageBox.warning(self, "unblend не найден", - f"Команда '{cfg.separate.unblend_path}' не найдена в PATH.") + + if task == "main": + has_work = any([ + cfg.shazam.enabled, cfg.lyrics.enabled, + cfg.convert.enabled, cfg.separate.enabled, cfg.transcribe.enabled, + ]) + if not has_work: + QMessageBox.warning(self, "Нет этапов", "Включите хотя бы один этап.") + return + needs_ffmpeg = cfg.convert.enabled or cfg.shazam.enabled + if needs_ffmpeg and not shutil.which(cfg.convert.ffmpeg_path): + QMessageBox.warning(self, "ffmpeg не найден", + f"Команда '{cfg.convert.ffmpeg_path}' не найдена в PATH.") + return + if cfg.separate.enabled and not shutil.which(cfg.separate.unblend_path): + QMessageBox.warning(self, "unblend не найден", + f"Команда '{cfg.separate.unblend_path}' не найдена в PATH.") + return + + if task == "lrc" and cfg.transcribe.engine != "whisperx": + QMessageBox.warning( + self, "LRC требует WhisperX", + "Align-модель доступна только для WhisperX. Переключите движок на whisperx." + ) return self.log_view.clear() self._log_device_info() - if cfg.transcribe.enabled and cfg.transcribe.language is None: - self._append_log("⚠ Язык = auto: WhisperX определит язык на каждом файле.") - self.progress.setValue(0) self.progress.setFormat("Запуск…") - self.start_btn.setEnabled(False) - self.stop_btn.setEnabled(True) + self._set_busy(True) - self._worker = PipelineWorker(cfg) + self._worker = PipelineWorker(cfg, task=task) self._worker.log.connect(self._append_log) self._worker.progress.connect(self._on_progress) self._worker.finished_ok.connect(self._on_finished) @@ -568,16 +626,14 @@ class MainWindow(QMainWindow): self.progress.setFormat(f"{label} — %v / %m") def _on_finished(self, results: list) -> None: - self.start_btn.setEnabled(True) - self.stop_btn.setEnabled(False) + self._set_busy(False) ok = sum(1 for r in results if r.error is None) fail = len(results) - ok self.progress.setFormat(f"Готово — успешно {ok}, с ошибкой {fail}") self._append_log(f"\n✅ Готово. Успешно: {ok}, с ошибкой: {fail}.") def _on_failed(self, message: str) -> None: - self.start_btn.setEnabled(True) - self.stop_btn.setEnabled(False) + self._set_busy(False) self.progress.setFormat("Ошибка") self._append_log(f"\n✗ Ошибка пайплайна: {message}") QMessageBox.critical(self, "Ошибка", message) diff --git a/gui/worker.py b/gui/worker.py index 28aa423..9fe255b 100644 --- a/gui/worker.py +++ b/gui/worker.py @@ -1,4 +1,4 @@ -"""QThread wrapper around core.pipeline.run_pipeline.""" +"""QThread wrapper. Supports three task types: main, lrc, embed.""" from __future__ import annotations import threading @@ -6,7 +6,6 @@ import threading from PySide6.QtCore import QThread, Signal from core.config import PipelineConfig -from core.pipeline import run_pipeline class PipelineWorker(QThread): @@ -15,9 +14,10 @@ class PipelineWorker(QThread): finished_ok = Signal(list) failed = Signal(str) - def __init__(self, config: PipelineConfig, parent=None): + def __init__(self, config: PipelineConfig, task: str = "main", parent=None): super().__init__(parent) self._config = config + self._task = task self._stop_event = threading.Event() def stop(self) -> None: @@ -25,7 +25,14 @@ class PipelineWorker(QThread): def run(self) -> None: try: - results = run_pipeline( + if self._task == "lrc": + from core.lrc_pipeline import run_lrc_pipeline as runner + elif self._task == "embed": + from core.lrc_pipeline import run_embed_pipeline as runner + else: + from core.pipeline import run_pipeline as runner + + results = runner( self._config, log=self.log.emit, on_progress=lambda d, t, s: self.progress.emit(d, t, s),