From f3dc2a449d2e618b30acf9490b95880eec077ad7 Mon Sep 17 00:00:00 2001 From: Luffy2208 Date: Tue, 23 Jun 2026 21:15:25 +0530 Subject: [PATCH] feat(export): implement Neo4j Bulk CSV Exporter and update registry docs (#261) (#665) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * feat(export): implement Neo4j Bulk CSV Exporter and update registry docs (#261) * fix(export): address review bugs in Neo4j CSV exporter - _write_csv: filter **options to known csv.writer dialect params only, preventing TypeError when callers pass kwargs like delimiter= or encoding= that would reach csv.writer twice or as unknown arguments - export_neo4j_csv: split kwargs into constructor-level init_params vs per-call call_kwargs before forwarding, eliminating the double-pass that caused dialect params to collide inside _write_csv - _prepare_export: remove dead node_id_lookup dict that was built but never consumed by any caller - export_knowledge_graph dispatch: drop the ambiguous "neo4j" format alias (kept "neo4j_csv" and "neo4j-csv"); "neo4j" conflicts with the codebase's established meaning of the live Bolt/Cypher store backend; add inline comment clarifying that file_path is treated as an output directory for this format - export_usage.md: fix all three wrong API examples — constructor params node_label_sep/strict_validation corrected to label_separator/strict, non-existent nodes_path/rels_path kwargs removed, convenience-method example updated to show the correct positional output_dir argument Co-Authored-By: KaifAhmad1 * docs(changelog): add Neo4j Bulk CSV Export entry for PR #665 Documents the new Neo4jCSVExporter feature contributed by @Luffy2208 and the five follow-up bug fixes (TypeError on dialect kwargs, double-pass kwargs split, dead node_id_lookup removal, ambiguous format="neo4j" alias removal, and wrong API examples in docs). Co-Authored-By: KaifAhmad1 --------- Co-authored-by: KaifAhmad1 Co-authored-by: KaifAhmad1 --- CHANGELOG.md | 33 + semantica/export/__init__.py | 15 +- semantica/export/export_usage.md | 74 +- semantica/export/methods.py | 78 +- semantica/export/neo4j_csv_exporter.py | 905 ++++++++++++++++++++++++ semantica/export/registry.py | 16 +- tests/export/test_neo4j_csv_exporter.py | 303 ++++++++ 7 files changed, 1410 insertions(+), 14 deletions(-) create mode 100644 semantica/export/neo4j_csv_exporter.py create mode 100644 tests/export/test_neo4j_csv_exporter.py diff --git a/CHANGELOG.md b/CHANGELOG.md index b88f9c60..59513b39 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -11,6 +11,39 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Added +- **Neo4j Bulk CSV Export** (#665) by @Luffy2208 + - Added `Neo4jCSVExporter` for generating Neo4j bulk-import CSV files compatible with `neo4j-admin database import` + - Produces deterministic `nodes.csv` and `relationships.csv` with stable node IDs — reuses existing graph IDs or derives reproducible SHA-256 content-based IDs when none are present + - Multi-label support via Neo4j `:LABEL` convention with configurable `label_separator` (default `;`) + - Alphabetically sorted property columns and deterministic row ordering for reproducible output across permuted inputs + - Relationship endpoint resolution: aliases (`name`, `text`, `label`) automatically mapped to stable node IDs + - Nested property serialisation to canonical JSON; flat scalar values written directly + - `dry_run()` method for pre-flight CSV validation without writing files + - `validate_export()` for post-write integrity checks (unique `:id`, consistent column widths, valid endpoint references) + - `export_nodes()` and `export_relationships()` for partial exports + - `strict=True` mode raises `ValidationError` on unresolved relationship endpoints + - `export_neo4j_csv()` convenience function and `format="neo4j_csv"` / `format="neo4j-csv"` dispatch in `export_knowledge_graph()` + - Registry integration under the `neo4j_csv` task namespace + - Documentation added to `semantica/export/export_usage.md` with usage examples, mapping assumptions, and `neo4j-admin` import command + - 13 tests covering headers, node/relationship CSV structure, multi-label, missing properties, deterministic output, CSV quoting/escaping, Unicode, empty graphs, dry-run, duplicate ID detection, ambiguous alias handling, nested property serialisation, and `KnowledgeGraph` integration + +### Fixed + +- **Neo4j CSV exporter `_write_csv` crashed with `TypeError` on dialect kwargs** (#665) by @KaifAhmad1 + - Passing `delimiter=`, `encoding=`, or any caller kwarg to `export_neo4j_csv` caused `csv.writer` to receive unknown or duplicate keyword arguments; `_write_csv` now whitelists only valid `csv.writer` dialect params (`quotechar`, `doublequote`, `skipinitialspace`, `escapechar`, `strict`) + +- **`export_neo4j_csv` double-passed kwargs to both the constructor and `export()`** (#665) by @KaifAhmad1 + - Constructor-level settings (`node_file_name`, `relationship_file_name`, `encoding`, `delimiter`, `label_separator`, `strict`) were merged into config for the constructor then re-forwarded as `**kwargs` to `export_knowledge_graph`, causing dialect params to collide; kwargs are now split into `init_kwargs` and `call_kwargs` before forwarding + +- **Dead `node_id_lookup` dict removed from `_prepare_export`** (#665) by @KaifAhmad1 + - The `{original_index → stable_id}` mapping was built on every export but never consumed; removed to avoid misleading future readers + +- **Dropped ambiguous `format="neo4j"` alias from `export_knowledge_graph` dispatch** (#665) by @KaifAhmad1 + - `"neo4j"` is used throughout the codebase to identify the live Bolt/Cypher graph store backend; routing it silently to the offline bulk-CSV exporter would have confused callers; only `"neo4j_csv"` and `"neo4j-csv"` are accepted + +- **`export_usage.md` documented non-existent constructor and function parameters** (#665) by @KaifAhmad1 + - Examples showed `node_label_sep` (correct: `label_separator`), `strict_validation` (correct: `strict`), and `nodes_path`/`rels_path` kwargs that do not exist; all three examples corrected to match the actual API + - **Public API Ingestion Support** (#602) by @Luffy2208 - Added `PublicAPIIngestor` class built on top of `RESTIngestor` for credential-free REST endpoints - Added `PublicAPIExample` and `PublicAPIExamples` catalog with 6 pre-configured no-auth examples: diff --git a/semantica/export/__init__.py b/semantica/export/__init__.py index 7a9e5285..77e9dd67 100644 --- a/semantica/export/__init__.py +++ b/semantica/export/__init__.py @@ -61,6 +61,14 @@ CSV Export: - Multi-file Export: Knowledge graph split into multiple CSV files (entities, relationships) +Neo4j Bulk CSV Export: + - Neo4j Admin Import CSV: Deterministic nodes.csv and relationships.csv + files compatible with neo4j-admin database import + - Stable Node IDs: Reuse existing graph IDs or derive deterministic IDs + from node contents + - Label and Property Mapping: Multi-label :LABEL values and sorted + property columns + Graph Export: - GraphML Serialization: GraphML format generation for graph visualization tools - GEXF Serialization: GEXF format generation for Gephi and similar tools @@ -124,6 +132,7 @@ Main Classes: - RDFExporter: RDF format export (Turtle, RDF/XML, JSON-LD) - JSONExporter: JSON and JSON-LD format export - CSVExporter: CSV format export for tabular data + - Neo4jCSVExporter: Neo4j bulk import CSV export - ParquetExporter: Parquet format export for analytics and data warehousing - GraphExporter: Graph format export (GraphML, GEXF, DOT) - YAMLExporter: YAML format export for semantic networks @@ -162,8 +171,8 @@ License: MIT """ from .arango_aql_exporter import ArangoAQLExporter -from .distance_exporter import DistanceExporter from .config import ExportConfig, export_config +from .distance_exporter import DistanceExporter try: from .arrow_exporter import ArrowExporter @@ -188,6 +197,7 @@ from .methods import ( export_graph, export_json, export_lpg, + export_neo4j_csv, export_owl, export_parquet, export_rdf, @@ -197,6 +207,7 @@ from .methods import ( get_export_method, list_available_methods, ) +from .neo4j_csv_exporter import Neo4jCSVExporter from .owl_exporter import OWLExporter try: @@ -228,6 +239,7 @@ __all__ = [ "NamespaceManager", "JSONExporter", "CSVExporter", + "Neo4jCSVExporter", "ParquetExporter", "GraphExporter", "SemanticNetworkYAMLExporter", @@ -249,6 +261,7 @@ __all__ = [ "export_owl", "export_vector", "export_lpg", + "export_neo4j_csv", "export_arango", "generate_report", "get_export_method", diff --git a/semantica/export/export_usage.md b/semantica/export/export_usage.md index c6001823..140c5a01 100644 --- a/semantica/export/export_usage.md +++ b/semantica/export/export_usage.md @@ -13,6 +13,8 @@ This guide demonstrates how to use the export module for exporting knowledge gra 7. [OWL Export](#owl-export) 8. [Vector Export](#vector-export) 9. [LPG Export](#lpg-export) + - [Cypher Query Export](#cypher-query-export) + - [Neo4j Bulk CSV Export](#neo4j-bulk-csv-export) 10. [Report Generation](#report-generation) 11. [Knowledge Graph Export](#knowledge-graph-export) 12. [Using Methods](#using-methods) @@ -442,6 +444,63 @@ exporter.export_knowledge_graph(kg, "graph.cypher") # mgconsole < graph.cypher ``` +### Neo4j Bulk CSV Export + +The `Neo4jCSVExporter` generates node and relationship CSV files specifically structured for Neo4j's high-performance bulk import utility (`neo4j-admin database import`). + +#### Basic Usage + +```python +from semantica.export import Neo4jCSVExporter + +# Initialize the exporter +exporter = Neo4jCSVExporter( + label_separator=";", # Separator for multi-label nodes (default: ";") + strict=True # Raise on unresolved relationship endpoints (default: True) +) + +# Export a knowledge graph — output_dir receives nodes.csv and relationships.csv +exporter.export_knowledge_graph(kg, "neo4j_import/") +``` + +#### Convenience Method + +You can also use the convenience wrapper `export_neo4j_csv`: + +```python +from semantica.export.methods import export_neo4j_csv + +export_neo4j_csv(kg, "neo4j_import/") +``` + +Pass `validate=True` to run a post-export integrity check before returning: + +```python +export_neo4j_csv(kg, "neo4j_import/", validate=True) +``` + +#### Importing into Neo4j + +Once the CSV files are generated, they can be imported into a new Neo4j database using the `neo4j-admin database import` command: + +```bash +neo4j-admin database import full \ + --nodes=nodes.csv \ + --relationships=relationships.csv \ + neo4j +``` + +#### Mapping Assumptions & Rules + +- **Header Specification**: + - Node CSV headers are generated as `:id` (node identifier) and `:LABEL` (labels). + - Relationship CSV headers are generated as `:START_ID` (source node), `:END_ID` (target node), and `:TYPE` (relationship type). + - Custom attributes/properties are written as standard columns. +- **Node Labels**: Supports multiple labels per node. Labels are serialized into a single string column using the configured `label_separator` (default `;`). +- **Stable Node IDs**: Reuses `id` (or `entity_id` / `node_id`) from node dictionaries. If a node is missing an ID, a stable content-derived ID is generated via a SHA-256 hash of its properties, ensuring reproducibility. +- **Relationship Endpoint Resolution**: If a relationship refers to nodes by their `name`, `text`, or `label` rather than their exact `id`, the exporter resolves these aliases automatically to their stable node IDs to guarantee that `:START_ID` and `:END_ID` strictly match existing node IDs. +- **Property Serialization**: Flat values (strings, numbers, booleans) are serialized directly. Nested properties (e.g. dicts, lists, sets) are serialized into a canonical JSON representation. + ## Report Generation ### HTML Report @@ -553,7 +612,8 @@ from semantica.export.methods import ( export_yaml, export_owl, export_vector, - export_lpg + export_lpg, + export_neo4j_csv ) # RDF export @@ -579,6 +639,9 @@ export_vector(vectors, "vectors.json", format="json") # LPG export export_lpg(kg, "graph.cypher", method="cypher") + +# Neo4j CSV bulk export +export_neo4j_csv(kg, nodes_path="nodes.csv", rels_path="relationships.csv") ``` ## Using Registry @@ -799,16 +862,16 @@ from semantica.export.methods import export_json def custom_jsonld_export(data, file_path, **kwargs): """Custom JSON-LD export with specific context.""" from semantica.export import JSONExporter - + exporter = JSONExporter(format="json-ld") - + # Add custom context if isinstance(data, dict) and "@context" not in data: data["@context"] = { "@vocab": "http://example.org/vocab#", "ex": "http://example.org/ns#" } - + exporter.export(data, file_path, format="json-ld", **kwargs) # Register custom method @@ -921,7 +984,7 @@ for format_name, file_path, export_func, kwargs in export_configs: 2. **Parallel Export**: Export to multiple formats in parallel ```python from concurrent.futures import ThreadPoolExecutor - + with ThreadPoolExecutor() as executor: executor.submit(export_json, kg, "output.json") executor.submit(export_rdf, kg, "output.ttl", format="turtle") @@ -954,4 +1017,3 @@ for format_name, file_path, export_func, kwargs in export_configs: with gzip.open("output.json.gz", "wb") as f_out: f_out.writelines(f_in) ``` - diff --git a/semantica/export/methods.py b/semantica/export/methods.py index 7751c7e6..6cec22c4 100644 --- a/semantica/export/methods.py +++ b/semantica/export/methods.py @@ -47,6 +47,9 @@ LPG Export: - "cypher": Cypher query format for Neo4j/Memgraph - "lpg": Labeled Property Graph format +Neo4j Bulk CSV Export: + - "neo4j_csv": CSV files for ``neo4j-admin database import`` + Report Generation: - "html": HTML report format - "markdown": Markdown report format @@ -168,6 +171,7 @@ from .csv_exporter import CSVExporter from .graph_exporter import GraphExporter from .json_exporter import JSONExporter from .lpg_exporter import LPGExporter +from .neo4j_csv_exporter import Neo4jCSVExporter from .owl_exporter import OWLExporter try: @@ -693,6 +697,63 @@ def export_lpg( raise +def export_neo4j_csv( + knowledge_graph: Any, + output_dir: Union[str, Path], + method: str = "default", + **kwargs, +) -> Dict[str, Path]: + """ + Export knowledge graph to Neo4j bulk-import CSV files. + + This wrapper writes deterministic ``nodes.csv`` and ``relationships.csv`` + files that can be imported with ``neo4j-admin database import``. + + Args: + knowledge_graph: Knowledge graph dictionary or object with graph attributes. + output_dir: Output directory for Neo4j CSV files. + method: Export method (default: "default"). + **kwargs: Additional options passed to ``Neo4jCSVExporter``. + + Returns: + Mapping with ``"nodes"`` and ``"relationships"`` output paths. + """ + custom_method = method_registry.get("neo4j_csv", method) + if custom_method and custom_method is not export_neo4j_csv: + try: + return custom_method(knowledge_graph, output_dir, **kwargs) + except Exception as e: + logger.warning( + f"Custom method {method} failed: {e}, falling back to default" + ) + + try: + config = export_config.get_method_config("neo4j_csv") + + # Split kwargs: constructor-level settings vs per-call export() options. + # Passing all kwargs to both the constructor AND export_knowledge_graph + # causes dialect params like delimiter to reach csv.writer twice. + _init_params = frozenset({ + "node_file_name", + "relationship_file_name", + "encoding", + "delimiter", + "label_separator", + "strict", + "config", + }) + init_kwargs = {k: v for k, v in kwargs.items() if k in _init_params} + call_kwargs = {k: v for k, v in kwargs.items() if k not in _init_params} + + config.update(init_kwargs) + exporter = Neo4jCSVExporter(**config) + return exporter.export_knowledge_graph(knowledge_graph, output_dir, **call_kwargs) + + except Exception as e: + logger.error(f"Failed to export Neo4j CSV: {e}") + raise + + def export_arango( knowledge_graph: Dict[str, Any], file_path: Union[str, Path], @@ -713,12 +774,18 @@ def export_arango( - vertex_collection: Override default vertex collection name - edge_collection: Override default edge collection name - batch_size: Batch size for INSERT operations - - include_collection_creation: Whether to include collection creation statements + - include_collection_creation: Whether to include collection + creation statements Examples: >>> from semantica.export.methods import export_arango >>> export_arango(kg, "output.aql") - >>> export_arango(kg, "graph.aql", vertex_collection="nodes", edge_collection="links") + >>> export_arango( + ... kg, + ... "graph.aql", + ... vertex_collection="nodes", + ... edge_collection="links", + ... ) """ # Check for custom method in registry custom_method = method_registry.get("arango", method) @@ -872,6 +939,10 @@ def export_knowledge_graph( export_owl(knowledge_graph, file_path, format=format, method=method, **kwargs) elif format == "cypher": export_lpg(knowledge_graph, file_path, method=method, **kwargs) + elif format in ["neo4j_csv", "neo4j-csv"]: + # file_path is treated as the output directory; nodes.csv and + # relationships.csv are written inside it. + export_neo4j_csv(knowledge_graph, file_path, method=method, **kwargs) elif format == "aql": export_arango(knowledge_graph, file_path, method=method, **kwargs) elif format == "parquet": @@ -944,6 +1015,9 @@ method_registry.register("vector", "json", export_vector) method_registry.register("vector", "numpy", export_vector) method_registry.register("lpg", "default", export_lpg) method_registry.register("lpg", "cypher", export_lpg) +method_registry.register("neo4j_csv", "default", export_neo4j_csv) +method_registry.register("neo4j_csv", "neo4j_csv", export_neo4j_csv) +method_registry.register("neo4j_csv", "neo4j-csv", export_neo4j_csv) method_registry.register("arango", "default", export_arango) method_registry.register("arango", "aql", export_arango) method_registry.register("report", "default", generate_report) diff --git a/semantica/export/neo4j_csv_exporter.py b/semantica/export/neo4j_csv_exporter.py new file mode 100644 index 00000000..7a3461d1 --- /dev/null +++ b/semantica/export/neo4j_csv_exporter.py @@ -0,0 +1,905 @@ +""" +Neo4j bulk CSV export for ``neo4j-admin database import``. + +This module writes Semantica knowledge graphs to the two CSV files expected by +Neo4j's offline bulk importer: + +``nodes.csv``:: + + :id,:LABEL,name,type + 1,Person;Engineer,Alice,user + +``relationships.csv``:: + + :START_ID,:END_ID,:TYPE,since + 1,2,KNOWS,2024 + +Example import command:: + + neo4j-admin database import full \\ + --nodes=nodes.csv \\ + --relationships=relationships.csv \\ + neo4j + +The exporter accepts the canonical Semantica ``{"entities": ..., "relationships": ...}`` +shape, graph-style ``{"nodes": ..., "edges": ...}`` dictionaries, and objects exposing +``.entities``/``.relationships`` or ``.nodes``/``.edges`` attributes. Existing node IDs +are reused when present; otherwise deterministic IDs are derived from node contents. +""" + +from __future__ import annotations + +import csv +import hashlib +import json +from dataclasses import asdict, is_dataclass +from pathlib import Path +from typing import Any, Dict, Iterable, List, Optional, Sequence, Union + +from ..utils.exceptions import ProcessingError, ValidationError +from ..utils.helpers import ensure_directory +from ..utils.logging import get_logger +from ..utils.progress_tracker import get_progress_tracker + + +class Neo4jCSVExporter: + """ + Export knowledge graphs to Neo4j bulk-import CSV files. + + The exporter creates deterministic ``nodes.csv`` and ``relationships.csv`` files + under the requested output directory. Node property columns and relationship + property columns are generated from the graph data and sorted alphabetically. + + Args: + node_file_name: File name for node rows (default: ``nodes.csv``). + relationship_file_name: File name for relationship rows + (default: ``relationships.csv``). + encoding: Output file encoding (default: ``utf-8``). + delimiter: CSV delimiter (default: ``,``). + label_separator: Neo4j label separator for ``:LABEL`` values + (default: ``;``). + strict: If true, raise on relationships that cannot be connected to an + exported node. If false, unresolved endpoint values are written as-is. + config: Optional configuration dictionary, merged with keyword overrides. + + Example: + >>> from semantica.export import Neo4jCSVExporter + >>> kg = { + ... "entities": [{"id": "1", "labels": ["Person"], "name": "Alice"}], + ... "relationships": [], + ... } + >>> Neo4jCSVExporter().export_knowledge_graph(kg, "neo4j_import") + """ + + NODE_ID_HEADER = ":id" + NODE_LABEL_HEADER = ":LABEL" + REL_START_HEADER = ":START_ID" + REL_END_HEADER = ":END_ID" + REL_TYPE_HEADER = ":TYPE" + + _NODE_ID_KEYS = ("id", "entity_id", "node_id", "_id", "key", "uid") + _REL_ID_KEYS = ("id", "relationship_id", "edge_id", "_id", "key", "uid") + _SOURCE_KEYS = ( + "source", + "source_id", + "from", + "from_id", + "subject", + "start", + "start_id", + ) + _TARGET_KEYS = ( + "target", + "target_id", + "to", + "to_id", + "object", + "end", + "end_id", + ) + _REL_TYPE_KEYS = ( + "type", + "relationship_type", + "relation_type", + "predicate", + "label", + ) + _NODE_LABEL_KEYS = ("labels", "node_labels") + + _NODE_STRUCTURAL_KEYS = set(_NODE_ID_KEYS) | set(_NODE_LABEL_KEYS) + _REL_STRUCTURAL_KEYS = ( + set(_REL_ID_KEYS) | set(_SOURCE_KEYS) | set(_TARGET_KEYS) | set(_REL_TYPE_KEYS) + ) + + def __init__( + self, + node_file_name: str = "nodes.csv", + relationship_file_name: str = "relationships.csv", + encoding: str = "utf-8", + delimiter: str = ",", + label_separator: str = ";", + strict: bool = True, + config: Optional[Dict[str, Any]] = None, + **kwargs: Any, + ) -> None: + self.logger = get_logger("neo4j_csv_exporter") + self.config = config or {} + self.config.update(kwargs) + + self.node_file_name = self.config.get("node_file_name", node_file_name) + self.relationship_file_name = self.config.get( + "relationship_file_name", relationship_file_name + ) + self.encoding = self.config.get("encoding", encoding) + self.delimiter = self.config.get("delimiter", delimiter) + self.label_separator = self.config.get("label_separator", label_separator) + self.strict = bool(self.config.get("strict", strict)) + + self._validate_file_name(self.node_file_name, "node_file_name") + self._validate_file_name(self.relationship_file_name, "relationship_file_name") + if not self.label_separator: + raise ValueError("label_separator cannot be empty") + if not self.delimiter: + raise ValueError("delimiter cannot be empty") + + self.progress_tracker = get_progress_tracker() + if not self.progress_tracker.enabled: + self.progress_tracker.enabled = True + + self.logger.debug( + "Neo4j CSV exporter initialized: nodes=%s, relationships=%s", + self.node_file_name, + self.relationship_file_name, + ) + + def export( + self, + knowledge_graph: Any, + output_dir: Union[str, Path], + validate: bool = False, + **options: Any, + ) -> Dict[str, Path]: + """ + Export a graph to Neo4j bulk CSV files. + + Args: + knowledge_graph: Graph dictionary or object with graph attributes. + output_dir: Directory where ``nodes.csv`` and ``relationships.csv`` + will be written. + validate: If true, validate written files and raise ``ValidationError`` + if they do not meet basic Neo4j import expectations. + **options: Optional overrides for ``node_file_name``, + ``relationship_file_name``, ``strict``, or CSV writer options. + + Returns: + Mapping with ``"nodes"`` and ``"relationships"`` output paths. + """ + output_dir = Path(output_dir) + ensure_directory(output_dir) + + node_file_name = options.pop("node_file_name", self.node_file_name) + relationship_file_name = options.pop( + "relationship_file_name", self.relationship_file_name + ) + strict = bool(options.pop("strict", self.strict)) + self._validate_file_name(node_file_name, "node_file_name") + self._validate_file_name(relationship_file_name, "relationship_file_name") + + nodes_path = output_dir / node_file_name + relationships_path = output_dir / relationship_file_name + + tracking_id = self.progress_tracker.start_tracking( + file=str(output_dir), + module="export", + submodule="Neo4jCSVExporter", + message=f"Exporting graph to Neo4j CSV directory: {output_dir}", + ) + + try: + prepared = self._prepare_export(knowledge_graph, strict=strict) + + self._write_csv( + nodes_path, + [self.NODE_ID_HEADER, self.NODE_LABEL_HEADER] + + prepared["node_columns"], + prepared["node_rows"], + **options, + ) + self._write_csv( + relationships_path, + [ + self.REL_START_HEADER, + self.REL_END_HEADER, + self.REL_TYPE_HEADER, + ] + + prepared["relationship_columns"], + prepared["relationship_rows"], + **options, + ) + + exported = {"nodes": nodes_path, "relationships": relationships_path} + if validate: + validation = self.validate_export( + output_dir, + node_file_name=node_file_name, + relationship_file_name=relationship_file_name, + ) + if not validation["valid"]: + raise ValidationError( + "Neo4j CSV export validation failed", + errors=validation["errors"], + ) + + self.progress_tracker.stop_tracking( + tracking_id, + status="completed", + message=( + f"Exported {len(prepared['node_rows'])} node row(s) and " + f"{len(prepared['relationship_rows'])} relationship row(s)" + ), + ) + self.logger.info("Exported Neo4j CSV files to: %s", output_dir) + return exported + + except Exception as exc: + self.progress_tracker.stop_tracking( + tracking_id, status="failed", message=str(exc) + ) + if isinstance(exc, (ValidationError, ProcessingError)): + raise + raise ProcessingError(f"Failed to export Neo4j CSV files: {exc}") from exc + + def export_knowledge_graph( + self, + knowledge_graph: Any, + output_dir: Union[str, Path], + **options: Any, + ) -> Dict[str, Path]: + """Convenience wrapper around :meth:`export`.""" + return self.export(knowledge_graph, output_dir, **options) + + def export_nodes( + self, + nodes: Sequence[Any], + output_dir: Union[str, Path], + **options: Any, + ) -> Dict[str, Path]: + """Export node rows with an empty relationship file.""" + return self.export( + {"entities": list(nodes), "relationships": []}, output_dir, **options + ) + + def export_relationships( + self, + relationships: Sequence[Any], + output_dir: Union[str, Path], + nodes: Optional[Sequence[Any]] = None, + **options: Any, + ) -> Dict[str, Path]: + """Export relationship rows, optionally with the node rows they reference.""" + return self.export( + {"entities": list(nodes or []), "relationships": list(relationships)}, + output_dir, + **options, + ) + + def dry_run(self, knowledge_graph: Any, **options: Any) -> Dict[str, Any]: + """ + Prepare and validate CSV content without writing files. + + Returns a summary containing headers, row counts, and validation errors. + This is useful in CI or before running ``neo4j-admin database import``. + """ + strict = bool(options.pop("strict", self.strict)) + prepared = self._prepare_export(knowledge_graph, strict=strict) + validation = self._validate_prepared(prepared) + return { + "valid": validation["valid"], + "errors": validation["errors"], + "node_header": [self.NODE_ID_HEADER, self.NODE_LABEL_HEADER] + + prepared["node_columns"], + "relationship_header": [ + self.REL_START_HEADER, + self.REL_END_HEADER, + self.REL_TYPE_HEADER, + ] + + prepared["relationship_columns"], + "node_count": len(prepared["node_rows"]), + "relationship_count": len(prepared["relationship_rows"]), + } + + def validate_export( + self, + output_dir: Union[str, Path], + node_file_name: Optional[str] = None, + relationship_file_name: Optional[str] = None, + ) -> Dict[str, Any]: + """ + Validate written CSV files against core Neo4j import expectations. + + The validation checks required headers, consistent row widths, unique node + IDs, non-empty relationship endpoints/types, and relationship endpoint + references to exported node IDs. + """ + output_dir = Path(output_dir) + nodes_path = output_dir / (node_file_name or self.node_file_name) + relationships_path = output_dir / ( + relationship_file_name or self.relationship_file_name + ) + + errors: List[str] = [] + node_ids: set = set() + + if not nodes_path.exists(): + errors.append(f"Missing node CSV file: {nodes_path}") + else: + try: + with open( + nodes_path, "r", encoding=self.encoding, newline="" + ) as handle: + reader = csv.reader(handle, delimiter=self.delimiter) + rows = list(reader) + if not rows: + errors.append("Node CSV is empty") + else: + header = rows[0] + if len(header) < 2 or header[:2] != [ + self.NODE_ID_HEADER, + self.NODE_LABEL_HEADER, + ]: + errors.append("Node CSV header must start with ':id,:LABEL'") + for row_number, row in enumerate(rows[1:], start=2): + if len(row) != len(header): + errors.append( + f"Node row {row_number} has {len(row)} columns; " + f"expected {len(header)}" + ) + continue + node_id = row[0] + if not node_id: + errors.append(f"Node row {row_number} has an empty :id") + elif node_id in node_ids: + errors.append(f"Duplicate node :id {node_id!r}") + else: + node_ids.add(node_id) + except csv.Error as exc: + errors.append(f"Node CSV parse error: {exc}") + + if not relationships_path.exists(): + errors.append(f"Missing relationship CSV file: {relationships_path}") + else: + try: + with open( + relationships_path, + "r", + encoding=self.encoding, + newline="", + ) as handle: + reader = csv.reader(handle, delimiter=self.delimiter) + rows = list(reader) + if not rows: + errors.append("Relationship CSV is empty") + else: + header = rows[0] + expected = [ + self.REL_START_HEADER, + self.REL_END_HEADER, + self.REL_TYPE_HEADER, + ] + if len(header) < 3 or header[:3] != expected: + errors.append( + "Relationship CSV header must start with " + "':START_ID,:END_ID,:TYPE'" + ) + for row_number, row in enumerate(rows[1:], start=2): + if len(row) != len(header): + errors.append( + f"Relationship row {row_number} has {len(row)} " + f"columns; expected {len(header)}" + ) + continue + start_id, end_id, rel_type = row[:3] + if not start_id or not end_id: + errors.append( + f"Relationship row {row_number} has an empty endpoint" + ) + if not rel_type: + errors.append( + f"Relationship row {row_number} has an empty :TYPE" + ) + if node_ids and start_id and start_id not in node_ids: + errors.append( + f"Relationship row {row_number} references unknown " + f":START_ID {start_id!r}" + ) + if node_ids and end_id and end_id not in node_ids: + errors.append( + f"Relationship row {row_number} references unknown " + f":END_ID {end_id!r}" + ) + except csv.Error as exc: + errors.append(f"Relationship CSV parse error: {exc}") + + return {"valid": not errors, "errors": errors} + + def _prepare_export(self, graph: Any, strict: bool) -> Dict[str, Any]: + normalized = self._normalize_graph(graph) + node_infos = self._prepare_nodes(normalized["nodes"]) + + node_columns = sorted( + {key for node in node_infos for key in node["properties"].keys()} + ) + + alias_lookup = self._build_alias_lookup(node_infos) + node_id_set = {node["id"] for node in node_infos} + + relationship_infos = self._prepare_relationships( + normalized["relationships"], + alias_lookup=alias_lookup, + node_id_set=node_id_set, + strict=strict, + ) + relationship_columns = sorted( + { + key + for relationship in relationship_infos + for key in relationship["properties"].keys() + } + ) + + node_rows = [] + for node in sorted(node_infos, key=lambda item: item["id"]): + node_rows.append( + [node["id"], self.label_separator.join(node["labels"])] + + [ + self._serialize_value(node["properties"].get(column)) + for column in node_columns + ] + ) + + relationship_rows = [] + for relationship in sorted( + relationship_infos, + key=lambda item: ( + item["start_id"], + item["end_id"], + item["type"], + self._canonical_json(item["properties"]), + ), + ): + relationship_rows.append( + [ + relationship["start_id"], + relationship["end_id"], + relationship["type"], + ] + + [ + self._serialize_value(relationship["properties"].get(column)) + for column in relationship_columns + ] + ) + + prepared = { + "node_rows": node_rows, + "relationship_rows": relationship_rows, + "node_columns": node_columns, + "relationship_columns": relationship_columns, + } + validation = self._validate_prepared(prepared) + if not validation["valid"]: + raise ValidationError( + "Neo4j CSV export data is invalid", + errors=validation["errors"], + ) + return prepared + + def _normalize_graph(self, graph: Any) -> Dict[str, List[Dict[str, Any]]]: + if isinstance(graph, dict): + nodes = graph.get("nodes") or graph.get("entities") or [] + relationships = graph.get("edges") or graph.get("relationships") or [] + else: + nodes = getattr(graph, "nodes", None) + if nodes is None: + nodes = getattr(graph, "entities", None) + relationships = getattr(graph, "edges", None) + if relationships is None: + relationships = getattr(graph, "relationships", None) + if nodes is None and relationships is None: + raise ProcessingError( + f"Cannot export object of type '{type(graph).__name__}': " + "expected a dict with 'entities'/'relationships' or " + "'nodes'/'edges', or an object exposing equivalent attributes." + ) + + return { + "nodes": [self._record_to_dict(node) for node in list(nodes or [])], + "relationships": [ + self._record_to_dict(relationship) + for relationship in list(relationships or []) + ], + } + + def _prepare_nodes(self, nodes: Sequence[Dict[str, Any]]) -> List[Dict[str, Any]]: + explicit_ids: Dict[str, int] = {} + node_infos: List[Dict[str, Any]] = [] + + for index, node in enumerate(nodes): + explicit_id = self._first_value(node, self._NODE_ID_KEYS) + has_explicit_id = explicit_id is not None and str(explicit_id) != "" + base_id = ( + str(explicit_id) if has_explicit_id else self._derived_node_id(node) + ) + + if has_explicit_id: + if base_id in explicit_ids: + raise ValidationError( + f"Duplicate node ID {base_id!r}", + first_index=explicit_ids[base_id], + duplicate_index=index, + ) + explicit_ids[base_id] = index + + node_infos.append( + { + "base_id": base_id, + "has_explicit_id": has_explicit_id, + "original_index": index, + "record": node, + "canonical": self._canonical_json(node), + "labels": self._extract_node_labels(node), + "properties": self._extract_properties( + node, structural_keys=self._NODE_STRUCTURAL_KEYS + ), + } + ) + + used_ids = set(explicit_ids.keys()) + generated_counts: Dict[str, int] = {} + + for node in sorted( + node_infos, + key=lambda item: ( + item["base_id"], + item["canonical"], + item["original_index"], + ), + ): + if node["has_explicit_id"]: + node["id"] = node["base_id"] + continue + + base_id = node["base_id"] + generated_counts[base_id] = generated_counts.get(base_id, 0) + 1 + suffix = generated_counts[base_id] + candidate = base_id if suffix == 1 else f"{base_id}_{suffix}" + while candidate in used_ids: + suffix += 1 + generated_counts[base_id] = suffix + candidate = f"{base_id}_{suffix}" + used_ids.add(candidate) + node["id"] = candidate + + return node_infos + + def _prepare_relationships( + self, + relationships: Sequence[Dict[str, Any]], + alias_lookup: Dict[str, str], + node_id_set: set, + strict: bool, + ) -> List[Dict[str, Any]]: + relationship_infos = [] + + for index, relationship in enumerate(relationships): + raw_start = self._first_value(relationship, self._SOURCE_KEYS) + raw_end = self._first_value(relationship, self._TARGET_KEYS) + if raw_start is None or raw_end is None: + raise ValidationError( + f"Relationship {index} is missing source or target", + relationship=relationship, + ) + + start_id = self._resolve_endpoint(raw_start, alias_lookup) + end_id = self._resolve_endpoint(raw_end, alias_lookup) + rel_type = self._extract_relationship_type(relationship) + + unresolved = [ + endpoint + for endpoint in (start_id, end_id) + if endpoint not in node_id_set + ] + if unresolved and strict: + raise ValidationError( + f"Relationship {index} references unknown node ID(s): " + f"{', '.join(repr(endpoint) for endpoint in unresolved)}", + relationship=relationship, + ) + + relationship_infos.append( + { + "start_id": start_id, + "end_id": end_id, + "type": rel_type, + "properties": self._extract_properties( + relationship, structural_keys=self._REL_STRUCTURAL_KEYS + ), + } + ) + + return relationship_infos + + def _build_alias_lookup( + self, node_infos: Sequence[Dict[str, Any]] + ) -> Dict[str, str]: + aliases: Dict[str, str] = {} + ambiguous = set() + + for node in node_infos: + for alias in self._node_aliases(node["record"], node["id"]): + if alias in aliases and aliases[alias] != node["id"]: + ambiguous.add(alias) + else: + aliases[alias] = node["id"] + + for alias in ambiguous: + aliases.pop(alias, None) + + return aliases + + def _node_aliases(self, node: Dict[str, Any], stable_id: str) -> Iterable[str]: + raw_aliases: List[Any] = [stable_id] + raw_aliases.extend(node.get(key) for key in self._NODE_ID_KEYS) + raw_aliases.extend( + node.get(key) + for key in ("name", "text", "label") + if not isinstance(node.get(key), (list, tuple, set, dict)) + ) + + nested_properties = node.get("properties") + if isinstance(nested_properties, dict): + raw_aliases.extend( + nested_properties.get(key) + for key in self._NODE_ID_KEYS + ("name", "text", "label") + ) + + for alias in raw_aliases: + if alias is not None and str(alias) != "": + yield str(alias) + + def _extract_node_labels(self, node: Dict[str, Any]) -> List[str]: + label_values: List[Any] = [] + + for key in self._NODE_LABEL_KEYS: + if key in node: + label_values.extend(self._coerce_sequence(node[key])) + + label_field = node.get("label") + if isinstance(label_field, (list, tuple, set)): + label_values.extend(self._coerce_sequence(label_field)) + + if not label_values: + node_type = node.get("type") or node.get("entity_type") + label_values.extend(self._coerce_sequence(node_type)) + + labels: List[str] = [] + seen = set() + for label in label_values: + token = self._sanitize_token(str(label), fallback="") + if token and token not in seen: + labels.append(token) + seen.add(token) + + return labels or ["Entity"] + + def _extract_relationship_type(self, relationship: Dict[str, Any]) -> str: + rel_type = self._first_value(relationship, self._REL_TYPE_KEYS) + return self._sanitize_token(str(rel_type or ""), fallback="RELATED_TO") + + def _extract_properties( + self, record: Dict[str, Any], structural_keys: set + ) -> Dict[str, Any]: + properties: Dict[str, Any] = {} + + nested_properties = record.get("properties") + if isinstance(nested_properties, dict): + for key, value in nested_properties.items(): + properties[str(key)] = value + + for key, value in record.items(): + if key in structural_keys or key == "properties": + continue + properties[str(key)] = value + + return properties + + def _resolve_endpoint(self, value: Any, alias_lookup: Dict[str, str]) -> str: + if isinstance(value, dict) or is_dataclass(value): + endpoint_record = self._record_to_dict(value) + endpoint_id = self._first_value(endpoint_record, self._NODE_ID_KEYS) + if endpoint_id is None: + endpoint_id = ( + endpoint_record.get("name") + or endpoint_record.get("text") + or endpoint_record.get("label") + ) + if endpoint_id is None: + endpoint_id = self._derived_node_id(endpoint_record) + value = endpoint_id + elif not isinstance(value, str) and hasattr(value, "__dict__"): + return self._resolve_endpoint(self._record_to_dict(value), alias_lookup) + + endpoint = str(value) + return alias_lookup.get(endpoint, endpoint) + + def _first_value(self, record: Dict[str, Any], keys: Sequence[str]) -> Any: + for key in keys: + if key in record and record[key] is not None: + return record[key] + return None + + def _record_to_dict(self, record: Any) -> Dict[str, Any]: + if isinstance(record, dict): + return dict(record) + if is_dataclass(record): + return asdict(record) + if hasattr(record, "__dict__"): + return { + key: value + for key, value in vars(record).items() + if not key.startswith("_") + } + raise ValidationError( + f"Expected graph records to be dictionaries or objects, got " + f"{type(record).__name__}" + ) + + def _derived_node_id(self, node: Dict[str, Any]) -> str: + digest = hashlib.sha256(self._canonical_json(node).encode("utf-8")).hexdigest() + return f"n_{digest[:16]}" + + def _canonical_json(self, value: Any) -> str: + return json.dumps( + self._to_jsonable(value), + sort_keys=True, + ensure_ascii=False, + separators=(",", ":"), + ) + + def _to_jsonable(self, value: Any) -> Any: + if isinstance(value, dict): + return {str(key): self._to_jsonable(val) for key, val in value.items()} + if isinstance(value, (list, tuple)): + return [self._to_jsonable(item) for item in value] + if isinstance(value, set): + return sorted(self._to_jsonable(item) for item in value) + if is_dataclass(value): + return self._to_jsonable(asdict(value)) + if isinstance(value, (str, int, float, bool)) or value is None: + return value + return str(value) + + def _serialize_value(self, value: Any) -> str: + if value is None: + return "" + if isinstance(value, bool): + return "true" if value else "false" + if isinstance(value, (dict, list, tuple, set)) or is_dataclass(value): + return self._canonical_json(value) + return str(value) + + def _coerce_sequence(self, value: Any) -> List[Any]: + if value is None: + return [] + if isinstance(value, str): + if self.label_separator in value: + return [part for part in value.split(self.label_separator) if part] + return [value] + if isinstance(value, set): + return sorted(value) + if isinstance(value, (list, tuple)): + return list(value) + return [value] + + def _sanitize_token(self, value: str, fallback: str) -> str: + token = value.strip() + if not token: + return fallback + + sanitized = [] + for char in token: + if char.isalnum() or char == "_": + sanitized.append(char) + else: + sanitized.append("_") + + result = "".join(sanitized).strip("_") + if not result: + return fallback + if not (result[0].isalpha() or result[0] == "_"): + result = f"_{result}" + return result + + # Valid csv.writer dialect params other than delimiter/lineterminator/quoting + # (those three are always supplied explicitly below). + _CSV_WRITER_PARAMS = frozenset( + {"quotechar", "doublequote", "skipinitialspace", "escapechar", "strict"} + ) + + def _write_csv( + self, + file_path: Path, + header: List[str], + rows: List[List[str]], + **options: Any, + ) -> None: + lineterminator = options.get("lineterminator", "\n") + quoting = options.get("quoting", csv.QUOTE_MINIMAL) + # Only forward recognised csv.writer dialect params so that arbitrary + # caller kwargs (e.g. encoding, validate, node_file_name) do not + # trigger TypeError: csv.writer() got unexpected keyword argument. + writer_options = { + key: value + for key, value in options.items() + if key in self._CSV_WRITER_PARAMS + } + + with open(file_path, "w", encoding=self.encoding, newline="") as handle: + writer = csv.writer( + handle, + delimiter=self.delimiter, + lineterminator=lineterminator, + quoting=quoting, + **writer_options, + ) + writer.writerow(header) + writer.writerows(rows) + + def _validate_prepared(self, prepared: Dict[str, Any]) -> Dict[str, Any]: + errors: List[str] = [] + node_ids = [row[0] for row in prepared["node_rows"]] + node_id_set = set(node_ids) + + if len(node_ids) != len(node_id_set): + errors.append("Node rows contain duplicate :id values") + + for row_number, row in enumerate(prepared["node_rows"], start=2): + expected = 2 + len(prepared["node_columns"]) + if len(row) != expected: + errors.append( + f"Node row {row_number} has {len(row)} columns; expected {expected}" + ) + if not row[0]: + errors.append(f"Node row {row_number} has an empty :id") + + for row_number, row in enumerate(prepared["relationship_rows"], start=2): + expected = 3 + len(prepared["relationship_columns"]) + if len(row) != expected: + errors.append( + f"Relationship row {row_number} has {len(row)} columns; " + f"expected {expected}" + ) + continue + start_id, end_id, rel_type = row[:3] + if not start_id or not end_id: + errors.append(f"Relationship row {row_number} has an empty endpoint") + if not rel_type: + errors.append(f"Relationship row {row_number} has an empty :TYPE") + if node_id_set and start_id and start_id not in node_id_set: + errors.append( + f"Relationship row {row_number} references unknown :START_ID " + f"{start_id!r}" + ) + if node_id_set and end_id and end_id not in node_id_set: + errors.append( + f"Relationship row {row_number} references unknown :END_ID " + f"{end_id!r}" + ) + + return {"valid": not errors, "errors": errors} + + def _validate_file_name(self, file_name: str, param_name: str) -> None: + if not file_name: + raise ValueError(f"{param_name} cannot be empty") + path = Path(file_name) + if path.name != file_name: + raise ValueError(f"{param_name} must be a file name, not a path") diff --git a/semantica/export/registry.py b/semantica/export/registry.py index 3fceb9cd..b7428c04 100644 --- a/semantica/export/registry.py +++ b/semantica/export/registry.py @@ -9,6 +9,7 @@ Supported Registration Types: * "rdf": RDF export methods * "json": JSON/JSON-LD export methods * "csv": CSV export methods + * "neo4j_csv": Neo4j Bulk CSV export methods * "graph": Graph format export methods * "yaml": YAML export methods * "owl": OWL export methods @@ -26,7 +27,8 @@ Algorithms Used: Key Features: - Method registry for custom export methods - - Task-based method organization (rdf, json, csv, graph, yaml, owl, vector, lpg, report, export) + - Task-based method organization (rdf, json, csv, neo4j_csv, graph, + yaml, owl, vector, lpg, report, export) - Dynamic registration and unregistration - Easy discovery of available methods - Support for community-contributed extensions @@ -43,7 +45,7 @@ Example Usage: >>> available = method_registry.list_all("json") """ -from typing import Any, Callable, Dict, List, Optional +from typing import Callable, Dict, List, Optional class MethodRegistry: @@ -53,6 +55,7 @@ class MethodRegistry: "rdf": {}, "json": {}, "csv": {}, + "neo4j_csv": {}, "graph": {}, "yaml": {}, "owl": {}, @@ -68,7 +71,8 @@ class MethodRegistry: Register a custom export method. Args: - task: Task type ("rdf", "json", "csv", "graph", "yaml", "owl", "vector", "lpg", "report", "export") + task: Task type ("rdf", "json", "csv", "neo4j_csv", "graph", + "yaml", "owl", "vector", "lpg", "report", "export") name: Method name method_func: Method function """ @@ -82,7 +86,8 @@ class MethodRegistry: Get method by task and name. Args: - task: Task type ("rdf", "json", "csv", "graph", "yaml", "owl", "vector", "lpg", "report", "export") + task: Task type ("rdf", "json", "csv", "neo4j_csv", "graph", + "yaml", "owl", "vector", "lpg", "report", "export") name: Method name Returns: @@ -111,7 +116,8 @@ class MethodRegistry: Unregister a method. Args: - task: Task type ("rdf", "json", "csv", "graph", "yaml", "owl", "vector", "lpg", "report", "export") + task: Task type ("rdf", "json", "csv", "neo4j_csv", "graph", + "yaml", "owl", "vector", "lpg", "report", "export") name: Method name """ if task in cls._methods and name in cls._methods[task]: diff --git a/tests/export/test_neo4j_csv_exporter.py b/tests/export/test_neo4j_csv_exporter.py new file mode 100644 index 00000000..529f7bd4 --- /dev/null +++ b/tests/export/test_neo4j_csv_exporter.py @@ -0,0 +1,303 @@ +"""Tests for Neo4j bulk CSV export.""" + +import csv +from pathlib import Path + +import pytest + +from semantica.export import Neo4jCSVExporter +from semantica.export.methods import export_knowledge_graph, export_neo4j_csv +from semantica.kg.knowledge_graph import KnowledgeGraph +from semantica.utils.exceptions import ValidationError + + +def _read_csv(path: Path): + with open(path, "r", encoding="utf-8", newline="") as handle: + return list(csv.reader(handle)) + + +def _sample_graph(): + return { + "entities": [ + { + "id": "2", + "labels": ["Person", "Engineer"], + "name": "Alice", + "type": "user", + "properties": { + "age": 30, + "email": "alice@example.com", + }, + }, + { + "id": "1", + "type": "Company", + "name": "Acme, Inc.", + "properties": {"founded": 2024}, + }, + { + "text": "No ID concept", + "type": "Concept", + "properties": { + "empty": "", + "quote": 'He said "hello"', + "unicode": "東京", + }, + }, + ], + "relationships": [ + { + "source": "2", + "target": "1", + "type": "WORKS_FOR", + "properties": {"role": "Engineer", "since": 2024}, + }, + { + "source": "Alice", + "target": "Acme, Inc.", + "relationship_type": "KNOWS", + "properties": {"note": 'line1\nline2, "quoted"'}, + }, + ], + } + + +def test_header_correctness_and_node_csv_structure(tmp_path): + exporter = Neo4jCSVExporter() + exporter.export_knowledge_graph(_sample_graph(), tmp_path) + + rows = _read_csv(tmp_path / "nodes.csv") + + assert rows[0] == [ + ":id", + ":LABEL", + "age", + "email", + "empty", + "founded", + "name", + "quote", + "text", + "type", + "unicode", + ] + + by_id = {row[0]: row for row in rows[1:]} + assert list(by_id) == sorted(by_id) + assert by_id["2"][1] == "Person;Engineer" + assert by_id["2"][2] == "30" + assert by_id["2"][6] == "Alice" + assert by_id["2"][9] == "user" + assert by_id["1"][1] == "Company" + assert by_id["1"][2] == "" + assert by_id["1"][5] == "2024" + + derived_ids = [node_id for node_id in by_id if node_id.startswith("n_")] + assert len(derived_ids) == 1 + derived_row = by_id[derived_ids[0]] + assert derived_row[4] == "" + assert derived_row[7] == 'He said "hello"' + assert derived_row[8] == "No ID concept" + assert derived_row[10] == "東京" + + +def test_relationship_csv_structure_and_properties(tmp_path): + exporter = Neo4jCSVExporter() + exporter.export(_sample_graph(), tmp_path) + + rows = _read_csv(tmp_path / "relationships.csv") + + assert rows[0] == [":START_ID", ":END_ID", ":TYPE", "note", "role", "since"] + assert rows[1] == ["2", "1", "KNOWS", 'line1\nline2, "quoted"', "", ""] + assert rows[2] == ["2", "1", "WORKS_FOR", "", "Engineer", "2024"] + + +def test_missing_properties_create_empty_cells(tmp_path): + exporter = Neo4jCSVExporter() + exporter.export( + { + "nodes": [ + {"id": "a", "type": "Thing", "properties": {"optional": "present"}}, + {"id": "b", "type": "Thing"}, + ], + "edges": [{"source": "a", "target": "b", "type": "RELATED"}], + }, + tmp_path, + ) + + rows = _read_csv(tmp_path / "nodes.csv") + optional_index = rows[0].index("optional") + by_id = {row[0]: row for row in rows[1:]} + + assert by_id["a"][optional_index] == "present" + assert by_id["b"][optional_index] == "" + + +def test_deterministic_output_with_permuted_graph(tmp_path): + graph_a = { + "entities": [ + {"name": "Bob", "type": "Person"}, + {"name": "Alice", "type": "Person"}, + ], + "relationships": [ + {"source": "Alice", "target": "Bob", "type": "KNOWS"}, + ], + } + graph_b = { + "entities": list(reversed(graph_a["entities"])), + "relationships": list(reversed(graph_a["relationships"])), + } + + exporter = Neo4jCSVExporter() + out_a = tmp_path / "a" + out_b = tmp_path / "b" + + exporter.export(graph_a, out_a) + exporter.export(graph_b, out_b) + + assert (out_a / "nodes.csv").read_text(encoding="utf-8") == ( + out_b / "nodes.csv" + ).read_text(encoding="utf-8") + assert (out_a / "relationships.csv").read_text(encoding="utf-8") == ( + out_b / "relationships.csv" + ).read_text(encoding="utf-8") + + +def test_csv_quoting_escaping_and_unicode(tmp_path): + exporter = Neo4jCSVExporter() + exporter.export(_sample_graph(), tmp_path) + + nodes_text = (tmp_path / "nodes.csv").read_text(encoding="utf-8") + relationships_text = (tmp_path / "relationships.csv").read_text(encoding="utf-8") + + assert '"Acme, Inc."' in nodes_text + assert '"He said ""hello"""' in nodes_text + assert "東京" in nodes_text + assert '"line1\nline2, ""quoted"""' in relationships_text + + assert _read_csv(tmp_path / "nodes.csv") + assert _read_csv(tmp_path / "relationships.csv") + + +def test_empty_graph_export_writes_importable_headers(tmp_path): + exporter = Neo4jCSVExporter() + exporter.export({"entities": [], "relationships": []}, tmp_path) + + assert _read_csv(tmp_path / "nodes.csv") == [[":id", ":LABEL"]] + assert _read_csv(tmp_path / "relationships.csv") == [ + [":START_ID", ":END_ID", ":TYPE"] + ] + + validation = exporter.validate_export(tmp_path) + assert validation == {"valid": True, "errors": []} + + +def test_dry_run_and_written_validation(tmp_path): + exporter = Neo4jCSVExporter() + summary = exporter.dry_run(_sample_graph()) + + assert summary["valid"] is True + assert summary["node_count"] == 3 + assert summary["relationship_count"] == 2 + assert summary["node_header"][:2] == [":id", ":LABEL"] + assert summary["relationship_header"][:3] == [ + ":START_ID", + ":END_ID", + ":TYPE", + ] + + exporter.export(_sample_graph(), tmp_path, validate=True) + assert exporter.validate_export(tmp_path)["valid"] is True + + +def test_knowledge_graph_dataclass_and_convenience_wrappers(tmp_path): + kg = KnowledgeGraph( + entities=[ + {"id": "a", "type": "Person", "name": "Alice"}, + {"id": "b", "type": "Person", "name": "Bob"}, + ], + relationships=[{"source": "a", "target": "b", "type": "KNOWS"}], + ) + + result = export_neo4j_csv(kg, tmp_path / "direct") + assert result["nodes"].name == "nodes.csv" + assert result["relationships"].name == "relationships.csv" + assert result["nodes"].exists() + + export_knowledge_graph(kg, tmp_path / "unified", format="neo4j_csv") + assert (tmp_path / "unified" / "nodes.csv").exists() + assert (tmp_path / "unified" / "relationships.csv").exists() + + +def test_invalid_relationship_endpoint_fails_strict_validation(tmp_path): + exporter = Neo4jCSVExporter() + + with pytest.raises(ValidationError): + exporter.export( + { + "entities": [{"id": "a", "type": "Thing"}], + "relationships": [ + {"source": "a", "target": "missing", "type": "RELATED"} + ], + }, + tmp_path, + ) + + +def test_ambiguous_aliases_are_not_resolved(tmp_path): + exporter = Neo4jCSVExporter() + + # Create two nodes with the same alias/name "Alice" + graph = { + "entities": [ + {"id": "node1", "name": "Alice", "type": "Person"}, + {"id": "node2", "name": "Alice", "type": "Person"}, + ], + "relationships": [{"source": "Alice", "target": "node1", "type": "KNOWS"}], + } + + # Verify that strict export raises ValidationError because the alias is ambiguous + with pytest.raises(ValidationError): + exporter.export(graph, tmp_path) + + +def test_duplicate_node_ids_fail_validation(tmp_path): + exporter = Neo4jCSVExporter() + + # Create two nodes with the same explicit ID + graph = { + "entities": [ + {"id": "node1", "name": "Alice", "type": "Person"}, + {"id": "node1", "name": "Bob", "type": "Person"}, + ], + "relationships": [], + } + + # Verify duplicate explicit node IDs raise ValidationError + with pytest.raises(ValidationError): + exporter.export(graph, tmp_path) + + +def test_nested_properties_are_json_serialized(tmp_path): + exporter = Neo4jCSVExporter() + + graph = { + "entities": [ + { + "id": "node1", + "type": "Person", + "properties": {"nested_dict": {"k": "v"}, "nested_list": [1, 2, 3]}, + } + ], + "relationships": [], + } + + exporter.export(graph, tmp_path) + + rows = _read_csv(tmp_path / "nodes.csv") + assert rows[0] == [":id", ":LABEL", "nested_dict", "nested_list", "type"] + + # Verify that the nested properties are serialized as deterministic JSON strings + by_id = {row[0]: row for row in rows[1:]} + assert by_id["node1"][2] == '{"k":"v"}' + assert by_id["node1"][3] == "[1,2,3]"