Add temporary GigaSpeech2 conversion tab

features/__init__.py

@@ -18,6 +18,6 @@ and submit it from the tab with `core.jobs.submit_job(...)`, returning
 To add a feature: create the module and append it to `FEATURES` (tab order).
 """
 
-from . import audio_compress, text_extract
+from . import audio_compress, gigaspeech2_convert, text_extract
 
-FEATURES = [text_extract, audio_compress]
+FEATURES = [text_extract, audio_compress, gigaspeech2_convert]

features/gigaspeech2_convert.py

@@ -0,0 +1,270 @@
+"""Temporary GigaSpeech2 converter: disk-indexed transcripts and sequential archives."""
+from __future__ import annotations
+
+import io
+import json
+import os
+import re
+import shutil
+import sqlite3
+import tarfile
+import tempfile
+import traceback
+from concurrent.futures import ThreadPoolExecutor
+from pathlib import Path, PurePosixPath
+from typing import Optional
+
+import gradio as gr
+import pyarrow as pa
+import pyarrow.parquet as pq
+import soundfile as sf
+from datasets import Audio, Features, Value
+from huggingface_hub import CommitOperationAdd, DatasetCard, HfApi, hf_hub_download
+
+from core.common import AppContext, SPLIT_RE
+from core.jobs import submit_job
+from features.audio_compress import AudioEncoder, KEEP_SR, _upsert_config
+
+TITLE = "Convert GigaSpeech2 (temporary)"
+SOURCE = "speechcolab/gigaspeech2"
+
+
+def source_plan(api, subset, source_split, revision):
+    info = api.dataset_info(SOURCE, revision=revision or None)
+    files = {s.rfilename for s in info.siblings}
+    base = f"data/{subset}"
+    transcript = f"{base}/{source_split}.tsv"
+    if source_split == "train_refined":
+        archives = sorted((f for f in files if re.fullmatch(rf"{base}/train/\d+\.tar\.gz", f)),
+                          key=lambda f: int(PurePosixPath(f).name.split('.')[0]))
+    else:
+        archives = [f"{base}/{source_split}.tar.gz"]
+    if transcript not in files or not archives or any(f not in files for f in archives):
+        raise ValueError("Source transcript or archives are missing.")
+    return info.sha, transcript, archives
+
+
+def index_transcript(path, db):
+    db.execute("CREATE TABLE transcript (segment_id TEXT PRIMARY KEY, text TEXT NOT NULL)")
+    count = 0
+    with open(path, encoding="utf-8-sig") as stream:
+        for line_number, line in enumerate(stream, 1):
+            line = line.rstrip("\r\n")
+            if not line:
+                continue
+            parts = line.split("\t", 1)
+            if len(parts) != 2 or not parts[0]:
+                raise ValueError(f"Invalid transcript at line {line_number}")
+            db.execute("INSERT INTO transcript VALUES (?, ?)", parts)
+            count += 1
+            if count % 10000 == 0:
+                db.commit()
+    db.commit()
+    return count
+
+
+def convert_archive(archive, destination, db, subset, fmt, level, workers):
+    features = Features({"segment_id": Value("string"), "text": Value("string"),
+                         "audio": Audio(sampling_rate=16000), "duration": Value("float64"),
+                         "language": Value("string")})
+    schema = features.arrow_schema
+    matched = skipped = 0
+    batch = []
+
+    def encode(item):
+        segment_id, text, raw, path = item
+        info = sf.info(io.BytesIO(raw))
+        if info.samplerate != 16000 or info.channels != 1:
+            raise ValueError(f"Unexpected audio format for {segment_id}: {info.samplerate}Hz/{info.channels}ch")
+        audio = {"bytes": raw, "path": path}
+        if fmt != "Original WAV":
+            # Independent encoders: their statistics are mutable and must not be shared.
+            audio = AudioEncoder(fmt, level, None, "PCM_16", KEEP_SR, False)(audio)
+        return {"segment_id": segment_id, "text": text, "audio": audio,
+                "duration": info.duration, "language": subset}
+
+    db.execute("CREATE TEMP TABLE seen (segment_id TEXT PRIMARY KEY)")
+    try:
+        with pq.ParquetWriter(destination, schema, compression="zstd") as writer, \
+                ThreadPoolExecutor(max_workers=workers) as pool, tarfile.open(archive, "r|gz") as tar:
+            def flush():
+                writer.write_table(pa.Table.from_pylist(list(pool.map(encode, batch)), schema=schema))
+                batch.clear()
+
+            for member in tar:
+                if not member.isfile() or not member.name.lower().endswith(".wav"):
+                    continue
+                path = PurePosixPath(member.name)
+                segment_id = path.stem
+                result = db.execute("SELECT text FROM transcript WHERE segment_id=?", (segment_id,)).fetchone()
+                if result is None:
+                    skipped += 1
+                    continue
+                db.execute("INSERT INTO seen VALUES (?)", (segment_id,))
+                if member.size > 64 * 1024 * 1024:
+                    raise ValueError(f"Audio member too large: {member.name}")
+                with tar.extractfile(member) as audio_file:
+                    raw = audio_file.read()
+                batch.append((segment_id, result[0], raw, path.name))
+                matched += 1
+                if len(batch) >= 32:
+                    flush()
+            if batch:
+                flush()
+        return matched, skipped
+    finally:
+        db.execute("DROP TABLE seen")
+        db.commit()
+
+
+def convert_and_push(token, subset, source_split, revision, out_repo, out_subset, out_split,
+                     private, fmt, level, num_workers, files_per_commit, max_archives,
+                     progress=lambda *a, **k: None):
+    logs = []
+    workdir = tempfile.mkdtemp(prefix="gigaspeech2-")
+
+    def log(message, status="⏳ Converting GigaSpeech2…"):
+        logs.append(message)
+        print(message, flush=True)
+        return status, "\n".join(logs)
+
+    try:
+        out_repo, out_subset, out_split = out_repo.strip(), out_subset.strip(), out_split.strip()
+        if out_repo == SOURCE or not re.fullmatch(r"[\w.-]+/[\w.-]+", out_repo):
+            raise ValueError("Choose a destination namespace/repo different from the source.")
+        if not re.fullmatch(r"[\w.-]+", out_subset) or out_subset in (".", "..") or not SPLIT_RE.fullmatch(out_split):
+            raise ValueError("Invalid destination subset or split.")
+        if subset not in ("vi", "th", "id") or source_split not in ("train_refined", "dev", "test"):
+            raise ValueError("Invalid source subset/split.")
+        workers, per_commit, limit = int(num_workers), int(files_per_commit), int(max_archives)
+        if workers < 1 or per_commit < 1 or limit < 0 or fmt not in ("Original WAV", "FLAC", "Opus") or not 0 <= float(level) <= 1:
+            raise ValueError("Workers/files per upload must be positive; archive limit must be nonnegative.")
+        api = HfApi(token=token)
+        sha, transcript, archives = source_plan(api, subset, source_split, revision.strip())
+        prefix = f"{out_subset}/{out_split}"
+        manifest_path = f".gigaspeech2/{out_subset}/{out_split}.json"
+        settings = dict(source=SOURCE, revision=sha, subset=subset, source_split=source_split,
+                        format=fmt, level=float(level) if fmt != "Original WAV" else None,
+                        schema_version=1, archives=archives)
+        api.create_repo(out_repo, repo_type="dataset", private=bool(private), exist_ok=True)
+        existing_files = set(api.list_repo_files(out_repo, repo_type="dataset"))
+        expected = {f"{prefix}-{i:05d}-of-{len(archives):05d}.parquet" for i in range(len(archives))}
+        split_files = {f for f in existing_files if f.startswith(prefix + "-") and f.endswith(".parquet")}
+        completed = {}
+        if manifest_path in existing_files:
+            manifest_local = hf_hub_download(out_repo, manifest_path, repo_type="dataset", token=token,
+                                             local_dir=f"{workdir}/manifest")
+            manifest = json.loads(Path(manifest_local).read_text())
+            if manifest["settings"] != settings:
+                raise ValueError("Destination conversion settings differ. Use another destination subset/split.")
+            completed = manifest["completed"]
+        elif split_files:
+            raise ValueError("Destination split contains shards without a conversion manifest; choose another split.")
+        if split_files - expected or split_files != set(completed):
+            raise ValueError("Destination shards and conversion manifest disagree.")
+        todo = [(i, archive) for i, archive in enumerate(archives)
+                if f"{prefix}-{i:05d}-of-{len(archives):05d}.parquet" not in completed]
+        if limit:
+            todo = todo[:limit]
+        yield log(f"Source {subset}/{source_split} @ {sha}; {len(archives)} archives, {len(completed)} already uploaded.")
+        if not todo:
+            yield log("No remaining archives.", "### ✅ Conversion already complete")
+            return
+        yield log(f"Downloading transcript: {transcript}")
+        local_tsv = hf_hub_download(SOURCE, transcript, revision=sha, repo_type="dataset", token=token,
+                                    local_dir=f"{workdir}/transcript")
+        with sqlite3.connect(f"{workdir}/transcripts.sqlite") as db:
+            count = index_transcript(local_tsv, db)
+            shutil.rmtree(f"{workdir}/transcript")
+            yield log(f"Indexed {count:,} transcript rows on disk.")
+            pending = []
+            for k, (i, archive) in enumerate(todo, 1):
+                progress((k - 1) / len(todo), desc=f"Archive {i + 1}/{len(archives)}: download")
+                yield log(f"Downloading [{i + 1}/{len(archives)}] {archive}")
+                local_tar = hf_hub_download(SOURCE, archive, revision=sha, repo_type="dataset", token=token,
+                                            local_dir=f"{workdir}/archive")
+                destination = f"{prefix}-{i:05d}-of-{len(archives):05d}.parquet"
+                local_parquet = f"{workdir}/{i:05d}.parquet"
+                yield log(f"Converting {archive} with {workers} workers…")
+                rows, skipped = convert_archive(local_tar, local_parquet, db, subset, fmt, float(level), workers)
+                shutil.rmtree(f"{workdir}/archive")
+                completed[destination] = dict(archive=archive, rows=rows, skipped=skipped)
+                pending.append((local_parquet, destination))
+                # No yield between conversion and upload when the batch is ready.
+                if len(pending) >= per_commit or k == len(todo):
+                    card = DatasetCard.load(out_repo, token=token) if "README.md" in existing_files else DatasetCard(
+                        "---\nlicense: apache-2.0\ntask_categories:\n- automatic-speech-recognition\n---\n"
+                        "# GigaSpeech2 converted to Parquet\n\n"
+                        f"Source: [{SOURCE}](https://huggingface.co/datasets/{SOURCE}). "
+                        "The original dataset's terms of access apply. "
+                        "Only audio with a matching transcript is included.\n")
+                    card.data.configs = _upsert_config(card.data.get("configs", []), out_subset, out_split, prefix + "-*.parquet")
+                    manifest_file = f"{workdir}/conversion.json"
+                    Path(manifest_file).write_text(json.dumps(dict(settings=settings, completed=completed), ensure_ascii=False))
+                    api.create_commit(out_repo, repo_type="dataset", operations=[
+                        *[CommitOperationAdd(path_in_repo=dst, path_or_fileobj=local) for local, dst in pending],
+                        CommitOperationAdd(path_in_repo=manifest_path, path_or_fileobj=manifest_file),
+                        CommitOperationAdd(path_in_repo="README.md", path_or_fileobj=str(card).encode()),
+                    ], commit_message=f"Convert GigaSpeech2 {subset}/{source_split}: through archive {i}")
+                    existing_files.add("README.md")
+                    for local, _ in pending:
+                        os.remove(local)
+                    pending.clear()
+                    yield log(f"Uploaded through archive {i}: {rows:,} rows; {skipped:,} audio excluded by transcript.")
+                else:
+                    yield log(f"Prepared archive {i}: {rows:,} rows; awaiting upload batch ({len(pending)}/{per_commit}).")
+            if len(completed) == len(archives):
+                converted_rows = sum(item["rows"] for item in completed.values())
+                if converted_rows != count:
+                    raise ValueError(f"Transcript coverage mismatch: {converted_rows:,} audio rows / {count:,} transcript rows. Inspect source archives before using this split.")
+            progress(1, desc="Finished")
+            yield log(f"Uploaded {len(completed)}/{len(archives)} archives.",
+                      f"### ✅ Conversion finished\n\n[{out_repo}](https://huggingface.co/datasets/{out_repo}) · "
+                      f"`{out_subset}/{out_split}` · {len(completed)}/{len(archives)} archives. "
+                      "Submit again to resume remaining archives.")
+    except Exception as exc:
+        logs.append(traceback.format_exc())
+        yield f"### ❌ Failed\n\n{exc}\n\nUploaded batches are kept; submit again to resume.", "\n".join(logs)
+    finally:
+        shutil.rmtree(workdir, ignore_errors=True)
+
+
+def submit_convert(oauth_token: Optional[gr.OAuthToken], session_token: Optional[str], subset, source_split,
+                   revision, out_repo, out_subset, out_split, private, fmt, level, num_workers,
+                   files_per_commit, max_archives):
+    params = {k: v for k, v in locals().items() if k not in ("oauth_token", "session_token")}
+    return submit_job(oauth_token, session_token, feature=TITLE,
+                      title=f"GigaSpeech2 {subset}/{source_split} → {out_repo}", fn=convert_and_push, params=params)
+
+
+def build(ctx: AppContext):
+    gr.Markdown("Convert **speechcolab/gigaspeech2** on this Space: transcript → one tar.gz at a time → Parquet → Hub. "
+                "Columns: `segment_id`, `text`, `audio` (embedded bytes), `duration`, `language`.")
+    with gr.Accordion("📥 Source", open=True):
+        with gr.Row():
+            subset = gr.Dropdown(["vi", "th", "id"], value="vi", label="Source subset (language)")
+            source_split = gr.Dropdown(["train_refined", "dev", "test"], value="train_refined", label="Source transcript / split")
+            revision = gr.Textbox(value="main", label="Source revision")
+    with gr.Accordion("📤 Destination", open=True):
+        out_repo = gr.Textbox(label="Target dataset repo", placeholder="your-name/gigaspeech2-parquet")
+        with gr.Row():
+            out_subset = gr.Textbox(value="vi", label="Target subset (config)")
+            out_split = gr.Textbox(value="train", label="Target split")
+            private = gr.Checkbox(value=True, label="Create private repo")
+    with gr.Accordion("⚙️ Conversion", open=True):
+        with gr.Row():
+            fmt = gr.Dropdown(["Original WAV", "FLAC", "Opus"], value="FLAC", label="Audio format")
+            level = gr.Slider(0, 1, value=0.6, step=0.01, label="Compression level")
+        with gr.Row():
+            num_workers = gr.Number(value=os.cpu_count() or 2, precision=0, label="Conversion workers")
+            files_per_commit = gr.Number(value=1, precision=0, label="Parquet files per upload", info="One Parquet per archive; pending files use temporary disk.")
+            max_archives = gr.Number(value=0, precision=0, label="Max archives in this run (0 = all)")
+    gr.Markdown("Resume skips uploaded archives and checks source revision / conversion settings. "
+                "Temporary files are removed after upload. Follow or cancel in **📋 Jobs**; cancellation takes effect between stages. "
+                "Access to the gated source dataset must already be granted to your signed-in account.")
+    subset.input(lambda value: value, inputs=[subset], outputs=[out_subset])
+    run = gr.Button("🚀 Submit job: convert GigaSpeech2", variant="primary", elem_classes=["run-btn"])
+    status = gr.Markdown()
+    run.click(submit_convert, inputs=[ctx.session_token, subset, source_split, revision, out_repo,
+              out_subset, out_split, private, fmt, level, num_workers, files_per_commit, max_archives],
+              outputs=[status, ctx.submitted_job], concurrency_limit=None)