scripts/convert_rdf_to_hf_dataset_unified.py

@@ -25,6 +25,10 @@ USAGE:
     python convert_rdf_to_hf_dataset_unified.py input.ttl output/ \
          --format turtle --description "My Dataset" --license "CC BY 4.0"
 
+    # Upload directly to HuggingFace Hub
+    python convert_rdf_to_hf_dataset_unified.py input.ttl \
+         --format turtle --push-to-hub --hub-repo-id username/dataset-name
+
 REQUIREMENTS:
     pip install rdflib datasets rich pyarrow
 """
@@ -67,6 +71,104 @@ from rich.progress import (
 
 logger = logging.getLogger(__name__)
 
+# =============================================================================
+# MODULE-LEVEL FUNCTIONS FOR MULTIPROCESSING
+# =============================================================================
+
+def process_geonames_lines(lines):
+    """Process a batch of lines from GeoNames file.
+
+    Module-level function for multiprocessing.
+
+    Args:
+        lines: List of lines containing one or more XML documents
+
+    Returns:
+        List of triple dictionaries
+    """
+    from rdflib import Graph, Literal, URIRef
+
+    triples = []
+    current_xml = []
+
+    for line in lines:
+        # Check if this is a document boundary (URL line)
+        if line.startswith("http://") or line.startswith("https://"):
+            # Process previous document if exists
+            if current_xml:
+                xml_str = "".join(current_xml)
+                try:
+                    # Parse the RDF/XML document
+                    graph = Graph()
+                    graph.parse(data=xml_str, format="xml")
+
+                    # Extract triples
+                    for s, p, o in graph:
+                        triples.append(extract_triple(s, p, o))
+
+                except Exception as e:
+                    # Skip malformed documents
+                    logger.error(f"Error parsing XML document: {e}. Skipping document.")
+                    pass
+
+                # Reset for next document
+                current_xml = []
+
+            # Skip the URL line itself
+            continue
+
+        # Accumulate XML content
+        current_xml.append(line)
+
+    # Process final document if exists
+    if current_xml:
+        xml_str = "".join(current_xml)
+        try:
+            graph = Graph()
+            graph.parse(data=xml_str, format="xml")
+
+            for s, p, o in graph:
+                triples.append(extract_triple(s, p, o))
+        except Exception as e:
+            logger.error(f"Error parsing final XML document: {e}. Skipping document.")
+            pass
+
+    return triples
+
+
+def process_ntriples_lines(lines):
+    """Process a batch of N-Triples lines.
+
+    Args:
+        lines: List of N-Triple lines
+
+    Returns:
+        List of triple dictionaries
+    """
+    from rdflib import Graph
+
+    triples = []
+
+    for line in lines:
+        line = line.strip()
+        if not line or line.startswith("#"):
+            continue
+
+        try:
+            # Parse single N-Triple line
+            mini_graph = Graph()
+            mini_graph.parse(data=line, format="nt")
+
+            for s, p, o in mini_graph:
+                triples.append(extract_triple(s, p, o))
+
+        except Exception as e:
+            # Skip malformed lines
+            logger.error(f"Error parsing N-Triple line: {line}. Skipping line. Error: {e}")
+            continue
+
+    return triples
+
 # =============================================================================
 # DATA CLASSES - Configuration and Results
 # =============================================================================
@@ -75,7 +177,7 @@ logger = logging.getLogger(__name__)
 class ConversionConfig:
     """Configuration for RDF to HuggingFace conversion."""
     input_path: Path
-    output_path: Path
+    output_path: Path | None
     rdf_format: str = "turtle"
     chunk_size: int = 10000
     num_workers: int | None = None
@@ -84,6 +186,8 @@ class ConversionConfig:
     test_size: float = 0.05
     clean_cache: bool = False
     verbose: bool = False
+    push_to_hub: bool = False
+    hub_repo_id: str | None = None
 
 @dataclass
 class ConversionResult:
@@ -196,6 +300,80 @@ class FileHandler:
             logger.error(f"Error getting file size for {file_path}: {e}")
             raise
 
+def batch_file_lines(file_path: Path, batch_size: int = 1000, format: str = "geonames"):
+    """Generator that yields batches of lines from file.
+
+    For GeoNames format, ensures complete XML documents are kept together.
+
+    Args:
+        file_path: Path to file
+        batch_size: Number of lines per batch
+        format: File format (geonames or ntriples)
+
+    Yields:
+        Batches of lines
+    """
+    # Handle compressed files
+    if file_path.suffix == ".gz":
+        try:
+            file_obj = gzip.open(file_path, "rt", encoding="utf-8")
+            # Test read to check if file is valid
+            test_line = file_obj.readline()
+            if not test_line and file_path.stat().st_size > 0:
+                raise EOFError("Compressed file appears to be empty or corrupted")
+            file_obj.seek(0)
+        except (gzip.BadGzipFile, EOFError, OSError) as e:
+            console = Console()
+            console.print(f"[red]Error: Compressed file is corrupted or incomplete: {e}[/red]")
+            console.print(f"[yellow]File: {file_path}[/yellow]")
+            console.print("[yellow]Please re-download the dataset or use an uncompressed version[/yellow]")
+            raise
+    else:
+        file_obj = open(file_path, encoding="utf-8")
+
+    try:
+        current_batch = []
+        current_doc = []
+
+        if format == "geonames":
+            # For GeoNames, keep documents together
+            for line in file_obj:
+                # Check if this is a document boundary
+                if line.startswith("http://") or line.startswith("https://"):
+                    if current_doc:
+                        # Add completed document to batch
+                        current_batch.extend(current_doc)
+                        current_doc = [line]
+
+                        # Yield batch if large enough
+                        if len(current_batch) >= batch_size:
+                            yield current_batch
+                            current_batch = []
+                    else:
+                        current_doc = [line]
+                else:
+                    current_doc.append(line)
+
+            # Yield final batch
+            if current_doc:
+                current_batch.extend(current_doc)
+            if current_batch:
+                yield current_batch
+
+        else:  # ntriples or other line-based format
+            for line in file_obj:
+                current_batch.append(line)
+                if len(current_batch) >= batch_size:
+                    yield current_batch
+                    current_batch = []
+
+            # Yield final batch
+            if current_batch:
+                yield current_batch
+
+    finally:
+        file_obj.close()
+
 class MetadataHandler:
     """Handle metadata for the HuggingFace dataset."""
     @staticmethod
@@ -290,9 +468,18 @@ class ConversionStrategy(ABC):
 
     def _save_dataset(self, dataset_dict: DatasetDict, config: ConversionConfig) -> None:
         try:
-            config.output_path.mkdir(parents=True, exist_ok=True)
-            dataset_dict.save_to_disk(str(config.output_path))
-            self.progress.print(f"Dataset saved to {config.output_path}", "green")
+            if config.push_to_hub:
+                if not config.hub_repo_id:
+                    raise ValueError("hub_repo_id is required when push_to_hub is True")
+                self.progress.print(f"Uploading to HuggingFace Hub: {config.hub_repo_id}...", "yellow")
+                dataset_dict.push_to_hub(config.hub_repo_id, private=False)
+                self.progress.print(f"✓ Successfully uploaded to {config.hub_repo_id}", "green")
+            else:
+                if not config.output_path:
+                    raise ValueError("output_path is required when push_to_hub is False")
+                config.output_path.mkdir(parents=True, exist_ok=True)
+                dataset_dict.save_to_disk(str(config.output_path))
+                self.progress.print(f"Dataset saved to {config.output_path}", "green")
         except Exception as e:
             logger.error(f"Error saving dataset: {e}")
             raise
@@ -383,6 +570,124 @@ class ConversionStrategy(ABC):
         if current_chunk:
             yield current_chunk
 
+    def _stream_geonames_parallel(self, config: ConversionConfig, num_workers: int) -> Iterator[list[RDFTriple]]:
+        """Stream GeoNames RDF file with parallel processing."""
+        self.progress.print(f"[yellow]Using parallel GeoNames parser with {num_workers} workers[/yellow]")
+
+        # Create batches of lines
+        batches = batch_file_lines(config.input_path, batch_size=config.chunk_size, format="geonames")
+
+        # Process batches in parallel
+        try:
+            with mp.Pool(processes=num_workers) as pool:
+                for triples in pool.imap_unordered(process_geonames_lines, batches, chunksize=1):
+                    if triples:
+                        yield triples
+        except Exception as e:
+            logger.error(f"Error processing GeoNames file: {e}")
+            raise
+
+    def _stream_ntriples_parallel(self, config: ConversionConfig, num_workers: int) -> Iterator[list[RDFTriple]]:
+        """Stream N-Triples file with parallel processing."""
+        self.progress.print(f"[yellow]Using parallel N-Triples parser with {num_workers} workers[/yellow]")
+
+        # Create batches of lines
+        batches = batch_file_lines(config.input_path, batch_size=config.chunk_size, format="ntriples")
+
+        # Process batches in parallel
+        try:
+            with mp.Pool(processes=num_workers) as pool:
+                for triples in pool.imap_unordered(process_ntriples_lines, batches, chunksize=1):
+                    if triples:
+                        yield triples
+        except Exception as e:
+            logger.error(f"Error processing N-Triples file: {e}")
+            raise
+
+    def _stream_turtle_chunks(self, config: ConversionConfig) -> Iterator[list[RDFTriple]]:
+        """Stream Turtle file in chunks using incremental parsing."""
+        self.progress.print("[yellow]Streaming Turtle file (line-by-line parser)[/yellow]")
+
+        # Open file with compression support
+        file_obj = self.file_handler.open_file(config.input_path)
+
+        try:
+            current_chunk: list[str] = []
+            triple_count = 0
+            line_count = 0
+
+            # Collect prefixes first
+            prefix_lines: list[str] = []
+            in_prefixes = True
+
+            for line_bytes in file_obj:
+                line_count += 1
+                # Ensure we have a string
+                line = line_bytes if isinstance(line_bytes, str) else line_bytes.decode('utf-8', errors='ignore')
+                stripped = line.strip()
+
+                # Skip empty lines and comments
+                if not stripped or stripped.startswith("#"):
+                    continue
+
+                # Collect prefix declarations
+                if in_prefixes and (stripped.startswith("@prefix") or stripped.startswith("@base")):
+                    prefix_lines.append(line)
+                    continue
+                elif in_prefixes:
+                    # End of prefixes, now we're in the data
+                    in_prefixes = False
+
+                # Parse in batches of lines ending with '.'
+                if stripped.endswith("."):
+                    triple_count += 1
+
+                    if triple_count >= config.chunk_size:
+                        # Try to parse this chunk
+                        chunk_text = "".join(prefix_lines) + "\n" + "".join(current_chunk) + "\n" + line
+
+                        try:
+                            graph = Graph()
+                            graph.parse(data=chunk_text, format="turtle")
+
+                            # Extract triples
+                            triples = [extract_triple(s, p, o) for s, p, o in graph]
+                            if triples:
+                                yield triples
+
+                        except Exception as e:
+                            logger.error(f"Error parsing Turtle chunk: {e}")
+
+                        # Reset for next chunk
+                        current_chunk = []
+                        triple_count = 0
+                    else:
+                        current_chunk.append(line)
+                else:
+                    # Part of a multi-line statement
+                    current_chunk.append(line)
+
+                # Log progress
+                if line_count % 100000 == 0:
+                    self.progress.print(f"[dim]Processed {line_count:,} lines...[/dim]")
+
+            # Process remaining lines
+            if current_chunk:
+                chunk_text = "".join(prefix_lines) + "\n" + "".join(current_chunk)
+                try:
+                    graph = Graph()
+                    graph.parse(data=chunk_text, format="turtle")
+
+                    triples = [extract_triple(s, p, o) for s, p, o in graph]
+                    if triples:
+                        yield triples
+
+                except Exception as e:
+                    logger.error(f"Error parsing final Turtle chunk: {e}")
+
+        finally:
+            file_obj.close()
+
 # =============================================================================
 # CONCRETE STRATEGIES
 # =============================================================================
@@ -402,6 +707,10 @@ class StandardStrategy(ConversionStrategy):
             file_size_mb = self.file_handler.get_file_size_mb(config.input_path)
             self.progress.print(f"Converting RDF file: {config.input_path}", "cyan")
             self.progress.print(f"File size: {file_size_mb:.2f} MB", "dim")
+            if config.push_to_hub:
+                self.progress.print(f"Destination: {config.hub_repo_id} (HuggingFace Hub)", "dim")
+            else:
+                self.progress.print(f"Output: {config.output_path}", "dim")
             self.progress.emit_progress(5)
 
             # Choose parsing method based on rdf_format
@@ -426,7 +735,8 @@ class StandardStrategy(ConversionStrategy):
             self._save_dataset(dataset_dict, config)
             self.progress.emit_progress(100)
             elapsed = time.time() - start_time
-            self._save_dataset_info_json(config, len(triples), dataset_dict, elapsed)
+            if config.output_path and not config.push_to_hub:
+                self._save_dataset_info_json(config, len(triples), dataset_dict, elapsed)
             self.progress.print(f"✓ Converted {len(triples):,} triples in {elapsed:.1f}s", "bold green")
             return ConversionResult(success=True, total_triples=len(triples),
                                     processing_time_seconds=elapsed,
@@ -455,10 +765,20 @@ class StreamingStrategy(ConversionStrategy):
             self.progress.print(f"Converting RDF file: {config.input_path}", "cyan")
             self.progress.print(f"File size: {file_size_mb:.2f} MB", "dim")
             self.progress.print(f"Chunk size: {config.chunk_size:,} triples", "dim")
+            if config.push_to_hub:
+                self.progress.print(f"Destination: {config.hub_repo_id} (HuggingFace Hub)", "dim")
+            else:
+                self.progress.print(f"Output: {config.output_path}", "dim")
             self.progress.emit_progress(5)
-            config.output_path.mkdir(parents=True, exist_ok=True)
-            temp_chunks_dir = config.output_path / "temp_chunks"
-            temp_chunks_dir.mkdir(exist_ok=True)
+
+            # Create temp directory for chunks
+            if config.push_to_hub:
+                temp_chunks_dir = Path(tempfile.mkdtemp(prefix="rdf_chunks_"))
+            else:
+                config.output_path.mkdir(parents=True, exist_ok=True)
+                temp_chunks_dir = config.output_path / "temp_chunks"
+                temp_chunks_dir.mkdir(exist_ok=True)
+
             total_triples = 0
             chunk_count = 0
             with self.progress.progress_bar("Processing RDF chunks...", total=None) as (progress, task):
@@ -474,9 +794,14 @@ class StreamingStrategy(ConversionStrategy):
             self.progress.emit_progress(80)
             dataset_dict = self._merge_parquet_chunks(temp_chunks_dir, config, create_splits=True)
             shutil.rmtree(temp_chunks_dir)
+
+            if config.push_to_hub:
+                self._save_dataset(dataset_dict, config)
+
             self.progress.emit_progress(100)
             elapsed = time.time() - start_time
-            self._save_dataset_info_json(config, total_triples, dataset_dict, elapsed)
+            if config.output_path and not config.push_to_hub:
+                self._save_dataset_info_json(config, total_triples, dataset_dict, elapsed)
             self.progress.print("✓ Converted to HuggingFace dataset", "green")
             return ConversionResult(success=True, total_triples=total_triples,
                                     processing_time_seconds=elapsed,
@@ -505,10 +830,20 @@ class StreamingTurtleStrategy(ConversionStrategy):
             self.progress.print(f"Converting Turtle: {config.input_path.name}", "cyan")
             self.progress.print(f"File size: {file_size_mb:.2f} MB", "dim")
             self.progress.print(f"Chunk size: {config.chunk_size:,} statements", "dim")
+            if config.push_to_hub:
+                self.progress.print(f"Destination: {config.hub_repo_id} (HuggingFace Hub)", "dim")
+            else:
+                self.progress.print(f"Output: {config.output_path}", "dim")
             self.progress.emit_progress(5)
-            config.output_path.mkdir(parents=True, exist_ok=True)
-            temp_chunks_dir = config.output_path / "temp_chunks"
-            temp_chunks_dir.mkdir(exist_ok=True)
+
+            # Create temp directory for chunks
+            if config.push_to_hub:
+                temp_chunks_dir = Path(tempfile.mkdtemp(prefix="rdf_chunks_"))
+            else:
+                config.output_path.mkdir(parents=True, exist_ok=True)
+                temp_chunks_dir = config.output_path / "temp_chunks"
+                temp_chunks_dir.mkdir(exist_ok=True)
+
             total_triples = 0
             chunk_count = 0
             # Use a specialized streaming method for Turtle with prefix handling.
@@ -537,9 +872,14 @@ class StreamingTurtleStrategy(ConversionStrategy):
             self.progress.emit_progress(80)
             dataset_dict = self._merge_parquet_chunks(temp_chunks_dir, config)
             shutil.rmtree(temp_chunks_dir)
+
+            if config.push_to_hub:
+                self._save_dataset(dataset_dict, config)
+
             self.progress.emit_progress(100)
             elapsed = time.time() - start_time
-            self._save_dataset_info_json(config, total_triples, dataset_dict, elapsed)
+            if config.output_path and not config.push_to_hub:
+                self._save_dataset_info_json(config, total_triples, dataset_dict, elapsed)
             self.progress.print("✓ Converted to HuggingFace dataset", "green")
             return ConversionResult(success=True, total_triples=total_triples,
                                     processing_time_seconds=elapsed,
@@ -568,9 +908,19 @@ class SimpleStreamingStrategy(ConversionStrategy):
             self.progress.print(f"Converting: {config.input_path.name}", "cyan")
             self.progress.print(f"File size: {file_size_mb:.2f} MB", "dim")
             self.progress.print(f"Chunk size: {config.chunk_size:,} triples", "dim")
-            config.output_path.mkdir(parents=True, exist_ok=True)
-            temp_chunks_dir = config.output_path / "temp_chunks"
-            temp_chunks_dir.mkdir(exist_ok=True)
+            if config.push_to_hub:
+                self.progress.print(f"Destination: {config.hub_repo_id} (HuggingFace Hub)", "dim")
+            else:
+                self.progress.print(f"Output: {config.output_path}", "dim")
+
+            # Create temp directory for chunks
+            if config.push_to_hub:
+                temp_chunks_dir = Path(tempfile.mkdtemp(prefix="rdf_chunks_"))
+            else:
+                config.output_path.mkdir(parents=True, exist_ok=True)
+                temp_chunks_dir = config.output_path / "temp_chunks"
+                temp_chunks_dir.mkdir(exist_ok=True)
+
             total_triples = 0
             chunk_count = 0
             with self.progress.progress_bar("Processing simple streaming chunks...", total=None) as (progress, task):
@@ -585,8 +935,13 @@ class SimpleStreamingStrategy(ConversionStrategy):
             self.progress.print(f"✓ Processed {total_triples:,} triples in {chunk_count} chunks", "green")
             dataset_dict = self._merge_parquet_chunks(temp_chunks_dir, config)
             shutil.rmtree(temp_chunks_dir)
+
+            if config.push_to_hub:
+                self._save_dataset(dataset_dict, config)
+
             elapsed = time.time() - start_time
-            self._save_dataset_info_json(config, total_triples, dataset_dict, elapsed)
+            if config.output_path and not config.push_to_hub:
+                self._save_dataset_info_json(config, total_triples, dataset_dict, elapsed)
             self.progress.print("✓ Converted to HuggingFace dataset", "green")
             return ConversionResult(success=True, total_triples=total_triples,
                                     processing_time_seconds=elapsed,
@@ -606,50 +961,129 @@ class ParallelStreamingStrategy(ConversionStrategy):
 
     @property
     def description(self) -> str:
-        return "Parallel streaming conversion using multiprocessing."
+        return "Parallel streaming conversion using multiprocessing with Dataset.from_generator()."
 
     def convert(self, config: ConversionConfig) -> ConversionResult:
         start_time = time.time()
-        num_workers = config.num_workers or mp.cpu_count()
+        num_workers = config.num_workers or max(1, mp.cpu_count() - 1)
         try:
             file_size_mb = self.file_handler.get_file_size_mb(config.input_path)
             self.progress.print(f"Converting: {config.input_path.name}", "cyan")
             self.progress.print(f"File size: {file_size_mb:.2f} MB", "dim")
-            self.progress.print(f"Chunk size: {config.chunk_size:,} lines", "dim")
+            self.progress.print(f"Chunk size: {config.chunk_size:,} lines per batch", "dim")
             self.progress.print(f"Workers: {num_workers} CPU cores", "dim")
+            if config.push_to_hub:
+                self.progress.print(f"Destination: {config.hub_repo_id} (HuggingFace Hub)", "dim")
+            else:
+                self.progress.print(f"Output: {config.output_path}", "dim")
             self.progress.emit_progress(5)
-            config.output_path.mkdir(parents=True, exist_ok=True)
-            temp_chunks_dir = config.output_path / "temp_chunks"
-            temp_chunks_dir.mkdir(exist_ok=True)
-            total_triples = 0
-            chunk_count = 0
+
+            # Check if this is GeoNames format
+            is_geonames = "geonames" in str(config.input_path).lower()
+
+            self.progress.print("\n[yellow]Creating dataset from streamed chunks...[/yellow]")
+            self.progress.emit_progress(10)
+
+            # Estimate total chunks based on file size (rough estimate)
             estimated_chunks = max(10, int(file_size_mb * 1024 * 1024 / (config.chunk_size * 100)))
-            with self.progress.progress_bar("Processing parallel chunks...", total=None) as (progress, task):
-                # For parallel processing, choose method based on format
-                if config.rdf_format in ("nt", "ntriples"):
-                    batches = list(self._stream_ntriples(config))
+
+            # Create generator function that yields individual triples
+            def dataset_generator():
+                triple_count = 0
+                chunk_count = 0
+                last_print_count = 0
+
+                # Select appropriate streaming method based on format
+                if is_geonames and config.rdf_format in ("xml", "application/rdf+xml"):
+                    # Check if it's the special GeoNames format (URLs followed by XML)
+                    with open(config.input_path, encoding="utf-8", errors="ignore") as f:
+                        first_line = f.readline().strip()
+                        if first_line.startswith("http://") or first_line.startswith("https://"):
+                            chunk_iter = self._stream_geonames_parallel(config, num_workers)
+                        else:
+                            chunk_iter = self._stream_generic_rdf(config)
+                elif config.rdf_format in ("nt", "ntriples"):
+                    chunk_iter = self._stream_ntriples_parallel(config, num_workers)
+                elif config.rdf_format in ("turtle", "ttl"):
+                    chunk_iter = self._stream_turtle_chunks(config)
                 else:
-                    batches = list(self._stream_generic_rdf(config))
-                # Use multiprocessing Pool for parallel processing the batches
-                with mp.Pool(processes=num_workers) as pool:
-                    results = pool.map(lambda b: b, batches)
-                    for batch in results:
-                        schema = self.schema_manager.get_arrow_schema()
-                        table = pa.Table.from_pylist(batch, schema=schema)
-                        chunk_file = temp_chunks_dir / f"chunk-{chunk_count:05d}.parquet"
-                        pq.write_table(table, chunk_file)
-                        total_triples += len(batch)
-                        chunk_count += 1
-                        progress.update(task, advance=1)
-            self.progress.print(f"✓ Processed {total_triples:,} triples in {chunk_count} chunks", "green")
-            self.progress.emit_progress(80)
-            dataset_dict = self._merge_parquet_chunks(temp_chunks_dir, config)
-            shutil.rmtree(temp_chunks_dir)
-            self.progress.emit_progress(100)
+                    # For other formats, use generic streaming
+                    chunk_iter = self._stream_generic_rdf(config)
+
+                # Yield individual triples from chunks
+                for chunk in chunk_iter:
+                    chunk_count += 1
+                    for triple in chunk:
+                        triple_count += 1
+                        yield triple
+
+                    if triple_count - last_print_count >= 100000:
+                        elapsed = time.time() - start_time
+                        rate = triple_count / elapsed if elapsed > 0 else 0
+                        progress_pct = 10 + min(60, int((chunk_count / estimated_chunks) * 60))
+                        self.progress.print(
+                            f"  Processing: {chunk_count:,} chunks • "
+                            f"{triple_count:,} triples • {rate:.0f} triples/sec",
+                        )
+                        self.progress.emit_progress(progress_pct)
+                        last_print_count = triple_count
+
+            # Create dataset using from_generator for true streaming
+            features = self.schema_manager.get_hf_features()
+            dataset = Dataset.from_generator(dataset_generator, features=features)
+
+            self.progress.emit_progress(75)
+            self.progress.print("[green]✓ Dataset generation complete[/green]")
+
+            # Get statistics after generation
+            total_triples = len(dataset)
+
+            # Wrap in DatasetDict
+            dataset_dict = DatasetDict({"data": dataset})
+
+            # Add metadata
+            if config.metadata:
+                dataset_dict = self.metadata_handler.add_metadata(dataset_dict, config.metadata)
+
             elapsed = time.time() - start_time
-            self._save_dataset_info_json(config, total_triples, dataset_dict, elapsed,
-                                         extra_info={"num_workers": num_workers})
-            self.progress.print("✓ Converted to HuggingFace dataset", "green")
+
+            if config.push_to_hub:
+                self.progress.print(f"[yellow]Uploading to HuggingFace Hub: {config.hub_repo_id}...[/yellow]")
+                self.progress.emit_progress(95)
+                dataset_dict.push_to_hub(config.hub_repo_id, private=False)
+                self.progress.print(f"[bold green]✓ Successfully uploaded to {config.hub_repo_id}[/bold green]")
+                self.progress.emit_progress(100)
+            else:
+                self.progress.print("[yellow]Saving final dataset...[/yellow]")
+                self.progress.emit_progress(95)
+                config.output_path.mkdir(parents=True, exist_ok=True)
+                dataset_dict.save_to_disk(str(config.output_path), num_proc=num_workers)
+                self.progress.emit_progress(100)
+
+                # Save dataset info
+                info = {
+                    "format": "parquet",
+                    "total_triples": total_triples,
+                    "data_size": len(dataset_dict["data"]),
+                    "source_format": config.rdf_format,
+                    "chunk_size": config.chunk_size,
+                    "num_workers": num_workers,
+                    "processing_time_seconds": round(elapsed, 2),
+                    "triples_per_second": round(total_triples / elapsed, 2) if elapsed > 0 else 0,
+                    "conversion_date": time.strftime("%Y-%m-%d %H:%M:%S"),
+                }
+                with open(config.output_path / "dataset_info.json", "w") as f:
+                    json.dump(info, f, indent=2)
+
+            self.progress.print("\n[bold green]✓ Successfully converted to HuggingFace dataset[/bold green]")
+            self.progress.print(f"[green]  Data split: {len(dataset_dict['data']):,} triples[/green]")
+            self.progress.print(f"[green]  Processing time: {elapsed:.1f} seconds[/green]")
+            self.progress.print(f"[green]  Speed: {total_triples / elapsed:.0f} triples/second[/green]")
+            if config.push_to_hub:
+                self.progress.print(f"[green]  Repository: {config.hub_repo_id}[/green]")
+            else:
+                self.progress.print(f"[green]  Location: {config.output_path}[/green]")
+
             return ConversionResult(success=True, total_triples=total_triples,
                                     processing_time_seconds=elapsed,
                                     output_path=config.output_path,
@@ -725,7 +1159,7 @@ Strategies:
         """
     )
     parser.add_argument("input", type=Path, help="Input RDF file")
-    parser.add_argument("output", type=Path, help="Output directory for HuggingFace dataset")
+    parser.add_argument("output", nargs="?", type=Path, help="Output directory for HuggingFace dataset (optional if --push-to-hub)")
     parser.add_argument("-f", "--format", default="turtle",
                         choices=["turtle", "nt", "ntriples", "xml", "n3", "trig", "nquads", "tsv"],
                         help="RDF format (default: turtle)")
@@ -747,10 +1181,20 @@ Strategies:
     parser.add_argument("--citation", type=str, help="Dataset citation")
     parser.add_argument("--homepage", type=str, help="Dataset homepage URL")
     parser.add_argument("--license", type=str, help="Dataset license")
+    parser.add_argument("--push-to-hub", action="store_true",
+                        help="Upload directly to HuggingFace Hub without saving to disk")
+    parser.add_argument("--hub-repo-id", type=str,
+                        help="HuggingFace Hub repository ID (required if --push-to-hub)")
     parser.add_argument("-v", "--verbose", action="store_true",
                         help="Verbose output including warnings")
     args = parser.parse_args()
 
+    # Validate arguments
+    if args.push_to_hub and not args.hub_repo_id:
+        parser.error("--hub-repo-id is required when using --push-to-hub")
+    if not args.push_to_hub and not args.output:
+        parser.error("output directory is required when not using --push-to-hub")
+
     log_level = logging.WARNING if args.verbose else logging.ERROR
     logging.basicConfig(level=log_level, format="%(levelname)s:%(name)s:%(message)s", force=True)
     logging.getLogger("rdflib").setLevel(log_level)
@@ -789,14 +1233,19 @@ Strategies:
         create_train_test_split=args.create_splits,
         test_size=args.test_size,
         clean_cache=args.clean_cache,
-        verbose=args.verbose
+        verbose=args.verbose,
+        push_to_hub=args.push_to_hub,
+        hub_repo_id=args.hub_repo_id
     )
 
     progress.print("\n" + "="*60, "cyan")
     progress.print("Unified RDF to HuggingFace Dataset Converter", "bold cyan")
     progress.print("="*60, "cyan")
     progress.print(f"Input: {config.input_path}", "")
-    progress.print(f"Output: {config.output_path}", "")
+    if config.push_to_hub:
+        progress.print(f"Destination: {config.hub_repo_id} (HuggingFace Hub)", "")
+    else:
+        progress.print(f"Output: {config.output_path}", "")
     progress.print(f"Format: {config.rdf_format}", "")
     progress.print(f"Strategy: {strategy.name}", "")
     progress.print("", "")
@@ -808,7 +1257,10 @@ Strategies:
         progress.print("="*60, "green")
         progress.print(f"Total triples: {result.total_triples:,}", "green")
         progress.print(f"Processing time: {result.processing_time_seconds:.1f}s", "green")
-        progress.print(f"Output: {result.output_path}", "green")
+        if config.push_to_hub:
+            progress.print(f"Repository: {config.hub_repo_id}", "green")
+        else:
+            progress.print(f"Output: {result.output_path}", "green")
         return 0
     else:
         progress.print("\n" + "="*60, "red")

scripts/upload_all_datasets.py

@@ -731,11 +731,11 @@ def get_rdf_format(dataset_info: DatasetInfo) -> str:
     # 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", 
+        "rdf/xml": "xml",
         "xml": "xml",
         "turtle": "turtle",
         "ttl": "turtle",
@@ -745,18 +745,18 @@ def get_rdf_format(dataset_info: DatasetInfo) -> str:
         "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"
 
@@ -831,10 +831,11 @@ This dataset was prepared using the CleverErnie GISM framework:
 python scripts/rdf_dataset_downloader.py {dataset_info.id} -o datasets/
 
 # Convert to HuggingFace format
-python scripts/convert_rdf_to_hf_dataset.py \\
+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)}
+    --format {get_rdf_format(dataset_info)} \\
+    --strategy auto
 
 # Upload to HuggingFace Hub
 python scripts/upload_all_datasets.py --dataset {dataset_info.id}
@@ -849,8 +850,9 @@ python scripts/upload_all_datasets.py --dataset {dataset_info.id}
 ### Conversion Details
 
 - Converted using: [CleverErnie GISM](https://github.com/cleverthis/cleverernie)
-- Conversion script: `scripts/convert_rdf_to_hf_dataset.py`
+- 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
 
@@ -1009,85 +1011,52 @@ def convert(args: argparse.Namespace, rdf_file: Path, dataset_id: str) -> None:
                 console.print(f"[green]Created dataset card: {readme_path}[/green]")
         else:
             # Original conversion logic for other datasets
-            # Determine if we should use streaming converter based on dataset size
-            use_streaming = False
-            streaming_script = (
+            # Use unified converter which automatically selects best strategy
+            unified_script = (
                 Path(__file__).parent
-                / "convert_rdf_to_hf_dataset_streaming_parallel.py"
+                / "convert_rdf_to_hf_dataset_unified.py"
             )
 
-            # Use streaming for medium and larger datasets, or specifically for geonames
-            if (
-                dataset_info.category in ["medium", "large", "xlarge"]
-                or "geonames" in dataset_id.lower()
-            ):
-                use_streaming = streaming_script.exists()
-                if use_streaming:
-                    console.print(
-                        "[dim]Using parallel streaming converter for "
-                        f"{dataset_info.category} dataset[/dim]"
-                    )
+            # 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:
-                script_name = (
-                    "convert_rdf_to_hf_dataset_streaming_parallel.py"
-                    if use_streaming
-                    else "convert_rdf_to_hf_dataset.py"
-                )
                 console.print(
-                    f"[dim]Would run: python scripts/{script_name} "
-                    "{rdf_file} {hf_dataset_dir}[/dim]"
+                    f"[dim]Would run: python scripts/convert_rdf_to_hf_dataset_unified.py "
+                    "{rdf_file} {hf_dataset_dir} --strategy auto[/dim]"
                 )
             else:
-                # Choose appropriate converter script
-                if use_streaming:
-                    # Use streaming converter with the appropriate chunk size
-                    # based on dataset size. Larger chunk sizes are more memory
-                    # efficient but still reasonable
-                    chunk_size = 50000  # Default for streaming
-                    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
+                # 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(streaming_script),
-                            str(rdf_file),
-                            str(hf_dataset_dir),
-                            "--format",
-                            get_rdf_format(dataset_info),
-                            "--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,
-                    )
-                else:
-                    # Use original converter for small datasets
-                    result = subprocess.run(
-                        [
-                            sys.executable,
-                            str(Path(__file__).parent / "convert_rdf_to_hf_dataset.py"),
-                            str(rdf_file),
-                            str(hf_dataset_dir),
-                            "--format",
-                            get_rdf_format(dataset_info),
-                            "--description",
-                            dataset_info.description,
-                            "--homepage",
-                            dataset_info.url,
-                            "--license",
-                            dataset_info.license if dataset_info.license else "Unknown",
-                        ],
-                        capture_output=False,
-                    )
+                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(
@@ -1290,13 +1259,16 @@ def main() -> int:
         print(f"`download` took {one_dataset_download_duration:.2f} seconds:")
 
         # Decompress
-        console.print(f"[bold cyan]Decompressing {dataset_id}...[/bold cyan]")
+        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]")
-            return 1
+            assert rdf_file is not None
+            result = decompress(rdf_file)
+            if not result:
+                console.print("[red]Decompression failed. Aborting.[/red]")
+                return 1
 
         one_dataset_decompress_time = time.monotonic()
         one_dataset_decompress_duration = one_dataset_decompress_time - one_dataset_download_time