from __future__ import annotations import csv import json import os import re import tempfile from collections import Counter 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, hf_hub_download, snapshot_download, ) AUDIO_SUFFIXES = { ".mp3", ".wav", ".ogg", ".opus", ".flac", ".m4a", ".aac", } DEFAULT_SHARD_TARGET_MB = 230 HARD_SHARD_LIMIT_MB = 260 DEFAULT_ALLOW_PATTERNS = [ "**/*.mp3", "**/*.MP3", "**/*.wav", "**/*.WAV", "**/*.ogg", "**/*.OGG", "**/*.opus", "**/*.OPUS", "**/*.flac", "**/*.FLAC", "**/*.m4a", "**/*.M4A", "**/*.aac", "**/*.AAC", "**/metadata.csv", "metadata.csv", ] TITLE_KEYS = ( "Title", "title", "book_title", "Book Title", "album", "Album", "name", "Name", "pretty_name", ) AUTHOR_KEYS = ( "Author", "author", "authors", "Authors", "writer", "Writer", "creator", "Creator", ) NARRATOR_KEYS = ( "Narrator", "narrator", "reader", "Reader", "voice", "Voice", "performer", "Performer", ) SOURCE_GROUP_KEYS = ( "Source Group", "source_group", "source-group", "source", "Source", "collection", "Collection", "category", "Category", ) LANGUAGE_KEYS = ( "language", "Language", "lang", "Lang", "locale", "Locale", ) # ----------------------------------------------------------------------------- # General helpers # ----------------------------------------------------------------------------- def normalize_repo_id(value: str) -> str: value = (value or "").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("/resolve/", 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 humanize_repo_name(repo_id: str) -> str: name = repo_id.split("/", 1)[-1] name = re.sub(r"[-_]+", " ", name).strip() return name[:1].upper() + name[1:] if name else repo_id def get_optional_token(token_from_ui: str | None) -> str | None: return (token_from_ui or "").strip() or os.environ.get("HF_TOKEN", "").strip() or None def get_token(token_from_ui: str | None) -> str: token = get_optional_token(token_from_ui) if not token: raise gr.Error( "HF token не знойдзены. Дадай HF_TOKEN у Space Settings → Secrets " "або ўвядзі token у поле HF token override." ) return token def suggest_output_repo(source_repo: str) -> str: if not source_repo: return "" if source_repo.endswith("-input"): return source_repo if "/" not in source_repo: return f"{source_repo}-input" owner, name = source_repo.split("/", 1) return f"{owner}/{name}-input" def unique_keep_order(values: list[str]) -> list[str]: seen = set() result = [] for value in values: value = value.strip() if value and value not in seen: seen.add(value) result.append(value) return result # ----------------------------------------------------------------------------- # Metadata prefill helpers # ----------------------------------------------------------------------------- def strip_frontmatter(readme_text: str) -> tuple[str, str]: """Return (frontmatter, markdown_body). YAML is parsed lightly by regex only.""" text = readme_text or "" if not text.startswith("---"): return "", text match = re.match(r"^---\s*\n(.*?)\n---\s*\n?(.*)$", text, flags=re.DOTALL) if not match: return "", text return match.group(1), match.group(2) def yaml_scalar_value(frontmatter: str, key: str) -> str: """ Very small YAML front-matter reader for common HF card fields. Handles: key: value key: - value key: [a, b] """ if not frontmatter: return "" lines = frontmatter.splitlines() for index, line in enumerate(lines): scalar = re.match(rf"^\s*{re.escape(key)}\s*:\s*(.*?)\s*$", line) if not scalar: continue value = scalar.group(1).strip().strip('"\'') if value: if value.startswith("[") and value.endswith("]"): value = value[1:-1].split(",", 1)[0].strip().strip('"\'') return value # List value on next lines. collected = [] for next_line in lines[index + 1 :]: if re.match(r"^\S[^:]*:\s*", next_line): break list_item = re.match(r"^\s*-\s*(.*?)\s*$", next_line) if list_item: item = list_item.group(1).strip().strip('"\'') if item: collected.append(item) if collected: return collected[0] return "" def markdown_h1(readme_body: str) -> str: for line in (readme_body or "").splitlines(): match = re.match(r"^#\s+(.+?)\s*$", line) if match: return match.group(1).strip() return "" def markdown_inline_field(readme_text: str, *labels: str) -> str: """ Finds simple card lines such as: Author: Іван Мележ - Author: Іван Мележ **Author:** Іван Мележ """ for label in labels: pattern = rf"(?im)^\s*(?:[-*]\s*)?(?:\*\*)?{re.escape(label)}(?:\*\*)?\s*:\s*(.+?)\s*$" match = re.search(pattern, readme_text or "") if match: value = match.group(1).strip() value = re.sub(r"^[`*_\s]+|[`*_\s]+$", "", value) if value: return value return "" def first_existing_value(row: dict[str, str], keys: tuple[str, ...]) -> str: for key in keys: value = row.get(key) if value: return str(value).strip() return "" def most_common_metadata_value(rows: list[dict[str, str]], keys: tuple[str, ...]) -> str: values = [] for row in rows: value = first_existing_value(row, keys) if value: values.append(value) if not values: return "" counter = Counter(values) return counter.most_common(1)[0][0] def read_metadata_csv_rows( repo_id: str, repo_files: list[str], token: str | None, max_rows: int = 500, ) -> list[dict[str, str]]: metadata_paths = [p for p in repo_files if Path(p).name.lower() == "metadata.csv"] rows: list[dict[str, str]] = [] for metadata_path in metadata_paths[:5]: try: local_path = hf_hub_download( repo_id=repo_id, repo_type="dataset", filename=metadata_path, token=token, ) with open(local_path, "r", encoding="utf-8-sig", newline="") as f: reader = csv.DictReader(f) for row in reader: cleaned_row = { str(k).strip(): str(v).strip() for k, v in row.items() if k is not None and v is not None and str(v).strip() } rows.append(cleaned_row) if len(rows) >= max_rows: return rows except Exception: continue return rows def read_readme_metadata(repo_id: str, token: str | None) -> dict[str, str]: for filename in ("README.md", "readme.md", "Readme.md"): try: local_path = hf_hub_download( repo_id=repo_id, repo_type="dataset", filename=filename, token=token, ) text = Path(local_path).read_text(encoding="utf-8", errors="ignore") frontmatter, body = strip_frontmatter(text) return { "title": ( yaml_scalar_value(frontmatter, "pretty_name") or markdown_inline_field(text, "Title", "title") or markdown_h1(body) ), "author": markdown_inline_field(text, "Author", "author", "Authors", "authors"), "narrator": markdown_inline_field(text, "Narrator", "narrator", "Reader", "reader"), "source_group": markdown_inline_field( text, "Source Group", "source_group", "Source", "source", "Collection", "collection", ), "language": ( yaml_scalar_value(frontmatter, "language") or yaml_scalar_value(frontmatter, "languages") or markdown_inline_field(text, "Language", "language", "Lang", "lang") ), } except Exception: continue return {} def infer_allow_patterns_from_repo_files(repo_files: list[str]) -> str: audio_exts = sorted({Path(p).suffix.lower() for p in repo_files if Path(p).suffix.lower() in AUDIO_SUFFIXES}) patterns: list[str] = [] # Preserve a useful explicit pattern for datasets split into part_*/ directories. has_part_dirs = any(re.search(r"(^|/)part_[^/]+/[^/]+$", p) for p in repo_files) if has_part_dirs: for ext in audio_exts: patterns.append(f"part_*/*{ext}") for ext in audio_exts: patterns.append(f"**/*{ext}") upper_ext = ext.upper() if upper_ext != ext: patterns.append(f"**/*{upper_ext}") metadata_paths = [p for p in repo_files if Path(p).name.lower() == "metadata.csv"] if metadata_paths: patterns.extend(metadata_paths) patterns.append("**/metadata.csv") patterns.append("metadata.csv") if not patterns: patterns = DEFAULT_ALLOW_PATTERNS.copy() return "\n".join(unique_keep_order(patterns)) def prefill_from_dataset_metadata(source_repo: str, hf_token: str): """ Fill Gradio fields from a pasted Hugging Face dataset URL/repo_id. Priority: 1. metadata.csv values, if present 2. README.md card/front matter 3. repo name fallback """ source_repo = normalize_repo_id(source_repo) if not source_repo: raise gr.Error( "Устаў спасылку на Hugging Face dataset або repo id, напрыклад " "`https://huggingface.co/datasets/owner/name`." ) token = get_optional_token(hf_token) api = HfApi(token=token) try: repo_files = api.list_repo_files( repo_id=source_repo, repo_type="dataset", ) except Exception as exc: raise gr.Error( "Не атрымалася прачытаць файлы dataset repo. Калі датасэт прыватны, " "дадай HF_TOKEN у Space Secrets або ў поле HF token override.\n\n" f"Дэталі: {exc}" ) readme_meta = read_readme_metadata(source_repo, token) metadata_rows = read_metadata_csv_rows(source_repo, repo_files, token) title_value = ( most_common_metadata_value(metadata_rows, TITLE_KEYS) or readme_meta.get("title", "") or humanize_repo_name(source_repo) ) author_value = ( most_common_metadata_value(metadata_rows, AUTHOR_KEYS) or readme_meta.get("author", "") ) narrator_value = ( most_common_metadata_value(metadata_rows, NARRATOR_KEYS) or readme_meta.get("narrator", "") ) source_group_value = ( most_common_metadata_value(metadata_rows, SOURCE_GROUP_KEYS) or readme_meta.get("source_group", "") or "Аўдыёкнігі" ) language_value = ( most_common_metadata_value(metadata_rows, LANGUAGE_KEYS) or readme_meta.get("language", "") or "be" ) allow_patterns_text = infer_allow_patterns_from_repo_files(repo_files) output_repo_value = suggest_output_repo(source_repo) log_lines = [ "DONE: палі запоўнены з metadata dataset-а.", f"Source repo: {source_repo}", f"Output repo: {output_repo_value}", f"Repo files detected: {len(repo_files)}", f"metadata.csv rows sampled: {len(metadata_rows)}", f"Title: {title_value}", f"Author: {author_value or '-'}", f"Narrator: {narrator_value or '-'}", f"Source Group: {source_group_value or '-'}", f"Language: {language_value or '-'}", ] return ( source_repo, output_repo_value, title_value, author_value, narrator_value, source_group_value, language_value, allow_patterns_text, "\n".join(log_lines), ) # ----------------------------------------------------------------------------- # Dataset conversion helpers # ----------------------------------------------------------------------------- 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 = DEFAULT_ALLOW_PATTERNS.copy() 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)) # ----------------------------------------------------------------------------- # UI # ----------------------------------------------------------------------------- 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-фармаце. 1. Устаў `Source dataset repo або URL`. 2. Націсні `Запоўніць з metadata dataset-а`. 3. Правер палі і націсні `Convert and push`. Выхадны фармат: ```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="https://huggingface.co/datasets/owner/name або owner/name", ) output_repo = gr.Textbox( label="Output dataset repo", value="archivartaunik/ivan-melezh-podykh-navalnitsy-valer-budzevich-input", placeholder="owner/name-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, ) with gr.Row(): fill_from_metadata_button = gr.Button( "Запоўніць з metadata dataset-а", variant="secondary", ) run_button = gr.Button( "Convert and push", variant="primary", ) log_output = gr.Textbox( label="Log", lines=25, max_lines=60, ) fill_from_metadata_button.click( fn=prefill_from_dataset_metadata, inputs=[ source_repo, hf_token, ], outputs=[ source_repo, output_repo, title, author, narrator, source_group, language, allow_patterns_text, log_output, ], ) 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()