Fix: per-batch progress, detailed log, slim output schema

app.py

@@ -1,7 +1,7 @@
 """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.
+Writes a ``<input>_classified`` dataset with id + audio + classification columns only.
 The audio column is kept as-is so the HF viewer shows an audio player.
 """
 
@@ -15,51 +15,68 @@ import time
 from pathlib import Path
 from typing import Any, Generator
 
-# HF env must be set before any HF import
+# Must be set before any HF import
 os.environ.setdefault("HF_XET_HIGH_PERFORMANCE", "1")
+os.environ.setdefault("HF_HUB_DOWNLOAD_TIMEOUT", "120")
 
 import gradio as gr
 import pandas as pd
 import pyarrow as pa
 import pyarrow.parquet as pq
+import torch
 from huggingface_hub import HfApi, hf_hub_download
 
-from music_detector import ChunkVerdict, judge_chunk_files_batched
+from music_detector import ChunkVerdict, _get_ast_runtime, judge_chunk_files_batched
 
 # ---------------------------------------------------------------------------
 # Constants
 # ---------------------------------------------------------------------------
 
-SHARD_MAX_BYTES = 260 * 1024 * 1024   # 260 MB — HF viewer limit
+N_CPUS = os.cpu_count() or 2
+torch.set_num_threads(N_CPUS)   # let BLAS use all CPUs for AST inference
+
 BATCH_SIZE = 32
 DEFAULT_THRESHOLD = 0.25
 DEFAULT_NAMESPACE = "fosters"
+LOG_MAX_LINES = 500             # keep only last N lines in the log box
 
-# 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
+# New classifier columns added to every output shard
 _CLASSIFIER_FIELDS = [
-    pa.field("music_score", pa.float32()),
-    pa.field("has_music", pa.bool_()),
+    pa.field("music_score",    pa.float32()),
+    pa.field("has_music",      pa.bool_()),
+    pa.field("contaminated",   pa.bool_()),
+    pa.field("flags",          pa.string()),
+    pa.field("top_labels",     pa.string()),
+    pa.field("rms_db",         pa.float32()),
     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_()),
+    pa.field("max_silence_sec",pa.float32()),
 ]
 
+_HF_FEATURES_CLASSIFIER = {
+    "id":             {"_type": "Value", "dtype": "string"},
+    "audio":          {"_type": "Audio"},
+    "music_score":    {"_type": "Value", "dtype": "float32"},
+    "has_music":      {"_type": "Value", "dtype": "bool"},
+    "contaminated":   {"_type": "Value", "dtype": "bool"},
+    "flags":          {"_type": "Value", "dtype": "string"},
+    "top_labels":     {"_type": "Value", "dtype": "string"},
+    "rms_db":         {"_type": "Value", "dtype": "float32"},
+    "clipping_ratio": {"_type": "Value", "dtype": "float32"},
+    "max_silence_sec":{"_type": "Value", "dtype": "float32"},
+}
+
+_OUTPUT_SCHEMA = pa.schema(
+    [
+        pa.field("id", pa.string()),
+        pa.field("audio", pa.struct([
+            pa.field("bytes", pa.binary()),
+            pa.field("path",  pa.string()),
+        ])),
+        *_CLASSIFIER_FIELDS,
+    ],
+    metadata={"huggingface": json.dumps({"info": {"features": _HF_FEATURES_CLASSIFIER}})},
+)
+
 
 # ---------------------------------------------------------------------------
 # Helpers
@@ -70,15 +87,17 @@ def _ts() -> str:
     return datetime.datetime.now().strftime("%H:%M:%S")
 
 
-def _default_out_repo(in_repo: str) -> str:
+def _default_out_repo(in_repo: str, suffix: 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"
+    name = parts[-1]
+    ns   = parts[0] if len(parts) > 1 else DEFAULT_NAMESPACE
+    suf  = suffix.strip()
+    if not suf.startswith("_"):
+        suf = "_" + suf
+    return f"{ns}/{name}{suf}"
 
 
 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
@@ -89,72 +108,57 @@ def _list_parquet_shards(repo_id: str, token: str | None) -> list[str]:
     )
 
 
-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},
+def _build_output_row(
+    chunk_id: str,
+    audio_bytes: bytes,
+    audio_path_hint: str,
+    verdict: ChunkVerdict,
+) -> dict[str, Any]:
+    return {
+        "id":             chunk_id,
+        "audio":          {"bytes": audio_bytes, "path": audio_path_hint},
+        "music_score":    verdict.music_score,
+        "has_music":      verdict.has_music,
+        "contaminated":   verdict.contaminated,
+        "flags":          ", ".join(verdict.flags),
+        "top_labels":     ", ".join(verdict.top_labels[:3]),
+        "rms_db":         verdict.rms_db,
+        "clipping_ratio": verdict.clipping_ratio,
+        "max_silence_sec":verdict.max_silence_sec,
+    }
+
+
+def _rows_to_table(rows: list[dict[str, Any]]) -> pa.Table:
+    audio_type = _OUTPUT_SCHEMA.field("audio").type
+    return pa.table(
+        {
+            "id":             pa.array([r["id"] for r in rows], type=pa.string()),
+            "audio":          pa.array(
+                [{"bytes": r["audio"]["bytes"], "path": r["audio"]["path"]} for r in rows],
+                type=audio_type,
+            ),
+            "music_score":    pa.array([r["music_score"]     for r in rows], type=pa.float32()),
+            "has_music":      pa.array([r["has_music"]       for r in rows], type=pa.bool_()),
+            "contaminated":   pa.array([r["contaminated"]    for r in rows], type=pa.bool_()),
+            "flags":          pa.array([r["flags"]           for r in rows], type=pa.string()),
+            "top_labels":     pa.array([r["top_labels"]      for r in rows], type=pa.string()),
+            "rms_db":         pa.array([r["rms_db"]          for r in rows], type=pa.float32()),
+            "clipping_ratio": pa.array([r["clipping_ratio"]  for r in rows], type=pa.float32()),
+            "max_silence_sec":pa.array([r["max_silence_sec"] for r in rows], type=pa.float32()),
+        },
+        schema=_OUTPUT_SCHEMA,
     )
 
-    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)
+def _fmt_verdict(chunk_id: str, v: ChunkVerdict) -> str:
+    status   = "🔴 MUSIC"    if v.has_music  else ("⚠️ TECH" if v.flags else "✅ ok")
+    labels   = ", ".join(v.top_labels[:2])
+    flags    = f"  [{', '.join(v.flags)}]" if v.flags else ""
+    return f"  {chunk_id}: score={v.music_score:.3f}  {labels}{flags}  {status}"
 
 
 # ---------------------------------------------------------------------------
-# Per-repo processing
+# Per-repo generator
 # ---------------------------------------------------------------------------
 
 
@@ -164,22 +168,35 @@ def _process_repo(
     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)
+) -> Generator[tuple[list[str], dict[str, Any] | None], None, None]:
+    """Generator: process one repo. Yields (log_lines, stats_or_None).
+
+    Yields after every BATCH_SIZE chunks so the UI stays responsive.
+    Final yield carries the completed 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)
+    # --- list shards ---
+    yield [f"[{_ts()}] {short}: listing parquet shards…"], None
+    try:
+        shard_names = _list_parquet_shards(repo_id, token)
+    except Exception as exc:
+        yield [f"[{_ts()}] {short}: ✗ cannot list shards: {exc}"], {"repo": repo_id, "error": str(exc)}
+        return
+
     if not shard_names:
-        log.append(f"[{_ts()}] {short}: ✗ no parquet shards found under data/")
-        return {"repo": repo_id, "error": "no shards"}
+        msg = f"[{_ts()}] {short}: ✗ no data/train-*.parquet shards found"
+        yield [msg], {"repo": repo_id, "error": "no shards"}
+        return
 
-    log.append(f"[{_ts()}] {short}: {len(shard_names)} shard(s) found")
+    yield [f"[{_ts()}] {short}: {len(shard_names)} shard(s) — creating output repo…"], None
     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}
+    stats: dict[str, Any] = {
+        "repo": repo_id, "out_repo": out_repo,
+        "total": 0, "contaminated": 0, "has_music": 0, "technical": 0,
+    }
 
     with tempfile.TemporaryDirectory() as tmp:
         out_data_dir = Path(tmp) / "data"
@@ -187,67 +204,112 @@ def _process_repo(
         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",
+            # --- download shard ---
+            yield [f"[{_ts()}] {short}: shard {shard_idx+1}/{len(shard_names)} — downloading…"], None
+            t_shard = time.time()
+            try:
+                shard_path = hf_hub_download(
+                    repo_id=repo_id, filename=shard_name,
+                    repo_type="dataset", token=token,
                 )
-            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"
+            except Exception as exc:
+                yield [f"[{_ts()}] {short}: ✗ download failed: {exc}"], None
+                continue
+
+            in_table  = pq.read_table(shard_path)
+            n_rows    = len(in_table)
+            audio_col = in_table.column("audio")
+            id_col    = in_table.column("id").to_pylist()
+
+            yield [
+                f"[{_ts()}] {short}: shard {shard_idx+1}/{len(shard_names)} — "
+                f"{n_rows} rows downloaded in {time.time()-t_shard:.1f}s, classifying…"
+            ], None
+
+            n_batches    = (n_rows + BATCH_SIZE - 1) // BATCH_SIZE
+            shard_rows: list[dict[str, Any]] = []
+
+            for batch_idx in range(n_batches):
+                b_start = batch_idx * BATCH_SIZE
+                b_end   = min(b_start + BATCH_SIZE, n_rows)
+                t_batch = time.time()
+
+                batch_ids: list[str]   = id_col[b_start:b_end]
+                batch_audio_bytes: list[bytes] = [
+                    (audio_col[i].as_py() or {}).get("bytes") or b""
+                    for i in range(b_start, b_end)
+                ]
+                batch_path_hints: list[str] = [
+                    (audio_col[i].as_py() or {}).get("path") or "chunk.mp3"
+                    for i in range(b_start, b_end)
+                ]
+
+                # write temp audio files for the batch
+                with tempfile.TemporaryDirectory() as audio_tmp:
+                    audio_files: list[Path] = []
+                    for k, (ab, ph) in enumerate(zip(batch_audio_bytes, batch_path_hints)):
+                        suffix = Path(ph).suffix or ".mp3"
+                        p = Path(audio_tmp) / f"chunk_{k:04d}{suffix}"
+                        p.write_bytes(ab)
+                        audio_files.append(p)
+
+                    verdicts = judge_chunk_files_batched(
+                        audio_files,
+                        music_threshold=music_threshold,
+                        batch_size=len(audio_files),
+                        device="cpu",
+                    )
+
+                # accumulate output rows
+                for chunk_id, audio_bytes, ph, v in zip(
+                    batch_ids, batch_audio_bytes, batch_path_hints, verdicts
+                ):
+                    shard_rows.append(_build_output_row(chunk_id, audio_bytes, ph, v))
+
+                # update stats
+                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)
+                stats["total"]       += b_end - b_start
+                stats["contaminated"]+= n_cont
+                stats["has_music"]   += n_music
+                stats["technical"]   += n_tech
+
+                elapsed = time.time() - t_batch
+
+                log_lines  = [""]   # blank separator
+                log_lines += [_fmt_verdict(cid, v) for cid, v in zip(batch_ids, verdicts)]
+                log_lines += [
+                    f"[{_ts()}] batch {batch_idx+1}/{n_batches}  "
+                    f"({b_end - b_start} chunks, {elapsed:.1f}s)  "
+                    f"shard {shard_idx+1}/{len(shard_names)}  "
+                    f"total {stats['total']} processed  "
+                    f"contaminated {stats['contaminated']} "
+                    f"({100*stats['contaminated']/max(stats['total'],1):.1f}%)"
+                ]
+                yield log_lines, None
+
+            # write shard
+            out_table = _rows_to_table(shard_rows)
+            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)
+            yield [
+                f"[{_ts()}] {short}: shard {shard_idx+1} written "
+                f"({len(shard_rows)} rows, {out_part.stat().st_size//1024} KB)"
+            ], None
 
-            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
+        # rename with final shard count
         n_shards = len(out_parts)
-        final_parts: list[Path] = []
+        final: 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)
+            dst = out_data_dir / f"train-{i:05d}-of-{n_shards:05d}.parquet"
+            p.rename(dst)
+            final.append(dst)
 
-        log.append(f"[{_ts()}] {short}: uploading {n_shards} shard(s) → {out_repo}…")
+        # upload
+        yield [f"[{_ts()}] {short}: uploading {n_shards} shard(s) → {out_repo}…"], None
+        t_up = time.time()
         api.upload_folder(
             folder_path=str(out_data_dir),
             repo_id=out_repo,
@@ -260,13 +322,10 @@ def _process_repo(
                 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
+        yield [
+            f"[{_ts()}] {short}: ✓ uploaded in {time.time()-t_up:.1f}s "
+            f"→ https://huggingface.co/datasets/{out_repo}"
+        ], stats
 
 
 # ---------------------------------------------------------------------------
@@ -292,94 +351,108 @@ def classify_repos(
         if r and "/" in r
     ]
     if not repos:
-        yield "No valid repo IDs found (expected owner/name format).", pd.DataFrame(), ""
+        yield "No valid repo IDs (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…")
+    log:          list[str]            = []
+    summary_rows: list[dict[str, Any]] = []
+    links:        list[str]            = []
+
+    def _emit() -> tuple[str, pd.DataFrame, str]:
+        log_text   = "\n".join(log[-LOG_MAX_LINES:])
+        df         = pd.DataFrame(summary_rows) if summary_rows else pd.DataFrame()
+        links_md   = "## Output datasets\n" + "\n".join(links) if links else ""
+        return log_text, df, links_md
+
+    log.append(f"[{_ts()}] Starting: {len(repos)} repo(s), threshold={music_threshold:.2f}")
+    log.append(f"[{_ts()}] CPUs={N_CPUS}, BATCH_SIZE={BATCH_SIZE}, "
+               f"HF_XET={'1' if os.environ.get('HF_XET_HIGH_PERFORMANCE') else 'off'}")
+
+    # Warm up model
+    progress(0.0, desc="Loading AST model…")
+    log.append(f"[{_ts()}] Loading AST model (MIT/ast-finetuned-audioset-10-10-0.4593)…")
+    yield _emit()
     try:
-        from music_detector import _get_ast_runtime
         _get_ast_runtime("cpu")
-        log.append(f"[{_ts()}] Model loaded ✓")
+        log.append(f"[{_ts()}] Model ready ✓")
     except Exception as exc:
-        log.append(f"[{_ts()}] ✗ Model load failed: {exc}")
-        yield "\n".join(log), pd.DataFrame(), ""
+        log.append(f"[{_ts()}] ✗ model load failed: {exc}")
+        yield _emit()
         return
+    yield _emit()
+
+    t_total = time.time()
+
+    for repo_idx, repo_id in enumerate(repos):
+        out_repo = _default_out_repo(repo_id, out_suffix)
+        log.append(f"\n[{_ts()}] {'─'*60}")
+        log.append(f"[{_ts()}] [{repo_idx+1}/{len(repos)}] {repo_id}")
+        log.append(f"[{_ts()}] output → {out_repo}")
+        yield _emit()
+
+        t_repo = time.time()
+        final_stats: dict[str, Any] | None = None
+
+        for new_lines, maybe_stats in _process_repo(
+            repo_id=repo_id,
+            out_repo=out_repo,
+            music_threshold=music_threshold,
+            private=private,
+            token=token,
+        ):
+            log.extend(new_lines)
+            if maybe_stats is not None:
+                final_stats = maybe_stats
+
+            # Update progress bar (rough estimate by chunk count)
+            total_done = sum(r.get("total", 0) for r in summary_rows)
+            if final_stats:
+                total_done += final_stats.get("total", 0)
+            progress(
+                (repo_idx + min(
+                    (final_stats or {}).get("total", 0) / max(
+                        (final_stats or {}).get("total", 1), 1
+                    ), 1.0
+                )) / len(repos),
+                desc=f"[{repo_idx+1}/{len(repos)}] {repo_id.split('/')[-1]}",
+            )
+            yield _emit()
 
-    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,
+        # Build summary row
+        if final_stats and "error" not in final_stats:
+            total = final_stats["total"]
+            cont  = final_stats["contaminated"]
+            summary_rows.append({
+                "repo":       repo_id.split("/")[-1],
+                "chunks":     total,
+                "contam_%":   f"{100*cont/max(total,1):.1f}",
+                "music":      final_stats["has_music"],
+                "technical":  final_stats["technical"],
+                "time_s":     f"{time.time()-t_repo:.0f}",
+                "output":     out_repo,
+            })
+            links.append(f"- [{out_repo}](https://huggingface.co/datasets/{out_repo})")
+            log.append(
+                f"[{_ts()}] {repo_id.split('/')[-1]}: DONE  "
+                f"total={total}  contaminated={cont} ({100*cont/max(total,1):.1f}%)  "
+                f"music={final_stats['has_music']}  technical={final_stats['technical']}  "
+                f"elapsed={time.time()-t_repo:.0f}s"
             )
-            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}")
+        elif final_stats:
+            err = final_stats.get("error", "unknown error")
             summary_rows.append({
-                "repo": repo_id.split("/")[-1],
-                "chunks": 0, "contaminated": 0, "contam_%": "—",
-                "music": 0, "technical": 0,
-                "output": f"ERROR: {exc}",
+                "repo": repo_id.split("/")[-1], "chunks": 0,
+                "contam_%": "—", "music": 0, "technical": 0, "time_s": "—",
+                "output": f"ERROR: {err}",
             })
+            log.append(f"[{_ts()}] {repo_id.split('/')[-1]}: ✗ {err}")
 
-        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
+        progress((repo_idx + 1) / len(repos), desc=f"Done {repo_idx+1}/{len(repos)}")
+        yield _emit()
 
-    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
+    log.append(f"\n[{_ts()}] {'='*60}")
+    log.append(f"[{_ts()}] All done in {time.time()-t_total:.0f}s")
+    yield _emit()
 
 
 # ---------------------------------------------------------------------------
@@ -389,20 +462,14 @@ def classify_repos(
 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).
+Scores audio chunks for **music contamination** and **technical defects** using
+[MIT/ast-finetuned-audioset-10-10-0.4593](https://huggingface.co/MIT/ast-finetuned-audioset-10-10-0.4593)
+(527 AudioSet classes, multi-label sigmoid).
 
-**Input:** chunk dataset(s) produced by the AudioSet pipeline
-(`fosters/some-book-chunks`, one per line).
+**Input** — chunk dataset repo IDs (one per line, `owner/name` format).
+**Output** — `<input>_classified` with: `id · audio · music_score · has_music · contaminated · flags · top_labels · rms_db · clipping_ratio · max_silence_sec`.
 
-**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.**
-
----
+Sort by `music_score` ↓ in the HF viewer and press play to spot-check flagged chunks.
 """
 
 with gr.Blocks(title="Chunk Classifier") as demo:
@@ -412,43 +479,41 @@ with gr.Blocks(title="Chunk Classifier") as demo:
         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,
+                placeholder="fosters/my-book-chunks\nfosters/another-book-chunks",
+                lines=6,
             )
         with gr.Column(scale=1):
             out_suffix = gr.Textbox(
-                label="Output suffix",
-                value="_classified",
+                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,
+                info="Lower = more sensitive. Default 0.25 biased to recall.",
             )
-            token_input = gr.Textbox(
+            private_toggle = gr.Checkbox(label="Private output", value=True)
+            token_input    = gr.Textbox(
                 label="HF Token (or set HF_TOKEN secret)",
-                type="password",
-                placeholder="hf_…",
+                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()
+    with gr.Row():
+        with gr.Column(scale=3):
+            log_out = gr.Textbox(
+                label="Processing log  (per-chunk detail, updates after every batch)",
+                interactive=False,
+                lines=30,
+                max_lines=60,
+            )
+        with gr.Column(scale=2):
+            summary_out = gr.Dataframe(
+                label="Per-repo summary",
+                headers=["repo", "chunks", "contam_%", "music", "technical", "time_s", "output"],
+                wrap=True,
+            )
+            links_out = gr.Markdown()
 
     run_btn.click(
         classify_repos,