diff --git a/core/config.py b/core/config.py index 7b54f16..3255e19 100644 --- a/core/config.py +++ b/core/config.py @@ -38,7 +38,23 @@ class TranscribeConfig: batch_size: int = 8 gigaam_provider: str = "auto" local_files_only: bool = False - vad_filter: bool = True + + +@dataclass +class ShazamConfig: + enabled: bool = False + + +@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 # difflib ratio, 0..1 + transcribe_if_identified_no_lyrics: bool = True @dataclass @@ -48,11 +64,13 @@ class PipelineConfig: recursive: bool = False keep_intermediate: bool = True overwrite: bool = True - mode: str = "sequential" # "sequential" | "staged" + 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) diff --git a/core/lyrics.py b/core/lyrics.py new file mode 100644 index 0000000..6c77aed --- /dev/null +++ b/core/lyrics.py @@ -0,0 +1,115 @@ +"""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 diff --git a/core/metadata.py b/core/metadata.py new file mode 100644 index 0000000..b49830f --- /dev/null +++ b/core/metadata.py @@ -0,0 +1,58 @@ +"""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), + ) diff --git a/core/pipeline.py b/core/pipeline.py index f41ec23..c548e31 100644 --- a/core/pipeline.py +++ b/core/pipeline.py @@ -1,24 +1,33 @@ -"""Orchestrate convert -> separate -> transcribe. +"""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: for each file, run all enabled stages before moving to the next. - - staged: run stage 1 over all files, then stage 2 over all files, then stage 3. - -Intermediate artifacts (convert.wav, vocals.wav) are written to a temp dir and -NEVER persisted to the output directory — unless `keep_intermediate=True`. -Only the artifact of the LAST enabled stage is moved into `//`. + - 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 dataclasses import dataclass +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 from .transcribe import transcribe @@ -28,7 +37,7 @@ AUDIO_EXTENSIONS = { } LogFn = Callable[[str], None] -ProgressFn = Callable[[int, int, str], None] # done, total, stage +ProgressFn = Callable[[int, int, str], None] @dataclass @@ -37,6 +46,8 @@ class FileResult: text: Optional[str] output_dir: Optional[Path] error: Optional[str] = None + source: Optional[str] = None + metadata: Optional[dict] = field(default=None) class Cancelled(Exception): @@ -45,68 +56,218 @@ class Cancelled(Exception): # ---------------------------------------------------------------- helpers -_ARTIFACT_NAMES = { - "convert": "convert.wav", - "separate": "vocals.wav", - "transcribe": "text.txt", -} - - 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 _final_stage(config: PipelineConfig) -> Optional[str]: - """The stage whose artifact is the final result (single element).""" - if config.transcribe.enabled: - return "transcribe" - if config.separate.enabled: - return "separate" - if config.convert.enabled: - return "convert" - return None - - def _check_cancel(stop_event: Optional[threading.Event]) -> None: if stop_event is not None and stop_event.is_set(): raise Cancelled() -def _persist_final(work_dir: Path, persist_dir: Path, final_stage: str, log: LogFn) -> None: - """Move the final-stage artifact from work_dir into persist_dir.""" - name = _ARTIFACT_NAMES[final_stage] - src = work_dir / name - if not src.exists(): - raise FileNotFoundError(f"Final artifact missing: {src}") - persist_dir.mkdir(parents=True, exist_ok=True) - dst = persist_dir / name - if dst.exists(): - dst.unlink() - shutil.move(str(src), str(dst)) - log(f" [persist] {final_stage} -> {persist_dir.name}/{name}") +def _text_head(text: str, n_words: int = 8) -> str: + return " ".join(text.split()[:n_words]) -def _work_dir_for( +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, - stem: str, - output_root: Path, - tmp_root: Optional[Path], -) -> tuple[Path, Path]: - """Return (work_dir, persist_dir). + 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]: + """Try to find the actual lyrics via LRCLIB using recognized text. Returns + the lyrics text on a confident match, else None.""" + if not text or len(text.split()) < 3: + return None + + head = _text_head(text, 8) + log(f" LRCLIB по тексту: {head!r}...") + + try: + candidates = search_by_text( + head, 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") - - keep_intermediate=True -> work_dir == persist_dir == output_root/stem - - keep_intermediate=False -> work_dir = tmp_root/stem, persist_dir = output_root/stem - """ - persist_dir = output_root / stem if config.keep_intermediate: - persist_dir.mkdir(parents=True, exist_ok=True) - return persist_dir, persist_dir - if tmp_root is None: - raise ValueError("tmp_root required when keep_intermediate=False") - work_dir = tmp_root / stem - work_dir.mkdir(parents=True, exist_ok=True) - return work_dir, persist_dir + 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 @@ -130,10 +291,14 @@ def run_pipeline( return [] mode = (config.mode or "sequential").lower() - log(f"Найдено треков: {len(files)} | режим: {mode} | " - f"keep_intermediate={config.keep_intermediate}") - if not config.keep_intermediate: - log(" Промежуточные файлы будут во временной папке и удалены после прогона.") + 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) @@ -150,7 +315,6 @@ def _run_sequential( stop_event: Optional[threading.Event], ) -> list[FileResult]: total = len(files) - final_stage = _final_stage(config) results: list[FileResult] = [] output_root = Path(config.output_dir) @@ -163,18 +327,43 @@ def _run_sequential( log(f"\n▶ [{idx}/{total}] {f.name}") tmp_root = Path(tempfile.mkdtemp(prefix=f"at_{f.stem}_")) - work_dir, persist_dir = _work_dir_for(config, f.stem, output_root, tmp_root) + track_out = output_root / f.stem + track_out.mkdir(parents=True, exist_ok=True) - audio_path: Path = f text: Optional[str] = None + source: Optional[str] = None error: Optional[str] = None - processed_ok = False + 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 = work_dir / "convert.wav" + target = tmp_root / "convert.wav" if config.overwrite or not target.exists(): convert_to_wav( f, target, @@ -188,7 +377,7 @@ def _run_sequential( if config.separate.enabled: _check_cancel(stop_event) log(" [2] Отделение вокала...") - target = work_dir / "vocals.wav" + target = tmp_root / "vocals.wav" with tempfile.TemporaryDirectory() as tmp: raw = separate_vocals( audio_path, Path(tmp), @@ -214,12 +403,28 @@ def _run_sequential( batch_size=config.transcribe.batch_size, gigaam_provider=config.transcribe.gigaam_provider, local_files_only=config.transcribe.local_files_only, - vad_filter=config.transcribe.vad_filter, ) - (work_dir / "text.txt").write_text(text, encoding="utf-8") - log(f" ✓ text.txt ({len(text)} символов)") + source = config.transcribe.engine + log(f" ✓ распознано {len(text)} символов") - processed_ok = True + 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("⏹ Отменено") @@ -228,17 +433,14 @@ def _run_sequential( error = f"{type(e).__name__}: {e}" log(f" ✗ Ошибка: {error}") finally: - if processed_ok and not config.keep_intermediate and final_stage is not None: - try: - _persist_final(work_dir, persist_dir, final_stage, log) - except Exception as e: # noqa: BLE001 - error = f"persist failed: {e}" - log(f" ✗ {error}") shutil.rmtree(tmp_root, ignore_errors=True) if on_progress: on_progress(idx, total, "file") - results.append(FileResult(f, text, persist_dir if not error else None, error)) + results.append( + FileResult(f, text, track_out if not error else None, error, source, metadata_dict) + ) + _log_summary(results, log) return results @@ -253,134 +455,217 @@ def _run_staged( ) -> list[FileResult]: output_root = Path(config.output_dir) total = len(files) - final_stage = _final_stage(config) - # One tmp_root for the whole batch: stage 2 needs convert.wav of every file, - # stage 3 needs vocals.wav of every file. All of it is removed at the end. - tmp_root_ctx = tempfile.mkdtemp(prefix="at_staged_") - tmp_root = Path(tmp_root_ctx) - - state: dict[Path, dict] = {} - for f in files: - work_dir, persist_dir = _work_dir_for(config, f.stem, output_root, tmp_root) - state[f] = { + state: dict[Path, dict] = { + f: { "audio_path": f, - "work_dir": work_dir, - "persist_dir": persist_dir, + "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] + 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() - try: - # ---- Stage 1: convert ---- - if config.convert.enabled: - log(f"\n══ Этап 1/3 · Конвертация ({total} файлов) ══") + # --- 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, "convert") + on_progress(i, total, "identify") try: - target = state[f]["work_dir"] / "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, + 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, ) - state[f]["audio_path"] = target - log(f" ✓ {f.name}") + # 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 2: separate ---- - if config.separate.enabled and not cancelled(): - todo = active() - log(f"\n══ Этап 2/3 · Отделение вокала ({len(todo)} файлов) ══") - for i, f in enumerate(todo, 1): - if cancelled(): - log("⏹ Отменено") - break - if on_progress: - on_progress(i, len(todo), "separate") - try: - target = state[f]["work_dir"] / "vocals.wav" - with tempfile.TemporaryDirectory() as tmp: - raw = separate_vocals( - state[f]["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)) - 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 3: transcribe ---- - if config.transcribe.enabled and not cancelled(): - todo = active() - log(f"\n══ Этап 3/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, - vad_filter=config.transcribe.vad_filter, + # --- 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]["work_dir"] / "text.txt").write_text(text, encoding="utf-8") - state[f]["text"] = text - 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}") + 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}") - # ---- Persist final artifacts ---- - if not config.keep_intermediate and final_stage is not None: - log(f"\n══ Перенос финальных файлов ({final_stage}) ══") - for f in files: - s = state[f] - if s["error"] is not None: - continue - try: - _persist_final(s["work_dir"], s["persist_dir"], final_stage, log) - except Exception as e: # noqa: BLE001 - s["error"] = f"persist failed: {e}" - log(f" ✗ {f.name}: {e}") + # --- Stage 2: separate --- + if config.separate.enabled and not cancelled(): + todo = active() + log(f"\n══ Этап 2 · Отделение вокала ({len(todo)} файлов) ══") + for i, f in enumerate(todo, 1): + if cancelled(): + log("⏹ Отменено") + 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 e: # noqa: BLE001 + state[f]["error"] = f"{type(e).__name__}: {e}" + log(f" ✗ {f.name}: {e}") - finally: - shutil.rmtree(tmp_root, ignore_errors=True) + # --- 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}") - return [ + # --- 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"], - state[f]["persist_dir"] if state[f]["error"] is None else None, + 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}") diff --git a/core/recognize.py b/core/recognize.py new file mode 100644 index 0000000..835eb3c --- /dev/null +++ b/core/recognize.py @@ -0,0 +1,52 @@ +"""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, + ) diff --git a/gui/main_window.py b/gui/main_window.py index a622277..3fd2705 100644 --- a/gui/main_window.py +++ b/gui/main_window.py @@ -8,13 +8,16 @@ from typing import Optional from PySide6.QtCore import Qt, QUrl from PySide6.QtGui import QColor, QDesktopServices from PySide6.QtWidgets import ( - QCheckBox, QComboBox, QFileDialog, QFormLayout, QGroupBox, + QCheckBox, QComboBox, QDoubleSpinBox, QFileDialog, QFormLayout, QGroupBox, QHBoxLayout, QLabel, QLineEdit, QMainWindow, QMessageBox, QPlainTextEdit, QProgressBar, QPushButton, QSpinBox, QTabWidget, QVBoxLayout, QWidget, ) -from core.config import ConvertConfig, PipelineConfig, SeparateConfig, TranscribeConfig +from core.config import ( + ConvertConfig, LyricsConfig, PipelineConfig, SeparateConfig, + ShazamConfig, TranscribeConfig, +) from gui.worker import PipelineWorker @@ -24,6 +27,7 @@ ENGINES = ["whisperx", "gigaam"] STAGE_LABELS = { "file": "Файлы", + "identify": "Идентификация", "convert": "Конвертация", "separate": "Отделение", "transcribe": "Транскрибация", @@ -36,7 +40,6 @@ MODE_LABELS = { def _detect_device_info() -> tuple[str, str]: - """Return (display_text, color_hex) describing the compute device.""" try: import torch except ImportError: @@ -57,7 +60,7 @@ class MainWindow(QMainWindow): def __init__(self) -> None: super().__init__() self.setWindowTitle("Song Transcriber") - self.resize(1100, 850) + self.resize(1140, 880) self._worker: Optional[PipelineWorker] = None self._build_ui() self._apply_defaults() @@ -74,19 +77,16 @@ class MainWindow(QMainWindow): root.setContentsMargins(18, 16, 18, 16) root.setSpacing(12) - # --- header row: title + device badge --- 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 settings --- top_box = QGroupBox("Общие настройки") top_layout = QFormLayout(top_box) top_layout.setSpacing(10) @@ -114,19 +114,21 @@ class MainWindow(QMainWindow): root.addWidget(top_box) - # --- stages --- 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) - # --- controls row --- row = QHBoxLayout() row.setSpacing(10) self.start_btn = QPushButton("▶ Запустить") @@ -136,7 +138,6 @@ class MainWindow(QMainWindow): 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("💾 Сохранить пресет") @@ -152,12 +153,10 @@ class MainWindow(QMainWindow): row.addWidget(self.load_preset_btn) root.addLayout(row) - # --- progress --- self.progress = QProgressBar() self.progress.setFormat("Ожидание") root.addWidget(self.progress) - # --- log --- self.log_view = QPlainTextEdit() self.log_view.setObjectName("log") self.log_view.setReadOnly(True) @@ -176,6 +175,64 @@ class MainWindow(QMainWindow): 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("Включён") + 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) @@ -241,8 +298,8 @@ class MainWindow(QMainWindow): self.lang_cb.setEditable(True) self.lang_cb.addItems(LANGUAGES) self.lang_cb.setToolTip( - "'auto' — WhisperX определяет язык на каждом файле отдельно.\n" - "Это заметно медленнее. Если знаешь язык — выбери его явно." + "'auto' — WhisperX определяет язык на каждом файле.\n" + "Если знаешь язык — выбери его явно, будет быстрее." ) form.addRow("Язык:", self.lang_cb) @@ -258,28 +315,19 @@ class MainWindow(QMainWindow): self.batch_spin.setRange(1, 512) self.batch_spin.setToolTip( "Размер батча сегментов ВНУТРИ одного файла (не файлов!).\n" - "Больше ≠ всегда быстрее: упирается в VRAM.\n" - "8–16 для 8 ГБ VRAM, 32–64 для 24 ГБ.\n" - "При 'CUDA out of memory' — уменьшайте." + "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( - "Если модель уже скачана — включите, чтобы не ждать сетевую проверку.\n" - "При первом запуске (модели нет) оставьте выключенным." + "Если модель уже скачана — включите, чтобы не ждать сетевую проверку." ) form.addRow("", self.local_files_only_cb) - - self.vad_filter_cb = QCheckBox("Использовать VAD (pyannote)") - self.vad_filter_cb.setChecked(True) - self.vad_filter_cb.setToolTip( - "Отключите, если загрузка модели виснет на 'Performing voice activity detection'." - ) - form.addRow("", self.vad_filter_cb) return w # ---------------- visual sync ---------------- @@ -294,9 +342,11 @@ class MainWindow(QMainWindow): active = QColor("#89b4fa") inactive = QColor("#585b70") bar = self.tabs.tabBar() - bar.setTabTextColor(0, active if self.convert_enabled_cb.isChecked() else inactive) - bar.setTabTextColor(1, active if self.separate_enabled_cb.isChecked() else inactive) - bar.setTabTextColor(2, active if self.transcribe_enabled_cb.isChecked() else inactive) + 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" @@ -320,6 +370,18 @@ class MainWindow(QMainWindow): if i >= 0: self.mode_cb.setCurrentIndex(i) + self.shazam_enabled_cb.setChecked(cfg.shazam.enabled) + + 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: @@ -344,7 +406,6 @@ class MainWindow(QMainWindow): 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) - self.vad_filter_cb.setChecked(cfg.transcribe.vad_filter) def _collect_config(self) -> PipelineConfig: cfg = PipelineConfig() @@ -355,6 +416,18 @@ class MainWindow(QMainWindow): cfg.overwrite = self.overwrite_cb.isChecked() cfg.mode = self.mode_cb.currentData() + cfg.shazam = ShazamConfig(enabled=self.shazam_enabled_cb.isChecked()) + + 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()), @@ -380,9 +453,8 @@ class MainWindow(QMainWindow): 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(), ) - cfg.transcribe.local_files_only = self.local_files_only_cb.isChecked() - cfg.transcribe.vad_filter = self.vad_filter_cb.isChecked() return cfg # ---------------- handlers ---------------- @@ -425,10 +497,15 @@ class MainWindow(QMainWindow): if not cfg.input_dir or not Path(cfg.input_dir).is_dir(): QMessageBox.warning(self, "Нет входной папки", "Выберите существующую входную папку.") return - if not (cfg.convert.enabled or cfg.separate.enabled or cfg.transcribe.enabled): + 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 - if cfg.convert.enabled and not shutil.which(cfg.convert.ffmpeg_path): + 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 @@ -440,8 +517,7 @@ class MainWindow(QMainWindow): self.log_view.clear() self._log_device_info() if cfg.transcribe.enabled and cfg.transcribe.language is None: - self._append_log("⚠ Язык = auto: WhisperX определит язык на каждом файле. " - "Если язык известен — задайте его явно, будет быстрее.") + self._append_log("⚠ Язык = auto: WhisperX определит язык на каждом файле.") self.progress.setValue(0) self.progress.setFormat("Запуск…") diff --git a/requirements.txt b/requirements.txt index 820d4d4..3630693 100644 --- a/requirements.txt +++ b/requirements.txt @@ -5,6 +5,11 @@ PySide6>=6.6 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: