scripts/rdf_to_hf_incremental.py

@@ -32,6 +32,7 @@ from __future__ import annotations
 import argparse
 import contextlib
 import hashlib
+import io
 import json
 import logging
 import multiprocessing as mp
@@ -42,6 +43,7 @@ import time
 import zipfile
 from collections.abc import Iterator
 from pathlib import Path
+from xml.etree import ElementTree as ET
 
 import pyarrow as pa
 import pyarrow.parquet as pq
@@ -51,7 +53,19 @@ from huggingface_hub import HfApi, hf_hub_download
 from rdflib import Graph, Literal, URIRef
 from rich.console import Console
 from rich.table import Table
-from src.streaming.extract_stream import (
+
+# Ensure project root is on sys.path
+PROJECT_ROOT = Path(__file__).resolve().parents[1]
+if str(PROJECT_ROOT) not in sys.path:
+    sys.path.insert(0, str(PROJECT_ROOT))
+
+from convert_rdf_to_hf_dataset_streaming_parallel import stream_rdf_chunks  # noqa: E402
+from dataset_registry import (  # noqa: E402
+    DATASET_REGISTRY,
+    DatasetInfo,
+    get_dataset_config,
+)
+from src.streaming.extract_stream import (  # noqa: E402
     ExtractConfig,
     _infer_name_from_url,
     iter_decompressed_bytes,
@@ -60,12 +74,7 @@ from src.streaming.extract_stream import (
     spool_zip_to_tempfile,
     detect_compression_format,
 )
-from src.streaming.http_stream import HTTPStreamConfig, stream_http_bytes
-
-# Ensure project root is on sys.path
-PROJECT_ROOT = Path(__file__).resolve().parents[1]
-if str(PROJECT_ROOT) not in sys.path:
-    sys.path.insert(0, str(PROJECT_ROOT))
+from src.streaming.http_stream import HTTPStreamConfig, stream_http_bytes  # noqa: E402
 
 
 # Setup logger
@@ -135,19 +144,17 @@ def stream_turtle_chunks_from_lines(
 
     current_chunk: list[str] = []
     triple_count = 0
-    line_count = 0
     prefix_lines: list[str] = []
     in_prefixes = True
 
-    for line in lines:
-        line_count += 1
+    for line_count, line in enumerate(lines, start=1):
         stripped = line.strip()
 
         if not stripped or stripped.startswith("#"):
             continue
 
-        if (in_prefixes
-                and (stripped.startswith("@prefix") or stripped.startswith("@base"))):
+        is_prefix = stripped.startswith("@prefix") or stripped.startswith("@base")
+        if in_prefixes and is_prefix:
             prefix_lines.append(line)
             continue
         elif in_prefixes:
@@ -158,17 +165,16 @@ def stream_turtle_chunks_from_lines(
 
             if triple_count >= chunk_size:
                 chunk_text = (
-                        "".join(prefix_lines)
-                        + "\n"
-                        + "".join(current_chunk)
-                        + "\n"
-                        + line
+                    "".join(prefix_lines) + "\n" +
+                    "".join(current_chunk) + "\n" + line
                 )
                 try:
                     graph = Graph()
                     graph.parse(data=chunk_text, format="turtle")
-                    triples = [_convert_rdf_triple_to_dict(s, p, o)
-                               for s, p, o in graph]
+                    triples = [
+                        _convert_rdf_triple_to_dict(s, p, o)
+                        for s, p, o in graph
+                    ]
                     if triples:
                         yield triples
                 except Exception as e:
@@ -227,6 +233,179 @@ def stream_ntriples_from_lines(lines: Iterator[str]) -> Iterator[list[dict]]:
         yield batch
 
 
+class _LineIteratorReader(io.RawIOBase):
+    """Adapter to convert line iterator to file-like object for XML parsing."""
+
+    def __init__(self, lines: Iterator[str]):
+        self._lines = lines
+        self._buffer = b''
+        self._exhausted = False
+
+    def readable(self) -> bool:
+        return True
+
+    def readinto(self, b):  # type: ignore[override]
+        """Read bytes from line iterator into buffer."""
+        if self._exhausted and not self._buffer:
+            return 0
+
+        # Fill buffer from line iterator
+        while len(self._buffer) < len(b) and not self._exhausted:
+            try:
+                line = next(self._lines)
+                self._buffer += line.encode('utf-8')
+            except StopIteration:
+                self._exhausted = True
+                break
+
+        # Copy buffer to output
+        n = min(len(self._buffer), len(b))
+        b[:n] = self._buffer[:n]
+        self._buffer = self._buffer[n:]
+        return n
+
+
+def _wrap_description_xml(description_xml: str) -> str:
+    """Wrap a Description element in a minimal valid XML document.
+
+    ElementTree preserves namespace URIs when serializing elements, so we
+    only need to provide the minimal RDF namespace declaration. Additional
+    namespaces are already included in the serialized element.
+
+    Args:
+        description_xml: Serialized Description element
+
+    Returns:
+        Complete XML document string
+    """
+    # Minimal RDF/XML wrapper - ElementTree preserves other namespaces
+    wrapped = (
+        '<?xml version="1.0" encoding="UTF-8"?>\n'
+        '<rdf:RDF xmlns:rdf="http://www.w3.org/1999/02/22-rdf-syntax-ns#">\n'
+        f'{description_xml}\n'
+        '</rdf:RDF>'
+    )
+    return wrapped
+
+
+def _parse_description_xml(xml_content: str) -> list[dict]:
+    """Parse a Description XML fragment and extract all triples.
+
+    Args:
+        xml_content: Complete XML document containing one Description element
+
+    Returns:
+        List of triple dictionaries
+    """
+    try:
+        graph = Graph()
+        graph.parse(data=xml_content, format="xml")
+
+        # Extract all triples from this Description
+        triples = [_convert_rdf_triple_to_dict(s, p, o) for s, p, o in graph]
+        return triples
+    except Exception as e:
+        logger.error(f"Error parsing Description XML: {e}")
+        logger.debug(f"XML content (first 500 chars): {xml_content[:500]}")
+        return []  # Return empty list on error, don't crash
+
+
+def stream_rdfxml_chunks_from_lines(
+    lines: Iterator[str],
+    chunk_size: int = DEFAULT_BATCH_SIZE,
+) -> Iterator[list[dict]]:
+    """Stream RDF/XML format from a line iterator using standard XML parser.
+
+    Uses xml.etree.ElementTree.iterparse for robust XML parsing that handles:
+    - XML comments, CDATA, processing instructions
+    - Multi-line elements and attributes
+    - Nested elements with correct depth tracking
+    - All XML edge cases per W3C specification
+
+    Args:
+        lines: Iterator yielding text lines of RDF/XML data
+        chunk_size: Number of Description elements to accumulate before yielding
+
+    Yields:
+        Lists of triple dictionaries
+    """
+    console = Console()
+    console.print("[yellow]Streaming RDF/XML with standard XML parser[/yellow]")
+
+    # Convert line iterator to file-like object for XML parser
+    file_obj = io.BufferedReader(_LineIteratorReader(lines))
+
+    triple_batch: list[dict] = []
+    batch_description_count = 0
+    total_descriptions = 0
+
+    try:
+        # Use iterparse to process XML incrementally
+        # Get 'end' events when elements are complete
+        for _event, elem in ET.iterparse(file_obj, events=('end',)):
+            # Process complete Description elements
+            if _is_description_element(elem):
+                total_descriptions += 1
+
+                # Serialize element to string (preserves namespaces)
+                description_xml = ET.tostring(elem, encoding='unicode')
+
+                # Wrap for rdflib parsing
+                wrapped_xml = _wrap_description_xml(description_xml)
+
+                # Parse and extract triples
+                triples = _parse_description_xml(wrapped_xml)
+                if triples:
+                    triple_batch.extend(triples)
+                    batch_description_count += 1
+
+                # Clear element to free memory
+                elem.clear()
+
+                # Yield when chunk_size reached
+                if batch_description_count >= chunk_size and triple_batch:
+                    yield triple_batch
+                    triple_batch = []
+                    batch_description_count = 0
+
+                # Progress logging every N descriptions
+                if total_descriptions % PROGRESS_LOG_INTERVAL == 0:
+                    console.print(
+                        f"[dim]Processed {total_descriptions:,} "
+                        f"Description elements...[/dim]"
+                    )
+
+    except ET.ParseError as e:
+        logger.error(f"XML parsing error: {e}")
+        logger.error("This may indicate malformed XML in the source file")
+
+    # Yield final batch
+    if triple_batch:
+        yield triple_batch
+
+    console.print(
+        f"[green]Completed: processed {total_descriptions:,} "
+        f"Description elements[/green]"
+    )
+
+
+def _is_description_element(elem: ET.Element) -> bool:
+    """Check if element is an rdf:Description element.
+
+    Args:
+        elem: XML element
+
+    Returns:
+        True if element is rdf:Description
+    """
+    # Handle both namespaced and non-namespaced forms
+    tag = elem.tag
+    return (
+        tag.endswith('Description') and
+        ('rdf' in tag.lower() or 'Description' in tag)
+    )
+
+
 def stream_rdf_chunks_from_lines(
     lines: Iterator[str],
     format: str = "turtle",
@@ -236,7 +415,8 @@ def stream_rdf_chunks_from_lines(
 
     Args:
         lines: Iterator yielding text lines of RDF data
-        format: RDF serialization format (supported: nt, ntriples, turtle, ttl)
+        format: RDF serialization format
+            (supported: nt, ntriples, turtle, ttl, xml, rdf, rdfxml)
         chunk_size: Number of triples per chunk for Turtle parsing
 
     Yields:
@@ -249,10 +429,12 @@ def stream_rdf_chunks_from_lines(
         yield from stream_ntriples_from_lines(lines)
     elif format in ("turtle", "ttl"):
         yield from stream_turtle_chunks_from_lines(lines, chunk_size=chunk_size)
+    elif format in ("xml", "rdf", "rdfxml"):
+        yield from stream_rdfxml_chunks_from_lines(lines, chunk_size=chunk_size)
     else:
         raise ValueError(
             f"Streaming line-iterator parser not implemented for format: {format}. "
-            f"Supported formats: nt, ntriples, turtle, ttl"
+            f"Supported formats: nt, ntriples, turtle, ttl, xml, rdf, rdfxml"
         )
 
 
@@ -261,9 +443,11 @@ def stream_rdf_chunks_from_lines(
 # ============================================================================
 
 
-def _pick_zip_member(zf: zipfile.ZipFile,
-                     *,
-                     prefer_exts: tuple[str, ...] = (".ttl", ".nt")) -> str:
+def _pick_zip_member(
+    zf: zipfile.ZipFile,
+    *,
+    prefer_exts: tuple[str, ...] = (".ttl", ".nt", ".rdf", ".xml"),
+) -> str:
     """Select the best RDF member file from a ZIP archive."""
     candidates = []
     for info in zf.infolist():
@@ -314,15 +498,15 @@ def open_rdf_line_stream_from_url(
 
     # TAR archives
     if lower.endswith((".tar.gz", ".tgz", ".tar")):
-        for member_name, member_bytes in iter_tar_member_bytes(name, byte_iter):
-            if member_name.lower().endswith((".ttl", ".nt")):
+        for member_name, member_bytes in iter_tar_member_bytes(byte_iter, tar_name=name):
+            if member_name.lower().endswith((".ttl", ".nt", ".rdf", ".xml")):
                 lines = iter_text_lines(
                     member_bytes,
                     encoding=extract_cfg.encoding,
-                    errors=extract_cfg.errors
+                    errors=extract_cfg.errors,
                 )
                 return lines, f"{name}:{member_name}"
-        raise ValueError("No .ttl/.nt member found in tar archive stream.")
+        raise ValueError("No .ttl/.nt/.rdf/.xml member found in tar archive stream.")
 
     # ZIP archives
     if lower.endswith(".zip"):
@@ -342,15 +526,11 @@ def open_rdf_line_stream_from_url(
                 os.remove(tmp_path)
 
     # Compressed files and plain text
-    compression_format = detect_compression_format(name)
-
-    decompressed = iter_decompressed_bytes(
-        byte_iter = byte_iter,
-    )
+    decompressed = iter_decompressed_bytes(byte_iter, compressed_name=name)
     lines = iter_text_lines(
         decompressed,
         encoding=extract_cfg.encoding,
-        errors=extract_cfg.errors
+        errors=extract_cfg.errors,
     )
     return lines, name
 
@@ -415,22 +595,23 @@ def create_rdf_chunk_iterator(
     if input_type == "file":
         # Local file mode: use optimized file-based streaming with multiprocessing
         console.print("[cyan]Mode: Local file streaming[/cyan]")
-        console.print(f"[dim]Using parallel file-based parser ({num_workers} "
-                      "workers)[/dim]")
+        console.print(
+            f"[dim]Using parallel file-based parser "
+            f"({num_workers} workers)[/dim]"
+        )
         file_path = Path(input_value)
         chunk_iter = stream_rdf_chunks(
-            file_path,
-            rdf_format,
-            DEFAULT_BATCH_SIZE,
-            num_workers
+            file_path, rdf_format, DEFAULT_BATCH_SIZE, num_workers
         )
         return chunk_iter, f"file:{file_path.name}"
 
     elif input_type == "registry":
         # Streaming download mode: use line-based streaming
         console.print("[cyan]Mode: Streaming download from registry[/cyan]")
-        console.print("[dim]Using line-based streaming parser (no "
-                      "multiprocessing)[/dim]")
+        console.print(
+            "[dim]Using line-based streaming parser "
+            "(no multiprocessing)[/dim]"
+        )
 
         ds = DATASET_REGISTRY[input_value]
         url = ds.url
@@ -448,9 +629,7 @@ def create_rdf_chunk_iterator(
         console.print(f"[green]✓ Streaming source ready:[/green] {source_name}")
 
         chunk_iter = stream_rdf_chunks_from_lines(
-            lines,
-            format=rdf_format,
-            chunk_size=DEFAULT_BATCH_SIZE
+            lines, format=rdf_format, chunk_size=DEFAULT_BATCH_SIZE
         )
         return chunk_iter, source_name
 
@@ -555,9 +734,12 @@ def convert_and_upload_incremental(
             except Exception as e:
                 error_str = str(e).lower()
                 # Check if it's a rate limit error (429)
-                if ("429" in error_str
-                        or "too many requests" in error_str
-                        or "rate limit" in error_str):
+                is_rate_limit = (
+                    "429" in error_str
+                    or "too many requests" in error_str
+                    or "rate limit" in error_str
+                )
+                if is_rate_limit:
                     if attempt < max_retries - 1:
                         # Exponential backoff with jitter
                         delay = min(base_delay * (2 ** attempt), max_delay)
@@ -572,8 +754,10 @@ def convert_and_upload_incremental(
                         time.sleep(wait_time)
                         buffer.seek(0)  # Reset buffer position for retry
                     else:
-                        console.print("[red]✗ Max retries reached for shard "
-                                      f"{current_shard_idx}[/red]")
+                        console.print(
+                            f"[red]✗ Max retries reached for "
+                            f"shard {current_shard_idx}[/red]"
+                        )
                         raise
                 else:
                     # Not a rate limit error, re-raise immediately
@@ -625,8 +809,10 @@ def convert_and_upload_incremental(
 
                     ensure_writer()
                     finalize_and_upload(shard_idx, shard_rows)
-                    console.print("[green]✓ Uploaded shard "
-                                  f"{shard_idx} ({shard_rows:,} rows)[/green]")
+                    console.print(
+                        f"[green]✓ Uploaded shard {shard_idx} "
+                        f"({shard_rows:,} rows)[/green]"
+                    )
                     # Add small delay to avoid overwhelming HuggingFace API
                     time.sleep(1.0)
 
@@ -634,8 +820,10 @@ def convert_and_upload_incremental(
                 shard_rows = 0
 
                 if max_shards is not None and shard_idx >= max_shards:
-                    console.print("[yellow]Stopping after "
-                                  f"max_shards={max_shards}[/yellow]")
+                    console.print(
+                        f"[yellow]Stopping after "
+                        f"max_shards={max_shards}[/yellow]"
+                    )
                     break
 
         if max_shards is not None and shard_idx >= max_shards:
@@ -651,8 +839,10 @@ def convert_and_upload_incremental(
 
         ensure_writer()
         finalize_and_upload(shard_idx, shard_rows)
-        console.print("[green]✓ Uploaded final shard "
-                      f"{shard_idx} ({shard_rows:,} rows)[/green]")
+        console.print(
+            f"[green]✓ Uploaded final shard {shard_idx} "
+            f"({shard_rows:,} rows)[/green]"
+        )
         shard_idx += 1
 
     elapsed = time.time() - start_time
@@ -703,11 +893,15 @@ def list_datasets() -> None:
     for info in DATASET_REGISTRY.values():
         by_category[info.category] = by_category.get(info.category, 0) + 1
 
-    console.print(f"\n[bold]Total:[/bold] {total} datasets "
-                  f"({available_count} available)")
+    console.print(
+        f"\n[bold]Total:[/bold] {total} datasets "
+        f"({available_count} available)"
+    )
     console.print("[bold]By category:[/bold] ", end="")
-    console.print(", ".join(f"{cat}: {count}"
-                            for cat, count in sorted(by_category.items())))
+    cat_list = ", ".join(
+        f"{cat}: {count}" for cat, count in sorted(by_category.items())
+    )
+    console.print(cat_list)
     console.print()
 
 
@@ -796,14 +990,16 @@ size_categories:
 
 ### Dataset Summary
 
-This dataset contains RDF triples from {dataset_info.name} converted to HuggingFace
-dataset format for easy use in machine learning pipelines.
+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
+- **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"}
+- **Original License:** {dataset_info.license if dataset_info.license else
+  "See original source"}
 
 ### Recommended Use
 
@@ -813,7 +1009,8 @@ dataset format for easy use in machine learning pipelines.
 
 ## RDF Format
 
-This dataset uses a standard lossless format for representing RDF triples. Each triple is a row with 6 fields:
+This dataset uses a standard lossless format for representing RDF triples.
+Each triple is a row with 6 fields:
 - `subject`: Subject URI or blank node
 - `predicate`: Predicate URI
 - `object`: Object value (URI, literal, or blank node)
@@ -859,7 +1056,10 @@ This dataset is part of the CleverThis knowledge graph collection.
 def main() -> int:
     """Unified RDF to HuggingFace converter with automatic mode detection."""
     parser = argparse.ArgumentParser(
-        description="Convert RDF to HuggingFace with incremental Parquet writing",
+        description=(
+            "Convert RDF to HuggingFace dataset "
+            "with incremental Parquet writing"
+        ),
         formatter_class=argparse.RawDescriptionHelpFormatter,
         epilog="""
 Input Modes:
@@ -879,7 +1079,8 @@ Examples:
   python rdf_to_hf_incremental.py /data/wordnet.nt username/wordnet-dataset
 
   # Process local file with explicit format
-  python rdf_to_hf_incremental.py /data/wordnet.nt username/wordnet-dataset --format ntriples
+  python rdf_to_hf_incremental.py /data/wordnet.nt username/wordnet-dataset \\
+    --format ntriples
 
   # Production with checkpointing
   python rdf_to_hf_incremental.py dbpedia-core username/dbpedia \\
@@ -963,7 +1164,10 @@ Features:
         "--num-workers",
         type=int,
         default=None,
-        help="Number of processes for local file parsing (default: CPU count - 1)",
+        help=(
+            "Number of worker processes for local file parsing "
+            "(default: CPU count - 1)"
+        ),
     )
     parser.add_argument(
         "--verify-after",
@@ -1008,10 +1212,14 @@ Features:
 
     # Validate required arguments if not in list mode
     if not args.input or not args.repo_id:
-        console.print("[red]Error: input and repo_id are required "
-                      "(unless using --list)[/red]")
-        console.print("[yellow]Usage: rdf_to_hf_incremental.py INPUT REPO_ID "
-                      "[OPTIONS][/yellow]")
+        console.print(
+            "[red]Error: input and repo_id are required "
+            "(unless using --list)[/red]"
+        )
+        console.print(
+            "[yellow]Usage: rdf_to_hf_incremental.py "
+            "INPUT REPO_ID [OPTIONS][/yellow]"
+        )
         console.print("[yellow]   or: rdf_to_hf_incremental.py --list[/yellow]")
         return 1
 
@@ -1028,42 +1236,72 @@ Features:
     if not rdf_format:
         if input_type == "registry":
             ds = DATASET_REGISTRY[input_value]
-            rdf_format = ds.format
-            console.print("[dim]Format auto-detected from registry: "
-                          f"{rdf_format}[/dim]")
+            # Map registry format to streaming parser format
+            registry_format = ds.format.lower()
+            format_mapping = {
+                "rdf": "rdf",
+                "rdf/xml": "rdf",
+                "xml": "rdf",
+                "turtle": "turtle",
+                "ttl": "turtle",
+                "nt": "ntriples",
+                "ntriples": "ntriples",
+            }
+            rdf_format = format_mapping.get(registry_format, registry_format)
+            console.print(
+                f"[dim]Format auto-detected from registry: "
+                f"{registry_format} → {rdf_format}[/dim]"
+            )
         elif input_type == "file":
-            # Infer from file extension, default to turtle
-            ext = Path(input_value).suffix.lower()
+            # Infer from file extension, handling compressed files
+            file_path = Path(input_value)
+            # Get stem (filename without extension) and check for double extensions
+            stem = file_path.stem.lower()
+            ext = file_path.suffix.lower()
+
+            # Handle compressed files: .rdf.xz, .ttl.gz, etc.
+            if ext in (".gz", ".bz2", ".xz", ".zip"):
+                # Check if stem has an RDF extension
+                stem_path = Path(stem)
+                if stem_path.suffix.lower() in (".rdf", ".xml", ".ttl", ".nt"):
+                    ext = stem_path.suffix.lower()
+
             format_map = {
                 ".ttl": "turtle",
                 ".nt": "ntriples",
                 ".nq": "nquads",
-                ".rdf": "xml",
-                ".xml": "xml",
+                ".rdf": "rdf",
+                ".xml": "rdf",
                 ".jsonld": "json-ld",
                 ".n3": "n3",
                 ".trig": "trig"
             }
             rdf_format = format_map.get(ext, "turtle")
-            console.print(f"[dim]Format inferred from extension: {rdf_format}[/dim]")
+            console.print(
+                f"[dim]Format inferred from extension: "
+                f"{ext} → {rdf_format}[/dim]"
+            )
 
-    console.print(f"[dim]Rows per shard: {args.rows_per_shard:,}, "
-                  f"Record batch size: {args.record_batch_size:,}[/dim]")
+    console.print(
+        f"[dim]Rows per shard: {args.rows_per_shard:,}, "
+        f"Record batch size: {args.record_batch_size:,}[/dim]"
+    )
 
     # Handle --dry-run mode
     if args.dry_run:
         console.print("\n[bold yellow]═══ DRY RUN MODE ═══[/bold yellow]")
-        console.print("[yellow]Showing what would be done:[/yellow]\n")
+        console.print(
+            "[yellow]Showing what would be done without executing:[/yellow]\n"
+        )
 
         console.print(f"[cyan]Input type:[/cyan] {input_type}")
         console.print(f"[cyan]Input value:[/cyan] {input_value}")
         console.print(f"[cyan]RDF format:[/cyan] {rdf_format}")
         console.print(f"[cyan]Target repo:[/cyan] {args.repo_id}")
         console.print(f"[cyan]Rows per shard:[/cyan] {args.rows_per_shard:,}")
-        console.print("[cyan]Record batch size:[/cyan] "
-                      f"{args.record_batch_size:,}")
-        console.print("[cyan]Num workers:[/cyan] "
-                      f"{args.num_workers or 'auto (CPU count - 1)'}")
+        console.print(f"[cyan]Record batch size:[/cyan] {args.record_batch_size:,}")
+        num_workers_display = args.num_workers or 'auto (CPU count - 1)'
+        console.print(f"[cyan]Num workers:[/cyan] {num_workers_display}")
         console.print(f"[cyan]Checkpoint file:[/cyan] {args.checkpoint or 'None'}")
 
         if input_type == "registry":
@@ -1078,21 +1316,33 @@ Features:
                 # Handle both string and numeric triple counts
                 if isinstance(ds.triples, (int, float)):
                     console.print(f"  Estimated triples: {int(ds.triples):,}")
-                    estimated_shards = (int(ds.triples) // args.rows_per_shard) + 1
-                    console.print(f"  Estimated shards: ~{estimated_shards:,}")
+                    estimated_shards = (
+                        (int(ds.triples) // args.rows_per_shard) + 1
+                    )
+                    console.print(
+                        f"  Estimated shards: ~{estimated_shards:,}"
+                    )
                 else:
                     console.print(f"  Estimated triples: {ds.triples}")
 
         console.print("\n[bold]Would perform:[/bold]")
-        console.print(f"  1. {'Stream from ' + DATASET_REGISTRY[input_value].url if input_type == 'registry' else 'Read from ' + input_value}") # noqa: E501
+        if input_type == 'registry':
+            action_1 = f"Stream from {DATASET_REGISTRY[input_value].url}"
+        else:
+            action_1 = f"Read from {input_value}"
+        console.print(f"  1. {action_1}")
         console.print(f"  2. Parse RDF triples in {rdf_format} format")
-        console.print("  3. Convert to Parquet shards "
-                      f"({args.rows_per_shard:,} rows each)")
+        console.print(
+            f"  3. Convert to Parquet shards "
+            f"({args.rows_per_shard:,} rows each)"
+        )
         console.print(f"  4. Upload shards to {args.repo_id}")
         console.print("  5. Create and upload README.md")
 
-        console.print("\n[bold green]✓ Dry run complete. "
-                      "Use without --dry-run to execute.[/bold green]")
+        console.print(
+            "\n[bold green]✓ Dry run complete. "
+            "Use without --dry-run to execute.[/bold green]"
+        )
         return 0
 
     # Initialize HuggingFace API
@@ -1100,8 +1350,9 @@ Features:
     if token:
         console.print("[dim]Using provided/environment token[/dim]")
     else:
-        console.print("[dim]Using cached credentials from "
-                      "'huggingface-cli login'[/dim]")
+        console.print(
+            "[dim]Using cached credentials from 'huggingface-cli login'[/dim]"
+        )
 
     try:
         api = HfApi(token=token)
@@ -1157,11 +1408,16 @@ Features:
             start_shard=start_shard,
         )
 
-        console.print("\n[bold green]✓ Successfully uploaded to "
-                      f"{args.repo_id}[/bold green]")
+        console.print(
+            f"\n[bold green]✓ Successfully uploaded to "
+            f"{args.repo_id}[/bold green]"
+        )
         console.print(f"[green]  Total rows: {total_rows:,}[/green]")
         console.print(f"[green]  Total shards: {total_shards}[/green]")
-        console.print(f"[green]  Repository: https://huggingface.co/datasets/{args.repo_id}[/green]")
+        console.print(
+            f"[green]  Repository: "
+            f"https://huggingface.co/datasets/{args.repo_id}[/green]"
+        )
 
         # Create and upload README.md (dataset card)
         if input_type == "registry":
@@ -1183,35 +1439,41 @@ Features:
                 console.print("[green]✓ Dataset card (README.md) "
                               "uploaded[/green]")
             except Exception as e:
-                console.print("[yellow]⚠ Warning: Could not upload README.md: "
-                              f"{e}[/yellow]")
-                console.print("[dim]You can manually create a README.md file "
-                              "for the dataset[/dim]")
+                console.print(
+                    f"[yellow]⚠ Warning: Could not upload README.md: "
+                    f"{e}[/yellow]"
+                )
+                console.print(
+                    "[dim]You can manually create a README.md file "
+                    "for the dataset[/dim]"
+                )
 
         # Verification
         if args.verify_after:
             console.print("\n[yellow]Verifying uploaded shards...[/yellow]")
             try:
                 repo_files = api.list_repo_files(
-                    repo_id=args.repo_id,
-                    repo_type="dataset"
+                    repo_id=args.repo_id, repo_type="dataset"
                 )
                 data_files = [f for f in repo_files if f.startswith("data/")]
-                console.print(f"[green]✓ Found {len(data_files)} shard files "
-                              "in repo[/green]")
+                console.print(
+                    f"[green]✓ Found {len(data_files)} shard files "
+                    f"in repo[/green]"
+                )
 
                 sample_n = min(args.verify_sample_shards, len(data_files))
                 for f in data_files[:sample_n]:
                     local_path = hf_hub_download(
                         repo_id=args.repo_id,
                         repo_type="dataset",
-                        filename=f
+                        filename=f,
                     )
                     pf = pq.ParquetFile(local_path)
                     console.print(f"[green]✓ Verified readable:[/green] {f}")
-                    console.print(
-                        "[dim]  Rows: "
-                        f"{pf.metadata.num_rows if pf.metadata else 'unknown'}[/dim]")
+                    num_rows = (
+                        pf.metadata.num_rows if pf.metadata else 'unknown'
+                    )
+                    console.print(f"[dim]  Rows: {num_rows}[/dim]")
             except Exception as e:
                 console.print(f"[red]✗ Verification failed: {e}[/red]")
 

src/streaming/http_stream.py

@@ -1,8 +1,11 @@
 from __future__ import annotations
 
 import time
+import urllib.error
+import urllib.request
 from collections.abc import Iterator
 from dataclasses import dataclass
+from urllib.parse import urlparse
 
 import httpx
 
@@ -18,12 +21,49 @@ class HTTPStreamConfig:
     follow_redirects: bool = True
 
 
+def _calculate_backoff_sleep(
+    attempt: int, base_s: float, max_s: float
+) -> float:
+    """Calculate exponential backoff sleep time."""
+    return min(base_s * (2 ** (attempt - 1)), max_s)
+
+
 def stream_http_bytes(
     url: str,
     *,
     headers: dict[str, str] | None = None,
     start_byte: int | None = None,
     config: HTTPStreamConfig | None = None,
+) -> Iterator[bytes]:
+    """Stream bytes from an HTTP(S) or FTP URL with retry and optional Range start.
+
+    - Yields chunks of bytes.
+    - If start_byte is provided, attempts to use Range requests (HTTP/HTTPS only).
+    - Retries on network/timeouts and 5xx responses with exponential backoff.
+    - Automatically detects protocol and routes to appropriate handler.
+    """
+    parsed = urlparse(url)
+    protocol = parsed.scheme.lower()
+
+    if protocol == "ftp":
+        yield from stream_ftp_bytes(url, config=config)
+    elif protocol in ("http", "https"):
+        yield from stream_http_bytes_impl(
+            url, headers=headers, start_byte=start_byte, config=config
+        )
+    else:
+        raise ValueError(
+            f"Unsupported protocol: {protocol}. "
+            f"Supported: http, https, ftp"
+        )
+
+
+def stream_http_bytes_impl(
+    url: str,
+    *,
+    headers: dict[str, str] | None = None,
+    start_byte: int | None = None,
+    config: HTTPStreamConfig | None = None,
 ) -> Iterator[bytes]:
     """Stream bytes from an HTTP(S) URL with retry and optional Range start.
 
@@ -57,10 +97,11 @@ def stream_http_bytes(
                         yield chunk
             return
         except (
-                httpx.TimeoutException,
-                httpx.NetworkError,
-                httpx.RemoteProtocolError,
-                httpx.HTTPStatusError) as e:
+            httpx.TimeoutException,
+            httpx.NetworkError,
+            httpx.RemoteProtocolError,
+            httpx.HTTPStatusError,
+        ) as e:
             # Retry on network/timeouts and 5xx. For 4xx (except 429), fail fast.
             if isinstance(e, httpx.HTTPStatusError):
                 status = e.response.status_code
@@ -70,10 +111,84 @@ def stream_http_bytes(
                 if status < 500 and status != 429:
                     raise
 
-            if attempt > cfg.max_retries:
+            if attempt >= cfg.max_retries:
                 raise
 
-            sleep_s = min(cfg.backoff_base_s * (2 ** (attempt - 1)), cfg.backoff_max_s)
+            sleep_s = _calculate_backoff_sleep(
+                attempt, cfg.backoff_base_s, cfg.backoff_max_s
+            )
+            time.sleep(sleep_s)
+
+
+def stream_ftp_bytes(
+    url: str,
+    *,
+    config: HTTPStreamConfig | None = None,
+) -> Iterator[bytes]:
+    """Stream bytes from an FTP URL with retry logic.
+
+    FTP Protocol Limitations:
+    - No Range header support (cannot resume interrupted downloads)
+    - No Content-Length in all cases (progress may be indeterminate)
+
+    Args:
+        url: FTP URL to stream from
+        config: Streaming configuration (chunk_size, timeouts, retries)
+
+    Yields:
+        Chunks of bytes from the FTP stream
+
+    Raises:
+        ValueError: If FTP authentication fails or file not found
+        urllib.error.URLError: On network errors after retries exhausted
+    """
+    cfg = config or HTTPStreamConfig()
+    attempt = 0
+
+    while True:
+        attempt += 1
+        try:
+            timeout = int(cfg.connect_timeout_s + cfg.read_timeout_s)
+            with urllib.request.urlopen(url, timeout=timeout) as response:
+                while True:
+                    chunk = response.read(cfg.chunk_size)
+                    if not chunk:
+                        break
+                    yield chunk
+            return
+
+        except urllib.error.URLError as e:
+            reason = str(e.reason) if hasattr(e, "reason") else str(e)
+
+            # FTP-specific error codes
+            if "530" in reason or "Login incorrect" in reason:
+                raise ValueError(f"FTP authentication failed: {url}") from e
+            elif "550" in reason or "No such file" in reason:
+                raise ValueError(f"File not found on FTP server: {url}") from e
+            elif "timed out" in reason.lower():
+                # Retry on timeout
+                if attempt >= cfg.max_retries:
+                    raise
+                sleep_s = _calculate_backoff_sleep(
+                    attempt, cfg.backoff_base_s, cfg.backoff_max_s
+                )
+                time.sleep(sleep_s)
+            else:
+                # Other errors: retry if we have attempts left
+                if attempt >= cfg.max_retries:
+                    raise
+                sleep_s = _calculate_backoff_sleep(
+                    attempt, cfg.backoff_base_s, cfg.backoff_max_s
+                )
+                time.sleep(sleep_s)
+
+        except Exception:
+            # Unexpected errors: retry if we have attempts left
+            if attempt >= cfg.max_retries:
+                raise
+            sleep_s = _calculate_backoff_sleep(
+                attempt, cfg.backoff_base_s, cfg.backoff_max_s
+            )
             time.sleep(sleep_s)