Fix: status bar instead of gr.Progress overlay, per-batch progress, total_rows from datasets-server

app.py

@@ -2,7 +2,6 @@
 
 Classifies audio chunks in one or more HF datasets using the AST AudioSet model.
 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.
 """
 
 from __future__ import annotations
@@ -23,58 +22,58 @@ import gradio as gr
 import pandas as pd
 import pyarrow as pa
 import pyarrow.parquet as pq
+import requests
 import torch
 from huggingface_hub import HfApi, hf_hub_download
 
-from music_detector import ChunkVerdict, _get_ast_runtime, judge_chunk_files_batched
+from music_detector import _get_ast_runtime, judge_chunk_files_batched
 
 # ---------------------------------------------------------------------------
 # Constants
 # ---------------------------------------------------------------------------
 
 N_CPUS = os.cpu_count() or 2
-torch.set_num_threads(N_CPUS)   # let BLAS use all CPUs for AST inference
+torch.set_num_threads(N_CPUS)
 
-BATCH_SIZE = 32
+BATCH_SIZE      = 32
 DEFAULT_THRESHOLD = 0.25
 DEFAULT_NAMESPACE = "fosters"
-LOG_MAX_LINES = 500             # keep only last N lines in the log box
+LOG_MAX_LINES   = 600
+DATASETS_SERVER = "https://datasets-server.huggingface.co"
 
-# New classifier columns added to every output shard
 _CLASSIFIER_FIELDS = [
-    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("max_silence_sec",pa.float32()),
+    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("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"},
+_HF_FEATURES = {
+    "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"},
 }
 
+_AUDIO_PA_TYPE = pa.struct([pa.field("bytes", pa.binary()), pa.field("path", pa.string())])
+
 _OUTPUT_SCHEMA = pa.schema(
     [
         pa.field("id", pa.string()),
-        pa.field("audio", pa.struct([
-            pa.field("bytes", pa.binary()),
-            pa.field("path",  pa.string()),
-        ])),
+        pa.field("audio", _AUDIO_PA_TYPE),
         *_CLASSIFIER_FIELDS,
     ],
-    metadata={"huggingface": json.dumps({"info": {"features": _HF_FEATURES_CLASSIFIER}})},
+    metadata={"huggingface": json.dumps({"info": {"features": _HF_FEATURES}})},
 )
 
 
@@ -89,9 +88,9 @@ def _ts() -> str:
 
 def _default_out_repo(in_repo: str, suffix: str) -> str:
     parts = in_repo.split("/")
-    name = parts[-1]
-    ns   = parts[0] if len(parts) > 1 else DEFAULT_NAMESPACE
-    suf  = suffix.strip()
+    ns    = parts[0] if len(parts) > 1 else DEFAULT_NAMESPACE
+    name  = parts[-1]
+    suf   = suffix.strip()
     if not suf.startswith("_"):
         suf = "_" + suf
     return f"{ns}/{name}{suf}"
@@ -108,59 +107,63 @@ def _list_parquet_shards(repo_id: str, token: str | None) -> list[str]:
     )
 
 
-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 _fetch_total_rows(repo_id: str, token: str | None) -> int | None:
+    """Get total row count via datasets-server (no download required)."""
+    try:
+        headers = {"Authorization": f"Bearer {token}"} if token else {}
+        r = requests.get(
+            f"{DATASETS_SERVER}/info",
+            params={"dataset": repo_id, "config": "default"},
+            headers=headers,
+            timeout=20,
+        )
+        if r.ok:
+            splits = r.json().get("dataset_info", {}).get("default", {}).get("splits", {})
+            return splits.get("train", {}).get("num_examples")
+    except Exception:
+        pass
+    return None
 
 
 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,
+            "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_PA_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()),
+            "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,
     )
 
 
-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}"
+def _fmt_chunk(chunk_id: str, v: Any) -> str:  # v: ChunkVerdict
+    if v.has_music:
+        icon = "🔴"
+    elif v.flags:
+        icon = "⚠️"
+    else:
+        icon = "✅"
+    labels = ", ".join(v.top_labels[:2])
+    flags  = f"  [{', '.join(v.flags)}]" if v.flags else ""
+    return f"  {icon} {chunk_id}  score={v.music_score:.3f}  {labels}{flags}"
 
 
 # ---------------------------------------------------------------------------
-# Per-repo generator
+# Per-repo processing generator
 # ---------------------------------------------------------------------------
 
+# Yields: (new_log_lines: list[str], status: str, chunks_done: int, stats: dict | None)
+# chunks_done counts chunks processed so far in this repo (for caller progress math)
 
 def _process_repo(
     repo_id: str,
@@ -168,29 +171,33 @@ def _process_repo(
     music_threshold: float,
     private: bool,
     token: str | None,
-) -> Generator[tuple[list[str], dict[str, Any] | None], None, None]:
-    """Generator: process one repo. Yields (log_lines, stats_or_None).
+) -> Generator[tuple[list[str], str, int, dict[str, Any] | None], None, 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]
 
-    # --- list shards ---
-    yield [f"[{_ts()}] {short}: listing parquet shards…"], None
+    # List shards
+    yield [f"[{_ts()}] {short}: listing shards…"], "Listing shards…", 0, 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)}
+        err = str(exc)
+        yield [f"[{_ts()}] {short}: ✗ {err}"], f"Error: {err}", 0, {"repo": repo_id, "error": err}
         return
 
     if not shard_names:
-        msg = f"[{_ts()}] {short}: ✗ no data/train-*.parquet shards found"
-        yield [msg], {"repo": repo_id, "error": "no shards"}
+        msg = "no data/train-*.parquet shards found"
+        yield [f"[{_ts()}] {short}: ✗ {msg}"], f"Error: {msg}", 0, {"repo": repo_id, "error": msg}
         return
 
-    yield [f"[{_ts()}] {short}: {len(shard_names)} shard(s) — creating output repo…"], None
+    # Get total row count (for progress display, not critical)
+    total_rows = _fetch_total_rows(repo_id, token)
+    total_str  = str(total_rows) if total_rows else "?"
+
+    yield [
+        f"[{_ts()}] {short}: {len(shard_names)} shard(s), {total_str} rows total"
+    ], f"Preparing… {len(shard_names)} shards, {total_str} chunks", 0, None
+
     api.create_repo(repo_id=out_repo, repo_type="dataset", exist_ok=True, private=private)
 
     stats: dict[str, Any] = {
@@ -204,29 +211,33 @@ def _process_repo(
         out_parts: list[Path] = []
 
         for shard_idx, shard_name in enumerate(shard_names):
-            # --- download shard ---
-            yield [f"[{_ts()}] {short}: shard {shard_idx+1}/{len(shard_names)} — downloading…"], None
-            t_shard = time.time()
+            yield [
+                f"[{_ts()}] {short}: shard {shard_idx+1}/{len(shard_names)} — downloading…"
+            ], f"[{shard_idx+1}/{len(shard_names)}] downloading shard…", stats["total"], None
+
+            t_dl = time.time()
             try:
                 shard_path = hf_hub_download(
                     repo_id=repo_id, filename=shard_name,
                     repo_type="dataset", token=token,
                 )
             except Exception as exc:
-                yield [f"[{_ts()}] {short}: ✗ download failed: {exc}"], None
+                yield [f"[{_ts()}] {short}: ✗ download failed: {exc}"], "Download error", stats["total"], 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()
+            n_batches = (n_rows + BATCH_SIZE - 1) // BATCH_SIZE
 
             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
+                f"{n_rows} rows, downloaded in {time.time()-t_dl:.1f}s"
+            ], (
+                f"[{shard_idx+1}/{len(shard_names)}] {n_rows} rows — starting inference…"
+            ), stats["total"], None
 
-            n_batches    = (n_rows + BATCH_SIZE - 1) // BATCH_SIZE
             shard_rows: list[dict[str, Any]] = []
 
             for batch_idx in range(n_batches):
@@ -234,81 +245,85 @@ def _process_repo(
                 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
+                batch_ids: list[str] = id_col[b_start:b_end]
+                batch_audio: list[dict] = [(audio_col[i].as_py() or {}) for i in range(b_start, b_end)]
+                batch_bytes: list[bytes] = [a.get("bytes") or b"" for a in batch_audio]
+                batch_paths: list[str]   = [a.get("path")  or "chunk.mp3" for a in batch_audio]
+
                 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}"
+                    files: list[Path] = []
+                    for k, (ab, ph) in enumerate(zip(batch_bytes, batch_paths)):
+                        p = Path(audio_tmp) / f"c{k:04d}{Path(ph).suffix or '.mp3'}"
                         p.write_bytes(ab)
-                        audio_files.append(p)
+                        files.append(p)
 
                     verdicts = judge_chunk_files_batched(
-                        audio_files,
+                        files,
                         music_threshold=music_threshold,
-                        batch_size=len(audio_files),
+                        batch_size=len(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))
+                for cid, ab, ph, v in zip(batch_ids, batch_bytes, batch_paths, verdicts):
+                    shard_rows.append({
+                        "id": cid, "audio_bytes": ab, "audio_path": ph,
+                        "music_score":     v.music_score,
+                        "has_music":       v.has_music,
+                        "contaminated":    v.contaminated,
+                        "flags":           ", ".join(v.flags),
+                        "top_labels":      ", ".join(v.top_labels[:3]),
+                        "rms_db":          v.rms_db,
+                        "clipping_ratio":  v.clipping_ratio,
+                        "max_silence_sec": v.max_silence_sec,
+                    })
 
-                # 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
+                stats["total"]        += b_end - b_start
+                stats["contaminated"] += n_cont
+                stats["has_music"]    += n_music
+                stats["technical"]    += n_tech
 
                 elapsed = time.time() - t_batch
+                done    = stats["total"]
+                cont_pct = 100 * stats["contaminated"] / max(done, 1)
+
+                # Build log lines: one per chunk + batch summary
+                chunk_lines = [_fmt_chunk(cid, v) for cid, v in zip(batch_ids, verdicts)]
+                summary_line = (
+                    f"[{_ts()}] batch {batch_idx+1}/{n_batches} "
+                    f"· shard {shard_idx+1}/{len(shard_names)} "
+                    f"· {elapsed:.1f}s "
+                    f"· total {done}/{total_str} chunks "
+                    f"· contaminated {stats['contaminated']} ({cont_pct:.1f}%)"
+                )
+
+                status = (
+                    f"[{shard_idx+1}/{len(shard_names)}] "
+                    f"batch {batch_idx+1}/{n_batches} "
+                    f"· {done}/{total_str} chunks "
+                    f"· {cont_pct:.1f}% contaminated"
+                )
+
+                yield [""] + chunk_lines + [summary_line], status, done, None
 
-                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
+            # 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
+                f"({len(shard_rows)} rows, {out_part.stat().st_size // 1024} KB)"
+            ], f"Shard {shard_idx+1} written, uploading…", stats["total"], None
 
-        # rename with final shard count
+        # Rename + upload
         n_shards = len(out_parts)
-        final: list[Path] = []
         for i, p in enumerate(out_parts):
-            dst = out_data_dir / f"train-{i:05d}-of-{n_shards:05d}.parquet"
-            p.rename(dst)
-            final.append(dst)
+            p.rename(out_data_dir / f"train-{i:05d}-of-{n_shards:05d}.parquet")
 
-        # upload
-        yield [f"[{_ts()}] {short}: uploading {n_shards} shard(s) → {out_repo}…"], None
+        yield [f"[{_ts()}] {short}: uploading {n_shards} shard(s) → {out_repo}…"], "Uploading…", stats["total"], None
         t_up = time.time()
         api.upload_folder(
             folder_path=str(out_data_dir),
@@ -322,28 +337,27 @@ def _process_repo(
                 f"contaminated={stats['contaminated']})"
             ),
         )
+
+        url = f"https://huggingface.co/datasets/{out_repo}"
         yield [
-            f"[{_ts()}] {short}: ✓ uploaded in {time.time()-t_up:.1f}s "
-            f"→ https://huggingface.co/datasets/{out_repo}"
-        ], stats
+            f"[{_ts()}] {short}: ✓ uploaded in {time.time()-t_up:.1f}s → {url}"
+        ], f"Done ✓  →  {out_repo}", stats["total"], stats
 
 
 # ---------------------------------------------------------------------------
-# Gradio handler
+# Gradio generator (no gr.Progress — avoids overlay)
 # ---------------------------------------------------------------------------
 
-
+# Outputs: log_text, status_text, summary_df, links_md
 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
+) -> Generator[tuple[str, str, pd.DataFrame, str], None, None]:
 
+    token = hf_token_input.strip() or os.environ.get("HF_TOKEN") or None
     repos = [
         r.strip()
         for line in repos_text.replace(",", "\n").splitlines()
@@ -351,49 +365,44 @@ def classify_repos(
         if r and "/" in r
     ]
     if not repos:
-        yield "No valid repo IDs (expected owner/name format).", pd.DataFrame(), ""
+        yield "No valid repo IDs (expected owner/name format).", "", pd.DataFrame(), ""
         return
 
-    log:          list[str]            = []
-    summary_rows: list[dict[str, Any]] = []
-    links:        list[str]            = []
+    log:  list[str]            = []
+    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
+    def _snapshot(status: str) -> tuple[str, str, pd.DataFrame, str]:
+        log_text = "\n".join(log[-LOG_MAX_LINES:])
+        df       = pd.DataFrame(rows) if rows else pd.DataFrame()
+        links_md = "## Output datasets\n" + "\n".join(links) if links else ""
+        return log_text, status, 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'}")
+    log.append(f"[{_ts()}] {len(repos)} repo(s) · threshold={music_threshold:.2f} · CPUs={N_CPUS}")
+    log.append(f"[{_ts()}] HF_XET_HIGH_PERFORMANCE={os.environ.get('HF_XET_HIGH_PERFORMANCE','off')}")
+    log.append(f"[{_ts()}] Loading AST model…")
+    yield _snapshot("Loading AST model…")
 
-    # 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:
         _get_ast_runtime("cpu")
         log.append(f"[{_ts()}] Model ready ✓")
     except Exception as exc:
         log.append(f"[{_ts()}] ✗ model load failed: {exc}")
-        yield _emit()
+        yield _snapshot(f"Error: {exc}")
         return
-    yield _emit()
 
+    yield _snapshot("Model ready")
     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()
+        log.append(f"\n[{_ts()}] {'─'*56}")
+        log.append(f"[{_ts()}] [{repo_idx+1}/{len(repos)}] {repo_id}  →  {out_repo}")
 
-        t_repo = time.time()
+        t_repo      = time.time()
         final_stats: dict[str, Any] | None = None
 
-        for new_lines, maybe_stats in _process_repo(
+        for new_lines, status, chunks_done, maybe_stats in _process_repo(
             repo_id=repo_id,
             out_repo=out_repo,
             music_threshold=music_threshold,
@@ -403,56 +412,39 @@ def classify_repos(
             log.extend(new_lines)
             if maybe_stats is not None:
                 final_stats = maybe_stats
+            yield _snapshot(f"[{repo_idx+1}/{len(repos)}] {status}")
 
-            # 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()
-
-        # Build summary row
+        # Summary row
         if final_stats and "error" not in final_stats:
-            total = final_stats["total"]
+            n     = 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,
+            rows.append({
+                "repo":      repo_id.split("/")[-1],
+                "chunks":    n,
+                "contam_%":  f"{100*cont/max(n,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"[{_ts()}] DONE  total={n}  contaminated={cont} ({100*cont/max(n,1):.1f}%)  "
                 f"music={final_stats['has_music']}  technical={final_stats['technical']}  "
                 f"elapsed={time.time()-t_repo:.0f}s"
             )
         elif final_stats:
-            err = final_stats.get("error", "unknown error")
-            summary_rows.append({
+            rows.append({
                 "repo": repo_id.split("/")[-1], "chunks": 0,
-                "contam_%": "—", "music": 0, "technical": 0, "time_s": "—",
-                "output": f"ERROR: {err}",
+                "contam_%": "—", "music": 0, "technical": 0,
+                "time_s": "—", "output": f"ERROR: {final_stats.get('error')}",
             })
-            log.append(f"[{_ts()}] {repo_id.split('/')[-1]}: ✗ {err}")
 
-        progress((repo_idx + 1) / len(repos), desc=f"Done {repo_idx+1}/{len(repos)}")
-        yield _emit()
+        yield _snapshot(f"[{repo_idx+1}/{len(repos)}] done")
 
-    log.append(f"\n[{_ts()}] {'='*60}")
+    log.append(f"\n[{_ts()}] {'='*56}")
     log.append(f"[{_ts()}] All done in {time.time()-t_total:.0f}s")
-    yield _emit()
+    yield _snapshot("All done ✓")
 
 
 # ---------------------------------------------------------------------------
@@ -466,10 +458,9 @@ 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 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`.
-
-Sort by `music_score` ↓ in the HF viewer and press play to spot-check flagged chunks.
+**Input** — chunk dataset repo IDs (one per line).
+**Output** — `<input>_classified` with `id · audio · music_score · has_music · contaminated · flags · top_labels · rms_db · clipping_ratio · max_silence_sec`.
+Sort by `music_score ↓` in the HF viewer and press play to spot-check.
 """
 
 with gr.Blocks(title="Chunk Classifier") as demo:
@@ -480,7 +471,7 @@ with gr.Blocks(title="Chunk Classifier") as demo:
             repos_input = gr.Textbox(
                 label="Dataset repo IDs (one per line)",
                 placeholder="fosters/my-book-chunks\nfosters/another-book-chunks",
-                lines=6,
+                lines=5,
             )
         with gr.Column(scale=1):
             out_suffix = gr.Textbox(
@@ -490,7 +481,7 @@ with gr.Blocks(title="Chunk Classifier") as demo:
             threshold = gr.Slider(
                 0.10, 0.60, value=DEFAULT_THRESHOLD, step=0.05,
                 label="Music threshold",
-                info="Lower = more sensitive. Default 0.25 biased to recall.",
+                info="Lower = more sensitive (recall-biased). Default 0.25.",
             )
             private_toggle = gr.Checkbox(label="Private output", value=True)
             token_input    = gr.Textbox(
@@ -499,17 +490,25 @@ with gr.Blocks(title="Chunk Classifier") as demo:
             )
             run_btn = gr.Button("▶ Classify", variant="primary", size="lg")
 
+    # Status line — replaces gr.Progress() which caused overlay issues
+    status_out = gr.Textbox(
+        label="Status",
+        interactive=False,
+        lines=1,
+        max_lines=1,
+    )
+
     with gr.Row():
         with gr.Column(scale=3):
             log_out = gr.Textbox(
-                label="Processing log  (per-chunk detail, updates after every batch)",
+                label="Processing log  (updates every batch ~30s)",
                 interactive=False,
-                lines=30,
+                lines=28,
                 max_lines=60,
             )
         with gr.Column(scale=2):
             summary_out = gr.Dataframe(
-                label="Per-repo summary",
+                label="Summary",
                 headers=["repo", "chunks", "contam_%", "music", "technical", "time_s", "output"],
                 wrap=True,
             )
@@ -518,7 +517,7 @@ with gr.Blocks(title="Chunk Classifier") as demo:
     run_btn.click(
         classify_repos,
         inputs=[repos_input, out_suffix, threshold, private_toggle, token_input],
-        outputs=[log_out, summary_out, links_out],
+        outputs=[log_out, status_out, summary_out, links_out],
     )
 
 demo.queue()