import csv
import hashlib
import json
import os
import shutil
import tarfile
import tempfile
import traceback
import zipfile
from pathlib import Path
from typing import Dict, Generator, List, Tuple
from urllib.parse import unquote, urlparse

import gdown
import gradio as gr
import gspread
import py7zr
import requests
from google.oauth2.service_account import Credentials
from gspread.exceptions import APIError, SpreadsheetNotFound, WorksheetNotFound
from huggingface_hub import HfApi
from slugify import slugify


SPREADSHEET_ID = os.getenv(
    "SPREADSHEET_ID",
    "1tLnmSx5PnpV16RYnc-hDbon6KieozYg3p3IVtBRQNz0",
)

WORKSHEET_GID = int(os.getenv("WORKSHEET_GID", "0"))

DEFAULT_COLLECTION_NAME = os.getenv("HF_COLLECTION_NAME", "audyaknihi")
DEFAULT_MAX_SPLIT_MB = int(os.getenv("MAX_SPLIT_MB", "250"))

REQUIRED_COLUMNS = [
    "File URL",
    "HF_Source",
    "Author",
    "Title",
    "Narrator",
    "Source Group",
    "Source",
]

AUDIO_EXTENSIONS = {
    ".wav",
    ".mp3",
    ".flac",
    ".ogg",
    ".oga",
    ".m4a",
    ".aac",
    ".opus",
    ".wma",
    ".aiff",
    ".aif",
    ".webm",
}

IGNORED_DIR_PARTS = {
    "__MACOSX",
    ".git",
}

IGNORED_FILE_NAMES = {
    ".DS_Store",
    "Thumbs.db",
}

SCOPES = [
    "https://www.googleapis.com/auth/spreadsheets",
    "https://www.googleapis.com/auth/drive",
]


SplitPlanItem = Tuple[Path, Path, int]
# (source_path, target_relative_path_inside_split, file_size_bytes)


def require_secret(name: str) -> None:
    if not os.getenv(name):
        raise RuntimeError(f"Не знойдзены HF Space Secret: {name}")


def get_service_account_info() -> Dict:
    require_secret("GOOGLE_SERVICE_ACCOUNT_JSON")

    raw = os.environ["GOOGLE_SERVICE_ACCOUNT_JSON"].strip()

    try:
        info = json.loads(raw)
    except json.JSONDecodeError as e:
        raise RuntimeError(
            "GOOGLE_SERVICE_ACCOUNT_JSON павінен быць валідным JSON. "
            "Калі ўстаўляеш файл у HF Secret, лепш ператварыць яго ў адзін радок."
        ) from e

    if info.get("type") != "service_account":
        raise RuntimeError(
            "GOOGLE_SERVICE_ACCOUNT_JSON не падобны да service account JSON: "
            "поле 'type' павінна быць 'service_account'."
        )

    if not info.get("client_email"):
        raise RuntimeError(
            "У GOOGLE_SERVICE_ACCOUNT_JSON няма поля 'client_email'. "
            "Правер, што гэта поўны JSON-ключ service account."
        )

    if not info.get("private_key"):
        raise RuntimeError(
            "У GOOGLE_SERVICE_ACCOUNT_JSON няма поля 'private_key'. "
            "Правер, што JSON устаўлены цалкам."
        )

    return info


def get_service_account_email_safe() -> str:
    try:
        return get_service_account_info().get("client_email", "")
    except Exception:
        return ""


def get_gspread_client() -> gspread.Client:
    info = get_service_account_info()
    creds = Credentials.from_service_account_info(info, scopes=SCOPES)
    return gspread.authorize(creds)


def explain_google_error(e: Exception) -> str:
    client_email = get_service_account_email_safe()

    lines = [
        "Дыягностыка Google Sheets:",
        f"- SPREADSHEET_ID: {SPREADSHEET_ID}",
        f"- WORKSHEET_GID: {WORKSHEET_GID}",
    ]

    if client_email:
        lines.append(f"- Service account client_email: {client_email}")

    if isinstance(e, SpreadsheetNotFound) or isinstance(e, PermissionError):
        lines.extend(
            [
                "",
                "Магчымыя прычыны:",
                "1. Не ўключаны Google Sheets API або Google Drive API ў project service account.",
                "2. Service account не мае доступу да табліцы.",
                "3. Няправільны SPREADSHEET_ID.",
                "",
                "Што зрабіць:",
                "1. Уключы Google Sheets API і Google Drive API ў Google Cloud project.",
                "2. Адкрый Google Sheet → Share.",
                f"3. Дадай гэты email як Editor: {client_email or '<client_email з JSON>'}",
                "4. Пачакай 1-3 хвіліны і запусці Space яшчэ раз.",
            ]
        )

    elif isinstance(e, WorksheetNotFound):
        lines.extend(
            [
                "",
                "Табліца адкрылася, але аркуш з такім gid не знойдзены.",
                "Калі працуеш з першым аркушам, пакінь WORKSHEET_GID=0.",
            ]
        )

    elif isinstance(e, APIError):
        lines.extend(
            [
                "",
                "Google API вярнуў памылку.",
                "Правер, што Google Sheets API і Google Drive API уключаны,",
                "і што service account мае правы Editor на табліцу.",
            ]
        )

    return "\n".join(lines)


def get_worksheet():
    gc = get_gspread_client()
    spreadsheet = gc.open_by_key(SPREADSHEET_ID)

    if WORKSHEET_GID == 0:
        return spreadsheet.sheet1

    try:
        ws = spreadsheet.get_worksheet_by_id(WORKSHEET_GID)
        if ws is not None:
            return ws
    except AttributeError:
        pass

    for ws in spreadsheet.worksheets():
        if getattr(ws, "id", None) == WORKSHEET_GID:
            return ws

    raise WorksheetNotFound(f"Worksheet gid={WORKSHEET_GID} not found")


def load_sheet_rows() -> Tuple[object, List[str], List[Dict[str, str]]]:
    ws = get_worksheet()
    values = ws.get_all_values()

    if not values:
        raise RuntimeError("Табліца пустая.")

    headers = [h.strip() for h in values[0]]
    missing = [c for c in REQUIRED_COLUMNS if c not in headers]

    if missing:
        raise RuntimeError(
            "У табліцы няма патрэбных слупкоў: " + ", ".join(missing)
        )

    rows = []

    for i, raw_row in enumerate(values[1:], start=2):
        row = {}

        for col_index, header in enumerate(headers):
            row[header] = raw_row[col_index].strip() if col_index < len(raw_row) else ""

        row["_sheet_row_number"] = str(i)
        rows.append(row)

    return ws, headers, rows


def hf_api_and_owner() -> Tuple[HfApi, str]:
    require_secret("HF_TOKEN")

    token = os.environ["HF_TOKEN"].strip()
    api = HfApi(token=token)

    owner = os.getenv("HF_OWNER", "").strip()
    if not owner:
        owner = api.whoami(token=token)["name"]

    return api, owner


def dataset_repo_name(row: Dict[str, str]) -> str:
    base = " ".join(
        [
            row.get("Author", ""),
            row.get("Title", ""),
            row.get("Narrator", ""),
        ]
    ).strip()

    slug = slugify(base, max_length=80)

    if not slug:
        digest = hashlib.sha1(base.encode("utf-8")).hexdigest()[:10]
        slug = f"dataset-{digest}"

    return slug


def normalize_collection_name(value: str) -> str:
    value = (value or "").strip()
    if not value:
        return DEFAULT_COLLECTION_NAME
    return value


def add_dataset_to_collection(repo_id: str, collection_name: str) -> str:
    api, owner = hf_api_and_owner()
    collection_name = normalize_collection_name(collection_name)

    if not hasattr(api, "create_collection") or not hasattr(api, "add_collection_item"):
        raise RuntimeError(
            "Усталяваная версія huggingface_hub не падтрымлівае Collections API. "
            "Абнаві requirements.txt: huggingface_hub>=0.24.0"
        )

    collection = api.create_collection(
        title=collection_name,
        namespace=owner,
        exists_ok=True,
    )

    collection_slug = collection.slug

    api.add_collection_item(
        collection_slug=collection_slug,
        item_id=repo_id,
        item_type="dataset",
        exists_ok=True,
    )

    return f"https://huggingface.co/collections/{collection_slug}"


def is_google_drive_url(url: str) -> bool:
    return "drive.google.com" in url or "docs.google.com" in url


def filename_from_url(url: str) -> str:
    parsed = urlparse(url)
    name = unquote(Path(parsed.path).name).strip()
    if name:
        return name
    return "source_file"


def download_file(url: str, out_dir: Path) -> Path:
    out_dir.mkdir(parents=True, exist_ok=True)

    if is_google_drive_url(url):
        out_path = out_dir / "source_archive"

        downloaded = gdown.download(
            url=url,
            output=str(out_path),
            quiet=False,
            fuzzy=True,
        )

        if not downloaded:
            raise RuntimeError(f"Не атрымалася спампаваць Google Drive файл: {url}")

        return Path(downloaded)

    guessed_name = filename_from_url(url)
    out_path = out_dir / guessed_name

    with requests.get(url, stream=True, timeout=120) as r:
        r.raise_for_status()

        content_disposition = r.headers.get("content-disposition", "")

        if "filename=" in content_disposition.lower():
            candidate = content_disposition.split("filename=")[-1].strip().strip('"')
            if candidate:
                out_path = out_dir / Path(candidate).name

        with open(out_path, "wb") as f:
            for chunk in r.iter_content(chunk_size=1024 * 1024):
                if chunk:
                    f.write(chunk)

    return out_path


def ensure_safe_path(base_dir: Path, target: Path) -> None:
    base = base_dir.resolve()
    resolved = target.resolve()

    if not str(resolved).startswith(str(base)):
        raise RuntimeError(f"Небяспечны шлях у архіве: {target}")


def extract_archive_or_copy_audio(source_path: Path, extract_dir: Path) -> None:
    extract_dir.mkdir(parents=True, exist_ok=True)

    if zipfile.is_zipfile(source_path):
        with zipfile.ZipFile(source_path) as zf:
            for member in zf.infolist():
                target = extract_dir / member.filename
                ensure_safe_path(extract_dir, target)

            zf.extractall(extract_dir)

        return

    if tarfile.is_tarfile(source_path):
        with tarfile.open(source_path) as tf:
            for member in tf.getmembers():
                target = extract_dir / member.name
                ensure_safe_path(extract_dir, target)

            try:
                tf.extractall(extract_dir, filter="data")
            except TypeError:
                tf.extractall(extract_dir)

        return

    if py7zr.is_7zfile(str(source_path)):
        with py7zr.SevenZipFile(source_path, mode="r") as z:
            z.extractall(path=extract_dir)

        return

    if source_path.suffix.lower() in AUDIO_EXTENSIONS:
        shutil.copy2(source_path, extract_dir / source_path.name)
        return

    raise RuntimeError(
        "Невядомы фармат файла. Падтрымліваюцца ZIP, TAR/TAR.GZ/TGZ, 7Z "
        "або асобны аўдыёфайл."
    )


def should_ignore_file(path: Path) -> bool:
    if path.name in IGNORED_FILE_NAMES:
        return True

    return any(part in IGNORED_DIR_PARTS for part in path.parts)


def collect_audio_files(root: Path) -> List[Path]:
    files = []

    for path in root.rglob("*"):
        if not path.is_file():
            continue

        if should_ignore_file(path.relative_to(root)):
            continue

        if path.suffix.lower() in AUDIO_EXTENSIONS:
            files.append(path)

    return sorted(files, key=lambda p: p.relative_to(root).as_posix().lower())


def common_top_level_dir(paths: List[Path], root: Path) -> str:
    first_parts = []

    for path in paths:
        rel = path.relative_to(root)

        if len(rel.parts) > 1:
            first_parts.append(rel.parts[0])
        else:
            return ""

    if first_parts and len(set(first_parts)) == 1:
        return first_parts[0]

    return ""


def target_relative_path(src: Path, extracted_dir: Path, top_dir_to_strip: str) -> Path:
    original_rel = src.relative_to(extracted_dir)

    if top_dir_to_strip and original_rel.parts[0] == top_dir_to_strip:
        rel = Path(*original_rel.parts[1:])
    else:
        rel = original_rel

    if not rel.parts:
        rel = Path(src.name)

    return rel


def make_split_plan(
    audio_files: List[Path],
    extracted_dir: Path,
    max_split_bytes: int,
) -> Tuple[List[Tuple[str, List[SplitPlanItem]]], List[str]]:
    warnings = []

    if max_split_bytes <= 0:
        max_split_bytes = DEFAULT_MAX_SPLIT_MB * 1024 * 1024

    top_dir_to_strip = common_top_level_dir(audio_files, extracted_dir)

    planned_items: List[SplitPlanItem] = []

    for src in audio_files:
        rel = target_relative_path(src, extracted_dir, top_dir_to_strip)
        size = src.stat().st_size
        planned_items.append((src, rel, size))

        if size > max_split_bytes:
            warnings.append(
                "⚠️ Адзін файл большы за ліміт split: "
                f"{rel.as_posix()} = {size / 1024 / 1024:.1f} MB. "
                "Такі split можа ўсё яшчэ не адкрывацца ў Dataset Viewer."
            )

    raw_splits: List[List[SplitPlanItem]] = []
    current: List[SplitPlanItem] = []
    current_size = 0

    for item in planned_items:
        _src, _rel, size = item

        if current and current_size + size > max_split_bytes:
            raw_splits.append(current)
            current = []
            current_size = 0

        current.append(item)
        current_size += size

    if current:
        raw_splits.append(current)

    if len(raw_splits) == 1:
        return [("train", raw_splits[0])], warnings

    named_splits = []

    for idx, items in enumerate(raw_splits, start=1):
        named_splits.append((f"part_{idx:05d}", items))

    return named_splits, warnings


def write_split_metadata(
    split_dir: Path,
    row: Dict[str, str],
    items: List[SplitPlanItem],
) -> None:
    metadata_path = split_dir / "metadata.csv"

    metadata_fields = [
        "file_name",
        "Author",
        "Title",
        "Narrator",
        "Source Group",
        "Source",
        "original_file_name",
        "original_extension",
        "file_size_bytes",
    ]

    with open(metadata_path, "w", newline="", encoding="utf-8") as f:
        writer = csv.DictWriter(f, fieldnames=metadata_fields)
        writer.writeheader()

        for src, rel, size in sorted(items, key=lambda x: x[1].as_posix().lower()):
            writer.writerow(
                {
                    "file_name": rel.as_posix(),
                    "Author": row.get("Author", ""),
                    "Title": row.get("Title", ""),
                    "Narrator": row.get("Narrator", ""),
                    "Source Group": row.get("Source Group", ""),
                    "Source": row.get("Source", ""),
                    "original_file_name": src.name,
                    "original_extension": src.suffix.lower(),
                    "file_size_bytes": str(size),
                }
            )


def copy_split_files_preserving_originals(
    split_dir: Path,
    items: List[SplitPlanItem],
) -> None:
    split_dir.mkdir(parents=True, exist_ok=True)

    for src, rel, _size in items:
        dst = split_dir / rel
        ensure_safe_path(split_dir, dst)
        dst.parent.mkdir(parents=True, exist_ok=True)
        shutil.copy2(src, dst)


def yaml_quote(value: str) -> str:
    return json.dumps(value, ensure_ascii=False)


def build_readme_text(
    row: Dict[str, str],
    repo_id: str,
    split_names: List[str],
    max_split_mb: int,
) -> str:
    title = row.get("Title", "").strip() or "Audio dataset"

    data_files_lines = []

    for split_name in split_names:
        data_files_lines.append(f"  - split: {yaml_quote(split_name)}")
        data_files_lines.append(f"    path: {yaml_quote(split_name + '/**')}")

    data_files_yaml = "\n".join(data_files_lines)

    split_list = "\n".join(f"- `{name}`" for name in split_names)

    return f"""---
license: other
tags:
- audio
- audiofolder
configs:
- config_name: default
  data_files:
{data_files_yaml}
---

# {title}

## Metadata

- Author: {row.get("Author", "")}
- Title: {row.get("Title", "")}
- Narrator: {row.get("Narrator", "")}
- Source Group: {row.get("Source Group", "")}
- Source: {row.get("Source", "")}

## Notes

The original audio files are preserved as-is:

- no conversion;
- no re-encoding;
- no filename changes inside each split folder, except removing one common top-level archive folder when present.

To avoid Hugging Face Dataset Viewer scan-size errors, the dataset is split into smaller folders.
Target maximum split size: about {max_split_mb} MB.

Each split folder contains its own `metadata.csv`, and its `file_name` column points exactly to the audio files in the same split folder.

## Splits

{split_list}

## Loading

```python
from datasets import load_dataset

ds = load_dataset("{repo_id}")
```
"""


def create_dataset_folder(
    row: Dict[str, str],
    extracted_dir: Path,
    dataset_dir: Path,
    repo_id: str,
    max_split_mb: int,
) -> Tuple[int, int, List[str]]:
    audio_files = collect_audio_files(extracted_dir)

    if not audio_files:
        raise RuntimeError("У архіве не знойдзены аўдыёфайлы.")

    max_split_bytes = max_split_mb * 1024 * 1024

    split_plan, warnings = make_split_plan(
        audio_files=audio_files,
        extracted_dir=extracted_dir,
        max_split_bytes=max_split_bytes,
    )

    total_files = 0
    split_names = []

    for split_name, items in split_plan:
        split_names.append(split_name)
        split_dir = dataset_dir / split_name

        copy_split_files_preserving_originals(split_dir, items)
        write_split_metadata(split_dir, row, items)

        total_files += len(items)

    readme = dataset_dir / "README.md"
    readme.write_text(
        build_readme_text(
            row=row,
            repo_id=repo_id,
            split_names=split_names,
            max_split_mb=max_split_mb,
        ),
        encoding="utf-8",
    )

    return total_files, len(split_names), warnings


def upload_dataset(dataset_dir: Path, repo_id: str) -> str:
    api, _owner = hf_api_and_owner()

    private = os.getenv("PRIVATE_DATASETS", "false").lower() in {
        "1",
        "true",
        "yes",
    }

    api.create_repo(
        repo_id=repo_id,
        repo_type="dataset",
        private=private,
        exist_ok=True,
    )

    api.upload_folder(
        folder_path=str(dataset_dir),
        repo_id=repo_id,
        repo_type="dataset",
        commit_message="Create dataset from source archive preserving original audio files with split folders",
        delete_patterns=[
            "train/**",
            "train_*/**",
            "part_*/**",
            "data/**",
            "README.md",
            "dataset_infos.json",
        ],
    )

    return f"https://huggingface.co/datasets/{repo_id}"


def process_row(
    row: Dict[str, str],
    collection_name: str,
    max_split_mb: int,
) -> Tuple[str, int, int, str, List[str]]:
    file_url = row.get("File URL", "").strip()

    if not file_url:
        raise RuntimeError("Пустое поле File URL.")

    _api, owner = hf_api_and_owner()
    repo_name = dataset_repo_name(row)
    repo_id = f"{owner}/{repo_name}"

    with tempfile.TemporaryDirectory() as tmp:
        tmp_dir = Path(tmp)

        download_dir = tmp_dir / "download"
        extract_dir = tmp_dir / "extracted"
        dataset_dir = tmp_dir / "dataset"

        download_dir.mkdir(parents=True, exist_ok=True)

        source_path = download_file(file_url, download_dir)
        extract_archive_or_copy_audio(source_path, extract_dir)

        file_count, split_count, warnings = create_dataset_folder(
            row=row,
            extracted_dir=extract_dir,
            dataset_dir=dataset_dir,
            repo_id=repo_id,
            max_split_mb=max_split_mb,
        )

        dataset_url = upload_dataset(dataset_dir, repo_id)

    collection_url = ""

    try:
        collection_url = add_dataset_to_collection(
            repo_id=repo_id,
            collection_name=collection_name,
        )
    except Exception as e:
        warnings.append(
            "⚠️ Dataset створаны, але не атрымалася дадаць у Collection: "
            f"{type(e).__name__}: {repr(e)}"
        )

    return dataset_url, file_count, split_count, collection_url, warnings


def normalize_max_rows(value) -> int:
    if value is None:
        return 1

    try:
        value = int(value)
    except Exception:
        return 1

    if value < 0:
        return 0

    return value


def normalize_max_split_mb(value) -> int:
    if value is None:
        return DEFAULT_MAX_SPLIT_MB

    try:
        value = int(value)
    except Exception:
        return DEFAULT_MAX_SPLIT_MB

    if value < 10:
        return 10

    if value > 290:
        return 290

    return value


def format_exception_block(
    title: str,
    e: Exception,
    include_google_help: bool = False,
) -> str:
    parts = [
        title,
        f"Тып памылкі: {type(e).__name__}",
        f"Тэкст памылкі: {repr(e)}",
    ]

    if include_google_help:
        parts.extend(
            [
                "",
                explain_google_error(e),
            ]
        )

    parts.extend(
        [
            "",
            "Traceback:",
            traceback.format_exc(),
        ]
    )

    return "\n".join(parts)


def run_pipeline(
    max_rows_value,
    force_reprocess,
    collection_name,
    max_split_mb_value,
) -> Generator[str, None, None]:
    max_rows = normalize_max_rows(max_rows_value)
    force_reprocess = bool(force_reprocess)
    collection_name = normalize_collection_name(collection_name)
    max_split_mb = normalize_max_split_mb(max_split_mb_value)

    log_lines = []

    def emit(line: str) -> str:
        log_lines.append(line)
        return "\n".join(log_lines)

    try:
        client_email = get_service_account_email_safe()
        if client_email:
            yield emit(f"Service account: {client_email}")

        yield emit(f"Hugging Face collection: {collection_name}")
        yield emit(f"Максімальны памер аднаго split: {max_split_mb} MB")

        ws, headers, rows = load_sheet_rows()
        hf_col = headers.index("HF_Source") + 1

    except Exception as e:
        yield emit(
            format_exception_block(
                title="❌ Памылка пры чытанні табліцы:",
                e=e,
                include_google_help=True,
            )
        )
        return

    total = len(rows)

    if max_rows == 0:
        yield emit(
            f"Знойдзена радкоў у табліцы: {total}. "
            "Рэжым: апрацаваць усе патрэбныя радкі."
        )
    else:
        yield emit(
            f"Знойдзена радкоў у табліцы: {total}. "
            f"Ліміт апрацоўкі: {max_rows}."
        )

    if force_reprocess:
        yield emit("⚠️ Уключаны рэжым перастварэння: радкі з HF_Source таксама будуць апрацаваныя.")
    else:
        yield emit("Рэжым: радкі з запоўненым HF_Source прапускаюцца.")

    processed = 0
    skipped = 0
    failed = 0
    attempted = 0

    for row in rows:
        row_number = int(row["_sheet_row_number"])
        title = row.get("Title", "").strip() or f"row {row_number}"

        if row.get("HF_Source", "").strip() and not force_reprocess:
            skipped += 1
            continue

        if not row.get("File URL", "").strip():
            skipped += 1
            yield emit(f"⏭️ Радок {row_number}: пустое поле File URL — {title}")
            continue

        if max_rows != 0 and attempted >= max_rows:
            yield emit(f"⏹️ Дасягнуты ліміт: {max_rows} радкоў.")
            break

        attempted += 1
        yield emit(f"▶️ Радок {row_number}: апрацоўваю — {title}")

        try:
            dataset_url, file_count, split_count, collection_url, warnings = process_row(
                row=row,
                collection_name=collection_name,
                max_split_mb=max_split_mb,
            )

            ws.update_cell(row_number, hf_col, dataset_url)

            processed += 1

            msg = (
                f"✅ Радок {row_number}: dataset абноўлены\n"
                f"Файлаў: {file_count}\n"
                f"Splits: {split_count}\n"
                f"Dataset: {dataset_url}"
            )

            if collection_url:
                msg += f"\nCollection: {collection_url}"

            if warnings:
                msg += "\n" + "\n".join(warnings)

            yield emit(msg)

        except Exception as e:
            failed += 1

            yield emit(
                format_exception_block(
                    title=f"❌ Радок {row_number}: памылка",
                    e=e,
                    include_google_help=False,
                )
            )

    yield emit(
        "\nГатова.\n"
        f"Спроб апрацоўкі: {attempted}.\n"
        f"Створана/абноўлена: {processed}.\n"
        f"Прапушчана: {skipped}.\n"
        f"Памылак: {failed}."
    )


with gr.Blocks(title="Archive to Hugging Face Dataset") as demo:
    gr.Markdown(
        """
# Archive → Hugging Face Dataset

Space апрацоўвае Google Sheet і стварае Hugging Face datasets з архіваў.

Правіла па змаўчанні:

- калі `HF_Source` запоўнена — радок прапускаецца;
- калі `HF_Source` пустое — спампоўваецца архіў з `File URL`;
- аўдыё распакоўваецца ў dataset;
- арыгінальныя аўдыёфайлы не канвертуюцца, не перакадуюцца і не пераймяноўваюцца;
- dataset дзеліцца на некалькі split-папак, каб Dataset Viewer не ўпіраўся ў scan-size limit;
- у кожнай split-папцы свой `metadata.csv` з дакладнымі шляхамі да файлаў;
- dataset дадаецца ў Hugging Face Collection;
- спасылка на dataset запісваецца назад у `HF_Source`.

`0` у полі колькасці радкоў азначае: апрацаваць усе патрэбныя радкі.
"""
    )

    collection_name_input = gr.Textbox(
        label="Назва калекцыі Hugging Face",
        value=DEFAULT_COLLECTION_NAME,
        placeholder=DEFAULT_COLLECTION_NAME,
    )

    max_rows_input = gr.Number(
        label="Колькі радкоў апрацаваць",
        value=1,
        precision=0,
    )

    max_split_mb_input = gr.Number(
        label="Максімальны памер аднаго split, MB",
        value=DEFAULT_MAX_SPLIT_MB,
        precision=0,
    )

    force_reprocess_input = gr.Checkbox(
        label="Перастварыць нават калі HF_Source ужо запоўнены",
        value=False,
    )

    run_btn = gr.Button("Запусціць апрацоўку")
    output = gr.Textbox(label="Лог", lines=38)

    run_btn.click(
        fn=run_pipeline,
        inputs=[
            max_rows_input,
            force_reprocess_input,
            collection_name_input,
            max_split_mb_input,
        ],
        outputs=output,
    )


demo.queue(default_concurrency_limit=1).launch()