Upload app(1).py

app(1).py

@@ -0,0 +1,801 @@
+from __future__ import annotations
+
+import csv
+import json
+import os
+import re
+import tempfile
+from pathlib import Path
+from typing import Any
+
+import gradio as gr
+import pyarrow as pa
+import pyarrow.parquet as pq
+from huggingface_hub import (
+    CommitOperationAdd,
+    CommitOperationDelete,
+    HfApi,
+    snapshot_download,
+)
+
+
+AUDIO_SUFFIXES = {
+    ".mp3",
+    ".wav",
+    ".ogg",
+    ".opus",
+    ".flac",
+    ".m4a",
+    ".aac",
+}
+
+DEFAULT_SHARD_TARGET_MB = 230
+HARD_SHARD_LIMIT_MB = 260
+
+
+def normalize_repo_id(value: str) -> str:
+    value = value.strip().rstrip("/")
+
+    if "huggingface.co/datasets/" in value:
+        value = value.split("huggingface.co/datasets/", 1)[1]
+
+    value = value.split("/tree/", 1)[0]
+    value = value.split("/blob/", 1)[0]
+    value = value.split("?", 1)[0]
+    value = value.strip("/")
+
+    return value
+
+
+def slugify(text: str, max_len: int = 80) -> str:
+    text = text.lower()
+    text = re.sub(r"[^\w\s.-]", "", text, flags=re.UNICODE)
+    text = re.sub(r"[\s_-]+", "_", text)
+    return text.strip("_.-")[:max_len] or "audio"
+
+
+def get_token(token_from_ui: str | None) -> str:
+    token = (token_from_ui or "").strip() or os.environ.get("HF_TOKEN", "").strip()
+
+    if not token:
+        raise gr.Error(
+            "HF token не знойдзены. Дадай HF_TOKEN у Space Settings → Secrets "
+            "або ўвядзі token у поле HF token override."
+        )
+
+    return token
+
+
+def discover_audio_files(local_dir: Path) -> list[Path]:
+    files = []
+
+    for path in local_dir.rglob("*"):
+        if path.is_file() and path.suffix.lower() in AUDIO_SUFFIXES:
+            files.append(path)
+
+    return sorted(files, key=lambda p: p.as_posix())
+
+
+def load_metadata_maps(local_dir: Path) -> dict[str, dict[str, str]]:
+    """
+    Reads metadata.csv files if they exist.
+
+    Supports:
+      file_name
+      filename
+      audio
+      path
+
+    Stores metadata by:
+      relative path
+      basename
+    """
+    result: dict[str, dict[str, str]] = {}
+
+    for metadata_path in local_dir.rglob("metadata.csv"):
+        try:
+            with metadata_path.open("r", encoding="utf-8-sig", newline="") as f:
+                reader = csv.DictReader(f)
+
+                for row in reader:
+                    raw_file = (
+                        row.get("file_name")
+                        or row.get("filename")
+                        or row.get("audio")
+                        or row.get("path")
+                        or ""
+                    ).strip()
+
+                    if not raw_file:
+                        continue
+
+                    raw_file = raw_file.replace("\\", "/")
+                    basename = Path(raw_file).name
+
+                    candidate = metadata_path.parent / raw_file
+
+                    try:
+                        relative = candidate.resolve().relative_to(local_dir.resolve()).as_posix()
+                    except Exception:
+                        relative = raw_file
+
+                    cleaned_row = {
+                        str(k).strip(): str(v).strip()
+                        for k, v in row.items()
+                        if k is not None and v is not None
+                    }
+
+                    result[relative] = cleaned_row
+                    result[basename] = cleaned_row
+
+        except Exception:
+            continue
+
+    return result
+
+
+def metadata_for_audio(
+    audio_path: Path,
+    local_dir: Path,
+    metadata_maps: dict[str, dict[str, str]],
+) -> dict[str, str]:
+    rel = audio_path.relative_to(local_dir).as_posix()
+    name = audio_path.name
+
+    return metadata_maps.get(rel) or metadata_maps.get(name) or {}
+
+
+def value_from_meta(
+    meta: dict[str, str],
+    *keys: str,
+    default: str = "",
+) -> str:
+    for key in keys:
+        value = meta.get(key)
+        if value:
+            return value
+    return default
+
+
+def build_schema() -> pa.Schema:
+    audio_type = pa.struct(
+        [
+            pa.field("bytes", pa.binary()),
+            pa.field("path", pa.string()),
+        ]
+    )
+
+    hf_meta = {
+        "info": {
+            "features": {
+                "id": {"_type": "Value", "dtype": "string"},
+                "audio": {"_type": "Audio"},
+                "title": {"_type": "Value", "dtype": "string"},
+                "language": {"_type": "Value", "dtype": "string"},
+                "file_name": {"_type": "Value", "dtype": "string"},
+                "filename": {"_type": "Value", "dtype": "string"},
+                "Author": {"_type": "Value", "dtype": "string"},
+                "Title": {"_type": "Value", "dtype": "string"},
+                "Narrator": {"_type": "Value", "dtype": "string"},
+                "Source Group": {"_type": "Value", "dtype": "string"},
+                "original_file_name": {"_type": "Value", "dtype": "string"},
+                "original_extension": {"_type": "Value", "dtype": "string"},
+                "file_size_bytes": {"_type": "Value", "dtype": "int64"},
+            }
+        }
+    }
+
+    return pa.schema(
+        [
+            pa.field("id", pa.string()),
+            pa.field("audio", audio_type),
+            pa.field("title", pa.string()),
+            pa.field("language", pa.string()),
+            pa.field("file_name", pa.string()),
+            pa.field("filename", pa.string()),
+            pa.field("Author", pa.string()),
+            pa.field("Title", pa.string()),
+            pa.field("Narrator", pa.string()),
+            pa.field("Source Group", pa.string()),
+            pa.field("original_file_name", pa.string()),
+            pa.field("original_extension", pa.string()),
+            pa.field("file_size_bytes", pa.int64()),
+        ],
+        metadata={
+            b"huggingface": json.dumps(hf_meta, ensure_ascii=False).encode("utf-8")
+        },
+    )
+
+
+def rows_to_table(rows: list[dict[str, Any]]) -> pa.Table:
+    schema = build_schema()
+    audio_type = 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,
+            ),
+            "title": pa.array([r["title"] for r in rows], type=pa.string()),
+            "language": pa.array([r["language"] for r in rows], type=pa.string()),
+            "file_name": pa.array([r["file_name"] for r in rows], type=pa.string()),
+            "filename": pa.array([r["filename"] for r in rows], type=pa.string()),
+            "Author": pa.array([r["Author"] for r in rows], type=pa.string()),
+            "Title": pa.array([r["Title"] for r in rows], type=pa.string()),
+            "Narrator": pa.array([r["Narrator"] for r in rows], type=pa.string()),
+            "Source Group": pa.array([r["Source Group"] for r in rows], type=pa.string()),
+            "original_file_name": pa.array(
+                [r["original_file_name"] for r in rows],
+                type=pa.string(),
+            ),
+            "original_extension": pa.array(
+                [r["original_extension"] for r in rows],
+                type=pa.string(),
+            ),
+            "file_size_bytes": pa.array(
+                [r["file_size_bytes"] for r in rows],
+                type=pa.int64(),
+            ),
+        },
+        schema=schema,
+    )
+
+
+def write_parquet_shard(rows: list[dict[str, Any]], path: Path) -> None:
+    table = rows_to_table(rows)
+
+    pq.write_table(
+        table,
+        path,
+        row_group_size=1,
+        compression="snappy",
+    )
+
+
+def make_dataset_card(
+    source_repo: str,
+    title: str,
+    language: str,
+    row_count: int,
+    shard_count: int,
+) -> str:
+    return f"""---
+configs:
+- config_name: default
+  data_files:
+  - split: train
+    path: data/train/*.parquet
+---
+
+# {title}
+
+This is a Hugging Face Parquet input dataset for an audio pipeline.
+
+Source dataset: `{source_repo}`
+
+## Format
+
+- config: `default`
+- split: `train`
+- format: `parquet`
+- id column: `id`
+- audio column: `audio`
+- rows: `{row_count}`
+- shards: `{shard_count}`
+- language: `{language}`
+
+The `audio` column is embedded into Parquet as Hugging Face `Audio`:
+
+```python
+audio = {{
+    "path": "file.mp3",
+    "bytes": b"..."
+}}
+```
+
+## Columns
+
+- `id`
+- `audio`
+- `title`
+- `language`
+- `file_name`
+- `filename`
+- `Author`
+- `Title`
+- `Narrator`
+- `Source Group`
+- `original_file_name`
+- `original_extension`
+- `file_size_bytes`
+"""
+
+
+def build_row(
+    index: int,
+    audio_path: Path,
+    local_dir: Path,
+    metadata_maps: dict[str, dict[str, str]],
+    default_title: str,
+    default_author: str,
+    default_narrator: str,
+    default_source_group: str,
+    default_language: str,
+) -> dict[str, Any]:
+    meta = metadata_for_audio(audio_path, local_dir, metadata_maps)
+
+    rel_path = audio_path.relative_to(local_dir).as_posix()
+    file_name = audio_path.name
+    stem = audio_path.stem
+    extension = audio_path.suffix.lower().lstrip(".")
+    file_bytes = audio_path.read_bytes()
+
+    row_id = value_from_meta(
+        meta,
+        "id",
+        "ID",
+        default=f"{index:05d}_{slugify(stem)}",
+    )
+
+    title = (
+        value_from_meta(meta, "title", "Title", default="")
+        or default_title
+        or stem
+    )
+
+    language = (
+        value_from_meta(meta, "language", "Language", "lang", default="")
+        or default_language
+        or "be"
+    )
+
+    author = (
+        value_from_meta(meta, "Author", "author", default="")
+        or default_author
+    )
+
+    narrator = (
+        value_from_meta(meta, "Narrator", "narrator", default="")
+        or default_narrator
+    )
+
+    source_group = (
+        value_from_meta(meta, "Source Group", "source_group", "source", default="")
+        or default_source_group
+    )
+
+    metadata_file_name = value_from_meta(
+        meta,
+        "file_name",
+        "filename",
+        default=file_name,
+    )
+
+    return {
+        "id": str(row_id),
+        "audio": {
+            "path": rel_path,
+            "bytes": file_bytes,
+        },
+        "title": str(title),
+        "language": str(language),
+        "file_name": str(metadata_file_name),
+        "filename": str(metadata_file_name),
+        "Author": str(author),
+        "Title": str(title),
+        "Narrator": str(narrator),
+        "Source Group": str(source_group),
+        "original_file_name": str(file_name),
+        "original_extension": str(extension),
+        "file_size_bytes": int(len(file_bytes)),
+    }
+
+
+def push_to_hub(
+    output_repo: str,
+    parquet_paths: list[Path],
+    readme_path: Path,
+    token: str,
+    private: bool,
+    overwrite_train: bool,
+) -> None:
+    api = HfApi(token=token)
+
+    api.create_repo(
+        repo_id=output_repo,
+        repo_type="dataset",
+        exist_ok=True,
+        private=private,
+    )
+
+    operations = []
+
+    if overwrite_train:
+        try:
+            existing_files = api.list_repo_files(
+                repo_id=output_repo,
+                repo_type="dataset",
+            )
+
+            for path in existing_files:
+                if path.startswith("data/train/") and path.endswith(".parquet"):
+                    operations.append(
+                        CommitOperationDelete(path_in_repo=path)
+                    )
+        except Exception:
+            pass
+
+    shard_count = len(parquet_paths)
+
+    for i, path in enumerate(parquet_paths):
+        path_in_repo = f"data/train/train-{i:05d}-of-{shard_count:05d}.parquet"
+
+        operations.append(
+            CommitOperationAdd(
+                path_in_repo=path_in_repo,
+                path_or_fileobj=str(path),
+            )
+        )
+
+    operations.append(
+        CommitOperationAdd(
+            path_in_repo="README.md",
+            path_or_fileobj=str(readme_path),
+        )
+    )
+
+    api.create_commit(
+        repo_id=output_repo,
+        repo_type="dataset",
+        operations=operations,
+        commit_message=f"Add train parquet input dataset: {shard_count} shard(s)",
+    )
+
+
+def convert_dataset(
+    source_repo: str,
+    output_repo: str,
+    title: str,
+    author: str,
+    narrator: str,
+    source_group: str,
+    language: str,
+    allow_patterns_text: str,
+    hf_token: str,
+    private: bool,
+    overwrite_train: bool,
+    shard_target_mb: int,
+):
+    logs: list[str] = []
+
+    def add_log(message: str) -> str:
+        logs.append(message)
+        return "\n".join(logs)
+
+    try:
+        token = get_token(hf_token)
+
+        source_repo = normalize_repo_id(source_repo)
+        output_repo = normalize_repo_id(output_repo)
+
+        if not source_repo:
+            raise gr.Error(
+                "Укажы зыходны dataset repo, напрыклад "
+                "`archivartaunik/ivan-melezh-podykh-navalnitsy-valer-budzevich`."
+            )
+
+        if not output_repo:
+            raise gr.Error(
+                "Укажы output dataset repo, напрыклад "
+                "`archivartaunik/ivan-melezh-podykh-navalnitsy-valer-budzevich-input`."
+            )
+
+        shard_target_bytes = int(shard_target_mb) * 1024 * 1024
+        hard_shard_limit_bytes = HARD_SHARD_LIMIT_MB * 1024 * 1024
+
+        allow_patterns = [
+            p.strip()
+            for p in allow_patterns_text.replace(",", "\n").splitlines()
+            if p.strip()
+        ]
+
+        if not allow_patterns:
+            allow_patterns = [
+                "**/*.mp3",
+                "**/*.MP3",
+                "**/*.wav",
+                "**/*.WAV",
+                "**/*.ogg",
+                "**/*.OGG",
+                "**/*.opus",
+                "**/*.flac",
+                "**/*.m4a",
+                "**/*.aac",
+                "**/metadata.csv",
+                "metadata.csv",
+            ]
+
+        yield add_log(f"Source repo: {source_repo}")
+        yield add_log(f"Output repo: {output_repo}")
+        yield add_log("Downloading source dataset files...")
+        yield add_log(f"Allow patterns: {allow_patterns}")
+
+        local_dir = Path(
+            snapshot_download(
+                repo_id=source_repo,
+                repo_type="dataset",
+                allow_patterns=allow_patterns,
+                token=token,
+            )
+        )
+
+        yield add_log(f"Downloaded to local cache: {local_dir}")
+
+        metadata_maps = load_metadata_maps(local_dir)
+        yield add_log(f"metadata.csv rows detected: {len(metadata_maps)} lookup keys")
+
+        audio_files = discover_audio_files(local_dir)
+
+        if not audio_files:
+            raise gr.Error(
+                "Аўдыяфайлы не знойдзены. Правер allow patterns, напрыклад `part_*/*.mp3` або `**/*.mp3`."
+            )
+
+        yield add_log(f"Audio files detected: {len(audio_files)}")
+
+        too_large = [
+            p for p in audio_files
+            if p.stat().st_size > hard_shard_limit_bytes
+        ]
+
+        if too_large:
+            examples = "\n".join(
+                f"- {p.relative_to(local_dir).as_posix()}: {p.stat().st_size / 1024 / 1024:.1f} MB"
+                for p in too_large[:10]
+            )
+
+            raise gr.Error(
+                "Ёсць асобныя аўдыяфайлы большыя за hard limit shard-а. "
+                "Іх трэба спачатку парэзаць на меншыя часткі.\n\n"
+                f"{examples}"
+            )
+
+        total_rows = 0
+        parquet_paths: list[Path] = []
+
+        with tempfile.TemporaryDirectory() as tmp:
+            tmp_dir = Path(tmp)
+
+            current_rows: list[dict[str, Any]] = []
+            current_bytes = 0
+
+            for index, audio_path in enumerate(audio_files, start=1):
+                file_size = audio_path.stat().st_size
+
+                if current_rows and current_bytes + file_size > shard_target_bytes:
+                    shard_path = tmp_dir / f"shard-{len(parquet_paths):05d}.parquet"
+                    write_parquet_shard(current_rows, shard_path)
+
+                    shard_size = shard_path.stat().st_size
+
+                    if shard_size > hard_shard_limit_bytes:
+                        raise gr.Error(
+                            f"Shard занадта вялікі: {shard_size / 1024 / 1024:.1f} MB. "
+                            "Паменшы shard target MB або парэж аўдыя на меншыя файлы."
+                        )
+
+                    parquet_paths.append(shard_path)
+
+                    yield add_log(
+                        f"Wrote shard {len(parquet_paths)}: "
+                        f"{shard_size / 1024 / 1024:.1f} MB, "
+                        f"{len(current_rows)} rows"
+                    )
+
+                    current_rows = []
+                    current_bytes = 0
+
+                row = build_row(
+                    index=index,
+                    audio_path=audio_path,
+                    local_dir=local_dir,
+                    metadata_maps=metadata_maps,
+                    default_title=title.strip(),
+                    default_author=author.strip(),
+                    default_narrator=narrator.strip(),
+                    default_source_group=source_group.strip(),
+                    default_language=language.strip() or "be",
+                )
+
+                current_rows.append(row)
+                current_bytes += file_size
+                total_rows += 1
+
+            if current_rows:
+                shard_path = tmp_dir / f"shard-{len(parquet_paths):05d}.parquet"
+                write_parquet_shard(current_rows, shard_path)
+
+                shard_size = shard_path.stat().st_size
+
+                if shard_size > hard_shard_limit_bytes:
+                    raise gr.Error(
+                        f"Last shard занадта вялікі: {shard_size / 1024 / 1024:.1f} MB. "
+                        "Паменшы shard target MB або парэж аўдыя на меншыя файлы."
+                    )
+
+                parquet_paths.append(shard_path)
+
+                yield add_log(
+                    f"Wrote shard {len(parquet_paths)}: "
+                    f"{shard_size / 1024 / 1024:.1f} MB, "
+                    f"{len(current_rows)} rows"
+                )
+
+            readme_path = tmp_dir / "README.md"
+            readme_path.write_text(
+                make_dataset_card(
+                    source_repo=source_repo,
+                    title=title.strip() or output_repo,
+                    language=language.strip() or "be",
+                    row_count=total_rows,
+                    shard_count=len(parquet_paths),
+                ),
+                encoding="utf-8",
+            )
+
+            yield add_log("Pushing Parquet dataset to Hub...")
+
+            push_to_hub(
+                output_repo=output_repo,
+                parquet_paths=parquet_paths,
+                readme_path=readme_path,
+                token=token,
+                private=private,
+                overwrite_train=overwrite_train,
+            )
+
+        yield add_log("")
+        yield add_log("DONE")
+        yield add_log(f"Rows: {total_rows}")
+        yield add_log(f"Shards: {len(parquet_paths)}")
+        yield add_log(f"Dataset: https://huggingface.co/datasets/{output_repo}")
+
+    except gr.Error:
+        raise
+    except Exception as exc:
+        raise gr.Error(str(exc))
+
+
+with gr.Blocks(title="Audio Dataset to HF Parquet Input") as demo:
+    gr.Markdown(
+        """
+# Audio Dataset → Hugging Face Parquet Input
+
+Гэты Space чытае аўдыя з зыходнага Hugging Face dataset repo і стварае новы dataset у Parquet-фармаце.
+
+Выхадны фармат:
+
+```text
+config: default
+split: train
+format: parquet
+id column: id
+audio column: audio
+path: data/train/train-xxxxx-of-yyyyy.parquet
+```
+
+Калонка `audio` захоўваецца як Hugging Face `Audio` з убудаванымі bytes.
+"""
+    )
+
+    with gr.Row():
+        source_repo = gr.Textbox(
+            label="Source dataset repo або URL",
+            value="archivartaunik/ivan-melezh-podykh-navalnitsy-valer-budzevich",
+            placeholder="archivartaunik/ivan-melezh-podykh-navalnitsy-valer-budzevich",
+        )
+
+        output_repo = gr.Textbox(
+            label="Output dataset repo",
+            value="archivartaunik/ivan-melezh-podykh-navalnitsy-valer-budzevich-input",
+            placeholder="archivartaunik/ivan-melezh-podykh-navalnitsy-valer-budzevich-input",
+        )
+
+    with gr.Row():
+        title = gr.Textbox(
+            label="Title",
+            value="Подых навальніцы",
+        )
+
+        author = gr.Textbox(
+            label="Author",
+            value="Іван Мележ",
+        )
+
+        narrator = gr.Textbox(
+            label="Narrator",
+            value="Валер Будзевіч",
+        )
+
+    with gr.Row():
+        source_group = gr.Textbox(
+            label="Source Group",
+            value="Аўдыёкнігі",
+        )
+
+        language = gr.Textbox(
+            label="Language",
+            value="be",
+        )
+
+    allow_patterns_text = gr.Textbox(
+        label="Allow patterns для зыходнага dataset",
+        value="part_*/*.mp3\n**/*.mp3\n**/metadata.csv\nmetadata.csv",
+        lines=5,
+    )
+
+    with gr.Row():
+        hf_token = gr.Textbox(
+            label="HF token override, optional",
+            type="password",
+            placeholder="Лепш дадаць HF_TOKEN у Space Settings → Secrets",
+        )
+
+        shard_target_mb = gr.Number(
+            label="Shard target MB",
+            value=DEFAULT_SHARD_TARGET_MB,
+            precision=0,
+        )
+
+    with gr.Row():
+        private = gr.Checkbox(
+            label="Create output dataset as private",
+            value=True,
+        )
+
+        overwrite_train = gr.Checkbox(
+            label="Delete old data/train/*.parquet before push",
+            value=True,
+        )
+
+    run_button = gr.Button(
+        "Convert and push",
+        variant="primary",
+    )
+
+    log_output = gr.Textbox(
+        label="Log",
+        lines=25,
+        max_lines=60,
+    )
+
+    run_button.click(
+        fn=convert_dataset,
+        inputs=[
+            source_repo,
+            output_repo,
+            title,
+            author,
+            narrator,
+            source_group,
+            language,
+            allow_patterns_text,
+            hf_token,
+            private,
+            overwrite_train,
+            shard_target_mb,
+        ],
+        outputs=log_output,
+    )
+
+
+if __name__ == "__main__":
+    demo.launch()