app.py

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()