From 652d6ba447fadd643cfd339d4ff00caee09a6538 Mon Sep 17 00:00:00 2001 From: KaifAhmad1 Date: Thu, 6 Nov 2025 17:54:26 +0530 Subject: [PATCH] refactor(deduplication): enhance documentation and code quality for all deduplication module files - Add comprehensive docstrings to all methods in duplicate_detector.py - Enhance entity_merger.py with detailed merge operation documentation - Improve similarity_calculator.py with multi-factor similarity docs - Refactor merge_strategy.py with complete strategy management docs - Add detailed cluster_builder.py documentation for clustering algorithms - Enhance module-level documentation in all files - Improve error messages and logging throughout - Add type hints and clearer variable names - Document all private methods for better maintainability - Add examples and usage patterns in docstrings --- semantica/deduplication/__init__.py | 29 +- semantica/deduplication/cluster_builder.py | 98 ++++- semantica/deduplication/duplicate_detector.py | 289 ++++++++++++--- semantica/deduplication/entity_merger.py | 346 +++++++++++++++--- semantica/deduplication/merge_strategy.py | 87 ++++- .../deduplication/similarity_calculator.py | 132 ++++++- 6 files changed, 834 insertions(+), 147 deletions(-) diff --git a/semantica/deduplication/__init__.py b/semantica/deduplication/__init__.py index a623ccd0..e6b16e4e 100644 --- a/semantica/deduplication/__init__.py +++ b/semantica/deduplication/__init__.py @@ -1,8 +1,33 @@ """ Advanced Deduplication Module -This module provides semantic entity deduplication and merging -to keep knowledge graphs clean and maintain single source of truth. +This module provides comprehensive semantic entity deduplication and merging +capabilities for the Semantica framework, helping keep knowledge graphs clean +and maintain a single source of truth. + +Key Features: + - Duplicate detection using similarity metrics + - Entity merging with configurable strategies + - Similarity calculation using multiple factors + - Cluster-based batch deduplication + - Provenance preservation during merges + +Main Components: + - DuplicateDetector: Detects duplicate entities and relationships + - EntityMerger: Merges duplicate entities using strategies + - SimilarityCalculator: Calculates multi-factor similarity + - MergeStrategyManager: Manages merge strategies and conflict resolution + - ClusterBuilder: Builds clusters for batch deduplication + +Example Usage: + >>> from semantica.deduplication import DuplicateDetector, EntityMerger + >>> detector = DuplicateDetector(similarity_threshold=0.8) + >>> duplicates = detector.detect_duplicates(entities) + >>> merger = EntityMerger() + >>> merged = merger.merge_duplicates(entities) + +Author: Semantica Contributors +License: MIT """ from .entity_merger import EntityMerger, MergeOperation diff --git a/semantica/deduplication/cluster_builder.py b/semantica/deduplication/cluster_builder.py index 2daa54bd..0a0cdee6 100644 --- a/semantica/deduplication/cluster_builder.py +++ b/semantica/deduplication/cluster_builder.py @@ -1,8 +1,24 @@ """ -Cluster Builder for Deduplication +Cluster Builder Module -Builds clusters of similar entities for batch deduplication -using clustering algorithms and similarity graphs. +This module provides cluster building capabilities for the Semantica framework, +creating clusters of similar entities for batch deduplication using clustering +algorithms and similarity graphs. + +Key Features: + - Graph-based clustering using union-find algorithm + - Hierarchical clustering for large datasets + - Cluster quality assessment and metrics + - Incremental cluster updates + - Configurable cluster size constraints + +Example Usage: + >>> from semantica.deduplication import ClusterBuilder + >>> builder = ClusterBuilder(similarity_threshold=0.8) + >>> result = builder.build_clusters(entities) + +Author: Semantica Contributors +License: MIT """ from typing import Any, Dict, List, Optional, Set, Tuple @@ -36,26 +52,76 @@ class ClusterResult: class ClusterBuilder: """ - Cluster building engine. + Cluster building engine for entity clustering. - • Builds entity clusters using similarity graphs - • Supports cluster-based deduplication workflows - • Assesses cluster quality - • Uses hierarchical clustering for large datasets - • Supports incremental cluster updates + This class builds clusters of similar entities for batch deduplication using + graph-based or hierarchical clustering algorithms. Clusters can be used for + efficient batch processing of duplicate detection and merging. + + Features: + - Graph-based clustering using union-find algorithm + - Hierarchical clustering for large datasets + - Cluster quality assessment and metrics + - Incremental cluster updates + - Configurable cluster size constraints + + Example Usage: + >>> builder = ClusterBuilder( + ... similarity_threshold=0.8, + ... min_cluster_size=2, + ... max_cluster_size=50 + ... ) + >>> result = builder.build_clusters(entities) + >>> print(f"Found {len(result.clusters)} clusters") """ - def __init__(self, config=None, **kwargs): - """Initialize cluster builder.""" + def __init__( + self, + similarity_threshold: float = 0.7, + min_cluster_size: int = 2, + max_cluster_size: int = 100, + use_hierarchical: bool = False, + config: Optional[Dict[str, Any]] = None, + **kwargs + ): + """ + Initialize cluster builder. + + Sets up the cluster builder with similarity calculator and clustering + configuration parameters. + + Args: + similarity_threshold: Minimum similarity for entities to be in same cluster + (0.0 to 1.0, default: 0.7) + min_cluster_size: Minimum number of entities in a valid cluster (default: 2) + max_cluster_size: Maximum number of entities in a cluster (default: 100) + use_hierarchical: Whether to use hierarchical clustering (default: False). + If False, uses faster graph-based clustering. + config: Configuration dictionary (merged with kwargs) + **kwargs: Additional configuration options: + - similarity: Configuration for SimilarityCalculator + """ self.logger = get_logger("cluster_builder") + + # Merge configuration self.config = config or {} self.config.update(kwargs) - self.similarity_calculator = SimilarityCalculator(**config.get("similarity", {})) - self.similarity_threshold = config.get("similarity_threshold", 0.7) - self.min_cluster_size = config.get("min_cluster_size", 2) - self.max_cluster_size = config.get("max_cluster_size", 100) - self.use_hierarchical = config.get("use_hierarchical", False) + # Initialize similarity calculator + similarity_config = self.config.get("similarity", {}) + self.similarity_calculator = SimilarityCalculator(**similarity_config) + + # Clustering parameters + self.similarity_threshold = similarity_threshold + self.min_cluster_size = min_cluster_size + self.max_cluster_size = max_cluster_size + self.use_hierarchical = use_hierarchical + + self.logger.debug( + f"Cluster builder initialized: threshold={similarity_threshold}, " + f"size_range=[{min_cluster_size}, {max_cluster_size}], " + f"hierarchical={use_hierarchical}" + ) def build_clusters( self, diff --git a/semantica/deduplication/duplicate_detector.py b/semantica/deduplication/duplicate_detector.py index 46c5ec33..53bce09d 100644 --- a/semantica/deduplication/duplicate_detector.py +++ b/semantica/deduplication/duplicate_detector.py @@ -1,8 +1,26 @@ """ -Duplicate Detector for Semantica framework. +Duplicate Detector Module -Detects duplicate entities and relationships in knowledge graphs -using similarity thresholds and clustering algorithms. +This module provides comprehensive duplicate detection capabilities for the Semantica +framework, identifying duplicate entities and relationships in knowledge graphs using +similarity thresholds, clustering algorithms, and confidence scoring. + +Key Features: + - Entity duplicate detection using similarity metrics + - Relationship duplicate detection + - Duplicate group formation using union-find algorithm + - Incremental duplicate detection for new entities + - Confidence scoring for duplicate candidates + - Representative entity selection from duplicate groups + +Example Usage: + >>> from semantica.deduplication import DuplicateDetector + >>> detector = DuplicateDetector(similarity_threshold=0.8) + >>> duplicates = detector.detect_duplicates(entities) + >>> groups = detector.detect_duplicate_groups(entities) + +Author: Semantica Contributors +License: MIT """ from typing import Any, Dict, List, Optional, Set, Tuple @@ -38,87 +56,201 @@ class DuplicateGroup: class DuplicateDetector: """ - Duplicate detection engine. + Duplicate detection engine for knowledge graphs. - • Detects duplicate entities using similarity - • Identifies duplicate relationships - • Uses cluster-based duplicate identification - • Supports batch duplicate detection - • Provides incremental duplicate detection - • Scores confidence for duplicate candidates + This class provides comprehensive duplicate detection capabilities, identifying + duplicate entities and relationships using similarity metrics, confidence scoring, + and group formation algorithms. + + Features: + - Entity duplicate detection using multi-factor similarity + - Relationship duplicate detection + - Duplicate group formation (union-find algorithm) + - Incremental detection for new entities + - Confidence scoring with multiple factors + - Representative entity selection + + Example Usage: + >>> detector = DuplicateDetector(similarity_threshold=0.8, confidence_threshold=0.7) + >>> candidates = detector.detect_duplicates(entities) + >>> groups = detector.detect_duplicate_groups(entities) """ - def __init__(self, config=None, **kwargs): - """Initialize duplicate detector.""" + def __init__( + self, + similarity_threshold: float = 0.7, + confidence_threshold: float = 0.6, + use_clustering: bool = True, + config: Optional[Dict[str, Any]] = None, + **kwargs + ): + """ + Initialize duplicate detector. + + Sets up the detector with similarity calculator and configurable thresholds + for duplicate detection and confidence scoring. + + Args: + similarity_threshold: Minimum similarity score to consider entities as duplicates + (0.0 to 1.0, default: 0.7) + confidence_threshold: Minimum confidence score for duplicate candidates + (0.0 to 1.0, default: 0.6) + use_clustering: Whether to use clustering for group formation (default: True) + config: Configuration dictionary (merged with kwargs) + **kwargs: Additional configuration options: + - similarity: Configuration for SimilarityCalculator + """ self.logger = get_logger("duplicate_detector") + + # Merge configuration self.config = config or {} self.config.update(kwargs) - self.similarity_calculator = SimilarityCalculator(**config.get("similarity", {})) - self.similarity_threshold = config.get("similarity_threshold", 0.7) - self.confidence_threshold = config.get("confidence_threshold", 0.6) - self.use_clustering = config.get("use_clustering", True) + # Initialize similarity calculator + similarity_config = self.config.get("similarity", {}) + self.similarity_calculator = SimilarityCalculator(**similarity_config) + + # Detection thresholds + self.similarity_threshold = similarity_threshold + self.confidence_threshold = confidence_threshold + self.use_clustering = use_clustering + + self.logger.debug( + f"Duplicate detector initialized: similarity_threshold={similarity_threshold}, " + f"confidence_threshold={confidence_threshold}" + ) def detect_duplicates( self, entities: List[Dict[str, Any]], + threshold: Optional[float] = None, **options ) -> List[DuplicateCandidate]: """ - Detect duplicate entities. + Detect duplicate entities from a list. + + This method compares all pairs of entities and identifies duplicates based + on similarity scores and confidence thresholds. Returns candidates sorted + by confidence (highest first). Args: - entities: List of entities to check - **options: Detection options + entities: List of entity dictionaries to check for duplicates. + Each entity should have at least a "name" field. + threshold: Minimum similarity threshold (overrides instance default) + **options: Additional detection options passed to similarity calculator Returns: - List of duplicate candidates + List of DuplicateCandidate objects, sorted by confidence (highest first). + Each candidate contains: + - entity1, entity2: The duplicate entity pair + - similarity_score: Similarity score (0.0 to 1.0) + - confidence: Confidence score (0.0 to 1.0) + - reasons: List of reasons why they're considered duplicates + - metadata: Additional metadata + + Example: + >>> entities = [ + ... {"id": "1", "name": "Apple Inc."}, + ... {"id": "2", "name": "Apple"}, + ... {"id": "3", "name": "Microsoft"} + ... ] + >>> candidates = detector.detect_duplicates(entities, threshold=0.8) + >>> # Returns candidates for Apple Inc. and Apple """ - threshold = options.get("threshold", self.similarity_threshold) - candidates = [] + # Use provided threshold or instance default + detection_threshold = threshold if threshold is not None else self.similarity_threshold - # Calculate similarity for all pairs - similarities = self.similarity_calculator.batch_calculate_similarity( - entities, - threshold=threshold + self.logger.info( + f"Detecting duplicates in {len(entities)} entities " + f"(threshold: {detection_threshold})" ) + # Calculate similarity for all entity pairs + similarities = self.similarity_calculator.batch_calculate_similarity( + entities, + threshold=detection_threshold + ) + + self.logger.debug(f"Found {len(similarities)} similar pairs above threshold") + + # Create duplicate candidates from similar pairs + candidates = [] for entity1, entity2, score in similarities: candidate = self._create_duplicate_candidate(entity1, entity2, score) + # Filter by confidence threshold if candidate.confidence >= self.confidence_threshold: candidates.append(candidate) - # Sort by confidence + # Sort by confidence (highest first) candidates.sort(key=lambda c: c.confidence, reverse=True) + self.logger.info( + f"Detected {len(candidates)} duplicate candidate(s) " + f"(confidence >= {self.confidence_threshold})" + ) + return candidates def detect_duplicate_groups( self, entities: List[Dict[str, Any]], + threshold: Optional[float] = None, **options ) -> List[DuplicateGroup]: """ Detect groups of duplicate entities. + This method identifies duplicate entities and groups them together using + a union-find algorithm. Each group represents entities that are duplicates + of each other, with confidence scores and representative entities. + + Process: + 1. Detect duplicate candidates using similarity + 2. Build groups using union-find (entities in same group are duplicates) + 3. Calculate group confidence scores + 4. Select representative entity for each group + Args: - entities: List of entities - **options: Detection options + entities: List of entity dictionaries to group + threshold: Minimum similarity threshold (overrides instance default) + **options: Additional detection options passed to detect_duplicates() Returns: - List of duplicate groups + List of DuplicateGroup objects, each containing: + - entities: List of duplicate entities in the group + - similarity_scores: Dict mapping entity pairs to similarity scores + - representative: Representative entity (most complete) + - confidence: Group confidence score (0.0 to 1.0) + - metadata: Additional group metadata + + Example: + >>> groups = detector.detect_duplicate_groups(entities, threshold=0.8) + >>> for group in groups: + ... print(f"Group: {len(group.entities)} entities, " + ... f"confidence: {group.confidence:.2f}") """ - candidates = self.detect_duplicates(entities, **options) + self.logger.info(f"Detecting duplicate groups from {len(entities)} entities") + + # Detect duplicate candidates + candidates = self.detect_duplicates(entities, threshold=threshold, **options) # Build groups using union-find approach + # This connects entities that are duplicates into groups groups = self._build_duplicate_groups(candidates) - # Calculate group metrics + self.logger.debug(f"Built {len(groups)} duplicate group(s)") + + # Calculate group metrics for each group for group in groups: group.confidence = self._calculate_group_confidence(group) group.representative = self._select_representative(group) + self.logger.info( + f"Detected {len(groups)} duplicate group(s) with " + f"{sum(len(g.entities) for g in groups)} total entities" + ) + return groups def detect_relationship_duplicates( @@ -153,39 +285,69 @@ class DuplicateDetector: self, new_entities: List[Dict[str, Any]], existing_entities: List[Dict[str, Any]], + threshold: Optional[float] = None, **options ) -> List[DuplicateCandidate]: """ Incremental duplicate detection for new entities. + This method efficiently detects duplicates between new entities and an + existing set of entities, avoiding the O(n²) comparison of all pairs. + Useful for streaming or incremental data processing scenarios. + Args: - new_entities: New entities to check - existing_entities: Existing entities to compare against - **options: Detection options + new_entities: List of new entity dictionaries to check for duplicates + existing_entities: List of existing entity dictionaries to compare against + threshold: Minimum similarity threshold (overrides instance default) + **options: Additional detection options Returns: - List of duplicate candidates + List of DuplicateCandidate objects representing duplicates between + new and existing entities, sorted by confidence (highest first). + + Example: + >>> new_entities = [{"id": "3", "name": "Apple Corp"}] + >>> existing = [{"id": "1", "name": "Apple Inc."}] + >>> candidates = detector.incremental_detect(new_entities, existing) + >>> # Returns candidates if Apple Corp and Apple Inc. are duplicates """ - candidates = [] - threshold = options.get("threshold", self.similarity_threshold) + detection_threshold = threshold if threshold is not None else self.similarity_threshold + self.logger.info( + f"Incremental detection: {len(new_entities)} new entities vs " + f"{len(existing_entities)} existing entities" + ) + + candidates = [] + + # Compare each new entity with all existing entities for new_entity in new_entities: for existing_entity in existing_entities: + # Calculate similarity similarity = self.similarity_calculator.calculate_similarity( new_entity, existing_entity ) - if similarity.score >= threshold: + # Check if above threshold + if similarity.score >= detection_threshold: candidate = self._create_duplicate_candidate( new_entity, existing_entity, similarity.score ) + # Filter by confidence threshold if candidate.confidence >= self.confidence_threshold: candidates.append(candidate) + # Sort by confidence (highest first) + candidates.sort(key=lambda c: c.confidence, reverse=True) + + self.logger.info( + f"Incremental detection found {len(candidates)} duplicate candidate(s)" + ) + return candidates def _create_duplicate_candidate( @@ -194,36 +356,62 @@ class DuplicateDetector: entity2: Dict[str, Any], similarity_score: float ) -> DuplicateCandidate: - """Create duplicate candidate from similarity result.""" + """ + Create duplicate candidate from similarity result. + + This method builds a DuplicateCandidate object by analyzing the similarity + score and additional factors (name match, property matches, type match) + to calculate a confidence score. + + Confidence Calculation: + - Base: similarity_score + - +0.1: Exact name match + - +0.05 per matching property value + - +0.05: Same entity type + - Capped at 1.0 + + Args: + entity1: First entity dictionary + entity2: Second entity dictionary + similarity_score: Base similarity score from similarity calculator + + Returns: + DuplicateCandidate object with calculated confidence and reasons + """ reasons = [] confidence = similarity_score - # Check name similarity - name1 = entity1.get("name", "").lower() - name2 = entity2.get("name", "").lower() - if name1 == name2: + # Check for exact name match (strong indicator) + name1 = entity1.get("name", "").lower().strip() + name2 = entity2.get("name", "").lower().strip() + if name1 == name2 and name1: # Non-empty exact match reasons.append("exact_name_match") confidence += 0.1 - # Check property matches + # Check property value matches props1 = entity1.get("properties", {}) props2 = entity2.get("properties", {}) common_props = set(props1.keys()) & set(props2.keys()) if common_props: + # Count properties with matching values prop_matches = sum( 1 for prop in common_props if props1.get(prop) == props2.get(prop) ) if prop_matches > 0: reasons.append(f"{prop_matches}_property_matches") + # Boost confidence for each matching property confidence += 0.05 * prop_matches - # Check type match - if entity1.get("type") == entity2.get("type"): + # Check entity type match + entity_type1 = entity1.get("type") + entity_type2 = entity2.get("type") + if entity_type1 and entity_type2 and entity_type1 == entity_type2: reasons.append("same_type") confidence += 0.05 + # Cap confidence at 1.0 confidence = min(1.0, confidence) return DuplicateCandidate( @@ -231,7 +419,12 @@ class DuplicateDetector: entity2=entity2, similarity_score=similarity_score, confidence=confidence, - reasons=reasons + reasons=reasons, + metadata={ + "name_match": name1 == name2, + "common_properties": len(common_props), + "type_match": entity_type1 == entity_type2 + } ) def _build_duplicate_groups( diff --git a/semantica/deduplication/entity_merger.py b/semantica/deduplication/entity_merger.py index a213b32a..78402467 100644 --- a/semantica/deduplication/entity_merger.py +++ b/semantica/deduplication/entity_merger.py @@ -1,8 +1,24 @@ """ -Entity Merger for Semantica framework. +Entity Merger Module -Performs semantic deduplication to merge semantically similar -entities and maintain graph cleanliness. +This module provides entity merging capabilities for the Semantica framework, +performing semantic deduplication to merge semantically similar entities and +maintain knowledge graph cleanliness. + +Key Features: + - Merge duplicate entities using configurable strategies + - Preserve provenance information during merges + - Incremental merging of new entities with existing ones + - Conflict resolution for property and relationship merging + - Merge history tracking and quality validation + +Example Usage: + >>> from semantica.deduplication import EntityMerger + >>> merger = EntityMerger(preserve_provenance=True) + >>> operations = merger.merge_duplicates(entities) + +Author: Semantica Contributors +License: MIT """ from typing import Any, Dict, List, Optional @@ -27,26 +43,66 @@ class MergeOperation: class EntityMerger: """ - Entity merging engine. + Entity merging engine for knowledge graphs. - • Calculates semantic similarity - • Detects and groups duplicates - • Merges entities with conflict resolution - • Merges properties and relationships - • Preserves provenance during merge + This class provides comprehensive entity merging capabilities, detecting duplicates, + applying merge strategies, resolving conflicts, and preserving provenance information. + + Features: + - Automatic duplicate detection and grouping + - Configurable merge strategies (keep_first, keep_most_complete, etc.) + - Property and relationship merging with conflict resolution + - Provenance preservation (tracks merged entities) + - Incremental merging for new entities + - Merge history tracking and quality validation + + Example Usage: + >>> merger = EntityMerger(preserve_provenance=True) + >>> operations = merger.merge_duplicates(entities, strategy=MergeStrategy.KEEP_MOST_COMPLETE) + >>> history = merger.get_merge_history() """ - def __init__(self, config=None, **kwargs): - """Initialize entity merger.""" + def __init__( + self, + preserve_provenance: bool = True, + config: Optional[Dict[str, Any]] = None, + **kwargs + ): + """ + Initialize entity merger. + + Sets up the merger with duplicate detector and merge strategy manager, + configured according to the provided options. + + Args: + preserve_provenance: Whether to preserve provenance information in merged entities + (default: True). When True, merged entities will contain + metadata about which entities were merged. + config: Configuration dictionary (merged with kwargs) + **kwargs: Additional configuration options: + - detector: Configuration for DuplicateDetector + - strategy: Configuration for MergeStrategyManager + """ self.logger = get_logger("entity_merger") + + # Merge configuration self.config = config or {} self.config.update(kwargs) - self.duplicate_detector = DuplicateDetector(**config.get("detector", {})) - self.merge_strategy_manager = MergeStrategyManager(**config.get("strategy", {})) + # Initialize components + detector_config = self.config.get("detector", {}) + strategy_config = self.config.get("strategy", {}) + self.duplicate_detector = DuplicateDetector(**detector_config) + self.merge_strategy_manager = MergeStrategyManager(**strategy_config) + + # Merge history tracking self.merge_history: List[MergeOperation] = [] - self.preserve_provenance = config.get("preserve_provenance", True) + self.preserve_provenance = preserve_provenance + + self.logger.debug( + f"Entity merger initialized (preserve_provenance: {preserve_provenance})" + ) def merge_duplicates( self, @@ -55,55 +111,105 @@ class EntityMerger: **options ) -> List[MergeOperation]: """ - Merge duplicate entities. + Merge duplicate entities from a list. + + This method detects duplicate groups, merges each group using the specified + strategy, and returns a list of merge operations. Provenance information + is preserved if enabled. + + Process: + 1. Detect duplicate groups using similarity thresholds + 2. For each group with 2+ entities: + - Apply merge strategy to combine entities + - Resolve property and relationship conflicts + - Add provenance information (if enabled) + 3. Track merge operations in history Args: - entities: List of entities to process - strategy: Merge strategy - **options: Merge options - + entities: List of entity dictionaries to merge. Entities should have + at least "id" or "name" fields. + strategy: Merge strategy to use (default: strategy manager's default). + Options: KEEP_FIRST, KEEP_LAST, KEEP_MOST_COMPLETE, + KEEP_HIGHEST_CONFIDENCE, MERGE_ALL + **options: Additional merge options passed to duplicate detector and + merge strategy manager: + - threshold: Similarity threshold for duplicate detection + - preserve_relationships: Whether to preserve all relationships + Returns: - List of merge operations + List of MergeOperation objects, each containing: + - source_entities: Original entities that were merged + - merged_entity: Resulting merged entity + - merge_result: Detailed merge result with conflicts + - metadata: Group confidence and similarity scores + + Example: + >>> entities = [ + ... {"id": "1", "name": "Apple Inc.", "type": "Company"}, + ... {"id": "2", "name": "Apple", "type": "Company"} + ... ] + >>> operations = merger.merge_duplicates( + ... entities, + ... strategy=MergeStrategy.KEEP_MOST_COMPLETE + ... ) + >>> merged = operations[0].merged_entity """ - # Detect duplicate groups + self.logger.info(f"Merging duplicates from {len(entities)} entities") + + # Detect duplicate groups using similarity duplicate_groups = self.duplicate_detector.detect_duplicate_groups( entities, **options ) + self.logger.debug(f"Found {len(duplicate_groups)} duplicate group(s)") + merge_operations = [] + # Merge each duplicate group for group in duplicate_groups: + # Skip groups with less than 2 entities (not duplicates) if len(group.entities) < 2: continue - # Merge group + self.logger.debug( + f"Merging group of {len(group.entities)} entities " + f"(confidence: {group.confidence:.2f})" + ) + + # Apply merge strategy to combine entities merge_result = self.merge_strategy_manager.merge_entities( group.entities, strategy=strategy, **options ) - # Preserve provenance + # Add provenance information if enabled if self.preserve_provenance: merge_result.merged_entity = self._add_provenance( merge_result.merged_entity, group.entities ) + # Create merge operation record operation = MergeOperation( source_entities=group.entities, merged_entity=merge_result.merged_entity, merge_result=merge_result, metadata={ "group_confidence": group.confidence, - "similarity_scores": group.similarity_scores + "similarity_scores": group.similarity_scores, + "strategy": strategy.value if strategy else "default" } ) merge_operations.append(operation) self.merge_history.append(operation) + self.logger.info( + f"Completed merging: {len(merge_operations)} merge operation(s) performed" + ) + return merge_operations def merge_entity_group( @@ -115,38 +221,79 @@ class EntityMerger: """ Merge a specific group of entities. + This method merges a pre-determined group of entities (typically from + a duplicate group) using the specified merge strategy. Unlike + merge_duplicates(), this method does not perform duplicate detection. + Args: - entities: Entities to merge - strategy: Merge strategy - **options: Merge options - + entities: List of entity dictionaries to merge (must have at least 2) + strategy: Merge strategy to use (default: strategy manager's default) + **options: Additional merge options passed to merge strategy manager + Returns: - MergeOperation result + MergeOperation object containing: + - source_entities: Original entities that were merged + - merged_entity: Resulting merged entity + - merge_result: Detailed merge result with conflicts + - metadata: Merge metadata + + Raises: + ValidationError: If less than 2 entities provided + + Example: + >>> entities = [ + ... {"id": "1", "name": "Apple Inc."}, + ... {"id": "2", "name": "Apple"} + ... ] + >>> operation = merger.merge_entity_group( + ... entities, + ... strategy=MergeStrategy.KEEP_MOST_COMPLETE + ... ) + >>> merged = operation.merged_entity """ if len(entities) < 2: - raise ValidationError("Need at least 2 entities to merge") + raise ValidationError( + f"Need at least 2 entities to merge, got {len(entities)}" + ) + self.logger.debug( + f"Merging group of {len(entities)} entities " + f"(strategy: {strategy.value if strategy else 'default'})" + ) + + # Apply merge strategy merge_result = self.merge_strategy_manager.merge_entities( entities, strategy=strategy, **options ) - # Preserve provenance + # Add provenance information if enabled if self.preserve_provenance: merge_result.merged_entity = self._add_provenance( merge_result.merged_entity, entities ) + # Create merge operation record operation = MergeOperation( source_entities=entities, merged_entity=merge_result.merged_entity, - merge_result=merge_result + merge_result=merge_result, + metadata={ + "strategy": strategy.value if strategy else "default", + "conflicts": len(merge_result.conflicts) + } ) + # Track in history self.merge_history.append(operation) + self.logger.info( + f"Successfully merged {len(entities)} entities " + f"({len(merge_result.conflicts)} conflict(s))" + ) + return operation def incremental_merge( @@ -158,42 +305,82 @@ class EntityMerger: """ Incremental merge of new entities with existing ones. + This method efficiently merges new entities with an existing set by detecting + duplicates between them and merging matching pairs. Useful for streaming + or incremental data processing where new entities are added over time. + + Process: + 1. Detect duplicates between new and existing entities + 2. For each duplicate pair: + - Merge the two entities + - Track which entities have been processed + - Avoid duplicate merges + 3. Return list of merge operations + Args: - new_entities: New entities to merge - existing_entities: Existing entities - **options: Merge options - + new_entities: List of new entity dictionaries to merge + existing_entities: List of existing entity dictionaries to merge with + **options: Additional merge options: + - threshold: Similarity threshold for duplicate detection + - strategy: Merge strategy to use + Returns: - List of merge operations + List of MergeOperation objects, one for each merged pair. + Entities that don't have duplicates remain unmerged. + + Example: + >>> new = [{"id": "3", "name": "Apple Corp"}] + >>> existing = [{"id": "1", "name": "Apple Inc."}] + >>> operations = merger.incremental_merge(new, existing) + >>> # Returns merge operation if Apple Corp and Apple Inc. are duplicates """ - # Detect duplicates between new and existing + self.logger.info( + f"Incremental merge: {len(new_entities)} new entities vs " + f"{len(existing_entities)} existing entities" + ) + + # Detect duplicates between new and existing entities candidates = self.duplicate_detector.incremental_detect( new_entities, existing_entities, **options ) - merge_operations = [] - processed_new = set() - processed_existing = set() + self.logger.debug(f"Found {len(candidates)} duplicate candidate(s)") + merge_operations = [] + processed_new = set() # Track processed new entity IDs + processed_existing = set() # Track processed existing entity IDs + + # Merge each duplicate pair for candidate in candidates: new_entity_id = candidate.entity1.get("id") or id(candidate.entity1) existing_entity_id = candidate.entity2.get("id") or id(candidate.entity2) + # Skip if either entity already processed (avoid duplicate merges) if new_entity_id in processed_new or existing_entity_id in processed_existing: + self.logger.debug( + f"Skipping merge: entity already processed " + f"(new: {new_entity_id}, existing: {existing_entity_id})" + ) continue - # Merge the pair + # Merge the duplicate pair operation = self.merge_entity_group( [candidate.entity1, candidate.entity2], **options ) merge_operations.append(operation) + + # Mark entities as processed processed_new.add(new_entity_id) processed_existing.add(existing_entity_id) + self.logger.info( + f"Incremental merge completed: {len(merge_operations)} merge operation(s)" + ) + return merge_operations def _add_provenance( @@ -201,11 +388,34 @@ class EntityMerger: merged_entity: Dict[str, Any], source_entities: List[Dict[str, Any]] ) -> Dict[str, Any]: - """Add provenance information to merged entity.""" - if "provenance" not in merged_entity.get("metadata", {}): - merged_entity.setdefault("metadata", {})["provenance"] = {} + """ + Add provenance information to merged entity. + + This method adds metadata about which entities were merged to create + the merged entity, preserving the history of the merge operation. + + Provenance Structure: + metadata.provenance: + - merged_from: List of source entity information (id, name, source) + - merge_count: Number of entities that were merged + + Args: + merged_entity: The merged entity dictionary to add provenance to + source_entities: List of source entities that were merged + + Returns: + Merged entity dictionary with provenance information added + """ + # Ensure metadata structure exists + if "metadata" not in merged_entity: + merged_entity["metadata"] = {} + + if "provenance" not in merged_entity["metadata"]: + merged_entity["metadata"]["provenance"] = {} provenance = merged_entity["metadata"]["provenance"] + + # Record source entities provenance["merged_from"] = [ { "id": e.get("id"), @@ -216,12 +426,52 @@ class EntityMerger: ] provenance["merge_count"] = len(source_entities) + self.logger.debug( + f"Added provenance for merge of {len(source_entities)} entity(ies)" + ) + return merged_entity def get_merge_history(self) -> List[MergeOperation]: - """Get merge operation history.""" - return self.merge_history + """ + Get merge operation history. + + Returns a list of all merge operations that have been performed by this + merger instance, in chronological order. + + Returns: + List of MergeOperation objects representing all merges performed. + Each operation contains source entities, merged entity, and metadata. + + Example: + >>> history = merger.get_merge_history() + >>> print(f"Total merges: {len(history)}") + >>> for op in history: + ... print(f"Merged {len(op.source_entities)} entities") + """ + return self.merge_history.copy() # Return copy to prevent external modification def validate_merge_quality(self, merge_operation: MergeOperation) -> Dict[str, Any]: - """Validate quality of merge operation.""" + """ + Validate quality of a merge operation. + + This method checks the quality of a merge operation by validating the + merged entity and checking for issues like missing required fields or + unresolved conflicts. + + Args: + merge_operation: MergeOperation object to validate + + Returns: + Dictionary containing validation results: + - valid: Whether merge is valid (bool) + - issues: List of validation issues found + - quality_score: Quality score (0.0 to 1.0) + + Example: + >>> operation = merger.merge_entity_group(entities) + >>> validation = merger.validate_merge_quality(operation) + >>> if not validation["valid"]: + ... print(f"Issues: {validation['issues']}") + """ return self.merge_strategy_manager.validate_merge(merge_operation.merge_result) diff --git a/semantica/deduplication/merge_strategy.py b/semantica/deduplication/merge_strategy.py index a0495d55..1def942d 100644 --- a/semantica/deduplication/merge_strategy.py +++ b/semantica/deduplication/merge_strategy.py @@ -1,8 +1,24 @@ """ -Merge Strategy Manager +Merge Strategy Manager Module -Manages different strategies for merging duplicate entities -including property merging rules and relationship preservation. +This module provides comprehensive merge strategy management for the Semantica +framework, handling different strategies for merging duplicate entities including +property merging rules, relationship preservation, and conflict resolution. + +Key Features: + - Multiple merge strategies (keep_first, keep_most_complete, etc.) + - Property-specific merge rules + - Custom conflict resolution functions + - Relationship preservation during merges + - Merge quality validation + +Example Usage: + >>> from semantica.deduplication import MergeStrategyManager, MergeStrategy + >>> manager = MergeStrategyManager() + >>> result = manager.merge_entities(entities, strategy=MergeStrategy.KEEP_MOST_COMPLETE) + +Author: Semantica Contributors +License: MIT """ from typing import Any, Dict, List, Optional, Callable @@ -46,25 +62,70 @@ class MergeResult: class MergeStrategyManager: """ - Merge strategy management engine. + Merge strategy management engine for entity merging. - • Manages property merging strategies - • Preserves relationships during merge - • Resolves merge conflicts - • Makes confidence-based merge decisions - • Validates merge quality - • Supports custom merge strategies + This class manages different strategies for merging duplicate entities, handling + property merging rules, relationship preservation, conflict resolution, and + merge quality validation. + + Features: + - Multiple merge strategies (keep_first, keep_most_complete, etc.) + - Property-specific merge rules with custom conflict resolution + - Relationship preservation during merges + - Automatic conflict detection and resolution + - Merge quality validation + - Support for custom merge strategies + + Example Usage: + >>> manager = MergeStrategyManager(default_strategy="keep_most_complete") + >>> manager.add_property_rule("name", MergeStrategy.KEEP_FIRST) + >>> result = manager.merge_entities(entities) """ - def __init__(self, config=None, **kwargs): - """Initialize merge strategy manager.""" + def __init__( + self, + default_strategy: str = "keep_most_complete", + config: Optional[Dict[str, Any]] = None, + **kwargs + ): + """ + Initialize merge strategy manager. + + Sets up the manager with a default merge strategy and empty property rules. + Property rules can be added later using add_property_rule(). + + Args: + default_strategy: Default merge strategy name (default: "keep_most_complete"). + Options: "keep_first", "keep_last", "keep_most_complete", + "keep_highest_confidence", "merge_all" + config: Configuration dictionary (merged with kwargs) + **kwargs: Additional configuration options + """ self.logger = get_logger("merge_strategy_manager") + + # Merge configuration self.config = config or {} self.config.update(kwargs) - self.default_strategy = MergeStrategy(config.get("default_strategy", "keep_most_complete")) + # Default merge strategy + strategy_name = self.config.get("default_strategy", default_strategy) + try: + self.default_strategy = MergeStrategy(strategy_name) + except ValueError: + self.logger.warning( + f"Invalid default strategy '{strategy_name}', using 'keep_most_complete'" + ) + self.default_strategy = MergeStrategy.KEEP_MOST_COMPLETE + + # Property-specific merge rules self.property_rules: Dict[str, PropertyMergeRule] = {} + + # Custom merge strategies (callable functions) self.custom_strategies: Dict[str, Callable] = {} + + self.logger.debug( + f"Merge strategy manager initialized (default: {self.default_strategy.value})" + ) def add_property_rule( self, diff --git a/semantica/deduplication/similarity_calculator.py b/semantica/deduplication/similarity_calculator.py index 6cafe969..c2d27038 100644 --- a/semantica/deduplication/similarity_calculator.py +++ b/semantica/deduplication/similarity_calculator.py @@ -1,8 +1,26 @@ """ -Similarity Calculator for Deduplication +Similarity Calculator Module -Calculates semantic similarity between entities to identify -potential duplicates using various similarity metrics. +This module provides comprehensive similarity calculation capabilities for the +Semantica framework, computing semantic similarity between entities using multiple +metrics including string similarity, property similarity, relationship similarity, +and embedding similarity. + +Key Features: + - Multi-factor similarity calculation (string, property, relationship, embedding) + - Multiple string similarity algorithms (Levenshtein, Jaro-Winkler, cosine) + - Weighted aggregation of similarity components + - Batch similarity calculation for efficiency + - Configurable similarity thresholds and weights + +Example Usage: + >>> from semantica.deduplication import SimilarityCalculator + >>> calculator = SimilarityCalculator() + >>> similarity = calculator.calculate_similarity(entity1, entity2) + >>> batch_results = calculator.batch_calculate_similarity(entities) + +Author: Semantica Contributors +License: MIT """ from typing import Any, Dict, List, Optional, Tuple @@ -25,26 +43,77 @@ class SimilarityResult: class SimilarityCalculator: """ - Similarity calculation engine. + Similarity calculation engine for entity comparison. - • Calculates semantic similarity using embeddings - • Uses string similarity metrics (Levenshtein, Jaro-Winkler) - • Compares property-based similarity - • Scores relationship-based similarity - • Aggregates multi-factor similarity + This class provides comprehensive similarity calculation using multiple factors: + string similarity, property similarity, relationship similarity, and embedding + similarity. Results are aggregated using configurable weights. + + Similarity Components: + - String similarity: Name/identifier comparison using various algorithms + - Property similarity: Comparison of entity properties + - Relationship similarity: Comparison of entity relationships + - Embedding similarity: Semantic similarity using vector embeddings + + Example Usage: + >>> calculator = SimilarityCalculator( + ... string_weight=0.4, + ... property_weight=0.3, + ... embedding_weight=0.3 + ... ) + >>> result = calculator.calculate_similarity(entity1, entity2) + >>> print(f"Similarity: {result.score:.2f}") """ - def __init__(self, config=None, **kwargs): - """Initialize similarity calculator.""" + def __init__( + self, + embedding_weight: float = 0.4, + string_weight: float = 0.3, + property_weight: float = 0.2, + relationship_weight: float = 0.1, + similarity_threshold: float = 0.7, + config: Optional[Dict[str, Any]] = None, + **kwargs + ): + """ + Initialize similarity calculator. + + Sets up the calculator with configurable weights for different similarity + components. Weights are normalized automatically if they don't sum to 1.0. + + Args: + embedding_weight: Weight for embedding similarity (default: 0.4) + string_weight: Weight for string similarity (default: 0.3) + property_weight: Weight for property similarity (default: 0.2) + relationship_weight: Weight for relationship similarity (default: 0.1) + similarity_threshold: Default similarity threshold for filtering (default: 0.7) + config: Configuration dictionary (merged with kwargs) + **kwargs: Additional configuration options + """ self.logger = get_logger("similarity_calculator") + + # Merge configuration self.config = config or {} self.config.update(kwargs) - self.embedding_weight = config.get("embedding_weight", 0.4) - self.string_weight = config.get("string_weight", 0.3) - self.property_weight = config.get("property_weight", 0.2) - self.relationship_weight = config.get("relationship_weight", 0.1) - self.similarity_threshold = config.get("similarity_threshold", 0.7) + # Component weights (used for weighted aggregation) + self.embedding_weight = embedding_weight + self.string_weight = string_weight + self.property_weight = property_weight + self.relationship_weight = relationship_weight + self.similarity_threshold = similarity_threshold + + # Validate weights sum to approximately 1.0 + total_weight = ( + self.embedding_weight + self.string_weight + + self.property_weight + self.relationship_weight + ) + if abs(total_weight - 1.0) > 0.01: + self.logger.debug( + f"Weights sum to {total_weight:.2f}, will be normalized during calculation" + ) + + self.logger.debug("Similarity calculator initialized") def calculate_similarity( self, @@ -55,13 +124,36 @@ class SimilarityCalculator: """ Calculate overall similarity between two entities. + This method computes a comprehensive similarity score by combining multiple + similarity factors: string similarity, property similarity, relationship + similarity, and embedding similarity (if available). Results are aggregated + using configurable weights. + + Similarity Components: + - String: Name/identifier similarity (Levenshtein, Jaro-Winkler, etc.) + - Property: Property value similarity + - Relationship: Relationship overlap (Jaccard similarity) + - Embedding: Cosine similarity of embeddings (if available) + Args: - entity1: First entity dictionary - entity2: Second entity dictionary - **options: Calculation options + entity1: First entity dictionary. Should have "name" field and optionally + "properties", "relationships", and "embedding" fields. + entity2: Second entity dictionary (same structure as entity1) + **options: Additional calculation options (currently unused) Returns: - SimilarityResult with overall score + SimilarityResult object containing: + - score: Overall similarity score (0.0 to 1.0) + - method: Calculation method used ("multi_factor") + - components: Dict of individual component scores + - metadata: Weights used for aggregation + + Example: + >>> entity1 = {"name": "Apple Inc.", "type": "Company"} + >>> entity2 = {"name": "Apple", "type": "Company"} + >>> result = calculator.calculate_similarity(entity1, entity2) + >>> print(f"Similarity: {result.score:.2f}") + >>> print(f"Components: {result.components}") """ components = {}