Phase 1: chunk music/quality classifier Space

README.md

@@ -1,13 +1,20 @@
 ---
 title: Chunk Classifier
-emoji: 💻
-colorFrom: blue
-colorTo: green
+emoji: 🎵
+colorFrom: purple
+colorTo: pink
 sdk: gradio
-sdk_version: 6.16.0
-python_version: '3.13'
 app_file: app.py
 pinned: false
 ---
 
-Check out the configuration reference at https://huggingface.co/docs/hub/spaces-config-reference
+Classify audio chunks for music contamination and technical defects using the
+[AST AudioSet](https://huggingface.co/MIT/ast-finetuned-audioset-10-10-0.4593) model (527 classes).
+
+**Input:** one or more HF chunk dataset repo IDs (e.g. `fosters/my-book-chunks`).
+
+**Output:** `<input>_classified` dataset — all rows kept, new columns added:
+`music_score`, `has_music`, `contaminated`, `flags`, `top_labels`, `rms_db`, `clipping_ratio`, `max_silence_sec`.
+The `audio` column is preserved as-is so you can listen in the HF viewer and sort by `music_score`.
+
+Set `HF_TOKEN` as a Space secret to read private datasets and write to `fosters/`.

app.py

@@ -0,0 +1,460 @@
+"""Chunk Classifier — HF Space.
+
+Classifies audio chunks in one or more HF datasets using the AST AudioSet model.
+Writes a ``<input>_classified`` dataset with all original rows + classification columns.
+The audio column is kept as-is so the HF viewer shows an audio player.
+"""
+
+from __future__ import annotations
+
+import datetime
+import json
+import os
+import tempfile
+import time
+from pathlib import Path
+from typing import Any, Generator
+
+# HF env must be set before any HF import
+os.environ.setdefault("HF_XET_HIGH_PERFORMANCE", "1")
+
+import gradio as gr
+import pandas as pd
+import pyarrow as pa
+import pyarrow.parquet as pq
+from huggingface_hub import HfApi, hf_hub_download
+
+from music_detector import ChunkVerdict, judge_chunk_files_batched
+
+# ---------------------------------------------------------------------------
+# Constants
+# ---------------------------------------------------------------------------
+
+SHARD_MAX_BYTES = 260 * 1024 * 1024   # 260 MB — HF viewer limit
+BATCH_SIZE = 32
+DEFAULT_THRESHOLD = 0.25
+DEFAULT_NAMESPACE = "fosters"
+
+# PyArrow type → HF dtype string
+_PA_TO_HF_DTYPE: dict[Any, str] = {
+    pa.string(): "string",
+    pa.large_string(): "string",
+    pa.float32(): "float32",
+    pa.float64(): "float64",
+    pa.bool_(): "bool",
+    pa.int8(): "int8",
+    pa.int16(): "int16",
+    pa.int32(): "int32",
+    pa.int64(): "int64",
+}
+
+# New classifier fields added to every output shard
+_CLASSIFIER_FIELDS = [
+    pa.field("music_score", pa.float32()),
+    pa.field("has_music", pa.bool_()),
+    pa.field("clipping_ratio", pa.float32()),
+    pa.field("rms_db", pa.float32()),
+    pa.field("max_silence_sec", pa.float32()),
+    pa.field("flags", pa.string()),
+    pa.field("top_labels", pa.string()),
+    pa.field("contaminated", pa.bool_()),
+]
+
+
+# ---------------------------------------------------------------------------
+# Helpers
+# ---------------------------------------------------------------------------
+
+
+def _ts() -> str:
+    return datetime.datetime.now().strftime("%H:%M:%S")
+
+
+def _default_out_repo(in_repo: str) -> str:
+    parts = in_repo.split("/")
+    name = parts[-1] if parts else in_repo
+    ns = parts[0] if len(parts) > 1 else DEFAULT_NAMESPACE
+    return f"{ns}/{name}_classified"
+
+
+def _list_parquet_shards(repo_id: str, token: str | None) -> list[str]:
+    """Return sorted list of data/train-*.parquet rfilenames in the dataset repo."""
+    api = HfApi(token=token)
+    return sorted(
+        f.rfilename
+        for f in api.list_repo_tree(repo_id, repo_type="dataset", recursive=True)
+        if hasattr(f, "rfilename")
+        and f.rfilename.startswith("data/train-")
+        and f.rfilename.endswith(".parquet")
+    )
+
+
+def _pa_field_to_hf(f: pa.Field) -> dict[str, Any]:
+    dtype = _PA_TO_HF_DTYPE.get(f.type, str(f.type))
+    return {"_type": "Value", "dtype": dtype}
+
+
+def _build_hf_features(in_schema: pa.Schema) -> dict[str, Any]:
+    """Build HF features dict: original columns + classifier columns.
+
+    Preserves existing HF metadata for original columns where available.
+    Always ensures the 'audio' column has ``{"_type": "Audio"}``.
+    """
+    # Try to load existing HF features from input schema metadata
+    existing: dict[str, Any] = {}
+    if in_schema.metadata and b"huggingface" in in_schema.metadata:
+        try:
+            meta = json.loads(in_schema.metadata[b"huggingface"].decode())
+            existing = meta.get("info", {}).get("features", {})
+        except Exception:
+            pass
+
+    features: dict[str, Any] = {}
+
+    # Original columns
+    for f in in_schema:
+        if f.name == "audio":
+            features[f.name] = {"_type": "Audio"}
+        elif f.name in existing:
+            features[f.name] = existing[f.name]
+        else:
+            features[f.name] = _pa_field_to_hf(f)
+
+    # Classifier columns
+    for f in _CLASSIFIER_FIELDS:
+        features[f.name] = _pa_field_to_hf(f)
+
+    return features
+
+
+def _append_classifier_columns(
+    in_table: pa.Table,
+    verdicts: list[ChunkVerdict],
+) -> pa.Table:
+    """Return a new table with all original columns + classifier columns + HF metadata."""
+    hf_features = _build_hf_features(in_table.schema)
+    hf_meta = json.dumps({"info": {"features": hf_features}})
+
+    new_schema = pa.schema(
+        list(in_table.schema) + _CLASSIFIER_FIELDS,
+        metadata={"huggingface": hf_meta},
+    )
+
+    data: dict[str, Any] = {name: in_table.column(name) for name in in_table.column_names}
+    data["music_score"] = pa.array([v.music_score for v in verdicts], type=pa.float32())
+    data["has_music"] = pa.array([v.has_music for v in verdicts], type=pa.bool_())
+    data["clipping_ratio"] = pa.array([v.clipping_ratio for v in verdicts], type=pa.float32())
+    data["rms_db"] = pa.array([v.rms_db for v in verdicts], type=pa.float32())
+    data["max_silence_sec"] = pa.array([v.max_silence_sec for v in verdicts], type=pa.float32())
+    data["flags"] = pa.array([",".join(v.flags) for v in verdicts], type=pa.string())
+    data["top_labels"] = pa.array([", ".join(v.top_labels[:3]) for v in verdicts], type=pa.string())
+    data["contaminated"] = pa.array([v.contaminated for v in verdicts], type=pa.bool_())
+
+    return pa.table(data, schema=new_schema)
+
+
+# ---------------------------------------------------------------------------
+# Per-repo processing
+# ---------------------------------------------------------------------------
+
+
+def _process_repo(
+    repo_id: str,
+    out_repo: str,
+    music_threshold: float,
+    private: bool,
+    token: str | None,
+    log: list[str],
+) -> dict[str, Any]:
+    """Download, classify, and push one repo. Returns per-repo stats dict."""
+    api = HfApi(token=token)
+    short = repo_id.split("/")[-1]
+
+    log.append(f"[{_ts()}] {short}: listing shards…")
+    shard_names = _list_parquet_shards(repo_id, token)
+    if not shard_names:
+        log.append(f"[{_ts()}] {short}: ✗ no parquet shards found under data/")
+        return {"repo": repo_id, "error": "no shards"}
+
+    log.append(f"[{_ts()}] {short}: {len(shard_names)} shard(s) found")
+    api.create_repo(repo_id=out_repo, repo_type="dataset", exist_ok=True, private=private)
+
+    stats = {"total": 0, "contaminated": 0, "has_music": 0, "technical": 0}
+
+    with tempfile.TemporaryDirectory() as tmp:
+        out_data_dir = Path(tmp) / "data"
+        out_data_dir.mkdir()
+        out_parts: list[Path] = []
+
+        for shard_idx, shard_name in enumerate(shard_names):
+            log.append(
+                f"[{_ts()}] {short}: shard {shard_idx + 1}/{len(shard_names)}"
+                f" — downloading…"
+            )
+            shard_path = hf_hub_download(
+                repo_id=repo_id, filename=shard_name,
+                repo_type="dataset", token=token,
+            )
+            in_table = pq.read_table(shard_path)
+            n_rows = len(in_table)
+            log.append(f"[{_ts()}] {short}: shard has {n_rows} rows — classifying…")
+
+            # Extract audio bytes + classify in a temp dir
+            with tempfile.TemporaryDirectory() as audio_tmp:
+                audio_files: list[Path] = []
+                audio_col = in_table.column("audio")
+                for i in range(n_rows):
+                    row_audio = audio_col[i].as_py()
+                    audio_bytes: bytes = row_audio.get("bytes") or b""
+                    suffix = Path(row_audio.get("path") or "chunk.mp3").suffix or ".mp3"
+                    audio_path = Path(audio_tmp) / f"chunk_{i:06d}{suffix}"
+                    audio_path.write_bytes(audio_bytes)
+                    audio_files.append(audio_path)
+
+                t0 = time.time()
+                verdicts = judge_chunk_files_batched(
+                    audio_files,
+                    music_threshold=music_threshold,
+                    batch_size=BATCH_SIZE,
+                    device="cpu",
+                )
+            elapsed = int(time.time() - t0)
+            n_cont = sum(v.contaminated for v in verdicts)
+            n_music = sum(v.has_music for v in verdicts)
+            n_tech = sum(bool(v.flags) and not v.has_music for v in verdicts)
+            log.append(
+                f"[{_ts()}] {short}: shard done in {elapsed}s — "
+                f"contaminated={n_cont}/{n_rows} "
+                f"(music={n_music}, technical={n_tech})"
+            )
+
+            # Build output shard
+            out_table = _append_classifier_columns(in_table, verdicts)
+            out_part = out_data_dir / f"part_{shard_idx:05d}.parquet"
+            pq.write_table(out_table, out_part, row_group_size=1, compression="snappy")
+            out_parts.append(out_part)
+
+            stats["total"] += n_rows
+            stats["contaminated"] += n_cont
+            stats["has_music"] += n_music
+            stats["technical"] += n_tech
+
+        # Rename parts with final shard count and upload
+        n_shards = len(out_parts)
+        final_parts: list[Path] = []
+        for i, p in enumerate(out_parts):
+            final = out_data_dir / f"train-{i:05d}-of-{n_shards:05d}.parquet"
+            p.rename(final)
+            final_parts.append(final)
+
+        log.append(f"[{_ts()}] {short}: uploading {n_shards} shard(s) → {out_repo}…")
+        api.upload_folder(
+            folder_path=str(out_data_dir),
+            repo_id=out_repo,
+            repo_type="dataset",
+            path_in_repo="data/",
+            delete_patterns=["data/*.parquet"],
+            commit_message=(
+                f"Classify {stats['total']} chunks "
+                f"(threshold={music_threshold:.2f}, "
+                f"contaminated={stats['contaminated']})"
+            ),
+        )
+
+    log.append(
+        f"[{_ts()}] {short}: ✓ done → https://huggingface.co/datasets/{out_repo}"
+    )
+    stats["repo"] = repo_id
+    stats["out_repo"] = out_repo
+    return stats
+
+
+# ---------------------------------------------------------------------------
+# Gradio handler
+# ---------------------------------------------------------------------------
+
+
+def classify_repos(
+    repos_text: str,
+    out_suffix: str,
+    music_threshold: float,
+    private: bool,
+    hf_token_input: str,
+    progress: gr.Progress = gr.Progress(),
+) -> Generator[tuple[str, pd.DataFrame, str], None, None]:
+    """Main Gradio generator. Yields (log_text, summary_df, links_md)."""
+    token = hf_token_input.strip() or os.environ.get("HF_TOKEN") or None
+
+    repos = [
+        r.strip()
+        for line in repos_text.replace(",", "\n").splitlines()
+        for r in [line.strip()]
+        if r and "/" in r
+    ]
+    if not repos:
+        yield "No valid repo IDs found (expected owner/name format).", pd.DataFrame(), ""
+        return
+
+    suffix = out_suffix.strip() or "_classified"
+    if not suffix.startswith("_"):
+        suffix = "_" + suffix
+
+    log: list[str] = [
+        f"[{_ts()}] Starting: {len(repos)} repo(s), threshold={music_threshold:.2f}",
+        f"[{_ts()}] Loading AST model (MIT/ast-finetuned-audioset-10-10-0.4593)…",
+    ]
+    yield "\n".join(log), pd.DataFrame(), ""
+
+    # Warm up the model before processing
+    progress(0.0, desc="Loading model…")
+    try:
+        from music_detector import _get_ast_runtime
+        _get_ast_runtime("cpu")
+        log.append(f"[{_ts()}] Model loaded ✓")
+    except Exception as exc:
+        log.append(f"[{_ts()}] ✗ Model load failed: {exc}")
+        yield "\n".join(log), pd.DataFrame(), ""
+        return
+
+    yield "\n".join(log), pd.DataFrame(), ""
+
+    summary_rows: list[dict[str, Any]] = []
+    links: list[str] = []
+
+    for i, repo_id in enumerate(repos):
+        out_repo = _default_out_repo(repo_id)[:-len("_classified")] + suffix
+        progress(
+            (i + 0.1) / len(repos),
+            desc=f"[{i+1}/{len(repos)}] {repo_id.split('/')[-1]}",
+        )
+        log.append(f"\n[{_ts()}] ── Processing {i+1}/{len(repos)}: {repo_id} ──")
+        yield "\n".join(log), pd.DataFrame(summary_rows) if summary_rows else pd.DataFrame(), ""
+
+        try:
+            stats = _process_repo(
+                repo_id=repo_id,
+                out_repo=out_repo,
+                music_threshold=music_threshold,
+                private=private,
+                token=token,
+                log=log,
+            )
+            if "error" not in stats:
+                total = stats["total"]
+                contaminated = stats["contaminated"]
+                pct = 100 * contaminated / total if total else 0
+                summary_rows.append({
+                    "repo": repo_id.split("/")[-1],
+                    "chunks": total,
+                    "contaminated": contaminated,
+                    "contam_%": f"{pct:.1f}",
+                    "music": stats["has_music"],
+                    "technical": stats["technical"],
+                    "output": out_repo,
+                })
+                links.append(
+                    f"- [{out_repo}](https://huggingface.co/datasets/{out_repo})"
+                )
+            else:
+                summary_rows.append({
+                    "repo": repo_id.split("/")[-1],
+                    "chunks": 0, "contaminated": 0, "contam_%": "—",
+                    "music": 0, "technical": 0,
+                    "output": f"ERROR: {stats['error']}",
+                })
+        except Exception as exc:
+            log.append(f"[{_ts()}] ✗ {repo_id}: {exc}")
+            summary_rows.append({
+                "repo": repo_id.split("/")[-1],
+                "chunks": 0, "contaminated": 0, "contam_%": "—",
+                "music": 0, "technical": 0,
+                "output": f"ERROR: {exc}",
+            })
+
+        progress((i + 1) / len(repos), desc=f"Done {i+1}/{len(repos)}")
+        summary_df = pd.DataFrame(summary_rows)
+        links_md = "## Output datasets\n" + "\n".join(links) if links else ""
+        yield "\n".join(log), summary_df, links_md
+
+    log.append(f"\n[{_ts()}] === All done ===")
+    summary_df = pd.DataFrame(summary_rows)
+    links_md = "## Output datasets\n" + "\n".join(links) if links else ""
+    yield "\n".join(log), summary_df, links_md
+
+
+# ---------------------------------------------------------------------------
+# UI
+# ---------------------------------------------------------------------------
+
+DESCRIPTION = """
+# 🎵 Chunk Classifier
+
+Scores audio chunks for **music contamination** and **technical defects** using the
+[AST AudioSet model](https://huggingface.co/MIT/ast-finetuned-audioset-10-10-0.4593)
+(527 classes, multi-label sigmoid).
+
+**Input:** chunk dataset(s) produced by the AudioSet pipeline
+(`fosters/some-book-chunks`, one per line).
+
+**Output:** `<input>_classified` — all rows kept, new columns added:
+`music_score · has_music · contaminated · flags · top_labels · rms_db · clipping_ratio · max_silence_sec`
+
+The `audio` column is unchanged → the HF viewer shows an audio player.
+**Sort by `music_score` descending to quickly review suspicious chunks.**
+
+---
+"""
+
+with gr.Blocks(title="Chunk Classifier") as demo:
+    gr.Markdown(DESCRIPTION)
+
+    with gr.Row():
+        with gr.Column(scale=2):
+            repos_input = gr.Textbox(
+                label="Dataset repo IDs (one per line)",
+                placeholder="fosters/my-audiobook-chunks\nfosters/another-book-chunks",
+                lines=8,
+            )
+        with gr.Column(scale=1):
+            out_suffix = gr.Textbox(
+                label="Output suffix",
+                value="_classified",
+                info="Appended to each input repo name",
+            )
+            threshold = gr.Slider(
+                0.10, 0.60, value=DEFAULT_THRESHOLD, step=0.05,
+                label="Music threshold",
+                info="Lower = more sensitive (flag more). Biased to recall.",
+            )
+            private_toggle = gr.Checkbox(
+                label="Private output datasets",
+                value=True,
+            )
+            token_input = gr.Textbox(
+                label="HF Token (or set HF_TOKEN secret)",
+                type="password",
+                placeholder="hf_…",
+            )
+            run_btn = gr.Button("▶ Classify", variant="primary", size="lg")
+
+    log_out = gr.Textbox(
+        label="Progress log",
+        interactive=False,
+        lines=18,
+        max_lines=100,
+    )
+    summary_out = gr.Dataframe(
+        label="Per-repo summary",
+        headers=["repo", "chunks", "contaminated", "contam_%", "music", "technical", "output"],
+        wrap=True,
+    )
+    links_out = gr.Markdown()
+
+    run_btn.click(
+        classify_repos,
+        inputs=[repos_input, out_suffix, threshold, private_toggle, token_input],
+        outputs=[log_out, summary_out, links_out],
+    )
+
+demo.queue()
+demo.launch(server_name="0.0.0.0")

music_detector.py

@@ -0,0 +1,346 @@
+"""Music and audio quality detector based on the AST AudioSet model.
+
+Self-contained: only torch, transformers, numpy, and ffmpeg (subprocess) at runtime.
+All torch/transformers imports are lazy so this module can be imported without
+those deps installed — they only need to be present at call time.
+
+Main API
+--------
+judge_chunk_files_batched(paths, *, music_threshold, batch_size, device)
+    -> list[ChunkVerdict]
+
+Pure helpers (_split_samples_into_windows, _music_score_from_probs, …)
+are top-level so they can be unit-tested without a model.
+"""
+
+from __future__ import annotations
+
+import importlib
+import re
+import subprocess
+import threading
+from concurrent.futures import ThreadPoolExecutor
+from dataclasses import dataclass, field
+from pathlib import Path
+from typing import Any
+
+import numpy as np
+
+
+MODEL_ID = "MIT/ast-finetuned-audioset-10-10-0.4593"
+SAMPLE_RATE = 16_000
+# AST native input length (~10.24 s at 16 kHz)
+MAX_CHUNK_SAMPLES = int(10.24 * SAMPLE_RATE)  # 163_840
+
+MUSIC_CLASSES = {
+    "Music",
+    "Musical instrument",
+    "Singing",
+    "Choir",
+    "Piano",
+    "Guitar",
+    "Drum",
+    "Jingle (music)",
+}
+
+_SILENCE_START_RE = re.compile(r"silence_start:\s*([0-9]+(?:\.[0-9]+)?)")
+_SILENCE_END_RE = re.compile(r"silence_end:\s*([0-9]+(?:\.[0-9]+)?)")
+
+_AST_LOCK = threading.Lock()
+# (feature_extractor, model, music_indices)
+_AST_RUNTIME: tuple[Any, Any, list[int]] | None = None
+
+
+# ---------------------------------------------------------------------------
+# Result type
+# ---------------------------------------------------------------------------
+
+
+@dataclass
+class ChunkVerdict:
+    music_score: float
+    has_music: bool
+    clipping_ratio: float | None
+    rms_db: float | None
+    max_silence_sec: float | None
+    flags: list[str] = field(default_factory=list)   # ["clipping","long_silence","low_loudness"]
+    top_labels: list[str] = field(default_factory=list)  # top AST labels for transparency
+    contaminated: bool = False                            # has_music OR any flag
+
+
+# ---------------------------------------------------------------------------
+# Lazy model runtime (mirrors silero_vad._get_silero_runtime pattern)
+# ---------------------------------------------------------------------------
+
+
+def _get_ast_runtime(device: str = "cpu") -> tuple[Any, Any, list[int]]:
+    """Load model + feature extractor once; cache in module global."""
+    global _AST_RUNTIME
+    with _AST_LOCK:
+        if _AST_RUNTIME is None:
+            transformers = importlib.import_module("transformers")
+            AutoFE = getattr(transformers, "AutoFeatureExtractor")
+            AutoModel = getattr(transformers, "AutoModelForAudioClassification")
+
+            fe = AutoFE.from_pretrained(MODEL_ID)
+            model = AutoModel.from_pretrained(MODEL_ID)
+            model.to(device).eval()
+
+            # Resolve music class indices by label name — never hardcode ints
+            music_indices = [
+                i for i, label in model.config.id2label.items()
+                if label in MUSIC_CLASSES
+            ]
+            if not music_indices:
+                raise RuntimeError(
+                    f"None of MUSIC_CLASSES matched in {MODEL_ID} id2label. "
+                    "Check that this model has AudioSet labels."
+                )
+            _AST_RUNTIME = (fe, model, music_indices)
+    return _AST_RUNTIME
+
+
+# ---------------------------------------------------------------------------
+# Audio decode (ffmpeg, same approach as quality_exp/layer1/audio_probe.py)
+# ---------------------------------------------------------------------------
+
+
+def _decode_mono16k(path: str | Path) -> np.ndarray | None:
+    """Decode any audio file to mono float32 at SAMPLE_RATE Hz via ffmpeg.
+
+    Returns None on any error (bad file, ffmpeg missing, timeout).
+    """
+    try:
+        cmd = [
+            "ffmpeg", "-i", str(path),
+            "-f", "f32le", "-ac", "1", "-ar", str(SAMPLE_RATE),
+            "-v", "quiet", "-",
+        ]
+        result = subprocess.run(cmd, capture_output=True, timeout=60)
+        if result.returncode != 0 or not result.stdout:
+            return None
+        return np.frombuffer(result.stdout, dtype=np.float32).copy()
+    except Exception:
+        return None
+
+
+# ---------------------------------------------------------------------------
+# Pure helpers — testable without a model
+# ---------------------------------------------------------------------------
+
+
+def _split_samples_into_windows(samples: np.ndarray) -> list[np.ndarray]:
+    """Return 1 or 2 AST-compatible windows from a chunk.
+
+    Chunks ≤ MAX_CHUNK_SAMPLES → one window.
+    Longer chunks → beginning window + end window (may overlap).
+    """
+    if len(samples) <= MAX_CHUNK_SAMPLES:
+        return [samples]
+    return [samples[:MAX_CHUNK_SAMPLES], samples[-MAX_CHUNK_SAMPLES:]]
+
+
+def _music_score_from_probs(probs: np.ndarray, music_indices: list[int]) -> float:
+    """Max sigmoid probability across music class indices."""
+    if not music_indices:
+        return 0.0
+    return float(probs[music_indices].max())
+
+
+def _top_labels_from_probs(
+    probs: np.ndarray,
+    id2label: dict[int, str],
+    k: int = 5,
+) -> list[str]:
+    """Top-k label names by descending probability."""
+    top_idx = probs.argsort()[-k:][::-1]
+    return [id2label[int(i)] for i in top_idx if int(i) in id2label]
+
+
+def _probe_duration(path: str | Path) -> float | None:
+    """Return audio duration in seconds via ffprobe, or None on error."""
+    try:
+        cmd = [
+            "ffprobe", "-v", "error",
+            "-show_entries", "format=duration",
+            "-of", "default=noprint_wrappers=1:nokey=1",
+            str(path),
+        ]
+        result = subprocess.run(cmd, capture_output=True, text=True, timeout=15)
+        return float(result.stdout.strip())
+    except Exception:
+        return None
+
+
+def _compute_max_silence(path: str | Path) -> float | None:
+    """Return longest silence gap in seconds via ffmpeg silencedetect, or None on error.
+
+    Handles trailing silence (file ends while still silent) by using the file
+    duration as the implicit silence_end.
+    """
+    try:
+        # Note: do NOT use -v quiet here — it suppresses silencedetect filter messages.
+        # Use -hide_banner + -nostats to keep stderr clean while preserving filter output.
+        cmd = [
+            "ffmpeg", "-hide_banner", "-nostats",
+            "-i", str(path),
+            "-af", "silencedetect=noise=-35dB:d=0.5",
+            "-f", "null", "-",
+        ]
+        result = subprocess.run(cmd, capture_output=True, text=True, timeout=60)
+        gaps: list[float] = []
+        pending: float | None = None
+        for line in result.stderr.splitlines():
+            m = _SILENCE_START_RE.search(line)
+            if m:
+                pending = float(m.group(1))
+                continue
+            m = _SILENCE_END_RE.search(line)
+            if m and pending is not None:
+                gap = float(m.group(1)) - pending
+                if gap > 0:
+                    gaps.append(gap)
+                pending = None
+
+        # Trailing silence: file ended before silence_end was emitted
+        if pending is not None:
+            duration = _probe_duration(path)
+            if duration is not None and duration > pending:
+                gaps.append(duration - pending)
+
+        return float(max(gaps)) if gaps else None
+    except Exception:
+        return None
+
+
+def _signal_checks(
+    path: str | Path,
+    samples: np.ndarray,
+    *,
+    clipping_threshold: float = 0.001,
+    loudness_threshold_db: float = -45.0,
+    silence_threshold_sec: float = 3.0,
+) -> tuple[float | None, float | None, float | None, list[str]]:
+    """Return (clipping_ratio, rms_db, max_silence_sec, flags).
+
+    flags is a list of triggered quality issues: "clipping", "low_loudness", "long_silence".
+    """
+    flags: list[str] = []
+
+    clipping_ratio: float | None = None
+    rms_db: float | None = None
+    if len(samples) > 0:
+        clipping_ratio = float(np.mean(np.abs(samples) > 0.99))
+        rms = float(np.sqrt(np.mean(samples ** 2)))
+        rms_db = float(20.0 * np.log10(max(rms, 1e-9)))
+        if clipping_ratio > clipping_threshold:
+            flags.append("clipping")
+        if rms_db < loudness_threshold_db:
+            flags.append("low_loudness")
+
+    max_silence_sec = _compute_max_silence(path)
+    if max_silence_sec is not None and max_silence_sec > silence_threshold_sec:
+        flags.append("long_silence")
+
+    return clipping_ratio, rms_db, max_silence_sec, flags
+
+
+# ---------------------------------------------------------------------------
+# Public batch API
+# ---------------------------------------------------------------------------
+
+
+def judge_chunk_files_batched(
+    audio_paths: list[str | Path],
+    *,
+    music_threshold: float = 0.25,
+    batch_size: int = 32,
+    device: str = "cpu",
+    n_decode_workers: int = 4,
+) -> list[ChunkVerdict]:
+    """Classify a list of audio files. Returns one ChunkVerdict per file.
+
+    Processing:
+    1. Parallel decode + signal checks (ffmpeg, I/O bound).
+    2. Batched AST inference over all windows (compute bound, single-threaded).
+    """
+    torch = importlib.import_module("torch")
+    fe, model, music_indices = _get_ast_runtime(device)
+    id2label: dict[int, str] = model.config.id2label
+
+    n = len(audio_paths)
+    if n == 0:
+        return []
+
+    # Step 1: parallel decode + signal checks
+    def _process_one(path: str | Path) -> tuple[
+        np.ndarray | None, float | None, float | None, float | None, list[str]
+    ]:
+        samples = _decode_mono16k(path)
+        if samples is None:
+            return None, None, None, None, ["decode_error"]
+        clipping, rms_db, max_silence, flags = _signal_checks(path, samples)
+        return samples, clipping, rms_db, max_silence, flags
+
+    workers = min(n, n_decode_workers)
+    with ThreadPoolExecutor(max_workers=workers) as ex:
+        file_results = list(ex.map(_process_one, audio_paths))
+
+    # Step 2: build flat window list for batched AST inference
+    # windows[i] → chunk index window_owners[i]
+    windows: list[np.ndarray] = []
+    window_owners: list[int] = []
+    for chunk_idx, (samples, *_) in enumerate(file_results):
+        wins = (
+            [np.zeros(MAX_CHUNK_SAMPLES, dtype=np.float32)]
+            if samples is None
+            else _split_samples_into_windows(samples)
+        )
+        for w in wins:
+            windows.append(w)
+            window_owners.append(chunk_idx)
+
+    # Step 3: batched AST inference
+    win_music_scores: list[float] = []
+    win_top_labels: list[list[str]] = []
+
+    for i in range(0, len(windows), batch_size):
+        batch_wins = [w for w in windows[i:i + batch_size]]
+        inputs = fe(batch_wins, sampling_rate=SAMPLE_RATE, return_tensors="pt", padding=True)
+        inputs = {k: v.to(model.device) for k, v in inputs.items()}
+        with torch.no_grad():
+            logits = model(**inputs).logits
+        probs_batch = torch.sigmoid(logits).cpu().numpy()  # (B, 527)
+
+        for row_probs in probs_batch:
+            win_music_scores.append(_music_score_from_probs(row_probs, music_indices))
+            win_top_labels.append(_top_labels_from_probs(row_probs, id2label))
+
+    # Step 4: max-pool windows → per-chunk score
+    chunk_music_scores = [0.0] * n
+    chunk_top_labels: list[list[str]] = [[] for _ in range(n)]
+    for win_idx, chunk_idx in enumerate(window_owners):
+        score = win_music_scores[win_idx]
+        if score > chunk_music_scores[chunk_idx]:
+            chunk_music_scores[chunk_idx] = score
+            chunk_top_labels[chunk_idx] = win_top_labels[win_idx]
+
+    # Step 5: assemble verdicts
+    verdicts: list[ChunkVerdict] = []
+    for chunk_idx in range(n):
+        samples, clipping, rms_db, max_silence, flags = file_results[chunk_idx]
+        music_score = chunk_music_scores[chunk_idx]
+        has_music = music_score >= music_threshold
+        contaminated = has_music or bool(flags)
+        verdicts.append(ChunkVerdict(
+            music_score=music_score,
+            has_music=has_music,
+            clipping_ratio=clipping,
+            rms_db=rms_db,
+            max_silence_sec=max_silence,
+            flags=flags,
+            top_labels=chunk_top_labels[chunk_idx],
+            contaminated=contaminated,
+        ))
+
+    return verdicts

requirements.txt

@@ -0,0 +1,8 @@
+gradio>=4.0
+torch>=2.0
+transformers>=4.40
+huggingface_hub>=0.24
+hf-xet>=1.0.0
+pyarrow>=15
+numpy
+soundfile