#!/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 download, decompress, 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 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 "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 " f"{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} -> " f"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}: " f"{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())