From 98232749fb5720dea1d61f05dd9aa9f8f8c3234e Mon Sep 17 00:00:00 2001 From: Luffy2208 Date: Tue, 19 May 2026 17:48:32 +0530 Subject: [PATCH] Add XML file ingestion support (#560) * Add XML file ingestion support * fix(xml-ingestor): add ingest_string test and document ingest() return keys - Add test_xml_ingestor_ingests_string to cover the public ingest_string() method which had no test coverage - Document all source_type return keys in the ingest() docstring so callers know to use result["xml"] rather than result["data"] for XML sources * docs(changelog): add unreleased entry for XML ingestion support (#560) --------- Co-authored-by: KaifAhmad1 --- CHANGELOG.md | 16 + docs/reference/ingest.md | 48 +- semantica/ingest/__init__.py | 10 + semantica/ingest/ingest_usage.md | 104 +++- semantica/ingest/methods.py | 111 +++- semantica/ingest/registry.py | 5 +- semantica/ingest/xml_ingestor.py | 814 ++++++++++++++++++++++++++++++ tests/ingest/test_xml_ingestor.py | 199 ++++++++ 8 files changed, 1287 insertions(+), 20 deletions(-) create mode 100644 semantica/ingest/xml_ingestor.py create mode 100644 tests/ingest/test_xml_ingestor.py diff --git a/CHANGELOG.md b/CHANGELOG.md index 1b83ba77..1df79221 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,22 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +### Added + +- **XML File Ingestion Support** (#560) by @Luffy2208 + - Added `XMLIngestor` class with `lxml` backend for parsing local XML files + - Nested element hierarchy and flat element list extraction + - Namespace and prefix extraction with collision handling + - Attribute and element metadata extraction + - Optional XSD schema validation with detailed error reporting + - Optional DTD validation (internal and external) + - Secure-by-default parser (`resolve_entities=False`, `no_network=True`) blocking XXE attacks + - `ingest_xml()` convenience function and `ingest_file(..., method="xml")` support + - Unified `.xml` auto-detection via `ingest("file.xml")` + - Directory ingestion with recursive scanning and `fail_fast` support + - `ingest_string()` for in-memory XML bytes/str ingestion + - Comprehensive test coverage (8/8 tests passing) + ### Fixed - **NERExtractor LLM method returning pattern-based output on custom gateways** (#554, PR #556) by @KaifAhmad1 diff --git a/docs/reference/ingest.md b/docs/reference/ingest.md index 55e8a5e1..57be5a0b 100644 --- a/docs/reference/ingest.md +++ b/docs/reference/ingest.md @@ -1,6 +1,6 @@ # Ingest -> **Universal data ingestion from files, web, feeds, streams, repos, emails, and databases.** +> **Universal data ingestion from files, XML, web, feeds, streams, repos, emails, and databases.** --- @@ -13,6 +13,7 @@ The **Ingest Module** is the entry point for loading data into Semantica. It pro **Data ingestion** is the process of loading data from various sources into Semantica for processing. The ingest module handles: - **File Systems**: Local files, cloud storage (S3, GCS, Azure) - **Analytics Files**: Apache Parquet files and partitioned datasets +- **Structured Files**: XML files with namespaces, attributes, XSD, and DTD validation - **Web Content**: Websites, RSS feeds, APIs - **Streams**: Real-time data from Kafka, RabbitMQ, etc. - **Databases**: SQL, NoSQL, and cloud data warehouses including Snowflake @@ -81,6 +82,12 @@ The **Ingest Module** is the entry point for loading data into Semantica. It pro Read Parquet files, schemas, metadata, and Hive-style partitioned directories +- :material-code-tags:{ .lg .middle } **XML Files** + + --- + + Parse XML into structured trees with namespaces, attributes, metadata, XSD, and DTD validation + !!! tip "When to Use" @@ -138,6 +145,19 @@ Handles Apache Parquet files and partitioned datasets. | `extract_schema(path)` | Extract column names, types, nullability, and schema metadata | | `extract_metadata(path)` | Extract row counts, row groups, compression, and partition info | +### XMLIngestor + +Handles XML files and directories. + +**Methods:** + +| Method | Description | +|--------|-------------| +| `ingest_file(path)` | Parse a single XML file into nested and flat structures | +| `ingest_directory(path)` | Parse XML files from a folder | +| `validate_file(path, schema_path=None, validate_dtd=False)` | Return XSD/DTD validation results | +| `extract_metadata(path)` | Extract root, namespace, element, attribute, and DTD metadata | + ### WebIngestor Handles web content. @@ -234,6 +254,7 @@ from semantica.ingest import ingest # Auto-detect source type ingest("doc.pdf", source_type="file") ingest("events.parquet") # Auto-detects Parquet +ingest("catalog.xml") # Auto-detects XML ingest("https://google.com", source_type="web") ingest("kafka://topic", source_type="stream") ``` @@ -260,6 +281,31 @@ metadata = ingestor.extract_metadata("events.parquet") partitioned = ingest_parquet("./warehouse/events", method="directory") ``` +### XML File Ingestion + +```python +from semantica.ingest import XMLIngestor, ingest_xml + +ingestor = XMLIngestor() + +# Parse a local XML file into a nested tree and flat element list +xml_data = ingestor.ingest_file("catalog.xml") + +print(xml_data.root_tag) +print(xml_data.namespaces) +print(xml_data.elements[0]["attributes"]) + +# Validate with an XSD schema +validated = ingest_xml("catalog.xml", schema_path="catalog.xsd") +print(validated.validation["schema"]["valid"]) + +# Validate an internal or local DTD declaration +dtd_report = ingestor.validate_file("catalog.xml", validate_dtd=True) + +# Extract metadata without returning the full tree +metadata = ingest_xml("catalog.xml", method="metadata") +``` + --- ## Configuration diff --git a/semantica/ingest/__init__.py b/semantica/ingest/__init__.py index 1eb910e2..54238342 100644 --- a/semantica/ingest/__init__.py +++ b/semantica/ingest/__init__.py @@ -94,6 +94,7 @@ Main Classes: - DBIngestor: Database export handling - OntologyIngestor: Ontology file processing - ParquetIngestor: Apache Parquet file and partitioned dataset processing + - XMLIngestor: XML file parsing, validation, and metadata extraction - MethodRegistry: Registry for custom ingestion methods - IngestConfig: Configuration manager for ingest module @@ -108,6 +109,7 @@ Convenience Functions: - ingest_database: Database ingestion wrapper - ingest_ontology: Ontology ingestion wrapper - ingest_parquet: Parquet ingestion wrapper + - ingest_xml: XML ingestion wrapper Example Usage: @@ -145,6 +147,7 @@ from .methods import ( ingest_repository, ingest_stream, ingest_web, + ingest_xml, list_available_methods, ) from .registry import MethodRegistry, method_registry @@ -204,6 +207,9 @@ _LAZY_EXPORTS: Dict[str, Tuple[str, str]] = { # Parquet ingestion "ParquetIngestor": (".parquet_ingestor", "ParquetIngestor"), "ParquetData": (".parquet_ingestor", "ParquetData"), + # XML ingestion + "XMLIngestor": (".xml_ingestor", "XMLIngestor"), + "XMLIngestionData": (".xml_ingestor", "XMLIngestionData"), } _OPTIONAL_DEPENDENCY_MESSAGES = { @@ -310,6 +316,9 @@ __all__ = [ # Parquet ingestion "ParquetIngestor", "ParquetData", + # XML ingestion + "XMLIngestor", + "XMLIngestionData", # Registry and Methods "MethodRegistry", "method_registry", @@ -323,6 +332,7 @@ __all__ = [ "ingest_database", "ingest_ontology", "ingest_parquet", + "ingest_xml", "ingest_mcp", "get_ingest_method", "list_available_methods", diff --git a/semantica/ingest/ingest_usage.md b/semantica/ingest/ingest_usage.md index b1877f8b..9fa60c38 100644 --- a/semantica/ingest/ingest_usage.md +++ b/semantica/ingest/ingest_usage.md @@ -1,24 +1,25 @@ # Ingest Module Usage Guide -This guide demonstrates how to use the ingest module for ingesting data from various sources including files, web content, feeds, streams, repositories, emails, and databases. +This guide demonstrates how to use the ingest module for ingesting data from various sources including files, XML, web content, feeds, streams, repositories, emails, and databases. ## Table of Contents 1. [Basic Usage](#basic-usage) 2. [File Ingestion](#file-ingestion) 3. [Parquet Ingestion](#parquet-ingestion) -4. [Web Ingestion](#web-ingestion) -5. [Feed Ingestion](#feed-ingestion) -6. [Stream Ingestion](#stream-ingestion) -7. [Repository Ingestion](#repository-ingestion) -8. [Email Ingestion](#email-ingestion) -9. [Database Ingestion](#database-ingestion) -10. [MCP Server Ingestion](#mcp-server-ingestion) -11. [Unified Ingestion](#unified-ingestion) -12. [Using Methods](#using-methods) -13. [Using Registry](#using-registry) -14. [Configuration](#configuration) -15. [Advanced Examples](#advanced-examples) +4. [XML Ingestion](#xml-ingestion) +5. [Web Ingestion](#web-ingestion) +6. [Feed Ingestion](#feed-ingestion) +7. [Stream Ingestion](#stream-ingestion) +8. [Repository Ingestion](#repository-ingestion) +9. [Email Ingestion](#email-ingestion) +10. [Database Ingestion](#database-ingestion) +11. [MCP Server Ingestion](#mcp-server-ingestion) +12. [Unified Ingestion](#unified-ingestion) +13. [Using Methods](#using-methods) +14. [Using Registry](#using-registry) +15. [Configuration](#configuration) +16. [Advanced Examples](#advanced-examples) ## Basic Usage @@ -33,6 +34,9 @@ result = ingest("document.pdf", source_type="file") # Ingest a Parquet file result = ingest("events.parquet") +# Ingest an XML file +result = ingest("catalog.xml") + # Ingest from web URL result = ingest("https://example.com", source_type="web") @@ -43,17 +47,21 @@ result = ingest("https://example.com/feed.xml", source_type="feed") ### Using Main Classes ```python -from semantica.ingest import FileIngestor, WebIngestor +from semantica.ingest import FileIngestor, WebIngestor, XMLIngestor # Create ingestor file_ingestor = FileIngestor() web_ingestor = WebIngestor(delay=1.0, respect_robots=True) +xml_ingestor = XMLIngestor() # Ingest files files = file_ingestor.ingest_directory("./documents", recursive=True) # Ingest web content content = web_ingestor.ingest_url("https://example.com") + +# Ingest XML content +xml_data = xml_ingestor.ingest_file("catalog.xml") ``` ## File Ingestion @@ -206,6 +214,66 @@ print(data.metadata["partition_columns"]) print(data.metadata["partition_values"]) ``` +## XML Ingestion + +### Single XML File + +```python +from semantica.ingest import XMLIngestor, ingest_xml + +# Using convenience function +xml_data = ingest_xml("catalog.xml") + +# Using class directly +ingestor = XMLIngestor() +xml_data = ingestor.ingest_file("catalog.xml") + +print(f"Root: {xml_data.root_tag}") +print(f"Namespaces: {xml_data.namespaces}") +print(f"Elements: {xml_data.metadata['element_count']}") +``` + +### XML Namespaces and Attributes + +```python +from semantica.ingest import XMLIngestor + +data = XMLIngestor().ingest_file("catalog.xml") + +for element in data.elements: + print(element["path"], element["tag"], element["attributes"]) +``` + +### XSD and DTD Validation + +```python +from semantica.ingest import XMLIngestor, ingest_xml + +# XSD validation runs automatically when schema_path is provided +validated = ingest_xml("catalog.xml", schema_path="catalog.xsd") +print(validated.validation["schema"]["valid"]) + +# Return a report without raising on validation errors +report = XMLIngestor().validate_file( + "catalog.xml", + schema_path="catalog.xsd", + validate_dtd=True, +) +print(report["is_valid"]) +``` + +### XML Metadata + +```python +from semantica.ingest import ingest_xml + +metadata = ingest_xml("catalog.xml", method="metadata") + +print(metadata["root_tag"]) +print(metadata["namespace_count"]) +print(metadata["tag_counts"]) +``` + ## Web Ingestion ### Single URL Ingestion @@ -1001,6 +1069,7 @@ from semantica.ingest import ingest # Auto-detect source type from source result = ingest("document.pdf") # Auto-detects file result = ingest("events.parquet") # Auto-detects Parquet +result = ingest("catalog.xml") # Auto-detects XML result = ingest("https://example.com") # Auto-detects web result = ingest("https://example.com/feed.xml") # Auto-detects feed result = ingest("postgresql://user:pass@localhost/db") # Auto-detects database @@ -1014,6 +1083,7 @@ from semantica.ingest import ingest # Explicit source type result = ingest("document.pdf", source_type="file") +result = ingest("catalog.xml", source_type="xml") result = ingest("https://example.com", source_type="web") result = ingest("https://example.com/feed.xml", source_type="feed") ``` @@ -1048,7 +1118,8 @@ from semantica.ingest.methods import ( ingest_email, ingest_database, ingest_mcp, - ingest_parquet + ingest_parquet, + ingest_xml, ) # File ingestion @@ -1075,6 +1146,9 @@ data = ingest_database("postgresql://user:pass@localhost/db", table="users") # Parquet ingestion events = ingest_parquet("events.parquet", columns=["event_id"], limit=1000) +# XML ingestion +xml_data = ingest_xml("catalog.xml", schema_path="catalog.xsd") + # MCP server ingestion via URL data = ingest_mcp("http://localhost:8000/mcp", method="resources") ``` diff --git a/semantica/ingest/methods.py b/semantica/ingest/methods.py index daf73268..9d56a41b 100644 --- a/semantica/ingest/methods.py +++ b/semantica/ingest/methods.py @@ -19,6 +19,12 @@ Parquet Ingestion: - "schema": Parquet schema extraction - "metadata": Parquet file or directory metadata extraction +XML Ingestion: + - "file": Single XML file ingestion with structured parsing + - "directory": Directory ingestion for XML files + - "metadata": XML structure and namespace metadata extraction + - "validate": XSD and DTD validation report + Web Ingestion: - "url": Single URL ingestion - "sitemap": Sitemap-based crawling @@ -138,6 +144,7 @@ Main Functions: - ingest_email: Email ingestion wrapper - ingest_database: Database ingestion wrapper - ingest_parquet: Parquet ingestion wrapper + - ingest_xml: XML ingestion wrapper - ingest: Unified ingestion function with source type dispatch - get_ingest_method: Get ingestion method by name - list_available_methods: List registered methods @@ -172,6 +179,7 @@ if TYPE_CHECKING: from .parquet_ingestor import ParquetData from .stream_ingestor import StreamProcessor from .web_ingestor import WebContent + from .xml_ingestor import XMLIngestionData logger = get_logger("ingest_methods") @@ -333,6 +341,77 @@ def ingest_parquet( raise +def ingest_xml( + source: Union[str, Path, List[Union[str, Path]]], + method: str = "file", + **kwargs, +) -> Union[ + XMLIngestionData, + List[XMLIngestionData], + Dict[str, Any], + List[Dict[str, Any]], +]: + """ + Ingest XML files from source (convenience function). + + Args: + source: XML file path, directory path, or list of XML file paths + method: Ingestion method: + - "file": Single XML file ingestion + - "directory": Directory ingestion with recursive scanning + - "metadata": Extract XML metadata without returning the full tree + - "validate": Return XSD/DTD validation report + **kwargs: Additional options passed to XMLIngestor + + Returns: + XMLIngestionData, list of XMLIngestionData, metadata dict, or validation dict + + Examples: + >>> from semantica.ingest.methods import ingest_xml + >>> data = ingest_xml("catalog.xml") + >>> report = ingest_xml( + ... "catalog.xml", method="validate", schema_path="catalog.xsd" + ... ) + """ + custom_method = method_registry.get("xml", method) + if custom_method and custom_method != ingest_xml: + try: + return custom_method(source, **kwargs) + except Exception as e: + logger.warning( + f"Custom method {method} failed: {e}, falling back to default" + ) + + try: + from .xml_ingestor import XMLIngestor + + config = ingest_config.get_method_config("xml") + config.update(kwargs) + ingestor = XMLIngestor(**config) + + def _run_single( + path: Union[str, Path], + ) -> Union[XMLIngestionData, Dict[str, Any]]: + if method == "metadata": + return ingestor.extract_metadata(path, **kwargs) + if method in {"validate", "validation"}: + return ingestor.validate_file(path, **kwargs) + return ingestor.ingest_file(path, **kwargs) + + if isinstance(source, list): + return [_run_single(path) for path in source] + + source_path = Path(source) + if method == "directory" or source_path.is_dir(): + return ingestor.ingest_directory(source_path, **kwargs) + + return _run_single(source_path) + + except Exception as e: + logger.error(f"Failed to ingest XML: {e}") + raise + + def ingest_web( source: Union[str, List[str]], method: str = "url", **kwargs ) -> Union[WebContent, List[WebContent], Dict[str, Any]]: @@ -1013,11 +1092,19 @@ def ingest( - "db": Database ingestion - "ontology": Ontology ingestion - "parquet": Apache Parquet file or directory ingestion + - "xml": XML file or directory ingestion method: Optional specific ingestion method **kwargs: Additional options passed to ingestor Returns: - Dict with ingestion results + Dict with ingestion results. The top-level key depends on source_type: + - "files": file ingestion + - "content": web ingestion + - "feeds": feed ingestion + - "emails": email ingestion + - "data": database, parquet, or MCP ingestion + - "ontology": ontology ingestion + - "xml": XML file or directory ingestion (use ``result["xml"]``) Examples: >>> from semantica.ingest.methods import ingest @@ -1027,6 +1114,9 @@ def ingest( >>> result = ingest("https://example.com", source_type="web") >>> # Auto-detect from source >>> result = ingest("https://example.com/feed.xml") # Auto-detects feed + >>> # XML ingestion — access via result["xml"] + >>> result = ingest("catalog.xml") + >>> xml_data = result["xml"] """ # Auto-detect source type if not specified if not source_type: @@ -1056,6 +1146,8 @@ def ingest( source_type = "ontology" elif source_str_lower.endswith((".parquet", ".pq")): source_type = "parquet" + elif source_str_lower.endswith(".xml"): + source_type = "xml" else: source_type = "file" elif ( @@ -1066,6 +1158,12 @@ def ingest( ) ): source_type = "parquet" + elif ( + isinstance(sources, list) + and sources + and all(str(source).lower().endswith(".xml") for source in sources) + ): + source_type = "xml" else: source_type = "file" @@ -1094,6 +1192,8 @@ def ingest( return {"data": ingest_database(sources, method=method, **kwargs)} elif source_type == "parquet": return {"data": ingest_parquet(sources, method=method or "file", **kwargs)} + elif source_type == "xml": + return {"xml": ingest_xml(sources, method=method or "file", **kwargs)} elif source_type == "ontology": return {"ontology": ingest_ontology(sources, method=method or "file", **kwargs)} elif source_type == "mcp": @@ -1108,7 +1208,7 @@ def get_ingest_method(task: str, name: str) -> Optional[Callable]: Args: task: Task type ("file", "web", "feed", "stream", "repo", "email", - "db", "mcp", "ingest") + "db", "mcp", "parquet", "xml", "ingest") name: Method name Returns: @@ -1178,6 +1278,13 @@ method_registry.register("parquet", "directory", ingest_parquet) method_registry.register("parquet", "schema", ingest_parquet) method_registry.register("parquet", "metadata", ingest_parquet) method_registry.register("file", "parquet", ingest_parquet) +method_registry.register("xml", "default", ingest_xml) +method_registry.register("xml", "file", ingest_xml) +method_registry.register("xml", "directory", ingest_xml) +method_registry.register("xml", "metadata", ingest_xml) +method_registry.register("xml", "validate", ingest_xml) +method_registry.register("xml", "validation", ingest_xml) +method_registry.register("file", "xml", ingest_xml) method_registry.register("mcp", "default", ingest_mcp) method_registry.register("mcp", "resources", ingest_mcp) method_registry.register("mcp", "tools", ingest_mcp) diff --git a/semantica/ingest/registry.py b/semantica/ingest/registry.py index 1d82f9ca..f24dec90 100644 --- a/semantica/ingest/registry.py +++ b/semantica/ingest/registry.py @@ -60,6 +60,7 @@ class MethodRegistry: "db": {}, "mcp": {}, "parquet": {}, + "xml": {}, "ingest": {}, } @@ -70,7 +71,7 @@ class MethodRegistry: Args: task: Task type such as "file", "web", "feed", "stream", - "repo", "email", "db", "mcp", "parquet", or "ingest" + "repo", "email", "db", "mcp", "parquet", "xml", or "ingest" name: Method name method_func: Method function """ @@ -85,7 +86,7 @@ class MethodRegistry: Args: task: Task type such as "file", "web", "feed", "stream", - "repo", "email", "db", "mcp", "parquet", or "ingest" + "repo", "email", "db", "mcp", "parquet", "xml", or "ingest" name: Method name Returns: diff --git a/semantica/ingest/xml_ingestor.py b/semantica/ingest/xml_ingestor.py new file mode 100644 index 00000000..57523538 --- /dev/null +++ b/semantica/ingest/xml_ingestor.py @@ -0,0 +1,814 @@ +""" +XML Ingestion Module + +This module provides dedicated XML ingestion for local XML files and directories. +It parses XML into structured dictionaries, extracts namespaces, element hierarchy, +attributes, and document metadata, and can optionally validate documents with XSD +schemas or DTD declarations. + +Example Usage: + >>> from semantica.ingest import XMLIngestor + >>> ingestor = XMLIngestor() + >>> data = ingestor.ingest_file("catalog.xml") + >>> data.root_tag + 'catalog' + >>> data.metadata["element_count"] + 12 +""" + +from __future__ import annotations + +import io +from dataclasses import dataclass, field +from datetime import datetime +from pathlib import Path +from typing import Any, Dict, List, Optional, Tuple, Union + +from lxml import etree + +from ..utils.constants import FILE_SIZE_LIMITS +from ..utils.exceptions import ProcessingError, ValidationError +from ..utils.logging import get_logger +from ..utils.progress_tracker import get_progress_tracker + +XML_NAMESPACE = "http://www.w3.org/XML/1998/namespace" + + +@dataclass +class XMLIngestionData: + """Structured XML ingestion result.""" + + root: Dict[str, Any] + elements: List[Dict[str, Any]] + namespaces: Dict[str, str] + source_path: str + root_tag: str + validation: Dict[str, Any] = field(default_factory=dict) + metadata: Dict[str, Any] = field(default_factory=dict) + ingested_at: datetime = field(default_factory=datetime.now) + + +class XMLIngestor: + """ + Dedicated XML file ingestion handler. + + Features: + - XML parsing into nested and flat structures + - Namespace and prefix extraction + - Element and attribute metadata extraction + - Optional XSD schema validation + - Optional DTD validation + - Directory ingestion for local XML files + """ + + SUPPORTED_EXTENSIONS = {".xml"} + + def __init__(self, config: Optional[Dict[str, Any]] = None, **kwargs): + """ + Initialize XML ingestor. + + Args: + config: Optional configuration dictionary + **kwargs: Additional configuration values + """ + self.logger = get_logger("xml_ingestor") + self.config = config or {} + self.config.update(kwargs) + self.progress_tracker = get_progress_tracker() + if not self.progress_tracker.enabled: + self.progress_tracker.enabled = True + + self.logger.debug("XML ingestor initialized") + + def ingest( + self, source: Union[str, Path], **options + ) -> Union[XMLIngestionData, List[XMLIngestionData]]: + """ + Ingest an XML file or directory. + + Args: + source: XML file or directory path + **options: Additional ingestion options + + Returns: + XMLIngestionData for a file, or a list for a directory + """ + source_path = Path(source) + if source_path.is_dir(): + return self.ingest_directory(source_path, **options) + return self.ingest_file(source_path, **options) + + def ingest_file(self, file_path: Union[str, Path], **options) -> XMLIngestionData: + """ + Ingest and parse a single XML file. + + Args: + file_path: Path to an XML file + **options: Ingestion options: + - schema_path/xsd_path/schema: Optional XSD schema file or XML string + - validate_schema: Validate against XSD when schema is provided + - validate_dtd: Validate against document DTD + - fail_on_validation_error: Raise on validation failure (default: True) + - include_tree: Include nested tree in result root (default: True) + - include_elements: Include flat element list (default: True) + - recover: Let lxml recover malformed XML when possible (default: False) + - include_comments: Include XML comments in the tree (default: False) + + Returns: + XMLIngestionData: Parsed XML data and metadata + """ + file_path = Path(file_path) + self._validate_file(file_path) + + tracking_id = self.progress_tracker.start_tracking( + file=str(file_path), + module="ingest", + submodule="XMLIngestor", + message=f"Ingesting XML: {file_path.name}", + ) + + try: + xml_bytes = file_path.read_bytes() + data = self._ingest_bytes( + xml_bytes, + source=str(file_path), + source_type="file", + file_path=file_path, + **options, + ) + + self.progress_tracker.stop_tracking( + tracking_id, + status="completed", + message=f"Ingested XML: {file_path.name}", + ) + return data + + except (ValidationError, ProcessingError): + self.progress_tracker.stop_tracking( + tracking_id, status="failed", message="XML ingestion failed" + ) + raise + except Exception as exc: + self.progress_tracker.stop_tracking( + tracking_id, status="failed", message=str(exc) + ) + self.logger.error(f"Failed to ingest XML {file_path}: {exc}") + raise ProcessingError(f"Failed to ingest XML: {exc}") from exc + + def ingest_string( + self, + xml_content: Union[str, bytes], + source: str = "string", + **options, + ) -> XMLIngestionData: + """ + Ingest XML content from a string or bytes object. + + Args: + xml_content: XML content + source: Source label to include in metadata + **options: Additional ingestion options + + Returns: + XMLIngestionData: Parsed XML data and metadata + """ + if isinstance(xml_content, str): + xml_bytes = xml_content.encode(options.get("encoding", "utf-8")) + else: + xml_bytes = xml_content + + return self._ingest_bytes( + xml_bytes, + source=source, + source_type="string", + file_path=None, + **options, + ) + + def ingest_directory( + self, + directory_path: Union[str, Path], + recursive: bool = True, + **options, + ) -> List[XMLIngestionData]: + """ + Ingest XML files from a directory. + + Args: + directory_path: Directory path + recursive: Whether to search subdirectories + **options: Additional ingestion options + + Returns: + List[XMLIngestionData]: Parsed XML files + """ + directory_path = Path(directory_path) + if not directory_path.exists(): + raise ValidationError(f"XML directory not found: {directory_path}") + if not directory_path.is_dir(): + raise ValidationError(f"Path is not a directory: {directory_path}") + + tracking_id = self.progress_tracker.start_tracking( + file=str(directory_path), + module="ingest", + submodule="XMLIngestor", + message=f"Ingesting XML directory: {directory_path.name}", + ) + + try: + xml_files = self._xml_files(directory_path, recursive=recursive) + results = [] + + self.progress_tracker.update_tracking( + tracking_id, message=f"Processing {len(xml_files)} XML files" + ) + + for index, xml_file in enumerate(xml_files, 1): + try: + results.append(self.ingest_file(xml_file, **options)) + self.progress_tracker.update_progress( + tracking_id, + processed=index, + total=len(xml_files), + message=f"Processing {index}/{len(xml_files)}: {xml_file.name}", + ) + except Exception as exc: + self.logger.error(f"Failed to ingest XML file {xml_file}: {exc}") + if self.config.get("fail_fast", False) or options.get( + "fail_fast", False + ): + raise ProcessingError( + f"Failed to ingest XML file: {exc}" + ) from exc + + self.progress_tracker.stop_tracking( + tracking_id, + status="completed", + message=f"Ingested {len(results)} XML files", + ) + return results + + except Exception as exc: + self.progress_tracker.stop_tracking( + tracking_id, status="failed", message=str(exc) + ) + raise + + def validate_file( + self, + file_path: Union[str, Path], + schema_path: Optional[Union[str, Path]] = None, + validate_dtd: bool = False, + **options, + ) -> Dict[str, Any]: + """ + Validate an XML file and return a validation report. + + Args: + file_path: XML file path + schema_path: Optional XSD schema path + validate_dtd: Whether to validate document DTD + **options: Additional parser options + + Returns: + Dict[str, Any]: Validation report + """ + if schema_path is not None: + options["schema_path"] = schema_path + + schema_source = self._schema_source(options) + options.setdefault("validate_schema", schema_source is not None) + options["validate_dtd"] = validate_dtd + options["fail_on_validation_error"] = False + options.setdefault("include_tree", False) + options.setdefault("include_elements", False) + + return self.ingest_file(file_path, **options).validation + + def extract_metadata( + self, file_path: Union[str, Path], **options + ) -> Dict[str, Any]: + """ + Extract XML document metadata without requiring callers to inspect the tree. + + Args: + file_path: XML file path + **options: Additional ingestion options + + Returns: + Dict[str, Any]: XML metadata + """ + options.setdefault("include_tree", False) + options.setdefault("include_elements", False) + return self.ingest_file(file_path, **options).metadata + + def _ingest_bytes( + self, + xml_bytes: bytes, + source: str, + source_type: str, + file_path: Optional[Path], + **options, + ) -> XMLIngestionData: + tree, parser_errors = self._parse_tree(xml_bytes, source, options) + validation = self._validate_tree(tree, source, options) + + root = tree.getroot() + namespaces = self._collect_namespaces(root) + prefix_by_namespace = self._prefix_by_namespace(namespaces) + root_data, elements, structure_metadata = self._build_structures( + root, + namespaces=namespaces, + prefix_by_namespace=prefix_by_namespace, + options=options, + ) + + metadata = self._document_metadata( + tree=tree, + source=source, + source_type=source_type, + file_path=file_path, + namespaces=namespaces, + structure_metadata=structure_metadata, + validation=validation, + parser_errors=parser_errors, + ) + + return XMLIngestionData( + root=root_data, + elements=elements, + namespaces=namespaces, + source_path=source, + root_tag=root_data["tag"], + validation=validation, + metadata=metadata, + ) + + def _parse_tree( + self, xml_bytes: bytes, source: str, options: Dict[str, Any] + ) -> Tuple[Any, List[str]]: + validate_dtd = bool( + options.get("validate_dtd", self.config.get("validate_dtd", False)) + ) + parser = etree.XMLParser( + remove_blank_text=bool( + options.get( + "remove_blank_text", self.config.get("remove_blank_text", True) + ) + ), + remove_comments=not bool(options.get("include_comments", False)), + resolve_entities=False, + no_network=not bool(options.get("allow_network", False)), + load_dtd=bool(options.get("load_dtd", validate_dtd)), + dtd_validation=False, + recover=bool(options.get("recover", self.config.get("recover", False))), + huge_tree=bool( + options.get("huge_tree", self.config.get("huge_tree", False)) + ), + ) + + try: + tree = etree.parse(io.BytesIO(xml_bytes), parser) + except etree.XMLSyntaxError as exc: + message = self._format_xml_error(exc) + raise ProcessingError(f"Malformed XML in {source}: {message}") from exc + + parser_errors = [str(error) for error in parser.error_log] + return tree, parser_errors + + def _validate_tree( + self, tree: Any, source: str, options: Dict[str, Any] + ) -> Dict[str, Any]: + schema_source = self._schema_source(options) + validate_schema_option = options.get("validate_schema") + validate_schema = ( + schema_source is not None + if validate_schema_option is None + else bool(validate_schema_option) + ) + validate_dtd = bool( + options.get("validate_dtd", self.config.get("validate_dtd", False)) + ) + fail_on_error = bool(options.get("fail_on_validation_error", True)) + + validation = { + "is_valid": True, + "schema": { + "validated": False, + "valid": None, + "source": self._schema_label(schema_source), + "errors": [], + }, + "dtd": { + "validated": False, + "valid": None, + "name": None, + "system_url": None, + "public_id": None, + "errors": [], + }, + } + + if validate_schema: + if schema_source is None: + raise ValidationError( + "XML schema validation requested but no XSD schema was provided", + validation_context={"source": source}, + ) + + schema = self._load_schema(schema_source) + schema_valid = schema.validate(tree) + schema_errors = [str(error) for error in schema.error_log] + validation["schema"].update( + { + "validated": True, + "valid": schema_valid, + "errors": schema_errors, + } + ) + validation["is_valid"] = validation["is_valid"] and schema_valid + + if not schema_valid and fail_on_error: + raise ValidationError( + self._validation_message( + "XML schema validation failed", source, schema_errors + ), + validation_context={"source": source}, + errors=schema_errors, + ) + + if validate_dtd: + dtd = tree.docinfo.internalDTD or tree.docinfo.externalDTD + if dtd is None: + dtd_valid = False + dtd_errors = ["No DTD declaration found in XML document."] + else: + dtd_valid = dtd.validate(tree) + dtd_errors = [str(error) for error in dtd.error_log] + validation["dtd"].update( + { + "name": getattr(dtd, "name", None), + "system_url": getattr(dtd, "system_url", None), + "public_id": getattr(dtd, "public_id", None), + } + ) + + validation["dtd"].update( + {"validated": True, "valid": dtd_valid, "errors": dtd_errors} + ) + validation["is_valid"] = validation["is_valid"] and dtd_valid + + if not dtd_valid and fail_on_error: + raise ValidationError( + self._validation_message( + "XML DTD validation failed", source, dtd_errors + ), + validation_context={"source": source}, + errors=dtd_errors, + ) + + return validation + + def _build_structures( + self, + root: Any, + namespaces: Dict[str, str], + prefix_by_namespace: Dict[str, str], + options: Dict[str, Any], + ) -> Tuple[Dict[str, Any], List[Dict[str, Any]], Dict[str, Any]]: + elements: List[Dict[str, Any]] = [] + stats = { + "element_count": 0, + "attribute_count": 0, + "max_depth": 0, + "tag_counts": {}, + "element_limit_reached": False, + } + + root_data = self._element_to_dict( + root, + parent_path="", + depth=0, + namespaces=namespaces, + prefix_by_namespace=prefix_by_namespace, + elements=elements, + stats=stats, + options=options, + ) + + unique_tags = sorted(stats["tag_counts"].keys()) + metadata = { + "element_count": stats["element_count"], + "attribute_count": stats["attribute_count"], + "max_depth": stats["max_depth"], + "unique_tags": unique_tags, + "tag_counts": stats["tag_counts"], + "flat_element_count": len(elements), + "element_limit_reached": stats["element_limit_reached"], + } + + return root_data, elements, metadata + + def _element_to_dict( + self, + element: Any, + parent_path: str, + depth: int, + namespaces: Dict[str, str], + prefix_by_namespace: Dict[str, str], + elements: List[Dict[str, Any]], + stats: Dict[str, Any], + options: Dict[str, Any], + ) -> Dict[str, Any]: + tag_info = self._name_info(element.tag, prefix_by_namespace) + path = ( + f"{parent_path}/{tag_info['tag']}" if parent_path else f"/{tag_info['tag']}" + ) + attributes, attribute_details = self._attributes_to_dict( + element.attrib, + prefix_by_namespace, + ) + text = element.text or "" + if options.get("strip_text", True): + text = text.strip() + + child_elements = [child for child in element if isinstance(child.tag, str)] + + element_data = { + "tag": tag_info["tag"], + "local_name": tag_info["local_name"], + "namespace": tag_info["namespace"], + "prefix": tag_info["prefix"], + "text": text, + "attributes": attributes, + "attribute_details": attribute_details, + "path": path, + "depth": depth, + "child_count": len(child_elements), + } + + stats["element_count"] += 1 + stats["attribute_count"] += len(attributes) + stats["max_depth"] = max(stats["max_depth"], depth) + stats["tag_counts"][tag_info["tag"]] = ( + stats["tag_counts"].get(tag_info["tag"], 0) + 1 + ) + + if options.get("include_elements", True): + max_elements = options.get("max_elements") + if max_elements is None or len(elements) < max_elements: + flat_element = dict(element_data) + flat_element.pop("children", None) + elements.append(flat_element) + else: + stats["element_limit_reached"] = True + + children = [ + self._element_to_dict( + child, + parent_path=path, + depth=depth + 1, + namespaces=namespaces, + prefix_by_namespace=prefix_by_namespace, + elements=elements, + stats=stats, + options=options, + ) + for child in child_elements + ] + + if options.get("include_tree", True): + element_data["children"] = children + + return element_data + + def _attributes_to_dict( + self, attributes: Dict[str, str], prefix_by_namespace: Dict[str, str] + ) -> Tuple[Dict[str, str], Dict[str, Dict[str, Optional[str]]]]: + values = {} + details = {} + + for raw_name, value in attributes.items(): + name_info = self._name_info(raw_name, prefix_by_namespace) + display_name = name_info["tag"] + values[display_name] = value + details[display_name] = { + "value": value, + "local_name": name_info["local_name"], + "namespace": name_info["namespace"], + "prefix": name_info["prefix"], + } + + return values, details + + def _name_info( + self, raw_name: Any, prefix_by_namespace: Dict[str, str] + ) -> Dict[str, Optional[str]]: + if not isinstance(raw_name, str): + return { + "tag": str(raw_name), + "local_name": str(raw_name), + "namespace": None, + "prefix": None, + } + + if raw_name.startswith("{"): + qname = etree.QName(raw_name) + namespace = qname.namespace + local_name = qname.localname + else: + namespace = None + local_name = raw_name + + prefix = prefix_by_namespace.get(namespace) if namespace else None + if namespace == XML_NAMESPACE: + prefix = "xml" + + if prefix and prefix != "default": + tag = f"{prefix}:{local_name}" + else: + tag = local_name + + return { + "tag": tag, + "local_name": local_name, + "namespace": namespace, + "prefix": prefix, + } + + def _collect_namespaces(self, root: Any) -> Dict[str, str]: + namespaces: Dict[str, str] = {} + + for element in root.iter(): + if not isinstance(element.tag, str): + continue + + for prefix, uri in (element.nsmap or {}).items(): + if not uri: + continue + key = prefix or "default" + self._add_namespace(namespaces, key, uri) + + for raw_name in element.attrib: + if isinstance(raw_name, str) and raw_name.startswith( + f"{{{XML_NAMESPACE}}}" + ): + self._add_namespace(namespaces, "xml", XML_NAMESPACE) + + return namespaces + + def _add_namespace(self, namespaces: Dict[str, str], prefix: str, uri: str) -> None: + if prefix not in namespaces: + namespaces[prefix] = uri + return + + if namespaces[prefix] == uri: + return + + index = 2 + while f"{prefix}_{index}" in namespaces: + index += 1 + namespaces[f"{prefix}_{index}"] = uri + + def _prefix_by_namespace(self, namespaces: Dict[str, str]) -> Dict[str, str]: + prefix_by_namespace: Dict[str, str] = {} + for prefix, uri in namespaces.items(): + prefix_by_namespace.setdefault(uri, prefix) + return prefix_by_namespace + + def _document_metadata( + self, + tree: Any, + source: str, + source_type: str, + file_path: Optional[Path], + namespaces: Dict[str, str], + structure_metadata: Dict[str, Any], + validation: Dict[str, Any], + parser_errors: List[str], + ) -> Dict[str, Any]: + root = tree.getroot() + docinfo = tree.docinfo + has_dtd = bool(docinfo.internalDTD or docinfo.externalDTD) + prefix_by_namespace = self._prefix_by_namespace(namespaces) + root_info = self._name_info(root.tag, prefix_by_namespace) + + metadata = { + "format": "xml", + "source": source, + "source_type": source_type, + "root_tag": root_info["tag"], + "root_element": root_info["local_name"], + "root_namespace": root_info["namespace"], + "root_prefix": root_info["prefix"], + "namespace_count": len(namespaces), + "namespaces": namespaces, + "has_dtd": has_dtd, + "dtd_name": docinfo.root_name, + "dtd_system_url": docinfo.system_url, + "dtd_public_id": docinfo.public_id, + "xml_version": docinfo.xml_version, + "encoding": docinfo.encoding, + "parser_errors": parser_errors, + "schema_validated": validation["schema"]["validated"], + "dtd_validated": validation["dtd"]["validated"], + "validation_passed": validation["is_valid"], + **structure_metadata, + } + + if file_path is not None: + metadata.update( + { + "file": str(file_path), + "file_name": file_path.name, + "file_size": file_path.stat().st_size, + "extension": file_path.suffix, + } + ) + + return metadata + + def _schema_source(self, options: Dict[str, Any]) -> Any: + return ( + options.get("schema_path") + or options.get("xsd_path") + or options.get("schema") + or self.config.get("schema_path") + or self.config.get("xsd_path") + or self.config.get("schema") + ) + + def _schema_label(self, schema_source: Any) -> Optional[str]: + if schema_source is None: + return None + if isinstance(schema_source, (str, Path)): + text = str(schema_source) + if text.lstrip().startswith("<"): + return "inline" + return text + return type(schema_source).__name__ + + def _load_schema(self, schema_source: Any) -> Any: + if isinstance(schema_source, etree.XMLSchema): + return schema_source + + parser = etree.XMLParser(resolve_entities=False, no_network=True) + + if isinstance(schema_source, Path): + if not schema_source.exists(): + raise ValidationError(f"XSD schema file not found: {schema_source}") + schema_doc = etree.parse(str(schema_source), parser) + return etree.XMLSchema(schema_doc) + + if isinstance(schema_source, bytes): + schema_doc = etree.fromstring(schema_source, parser) + return etree.XMLSchema(schema_doc) + + if isinstance(schema_source, str): + if schema_source.lstrip().startswith("<"): + schema_doc = etree.fromstring(schema_source.encode("utf-8"), parser) + return etree.XMLSchema(schema_doc) + + schema_path = Path(schema_source) + if not schema_path.exists(): + raise ValidationError(f"XSD schema file not found: {schema_source}") + schema_doc = etree.parse(str(schema_path), parser) + return etree.XMLSchema(schema_doc) + + raise ValidationError( + f"Unsupported XSD schema source: {type(schema_source).__name__}" + ) + + def _validation_message(self, prefix: str, source: str, errors: List[str]) -> str: + first_error = errors[0] if errors else "No detailed validation error available." + return f"{prefix} for {source}: {first_error}" + + def _format_xml_error(self, exc: etree.XMLSyntaxError) -> str: + if exc.error_log: + return str(exc.error_log.last_error) + return str(exc) + + def _validate_file(self, file_path: Path) -> None: + if not file_path.exists(): + raise ValidationError(f"XML file not found: {file_path}") + if not file_path.is_file(): + raise ValidationError(f"Path is not a file: {file_path}") + if file_path.suffix.lower() not in self.SUPPORTED_EXTENSIONS: + raise ValidationError(f"File is not an XML file: {file_path}") + + max_size = FILE_SIZE_LIMITS.get("MAX_DOCUMENT_SIZE", 104857600) + file_size = file_path.stat().st_size + if file_size > max_size: + raise ValidationError( + f"File size {file_size:,} bytes exceeds maximum {max_size:,} bytes " + f"({file_path.name})" + ) + + def _xml_files(self, directory_path: Path, recursive: bool) -> List[Path]: + iterator = directory_path.rglob("*") if recursive else directory_path.iterdir() + return sorted( + path + for path in iterator + if path.is_file() and path.suffix.lower() in self.SUPPORTED_EXTENSIONS + ) diff --git a/tests/ingest/test_xml_ingestor.py b/tests/ingest/test_xml_ingestor.py new file mode 100644 index 00000000..5b197a0d --- /dev/null +++ b/tests/ingest/test_xml_ingestor.py @@ -0,0 +1,199 @@ +from pathlib import Path + +import pytest + +from semantica.ingest import ( + XMLIngestionData, + XMLIngestor, + ingest, + ingest_file, + ingest_xml, +) +from semantica.ingest.xml_ingestor import XML_NAMESPACE +from semantica.utils.exceptions import ProcessingError, ValidationError + + +def test_xml_ingestor_extracts_structure_namespaces_and_attributes( + tmp_path: Path, +) -> None: + xml_file = tmp_path / "catalog.xml" + xml_file.write_text( + """ + + + Semantica + Ada + + +""", + encoding="utf-8", + ) + + data = XMLIngestor().ingest_file(xml_file) + + assert isinstance(data, XMLIngestionData) + assert data.root_tag == "catalog" + assert data.namespaces["default"] == "https://example.com/catalog" + assert data.namespaces["bk"] == "https://example.com/book" + assert data.namespaces["xml"] == XML_NAMESPACE + assert data.root["children"][0]["tag"] == "bk:book" + assert data.root["children"][0]["attributes"]["id"] == "b1" + assert data.root["children"][0]["attribute_details"]["xml:lang"]["value"] == "en" + assert ( + data.root["children"][0]["attribute_details"]["xml:lang"]["namespace"] + == XML_NAMESPACE + ) + assert data.elements[0]["tag"] == "catalog" + assert data.elements[1]["tag"] == "bk:book" + assert data.metadata["element_count"] == 4 + assert data.metadata["attribute_count"] == 3 + assert data.metadata["root_namespace"] == "https://example.com/catalog" + assert data.metadata["tag_counts"]["bk:author"] == 1 + + +def test_xml_ingestor_validates_xsd_schema(tmp_path: Path) -> None: + xml_file = tmp_path / "catalog.xml" + xml_file.write_text( + """ + + Semantica + + +""", + encoding="utf-8", + ) + xsd_file = tmp_path / "catalog.xsd" + xsd_file.write_text( + """ + + + + + + + + + + + + + + + + +""", + encoding="utf-8", + ) + + data = XMLIngestor().ingest_file(xml_file, schema_path=xsd_file) + + assert data.validation["is_valid"] is True + assert data.validation["schema"]["validated"] is True + assert data.validation["schema"]["valid"] is True + + +def test_xml_ingestor_reports_schema_validation_errors(tmp_path: Path) -> None: + xml_file = tmp_path / "catalog.xml" + xml_file.write_text("", encoding="utf-8") + xsd_file = tmp_path / "catalog.xsd" + xsd_file.write_text( + """ + + + + + + + + + + + + + + + +""", + encoding="utf-8", + ) + + report = XMLIngestor().validate_file(xml_file, schema_path=xsd_file) + + assert report["is_valid"] is False + assert report["schema"]["validated"] is True + assert report["schema"]["valid"] is False + assert report["schema"]["errors"] + + with pytest.raises(ValidationError, match="XML schema validation failed"): + XMLIngestor().ingest_file(xml_file, schema_path=xsd_file) + + +def test_xml_ingestor_validates_internal_dtd(tmp_path: Path) -> None: + xml_file = tmp_path / "note.xml" + xml_file.write_text( + """ + + +]> +AdaGrace +""", + encoding="utf-8", + ) + + report = XMLIngestor().validate_file(xml_file, validate_dtd=True) + + assert report["is_valid"] is True + assert report["dtd"]["validated"] is True + assert report["dtd"]["valid"] is True + assert report["dtd"]["name"] == "note" + + +def test_xml_ingestor_rejects_malformed_xml(tmp_path: Path) -> None: + xml_file = tmp_path / "broken.xml" + xml_file.write_text("", encoding="utf-8") + + with pytest.raises(ProcessingError, match="Malformed XML"): + XMLIngestor().ingest_file(xml_file) + + +def test_xml_ingest_methods_and_unified_dispatch(tmp_path: Path) -> None: + xml_file = tmp_path / "catalog.xml" + xml_file.write_text( + "Semantica", + encoding="utf-8", + ) + + direct = ingest_xml(xml_file) + metadata = ingest_xml(xml_file, method="metadata") + via_file_method = ingest_file(xml_file, method="xml") + unified = ingest(xml_file) + + assert isinstance(direct, XMLIngestionData) + assert metadata["root_tag"] == "catalog" + assert isinstance(via_file_method, XMLIngestionData) + assert isinstance(unified["xml"], XMLIngestionData) + + +def test_xml_ingestor_ingests_string() -> None: + data = XMLIngestor().ingest_string("hello") + + assert isinstance(data, XMLIngestionData) + assert data.root_tag == "root" + assert data.metadata["source_type"] == "string" + assert data.elements[1]["attributes"]["id"] == "1" + assert data.elements[1]["text"] == "hello" + + +def test_xml_ingestor_ingests_directory(tmp_path: Path) -> None: + (tmp_path / "one.xml").write_text("", encoding="utf-8") + (tmp_path / "two.xml").write_text("", encoding="utf-8") + (tmp_path / "skip.txt").write_text("", encoding="utf-8") + + results = ingest_xml(tmp_path, method="directory", recursive=False) + + assert len(results) == 2 + assert {result.metadata["file_name"] for result in results} == { + "one.xml", + "two.xml", + }