dataset-uploader/scripts/upload_all_datasets.py

#!/usr/bin/env python3
"""
Download, process, and upload all RDF datasets to HuggingFace Hub.

WORKFLOW:
    For each dataset, the script executes three phases:

    Step 1: Download
        - Downloads raw RDF from public source URL
        - Extracts to usable RDF format
        - Progress bars with download speed

    Step 2: Decompress
        - Decompresses RDF files (if necessary)

    Step 3: Convert
        - Parses RDF file (Turtle, N-Triples, RDF/XML, etc.)
        - Converts to HuggingFace dataset format
        - Preserves all semantic information (lossless)
        - Creates dataset card (README.md) with metadata

    Step 4: Upload
        - Uploads to HuggingFace Hub under CleverThis organization
        - Creates/updates dataset repository
        - Uploads dataset card with comprehensive documentation
        - Makes dataset publicly accessible

DATASET CATEGORIES:
    small       < 1 GB      (WordNet, Schema.org, Gene Ontology)
    medium      1-10 GB     (YAGO 4.5, DisGeNET, GeoNames)
    large       10-100 GB   (DBpedia, UniProt, DBLP)
    xlarge      > 100 GB    (Wikidata, BabelNet, Bio2RDF)

DIRECTORY STRUCTURE:
    dataset_processing/
    ├── downloads/              # Raw RDF downloads
    │   ├── wordnet/
    │   │   ├── english-wordnet-2024.ttl.gz
    │   │   └── english-wordnet-2024.ttl
    │   └── yago-4.5/
    │       └── yago-4.5.0.2-tiny.ttl
    │
    └── hf_datasets/           # Converted HuggingFace datasets
        ├── wordnet/
        │   ├── data/
        │   ├── dataset_info.json
        │   └── README.md
        └── yago-4.5/
            ├── data/
            ├── dataset_info.json
            └── README.md

"""

from __future__ import annotations

import argparse
import os
import shutil
import subprocess
import sys
import time
from datetime import datetime
from pathlib import Path

from bloom_filter2 import BloomFilter
from dotenv import load_dotenv

from scripts.download_and_decompress import (
    decompress,
    download,
    get_dataset_info,
    get_most_recent_file,
)

env_path = Path(__file__).resolve().parent.parent / ".env"
if env_path.exists():
    load_dotenv(dotenv_path=env_path, override=False)

from dataset_registry import (  # noqa: E402
    DATASET_REGISTRY,
    DatasetInfo,
    get_dataset_config,
)
from dataset_validator import (  # noqa: E402
    _new_string_store,
    collect_parquet_strings,
    collect_source_strings,
    dataset_can_download_from_hf,
    dataset_exists_on_hf,
    dataset_has_required_columns,
    dataset_is_parquet_file,
    strings_in_parquet_not_source,
    strings_in_source_not_parquet,
)
from google_sheets_tracker import create_tracker  # noqa: E402
from huggingface_hub import HfApi  # noqa: E402
from rich.console import Console  # noqa: E402
from rich.table import Table  # noqa: E402

console = Console()


def parse_args() -> tuple[argparse.Namespace, argparse.ArgumentParser]:
    """Parse command-line arguments."""
    parser = argparse.ArgumentParser(
        description="Download, process, and upload all RDF datasets to HuggingFace Hub."
    )
    parser.add_argument(
        "--dataset",
        "-d",
        type=str,
        action="append",
        help="Specific dataset(s) to process (can be specified multiple times)",
    )
    parser.add_argument(
        "--category",
        "-c",
        choices=["small", "medium", "large", "xlarge", "all"],
        default=None,
        help="Process datasets by category",
    )
    parser.add_argument(
        "--base-dir",
        type=Path,
        default=Path("./dataset_processing"),
        help="Base directory for work",
    )
    parser.add_argument(
        "--skip-download", action="store_true", help="Skip download step"
    )
    parser.add_argument(
        "--skip-convert", action="store_true", help="Skip conversion step"
    )
    parser.add_argument("--skip-upload", action="store_true", help="Skip upload step")
    parser.add_argument(
        "--dry-run",
        action="store_true",
        help="Show what would be done without doing it",
    )
    parser.add_argument(
        "--list", action="store_true", help="List datasets that would be processed"
    )
    parser.add_argument(
        "--validate",
        action="store_true",
        help="Validate datasets on HuggingFace without processing",
    )
    parser.add_argument(
        "--rm",
        action="store_true",
        dest="remove_downloaded",
        help="Remove mode: wipe all downloads at start, "
        "delete each dataset after processing",
    )
    parser.add_argument(
        "--parallel",
        "-p",
        type=int,
        default=1,
        help="Number of datasets to process in parallel (default: 1, sequential)",
    )
    parser.add_argument(
        "--repeat",
        action="store_true",
        default=3,
        help="Repeat download/upload attempts",
    )
    parser.add_argument(
        "--sheet",
        action="store_true",
        help="Enable Google Sheets tracking for dataset status",
    )

    return parser.parse_args(), parser


def get_datasets_to_process(args: argparse.Namespace) -> list[str]:
    """
    Determine which datasets to process.
    Args:
        args: A list of arguments parsed by argparse

    Returns:
        Which datasets to process.
    """
    if args.dataset:
        datasets_to_process = args.dataset
    elif args.category == "all":
        datasets_to_process = list(DATASET_REGISTRY.keys())
    elif args.category:
        datasets_to_process = [
            d_id
            for d_id, d_info in DATASET_REGISTRY.items()
            if d_info.category == args.category
        ]
    else:
        datasets_to_process = []

    return datasets_to_process


def list_datasets(datasets_to_process: list[str]) -> None:
    """Print a list of datasets that would be processed."""
    console.print("\n[bold cyan]Datasets to Process:[/bold cyan]\n")
    table = Table(show_header=True)
    table.add_column("ID", style="cyan")
    table.add_column("Name", style="green")
    table.add_column("Category", style="yellow")
    table.add_column("Size", style="blue")
    table.add_column("Available", style="magenta")

    if len(datasets_to_process) == 0:
        list_of_datasets = list(DATASET_REGISTRY.keys())
    else:
        list_of_datasets = datasets_to_process

    for dataset_id in list_of_datasets:
        info = DATASET_REGISTRY[dataset_id]
        available = "[green]✓[/green]" if info.available else "[red]✗[/red]"
        table.add_row(
            dataset_id, info.name, info.category, f"{info.size_gb} GB", available
        )

    console.print(table)
    available_count = sum(
        1 for d_id in datasets_to_process if DATASET_REGISTRY[d_id].available
    )
    console.print(
        f"\n[bold]Total: {len(datasets_to_process)} datasets "
        f"({available_count} available)[/bold]"
    )


def check_dataset_exists_on_hf(dataset_id: str) -> tuple[bool, str | None]:
    """
    Check if dataset already exists on HuggingFace Hub.

    Checks both the dataset name and dataset name with '-v1' suffix.

    Args:
        dataset_id: Dataset identifier

    Returns:
        Tuple of (exists, existing_repo_id)
        - exists: True if dataset exists, False otherwise
        - existing_repo_id: Full repo ID if exists, None otherwise
    """
    try:
        config = get_dataset_config()
        organization = config.get("organization")
        if not organization:
            console.print(
                "[yellow]Warning: Organization not found in config file. "
                "Cannot check for existing datasets on HuggingFace.[/yellow]"
            )
            return (False, None)
    except Exception as e:
        console.print(
            f"[yellow]Warning: Could not read config file: {e}. "
            "Cannot check for existing datasets on HuggingFace.[/yellow]"
        )
        return (False, None)

    try:
        hf_token = os.getenv("HF_TOKEN") or os.getenv("HUGGING_FACE_HUB_TOKEN")
        api = HfApi(token=hf_token) if hf_token else HfApi()

        datasets = list(api.list_datasets(author=organization))

        repo_id_base = f"{organization}/{dataset_id}"
        repo_id_v1 = f"{organization}/{dataset_id}-v1"

        for dataset in datasets:
            if dataset.id == repo_id_base:
                return (True, repo_id_base)
            if dataset.id == repo_id_v1:
                return (True, repo_id_v1)

        return (False, None)
    except Exception as e:
        console.print(
            f"[yellow]Warning: Could not check HuggingFace for "
            f"existing dataset: {e}[/yellow]"
        )
        console.print("[yellow]Continuing with processing...[/yellow]")
        return (False, None)


def get_valid_hf_license(license_string: str | None) -> str:
    """Map dataset license to valid HuggingFace license identifier.

    Args:
        license_string: License string from dataset registry

    Returns:
        Valid HuggingFace license identifier
    """
    if not license_string:
        return "other"

    # Get license mappings from config
    config = get_dataset_config()
    license_map = config.get("license_mappings", {})

    # Convert keys to lowercase for comparison
    license_map_lower = {k.lower(): v for k, v in license_map.items()}

    normalized = license_string.lower().strip()
    return license_map_lower.get(normalized, "other")


def get_size_category(dataset_info: DatasetInfo) -> str:
    """Get HuggingFace size category.

    Args:
        dataset_info: Dataset information

    Returns:
        Size category string
    """
    if dataset_info.size_gb < 0.01:
        return "n<1K"
    elif dataset_info.size_gb < 1:
        return "1K<n<10K"
    elif dataset_info.size_gb < 10:
        return "10K<n<100K"
    elif dataset_info.size_gb < 100:
        return "100K<n<1M"
    elif dataset_info.size_gb < 1000:
        return "1M<n<10M"
    elif dataset_info.size_gb < 10000:
        return "10M<n<100M"
    else:
        return "100M<n"


def get_rdf_format_documentation(dataset_info: DatasetInfo) -> str:
    """Get comprehensive RDF format documentation for README.

    Args:
        dataset_info: Dataset information

    Returns:
        Markdown documentation for RDF format
    """
    return f"""
## Dataset Format: Lossless RDF Representation

This dataset uses a **standard lossless format** for representing RDF (Resource Description Framework)
data in HuggingFace Datasets. All semantic information from the original RDF knowledge graph is preserved,
enabling perfect round-trip conversion between RDF and HuggingFace formats.

### Schema

Each RDF triple is represented as a row with **6 fields**:

| Field | Type | Description | Example |
|-------|------|-------------|---------|
| `subject` | string | Subject of the triple (URI or blank node) | `"http://schema.org/Person"` |
| `predicate` | string | Predicate URI | `"http://www.w3.org/1999/02/22-rdf-syntax-ns#type"` |
| `object` | string | Object of the triple | `"John Doe"` or `"http://schema.org/Thing"` |
| `object_type` | string | Type of object: `"uri"`, `"literal"`, or `"blank_node"` | `"literal"` |
| `object_datatype` | string | XSD datatype URI (for typed literals) | `"http://www.w3.org/2001/XMLSchema#integer"` |
| `object_language` | string | Language tag (for language-tagged literals) | `"en"` |

### Example: RDF Triple Representation

**Original RDF (Turtle)**:
```turtle
<http://example.org/John> <http://schema.org/name> "John Doe"@en .
```

**HuggingFace Dataset Row**:
```python
{{
  "subject": "http://example.org/John",
  "predicate": "http://schema.org/name",
  "object": "John Doe",
  "object_type": "literal",
  "object_datatype": None,
  "object_language": "en"
}}
```

### Loading the Dataset

```python
from datasets import load_dataset

# Load the dataset
dataset = load_dataset("CleverThis/{dataset_info.id}")

# Access the data
data = dataset["data"]

# Iterate over triples
for row in data:
    subject = row["subject"]
    predicate = row["predicate"]
    obj = row["object"]
    obj_type = row["object_type"]

    print(f"Triple: ({{subject}}, {{predicate}}, {{obj}})")
    print(f"  Object type: {{obj_type}}")
    if row["object_language"]:
        print(f"  Language: {{row['object_language']}}")
    if row["object_datatype"]:
        print(f"  Datatype: {{row['object_datatype']}}")
```

### Converting Back to RDF

The dataset can be converted back to any RDF format (Turtle, N-Triples, RDF/XML,
etc.) with **zero information loss**:

```python
from datasets import load_dataset
from rdflib import Graph, URIRef, Literal, BNode

def convert_to_rdf(dataset_name, output_file="output.ttl", split="data"):
    \"\"\"Convert HuggingFace dataset back to RDF Turtle format.\"\"\"
    # Load dataset
    dataset = load_dataset(dataset_name)

    # Create RDF graph
    graph = Graph()

    # Convert each row to RDF triple
    for row in dataset[split]:
        # Subject
        if row["subject"].startswith("_:"):
            subject = BNode(row["subject"][2:])
        else:
            subject = URIRef(row["subject"])

        # Predicate (always URI)
        predicate = URIRef(row["predicate"])

        # Object (depends on object_type)
        if row["object_type"] == "uri":
            obj = URIRef(row["object"])
        elif row["object_type"] == "blank_node":
            obj = BNode(row["object"][2:])
        elif row["object_type"] == "literal":
            if row["object_datatype"]:
                obj = Literal(row["object"], datatype=URIRef(row["object_datatype"]))
            elif row["object_language"]:
                obj = Literal(row["object"], lang=row["object_language"])
            else:
                obj = Literal(row["object"])

        graph.add((subject, predicate, obj))

    # Serialize to Turtle (or any RDF format)
    graph.serialize(output_file, format="turtle")
    print(f"Exported {{len(graph)}} triples to {{output_file}}")
    return graph

# Usage
graph = convert_to_rdf("CleverThis/{dataset_info.id}", "reconstructed.ttl")
```

### Information Preservation Guarantee

This format preserves **100% of RDF information**:

- ✅ **URIs**: Exact string representation preserved
- ✅ **Literals**: Full text content preserved
- ✅ **Datatypes**: XSD and custom datatypes preserved
    (e.g., `xsd:integer`, `xsd:dateTime`)
- ✅ **Language Tags**: BCP 47 language tags preserved (e.g., `@en`, `@fr`, `@ja`)
- ✅ **Blank Nodes**: Node structure preserved (identifiers may change but
    graph isomorphism maintained)

**Round-trip guarantee**: Original RDF → HuggingFace → Reconstructed RDF
produces **semantically identical** graphs.

### Querying the Dataset

You can filter and query the dataset like any HuggingFace dataset:

```python
from datasets import load_dataset

dataset = load_dataset("CleverThis/{dataset_info.id}")

# Find all triples with English literals
english_literals = dataset["data"].filter(
    lambda x: x["object_type"] == "literal" and x["object_language"] == "en"
)
print(f"Found {{len(english_literals)}} English literals")

# Find all rdf:type statements
type_statements = dataset["data"].filter(
    lambda x: "rdf-syntax-ns#type" in x["predicate"]
)
print(f"Found {{len(type_statements)}} type statements")

# Convert to Pandas for analysis
import pandas as pd
df = dataset["data"].to_pandas()

# Analyze predicate distribution
print(df["predicate"].value_counts())
```

### Dataset Format

The dataset contains all triples in a single **data** split, suitable for
machine learning tasks such as:

- Knowledge graph completion
- Link prediction
- Entity embedding
- Relation extraction
- Graph neural networks

### Format Specification

For complete technical documentation of the RDF-to-HuggingFace format, see:

📖 [RDF to HuggingFace Format Specification](https://github.com/CleverThis/cleverernie/blob/master/docs/rdf_huggingface_format_specification.md)

The specification includes:
- Detailed schema definition
- All RDF node type mappings
- Performance benchmarks
- Edge cases and limitations
- Complete code examples

### Conversion Metadata

- **Source Format**: {dataset_info.format}
- **Original Size**: {dataset_info.size_gb} GB
- **Conversion Tool**: [CleverErnie RDF Pipeline](https://github.com/CleverThis/cleverernie)
- **Format Version**: 1.0
- **Conversion Date**: {datetime.now().strftime("%Y-%m-%d")}
"""  # noqa: E501


def get_rdf_format(dataset_info: DatasetInfo) -> str:
    """Get RDF format string for rdflib.

    Args:
        dataset_info: Dataset information

    Returns:
        Format string for rdflib
    """
    # Get format mappings from config
    config = get_dataset_config()
    format_map = config.get("format_mappings", {})

    # Default mappings for common formats
    default_mappings = {
        "rdf": "xml",
        "rdf/xml": "xml",
        "xml": "xml",
        "turtle": "turtle",
        "ttl": "turtle",
        "nt": "nt",
        "ntriples": "nt",
        "n3": "n3",
        "jsonld": "json-ld",
        "json-ld": "json-ld",
    }

    format_lower = dataset_info.format.lower()

    # First try config mappings, then default mappings, then fallback
    mapped_format = format_map.get(format_lower)
    if mapped_format:
        return mapped_format

    mapped_format = default_mappings.get(format_lower)
    if mapped_format:
        return mapped_format

    # Fallback to turtle for unknown formats
    return "turtle"


def create_dataset_card(dataset_info: DatasetInfo) -> str:
    """Create README.md content for the dataset card.

    Args:
        dataset_info: Dataset information

    Returns:
        Markdown content for dataset card
    """
    return f"""---
license: {get_valid_hf_license(dataset_info.license)}
task_categories:
- text-generation
- feature-extraction
language:
- en
tags:
- rdf
- knowledge-graph
- semantic-web
- triples
size_categories:
- {get_size_category(dataset_info)}
---

# {dataset_info.name}

## Dataset Description

{dataset_info.description}

**Original Source:** {dataset_info.url}

### Dataset Summary

This dataset contains RDF triples from {dataset_info.name} converted to HuggingFace
dataset format for easy use in machine learning pipelines.

- **Format:** Originally {dataset_info.format}, converted to HuggingFace Dataset
- **Size:** {dataset_info.size_gb} GB (extracted)
- **Entities:** {dataset_info.entities if dataset_info.entities else "N/A"}
- **Triples:** {dataset_info.triples if dataset_info.triples else "N/A"}
- **Original License:**
    {dataset_info.license if dataset_info.license else "See original source"}

### Recommended Use

{dataset_info.recommended_for}

{f"### Notes: {dataset_info.notes}" if dataset_info.notes else ""}

{get_rdf_format_documentation(dataset_info)}

## Citation

If you use this dataset, please cite the original source:

**Original Dataset:** {dataset_info.name}
**URL:** {dataset_info.url}
**License:** {dataset_info.license if dataset_info.license else "See original source"}

## Dataset Preparation

This dataset was prepared using the CleverErnie GISM framework:

```bash
# Download original dataset
python scripts/rdf_dataset_downloader.py {dataset_info.id} -o datasets/

# Convert to HuggingFace format
python scripts/convert_rdf_to_hf_dataset_unified.py \\
    datasets/{dataset_info.id}/[file] \\
    hf_datasets/{dataset_info.id} \\
    --format {get_rdf_format(dataset_info)} \\
    --strategy auto

# Upload to HuggingFace Hub
python scripts/upload_all_datasets.py --dataset {dataset_info.id}
```

## Additional Information

### Original Source

{dataset_info.url}

### Conversion Details

- Converted using: [CleverErnie GISM](https://github.com/cleverthis/cleverernie)
- Conversion script: `scripts/convert_rdf_to_hf_dataset_unified.py`
- Dataset format: Single 'data' split with all triples
- Strategy: Auto-selected based on dataset size and format

### Maintenance

This dataset is maintained by the CleverThis organization.
"""


def get_dataset_dir(args: argparse.Namespace, dataset_id: str) -> Path | None:
    hf_dataset_dir = args.base_dir / "hf_datasets" / dataset_id
    try:
        hf_dataset_dir.mkdir(parents=True, exist_ok=True)
    except OSError as e:
        console.print(f"[red]Error creating directory: {e}[/red]")
        console.print(f"[yellow]HF dataset dir: {hf_dataset_dir}[/yellow]")
        console.print("[yellow]Check permissions and disk space[/yellow]")
        return None

    return hf_dataset_dir


def convert(args: argparse.Namespace, rdf_file: Path, dataset_id: str) -> None:
    """
    This function converts the dataset to the HuggingFace dataset format.

    At the end of the function, the dataset should be available in the
    hf_datasets directory.

    If the conversion was unsuccessful, the function will throw an exception.
    """
    download_dir = rdf_file.parent

    dataset_info = get_dataset_info(dataset_id)
    if not dataset_info:
        raise ValueError(f"Dataset {dataset_id} not found in registry")

    hf_dataset_dir = get_dataset_dir(args, dataset_id)
    if hf_dataset_dir is None:
        raise ValueError("Could not retrieve dataset directory")

    if not args.skip_convert:
        # Special handling for FB15k-237
        if dataset_id == "fb15k-237":
            # Use specialized FB15k-237 converter that handles all splits
            fb15k_script = Path(__file__).parent / "convert_fb15k237_to_hf.py"
            fb15k_dir = download_dir / "fb15k-237" / "FB15K-237.2"

            if args.dry_run:
                console.print(
                    "[dim]Would run: python scripts/convert_fb15k237_to_hf.py "
                    "{fb15k_dir} {hf_dataset_dir}[/dim]"
                )
            else:
                result = subprocess.run(
                    [
                        sys.executable,
                        str(fb15k_script),
                        str(fb15k_dir),
                        str(hf_dataset_dir),
                    ],
                    capture_output=False,
                )

                if result.returncode != 0:
                    console.print(
                        f"[red]Conversion failed for {dataset_id} "
                        f"(exit code: {result.returncode})[/red]"
                    )
                    console.print(f"[yellow]Input file: {rdf_file}[/yellow]")
                    console.print(f"[yellow]Output dir: {hf_dataset_dir}[/yellow]")
                    console.print(
                        "[yellow]Try running conversion manually to see "
                        "detailed error messages[/yellow]"
                    )
                    raise ValueError("Conversion failed")

                # Create dataset card
                readme_path = hf_dataset_dir / "README.md"
                readme_path.write_text(create_dataset_card(dataset_info))
                console.print(f"[green]Created dataset card: {readme_path}[/green]")
        # Special handling for NELL-995
        elif dataset_id == "nell-995":
            # Use specialized NELL-995 converter that handles raw.kb format
            nell_script = Path(__file__).parent / "convert_nell995_to_hf.py"

            if args.dry_run:
                console.print(
                    "[dim]Would run: python scripts/convert_nell995_to_hf.py "
                    "{download_dir} {hf_dataset_dir}[/dim]"
                )
            else:
                result = subprocess.run(
                    [
                        sys.executable,
                        str(nell_script),
                        str(download_dir),
                        str(hf_dataset_dir),
                    ],
                    capture_output=False,
                )

                if result.returncode != 0:
                    console.print(
                        f"[red]Conversion failed for {dataset_id}. "
                        f"Return code: {result.returncode}[/red]"
                    )
                    raise ValueError("Conversion failed")

                # Create dataset card
                readme_path = hf_dataset_dir / "README.md"
                readme_path.write_text(create_dataset_card(dataset_info))
                console.print(f"[green]Created dataset card: {readme_path}[/green]")
        # Special handling for ConceptNet
        elif dataset_id == "conceptnet":
            # Use specialized ConceptNet converter that handles CSV format
            conceptnet_script = Path(__file__).parent / "convert_conceptnet_to_hf.py"

            # Find the downloaded CSV file
            csv_files = list(download_dir.rglob("*.csv")) + list(
                download_dir.rglob("*.csv.gz")
            )
            if not csv_files:
                console.print(
                    f"[red]No CSV files found for ConceptNet in {download_dir}[/red]"
                )
                raise ValueError("No CSV files found")

            csv_file = csv_files[0]
            console.print(f"[green]Found ConceptNet CSV file: {csv_file}[/green]")

            if args.dry_run:
                console.print(
                    "[dim]Would run: python scripts/convert_conceptnet_to_hf.py "
                    "{csv_file} {hf_dataset_dir}[/dim]"
                )
            else:
                result = subprocess.run(
                    [
                        sys.executable,
                        str(conceptnet_script),
                        str(csv_file),
                        str(hf_dataset_dir),
                    ],
                    capture_output=False,
                )

                if result.returncode != 0:
                    console.print(
                        f"[red]Conversion failed for {dataset_id}. "
                        f"Return code: {result.returncode}[/red]"
                    )
                    raise ValueError("Conversion failed")

                # Create dataset card
                readme_path = hf_dataset_dir / "README.md"
                readme_path.write_text(create_dataset_card(dataset_info))
                console.print(f"[green]Created dataset card: {readme_path}[/green]")
        else:
            # Original conversion logic for other datasets
            # Use unified converter which automatically selects best strategy
            unified_script = (
                Path(__file__).parent / "convert_rdf_to_hf_dataset_unified.py"
            )

            # Unified script will auto-select best strategy based on file size
            # and format
            console.print(
                "[dim]Using unified converter (auto-selects best strategy)[/dim]"
            )

            if args.dry_run:
                console.print(
                    "[dim]Would run: python "
                    "scripts/convert_rdf_to_hf_dataset_unified.py "
                    "{rdf_file} {hf_dataset_dir} --strategy auto[/dim]"
                )
            else:
                # Use unified converter with auto-strategy selection
                # Adjust chunk size based on dataset size
                chunk_size = 50000  # Default
                if dataset_info.size_gb > 10:
                    chunk_size = 100000  # Larger chunks for very large datasets
                elif dataset_info.size_gb < 1:
                    chunk_size = 10000  # Smaller chunks for small datasets

                result = subprocess.run(
                    [
                        sys.executable,
                        str(unified_script),
                        str(rdf_file),
                        str(hf_dataset_dir),
                        "--format",
                        get_rdf_format(dataset_info),
                        "--strategy",
                        "auto",  # Let the unified script auto-select best strategy
                        "--chunk-size",
                        str(chunk_size),
                        "--description",
                        dataset_info.description,
                        "--homepage",
                        dataset_info.url,
                        "--license",
                        dataset_info.license if dataset_info.license else "Unknown",
                    ],
                    capture_output=False,
                )

                if result.returncode != 0:
                    console.print(
                        f"[red]Conversion failed for {dataset_id}. "
                        f"Return code: {result.returncode}[/red]"
                    )


def upload(args: argparse.Namespace, dataset_id: str) -> None:
    """
    This function uploads the dataset to the HuggingFace Hub.

    This function throws a ValueError if the upload fails.
    """
    dataset_info = get_dataset_info(dataset_id)
    if not dataset_info:
        raise ValueError(f"Dataset {dataset_id} not found in registry")

    hf_dataset_dir = get_dataset_dir(args, dataset_id)
    if hf_dataset_dir is None:
        raise ValueError("Could not retrieve dataset directory")

    if not args.skip_upload:
        # Get repository ID from dataset info or config file
        if hasattr(dataset_info, "huggingface_repo") and dataset_info.huggingface_repo:
            repo_id = dataset_info.huggingface_repo
        else:
            try:
                config = get_dataset_config()
                organization = config.get("organization")
                if not organization:
                    raise ValueError(
                        "Organization not found in config file. "
                        "Please set 'organization' field in dataset_config.json"
                    )
                repo_id = f"{organization}/{dataset_id}"
            except Exception as e:
                raise ValueError(
                    f"Could not read organization from config file: {e}. "
                    "Please ensure dataset_config.json exists and contains "
                    "'organization' field."
                ) from e
        if args.dry_run:
            console.print(f"[dim]Would upload to: {repo_id}[/dim]")
        else:
            try:
                from datasets import load_from_disk
                from huggingface_hub import HfApi

                # Load dataset
                dataset = load_from_disk(str(hf_dataset_dir))

                # Upload
                console.print(f"[cyan]Uploading to {repo_id}...[/cyan]")
                dataset.push_to_hub(repo_id, private=False)

                # Upload README separately if it exists
                readme_path = hf_dataset_dir / "README.md"
                if readme_path.exists():
                    api = HfApi()
                    api.upload_file(
                        path_or_fileobj=str(readme_path),
                        path_in_repo="README.md",
                        repo_id=repo_id,
                        repo_type="dataset",
                    )
                else:
                    # Create README if it doesn't exist (e.g., when skipping conversion)
                    console.print(
                        "[yellow]README.md not found, creating it now...[/yellow]"
                    )
                    readme_path.write_text(create_dataset_card(dataset_info))
                    api = HfApi()
                    api.upload_file(
                        path_or_fileobj=str(readme_path),
                        path_in_repo="README.md",
                        repo_id=repo_id,
                        repo_type="dataset",
                    )

                console.print(
                    f"[bold green]✓ Successfully uploaded to {repo_id}[/bold green]"
                )
            except ImportError as e:
                console.print(f"[red]Missing required packages for upload: {e}[/red]")
                console.print(
                    "[yellow]Install with: "
                    "pip install datasets huggingface-hub[/yellow]"
                )
                raise ValueError("Missing required packages for upload") from e
            except FileNotFoundError as e:
                console.print(
                    f"[red]Dataset directory not found: {hf_dataset_dir}[/red]"
                )
                console.print(
                    "[yellow]Make sure conversion step completed successfully[/yellow]"
                )
                raise ValueError("Dataset directory not found") from e
            except PermissionError as e:
                console.print(
                    "[red]Permission denied uploading to HuggingFace Hub[/red]"
                )
                console.print(
                    "[yellow]Check your HuggingFace token: "
                    "huggingface-cli login[/yellow]"
                )
                console.print("[yellow]Or set HF_TOKEN environment variable[/yellow]")
                raise ValueError("Permission denied uploading to HuggingFace") from e
            except Exception as e:
                error_msg = str(e).lower()
                if "authentication" in error_msg or "token" in error_msg:
                    console.print(f"[red]Authentication failed: {e}[/red]")
                    console.print("[yellow]Run: huggingface-cli login[/yellow]")
                    console.print(
                        "[yellow]Or set HF_TOKEN environment variable[/yellow]"
                    )
                    raise ValueError("Authentication failed") from e
                if "network" in error_msg or "connection" in error_msg:
                    console.print(f"[red]Network error during upload: {e}[/red]")
                    console.print(
                        "[yellow]Check internet connection and try again[/yellow]"
                    )
                    raise ValueError("Network error during upload") from e
                if "disk" in error_msg or "space" in error_msg:
                    console.print(f"[red]Disk space error: {e}[/red]")
                    console.print("[yellow]Check available disk space[/yellow]")
                    raise ValueError("Disk space error") from e

                console.print(f"[red]Upload failed: {e}[/red]")
                console.print(f"[yellow]Repository: {repo_id}[/yellow]")
                console.print(f"[yellow]Dataset path: {hf_dataset_dir}[/yellow]")
                raise ValueError("Upload failed") from e
    else:
        console.print("[dim]Skipping upload[/dim]")


def is_sub_bloom_filter(bf1: BloomFilter, bf2: BloomFilter) -> bool:
    """
    Returns True if bf1 is a subset of bf2.
    """
    if not bf1._match_template(bf2):
        return False

    assert bf1.backend is not None
    assert bf2.backend is not None

    for i in range(0, bf1.num_bits_m):
        if bf1.backend.is_set(i) and not bf2.backend.is_set(i):
            return False

    return True


def validate_string_parity(
    dataset_id: str,
    parquet_files: list[Path],
    base_dir: Path,
) -> bool:
    """Validate that source strings and parquet strings match (Level 2/3)."""
    source_strings = collect_source_strings(dataset_id, base_dir)
    if not source_strings:
        console.print(
            f"[yellow]Warning: No source strings found for '{dataset_id}'. "
            "Check downloads directory and file formats.[/yellow]"
        )
        return False

    if len(parquet_files) == 1:
        parquet_strings = collect_parquet_strings(parquet_files[0])
    else:
        parquet_strings = _new_string_store()
        for parquet_file in parquet_files:
            parquet_strings.union(collect_parquet_strings(parquet_file))

    missing_in_parquet = not is_sub_bloom_filter(parquet_strings, source_strings)
    if missing_in_parquet:
        console.print("[red]There were strings in parquet not in the source.[/red]")
        missing_strings = strings_in_parquet_not_source(
            parquet_paths=parquet_files,
            source_strings=source_strings,
        )

        console.print(f"[yellow]Missing strings: {missing_strings}[/yellow]")
        return False

    missing_in_source = not is_sub_bloom_filter(source_strings, parquet_strings)
    if missing_in_source:
        console.print("[red]There were strings in the source not in parquet.[/red] ")
        missing_strings = strings_in_source_not_parquet(
            dataset_id=dataset_id,
            base_dir=base_dir,
            parquet_strings=parquet_strings,
        )

        console.print(f"[yellow]Missing strings: {missing_strings}[/yellow]")
        return False

    if not missing_in_parquet and not missing_in_source:
        console.print("[green]Validation: Strings match.[/green]")

    return not missing_in_parquet and not missing_in_source


def validate(datasets_to_process: list[str], base_dir: Path) -> bool:
    """Run Level 1 validation checks for datasets on HuggingFace."""
    validation_root = base_dir / "validation_cache"
    validation_root.mkdir(parents=True, exist_ok=True)

    all_valid = True

    for dataset_id in datasets_to_process:
        console.print(f"[bold cyan]Validating {dataset_id}...[/bold cyan]")

        exists = dataset_exists_on_hf(dataset_id)
        if not exists:
            console.print(
                f"[red]Validation failed: dataset '{dataset_id}' "
                "does not exist on HuggingFace.[/red]"
            )
            all_valid = False
            continue

        cache_dir = validation_root / dataset_id
        download_path = dataset_can_download_from_hf(
            dataset_id,
            cache_dir=cache_dir,
        )
        if not download_path:
            console.print(
                f"[red]Validation failed: dataset '{dataset_id}' "
                "could not be downloaded from HuggingFace.[/red]"
            )
            all_valid = False
            continue

        parquet_files = sorted(Path(download_path).rglob("*.parquet"))
        if not parquet_files:
            console.print(
                f"[red]Validation failed: no parquet files found for "
                f"'{dataset_id}'.[/red]"
            )
            all_valid = False
            continue

        for parquet_file in parquet_files:
            if not dataset_is_parquet_file(parquet_file):
                console.print(
                    f"[red]Validation failed: '{parquet_file}' is not "
                    "readable as parquet.[/red]"
                )
                all_valid = False
                continue

            if not dataset_has_required_columns(parquet_file):
                console.print(
                    f"[red]Validation failed: '{parquet_file}' is missing "
                    "required columns.[/red]"
                )
                all_valid = False
                continue

        if not validate_string_parity(dataset_id, parquet_files, base_dir):
            all_valid = False

        for parquet_file in parquet_files:
            try:
                parquet_file.unlink()
            except OSError as exc:
                console.print(f"[red]Failed to delete '{parquet_file}': {exc}[/red]")
                all_valid = False

    return all_valid


def remove_dir(download_dir: Path) -> None:
    """
    Removes a specified directory and its contents. Provides feedback on success or
    failure during the operation and warns the user if manual removal may be
    required.

    Parameters:
    download_dir (Path): A pathlib.Path object representing the directory to be removed.

    Raises:
    OSError: Raised when an error occurs during the directory removal process.

    Returns:
    None
    """
    if download_dir.exists():
        try:
            shutil.rmtree(download_dir)
            console.print(f"[green]✓ Removed {download_dir}[/green]")
        except OSError as e:
            console.print(
                f"[yellow]Warning: Could not remove {download_dir}: {e}[/yellow]"
            )
            console.print("[yellow]You may need to remove it manually[/yellow]")


def main() -> int:
    """
    The main function calls everything else.

    Returns:
        A Unix-style exit code. 0 == success.
    """
    args, parser = parse_args()

    # Fix for "AF_UNIX path too long" error in multiprocessing
    # This forces the temporary directory to be /tmp (short path) instead of
    # a potentially deep workspace path
    os.environ["TMPDIR"] = "/tmp"

    start_time = time.monotonic()

    # List datasets to process
    datasets_to_process = get_datasets_to_process(args)

    datasets_time = time.monotonic()
    datasets_duration = datasets_time - start_time

    print(f"`get_datasets_to_process` took {datasets_duration:.2f} seconds:")

    # List mode:
    if args.list:
        list_datasets(datasets_to_process)
        return 0

    if not datasets_to_process:
        parser.print_help()
        console.print(
            "[red]Error: No datasets specified. Use --dataset or --category.[/red]"
        )
        return 1

    if args.validate:
        validation_ok = validate(datasets_to_process, args.base_dir)
        return 0 if validation_ok else 1

    sheet_tracker = None
    if args.sheet:
        try:
            sheet_tracker = create_tracker(
                spreadsheet_id=None,
                worksheet_name=None,
                credentials_path=None,
            )
            if sheet_tracker:
                console.print("[green]✓ Google Sheets tracking enabled[/green]")
            else:
                console.print(
                    "[yellow]Warning: Could not initialize Google Sheets "
                    "tracker.[/yellow]"
                )
                console.print(
                    "[yellow]Check GOOGLE_SHEETS_ID and "
                    "GOOGLE_SHEETS_CREDENTIALS_PATH environment "
                    "variables.[/yellow]"
                )
                console.print("[yellow]Continuing without sheet tracking...[/yellow]")
        except Exception as e:
            console.print(
                f"[yellow]Warning: Error initializing Google Sheets "
                f"tracker: {e}[/yellow]"
            )
            console.print("[yellow]Continuing without sheet tracking...[/yellow]")

    downloaded = False
    rdf_file = None
    current_dataset_id = None

    try:
        for dataset_id in datasets_to_process:
            current_dataset_id = dataset_id
            one_dataset_start_time = time.monotonic()

            # Check if dataset already exists on HuggingFace (before processing)
            dataset_exists, existing_repo_id = check_dataset_exists_on_hf(dataset_id)
            if dataset_exists:
                console.print(
                    f"[yellow]⊘ Skipping '{dataset_id}': Already exists on "
                    f"HuggingFace as '{existing_repo_id}'[/yellow]"
                )
                if sheet_tracker:
                    try:
                        current_status = sheet_tracker.get_dataset_status(dataset_id)
                        current_status_value = (
                            current_status.get("status", "").upper().strip()
                        )

                        if current_status_value == "DONE":
                            console.print(
                                f"[dim]Sheet status already DONE for {dataset_id}[/dim]"
                            )
                    except Exception:
                        pass
                # Skip all processing steps and continue to next dataset
                continue

            if sheet_tracker:
                should_process, reason = sheet_tracker.check_should_process(
                    dataset_id, force=False
                )
                if not should_process:
                    console.print(
                        f"[yellow]⊘ Skipping '{dataset_id}': {reason}[/yellow]"
                    )
                    continue
                try:
                    sheet_tracker.mark_in_progress(dataset_id)
                    console.print(
                        f"[dim]Updated sheet: {dataset_id} -> IN PROGRESS[/dim]"
                    )
                except Exception:
                    pass

            # Download
            attempt_count = 0
            while attempt_count < args.repeat:
                attempt_count += 1
                console.print(
                    f"[bold cyan]Downloading {dataset_id}... "
                    f"Attempt {attempt_count}[/bold cyan]"
                )

                (downloaded, rerun, rdf_file) = download(
                    dataset_id=dataset_id,
                    base_dir=args.base_dir,
                    skip_download=args.skip_download,
                    dry_run=args.dry_run,
                )

                if not rerun:
                    break

            if (not downloaded or not rdf_file) and not args.dry_run:
                console.print("[red]Download failed. Aborting.[/red]")
                if sheet_tracker:
                    try:
                        error_msg = f"Download failed for {dataset_id}"
                        sheet_tracker.reset_to_not_started(dataset_id, error_msg)
                        console.print(
                            f"[dim]Updated sheet: {dataset_id} -> "
                            f"NOT STARTED (with error)[/dim]"
                        )
                    except Exception:
                        pass
                return 1

            one_dataset_download_time = time.monotonic()
            one_dataset_download_duration = (
                one_dataset_download_time - one_dataset_start_time
            )
            print(f"`download` took {one_dataset_download_duration:.2f} seconds:")

            # Decompress
            if args.dry_run:
                console.print(f"[dim][DRY RUN] Would decompress {dataset_id}...[/dim]")
            else:
                console.print(f"[bold cyan]Decompressing {dataset_id}...[/bold cyan]")

                assert rdf_file is not None
                result = decompress(rdf_file)
                if not result:
                    console.print("[red]Decompression failed. Aborting.[/red]")
                    if sheet_tracker:
                        try:
                            error_msg = f"Decompression failed for {dataset_id}"
                            sheet_tracker.reset_to_not_started(dataset_id, error_msg)
                            console.print(
                                f"[dim]Updated sheet: {dataset_id} -> "
                                f"NOT STARTED (with error)[/dim]"
                            )
                        except Exception:
                            pass
                    return 1

            one_dataset_decompress_time = time.monotonic()
            one_dataset_decompress_duration = (
                one_dataset_decompress_time - one_dataset_download_time
            )
            print(f"`decompress` took {one_dataset_decompress_duration:.2f} seconds:")

            # Convert
            assert rdf_file is not None

            if not args.dry_run:
                rdf_file = get_most_recent_file(rdf_file.parent)
                if rdf_file is None:
                    console.print("[red]Could not find RDF file. Aborting.[/red]")
                    if sheet_tracker:
                        try:
                            error_msg = f"Could not find RDF file for {dataset_id}"
                            sheet_tracker.reset_to_not_started(dataset_id, error_msg)
                            console.print(
                                f"[dim]Updated sheet: {dataset_id} -> "
                                f"NOT STARTED (with error)[/dim]"
                            )
                        except Exception:
                            pass
                    return 1

            console.print(
                "[bold cyan]Converting "
                f"{rdf_file if rdf_file else 'DRY RUN'}...[/bold cyan]"
            )
            # Pass a dummy path for dry run if rdf_file is None
            if rdf_file:
                convert_file = rdf_file
            else:
                convert_file = Path(
                    f"dataset_processing/downloads/{dataset_id}/dummy.ttl"
                )

            try:
                convert(args, convert_file, dataset_id)
            except ValueError as e:
                console.print(f"[red]Conversion failed: {e} Aborting.[/red]")
                if sheet_tracker:
                    try:
                        error_msg = f"Conversion failed for {dataset_id}: {e!r}"
                        sheet_tracker.reset_to_not_started(dataset_id, error_msg)
                        console.print(
                            f"[dim]Updated sheet: {dataset_id} -> "
                            f"NOT STARTED (with error)[/dim]"
                        )
                    except Exception:
                        pass
                return 1
            except Exception as e:
                if sheet_tracker:
                    try:
                        error_msg = f"Conversion error for {dataset_id}: {e!r}"
                        sheet_tracker.reset_to_not_started(dataset_id, error_msg)
                        console.print(
                            f"[dim]Updated sheet: {dataset_id} -> "
                            f"NOT STARTED (with error)[/dim]"
                        )
                    except Exception:
                        pass
                raise

            one_dataset_convert_time = time.monotonic()
            one_dataset_convert_duration = (
                one_dataset_convert_time - one_dataset_decompress_time
            )
            print(f"`convert` took {one_dataset_convert_duration:.2f} seconds:")

            # Upload
            upload_error = None
            attempt_count = 0
            while attempt_count < args.repeat:
                attempt_count += 1
                console.print(
                    f"[bold cyan]Uploading {dataset_id}... "
                    f"Attempt {attempt_count}[/bold cyan]"
                )
                try:
                    upload(args, dataset_id)
                    break
                except ValueError as e:
                    upload_error = str(e)
                    if attempt_count < args.repeat:
                        continue
                    else:
                        if sheet_tracker:
                            try:
                                error_msg = (
                                    f"Upload failed for {dataset_id}: {upload_error}"
                                )
                                sheet_tracker.reset_to_not_started(
                                    dataset_id, error_msg
                                )
                                console.print(
                                    f"[dim]Updated sheet: {dataset_id} -> "
                                    f"NOT STARTED (with error)[/dim]"
                                )
                            except Exception:
                                pass
                        console.print("[red]Upload failed. Aborting.[/red]")
                        return 1

            one_dataset_upload_time = time.monotonic()
            one_dataset_upload_duration = (
                one_dataset_upload_time - one_dataset_convert_time
            )
            print(f"`upload` took {one_dataset_upload_duration:.2f} seconds:")

            if sheet_tracker:
                try:
                    sheet_tracker.mark_completed(dataset_id)
                    console.print(f"[dim]Updated sheet: {dataset_id} -> DONE[/dim]")
                except Exception:
                    pass

            # Remove if needed.
            if args.remove_downloaded and not args.dry_run and rdf_file is not None:
                console.print(f"[bold cyan]Removing {rdf_file}[/bold cyan]")
                remove_dir(rdf_file.parent)

            validation_ok = validate(list(datasets_to_process), args.base_dir)
            if not validation_ok:
                console.print("[red]Validation failed for one or more datasets.[/red]")
                return 1

        # Success!
        console.print("[green]All done![/green]")

        total_duration = time.monotonic() - start_time
        print(f"`main` took {total_duration:.2f} seconds:")

        return 0

    except KeyboardInterrupt:
        console.print("\n[yellow]Script interrupted by user (Ctrl+C)[/yellow]")
        if sheet_tracker and current_dataset_id:
            try:
                error_msg = f"Script interrupted for {current_dataset_id}"
                sheet_tracker.reset_to_not_started(current_dataset_id, error_msg)
                console.print(
                    f"[dim]Updated sheet: {current_dataset_id} -> "
                    f"NOT STARTED (interrupted)[/dim]"
                )
            except Exception:
                pass
        return 130

    except Exception as e:
        console.print(f"\n[red]Unexpected error: {e}[/red]")
        if sheet_tracker and current_dataset_id:
            try:
                error_msg = f"Unexpected error for {current_dataset_id}: {e!r}"
                sheet_tracker.reset_to_not_started(current_dataset_id, error_msg)
                console.print(
                    f"[dim]Updated sheet: {current_dataset_id} -> "
                    f"NOT STARTED (error)[/dim]"
                )
            except Exception:
                pass
        raise


if __name__ == "__main__":
    exit(main())