From d2d6adafdbdec2c9057d79fbfeca58ac378dc0c5 Mon Sep 17 00:00:00 2001 From: KaifAhmad1 Date: Mon, 29 Dec 2025 13:11:22 +0530 Subject: [PATCH] Add comprehensive progress tracking with Jupyter/Colab support - Enhanced progress tracker with automatic Jupyter/Colab detection - Added detailed progress tracking to all deduplication modules - Added detailed progress tracking to all semantic_extract modules - Progress tracker now always enabled automatically - Shows remaining items, percentages, ETA, and processing rates - Works in both Jupyter notebooks and Google Colab - Dynamic update intervals based on dataset size - Improved display handling for Colab compatibility --- .../02_Threat_Intelligence_Hybrid_RAG.ipynb | 2 +- semantica/conflicts/conflict_analyzer.py | 88 ++++- semantica/conflicts/conflict_detector.py | 330 +++++++++++++++--- semantica/conflicts/conflict_resolver.py | 27 +- semantica/deduplication/cluster_builder.py | 147 ++++++-- semantica/deduplication/duplicate_detector.py | 92 +++-- semantica/deduplication/entity_merger.py | 34 +- semantica/deduplication/merge_strategy.py | 52 ++- .../deduplication/similarity_calculator.py | 37 +- .../semantic_extract/coreference_resolver.py | 61 +++- semantica/semantic_extract/event_detector.py | 74 +++- semantica/semantic_extract/ner_extractor.py | 25 +- .../semantic_extract/relation_extractor.py | 25 +- .../semantic_extract/semantic_analyzer.py | 107 ++++-- .../semantic_network_extractor.py | 86 ++++- .../semantic_extract/triplet_extractor.py | 24 +- semantica/utils/progress_tracker.py | 162 ++++++++- 17 files changed, 1148 insertions(+), 225 deletions(-) diff --git a/cookbook/use_cases/cybersecurity/02_Threat_Intelligence_Hybrid_RAG.ipynb b/cookbook/use_cases/cybersecurity/02_Threat_Intelligence_Hybrid_RAG.ipynb index 03efa28d..6c49684c 100644 --- a/cookbook/use_cases/cybersecurity/02_Threat_Intelligence_Hybrid_RAG.ipynb +++ b/cookbook/use_cases/cybersecurity/02_Threat_Intelligence_Hybrid_RAG.ipynb @@ -140,7 +140,7 @@ { "data": { "text/html": [ - "

🧠 Semantica - 📊 Current Progress

StatusActionModuleSubmoduleProgressETARateTime
❌Semantica is parsing🔍 parseDocumentParser---0.00s
❌Semantica is parsing🔍 parseDocumentParser---0.00s
❌Semantica is parsing🔍 parseDocumentParser---0.00s
❌Semantica is parsing🔍 parseDocumentParser---0.00s
❌Semantica is parsing🔍 parseDocumentParser---0.00s
✅Semantica is normalizing🔧 normalizeTextNormalizer100.0%--0.00s
✅Semantica is extracting🎯 semantic_extractNERExtractor100.0%--0.56s
✅Semantica is extracting🎯 semantic_extractRelationExtractor100.0%--0.46s
🔄Semantica is deduplicating🔄 deduplicationDuplicateDetector---634.12s
🔄Semantica is deduplicating🔄 deduplicationSimilarityCalculator---0.01s
" + "

🧠 Semantica - 📊 Current Progress

StatusActionModuleSubmoduleProgressETARateTime
❌Semantica is parsing🔍 parseDocumentParser---0.01s
❌Semantica is parsing🔍 parseDocumentParser---0.00s
❌Semantica is parsing🔍 parseDocumentParser---0.00s
❌Semantica is parsing🔍 parseDocumentParser---0.01s
❌Semantica is parsing🔍 parseDocumentParser---0.00s
✅Semantica is normalizing🔧 normalizeTextNormalizer100.0%--0.00s
✅Semantica is extracting🎯 semantic_extractNERExtractor100.0%--1.16s
✅Semantica is extracting🎯 semantic_extractRelationExtractor100.0%--0.67s
🔄Semantica is deduplicating🔄 deduplicationDuplicateDetector---723.21s
🔄Semantica is deduplicating🔄 deduplicationSimilarityCalculator---0.01s
" ], "text/plain": [ "" diff --git a/semantica/conflicts/conflict_analyzer.py b/semantica/conflicts/conflict_analyzer.py index 8e0de3c4..07792ce4 100644 --- a/semantica/conflicts/conflict_analyzer.py +++ b/semantica/conflicts/conflict_analyzer.py @@ -110,7 +110,10 @@ class ConflictAnalyzer: self.logger = get_logger("conflict_analyzer") self.config = config or {} self.config.update(kwargs) + # Initialize progress tracker and ensure it's enabled self.progress_tracker = get_progress_tracker() + if not self.progress_tracker.enabled: + self.progress_tracker.enabled = True def analyze_conflicts(self, conflicts: List[Conflict]) -> Dict[str, Any]: """ @@ -131,18 +134,85 @@ class ConflictAnalyzer: ) try: - self.progress_tracker.update_tracking( - tracking_id, message="Analyzing conflict patterns..." + total_steps = 6 # by_type, by_severity, by_source, by_entity, by_property, patterns, recommendations + current_step = 0 + + # Step 1: Analyze by type + current_step += 1 + remaining_steps = total_steps - current_step + self.progress_tracker.update_progress( + tracking_id, + processed=current_step, + total=total_steps, + message=f"Analyzing by type... ({current_step}/{total_steps}, remaining: {remaining_steps} steps)" ) + by_type = self._analyze_by_type(conflicts) + + # Step 2: Analyze by severity + current_step += 1 + remaining_steps = total_steps - current_step + self.progress_tracker.update_progress( + tracking_id, + processed=current_step, + total=total_steps, + message=f"Analyzing by severity... ({current_step}/{total_steps}, remaining: {remaining_steps} steps)" + ) + by_severity = self._analyze_by_severity(conflicts) + + # Step 3: Analyze by source + current_step += 1 + remaining_steps = total_steps - current_step + self.progress_tracker.update_progress( + tracking_id, + processed=current_step, + total=total_steps, + message=f"Analyzing by source... ({current_step}/{total_steps}, remaining: {remaining_steps} steps)" + ) + by_source = self._analyze_by_source(conflicts) + + # Step 4: Analyze by entity + current_step += 1 + remaining_steps = total_steps - current_step + self.progress_tracker.update_progress( + tracking_id, + processed=current_step, + total=total_steps, + message=f"Analyzing by entity... ({current_step}/{total_steps}, remaining: {remaining_steps} steps)" + ) + by_entity = self._analyze_by_entity(conflicts) + + # Step 5: Analyze by property + current_step += 1 + remaining_steps = total_steps - current_step + self.progress_tracker.update_progress( + tracking_id, + processed=current_step, + total=total_steps, + message=f"Analyzing by property... ({current_step}/{total_steps}, remaining: {remaining_steps} steps)" + ) + by_property = self._analyze_by_property(conflicts) + + # Step 6: Identify patterns and generate recommendations + current_step += 1 + remaining_steps = total_steps - current_step + self.progress_tracker.update_progress( + tracking_id, + processed=current_step, + total=total_steps, + message=f"Identifying patterns and generating recommendations... ({current_step}/{total_steps}, remaining: {remaining_steps} steps)" + ) + patterns = self._identify_patterns(conflicts) + recommendations = self._generate_recommendations(conflicts) + analysis = { "total_conflicts": len(conflicts), - "by_type": self._analyze_by_type(conflicts), - "by_severity": self._analyze_by_severity(conflicts), - "by_source": self._analyze_by_source(conflicts), - "by_entity": self._analyze_by_entity(conflicts), - "by_property": self._analyze_by_property(conflicts), - "patterns": self._identify_patterns(conflicts), - "recommendations": self._generate_recommendations(conflicts), + "by_type": by_type, + "by_severity": by_severity, + "by_source": by_source, + "by_entity": by_entity, + "by_property": by_property, + "patterns": patterns, + "recommendations": recommendations, } self.progress_tracker.stop_tracking( diff --git a/semantica/conflicts/conflict_detector.py b/semantica/conflicts/conflict_detector.py index dfa826ba..2458441f 100644 --- a/semantica/conflicts/conflict_detector.py +++ b/semantica/conflicts/conflict_detector.py @@ -217,7 +217,19 @@ class ConflictDetector: ) total_entities = len(entities) - update_interval = max(1, total_entities // 20) # Update every 5% + if total_entities <= 10: + update_interval = 1 # Update every item for small datasets + else: + update_interval = max(1, min(10, total_entities // 100)) + + # Initial progress update - ALWAYS show this + remaining = total_entities + self.progress_tracker.update_progress( + tracking_id, + processed=0, + total=total_entities, + message=f"Analyzing entities... 0/{total_entities} (remaining: {remaining})" + ) for i, entity in enumerate(entities): entity_id = entity.get("id") or entity.get("entity_id") @@ -231,18 +243,37 @@ class ConflictDetector: entity_groups[entity_id] = [] entity_groups[entity_id].append(entity) - # Update progress periodically - if (i + 1) % update_interval == 0 or (i + 1) == total_entities: + remaining = total_entities - (i + 1) + # Update progress: always update for small datasets, or at intervals for large ones + should_update = ( + (i + 1) % update_interval == 0 or + (i + 1) == total_entities or + i == 0 or + total_entities <= 10 # Always update for small datasets + ) + if should_update: self.progress_tracker.update_progress( tracking_id, processed=i + 1, total=total_entities, - message=f"Analyzing entities... {i + 1}/{total_entities}" + message=f"Analyzing entities... {i + 1}/{total_entities} (remaining: {remaining})" ) # Check each entity group for conflicts total_groups = len(entity_groups) - group_update_interval = max(1, total_groups // 20) # Update every 5% + if total_groups <= 10: + group_update_interval = 1 # Update every item for small datasets + else: + group_update_interval = max(1, min(10, total_groups // 100)) + + # Initial progress update for group checking + remaining_groups = total_groups + self.progress_tracker.update_progress( + tracking_id, + processed=0, + total=total_groups, + message=f"Checking entity groups for conflicts... 0/{total_groups} (remaining: {remaining_groups})" + ) for j, (entity_id, entity_list) in enumerate(entity_groups.items()): if len(entity_list) < 2: @@ -302,13 +333,20 @@ class ConflictDetector: f"has conflicting values: {unique_values}" ) - # Update progress periodically for group checking - if (j + 1) % group_update_interval == 0 or (j + 1) == total_groups: + remaining_groups = total_groups - (j + 1) + # Update progress: always update for small datasets, or at intervals for large ones + should_update = ( + (j + 1) % group_update_interval == 0 or + (j + 1) == total_groups or + j == 0 or + total_groups <= 10 # Always update for small datasets + ) + if should_update: self.progress_tracker.update_progress( tracking_id, processed=j + 1, total=total_groups, - message=f"Checking entity groups for conflicts... {j + 1}/{total_groups}" + message=f"Checking entity groups for conflicts... {j + 1}/{total_groups} (remaining: {remaining_groups})" ) self.progress_tracker.stop_tracking( @@ -365,7 +403,19 @@ class ConflictDetector: # Group relationships by ID rel_groups: Dict[str, List[Dict[str, Any]]] = {} total_rels = len(relationships) - update_interval = max(1, total_rels // 20) # Update every 5% + if total_rels <= 10: + update_interval = 1 # Update every item for small datasets + else: + update_interval = max(1, min(10, total_rels // 100)) + + # Initial progress update + remaining = total_rels + self.progress_tracker.update_progress( + tracking_id, + processed=0, + total=total_rels, + message=f"Grouping relationships... 0/{total_rels} (remaining: {remaining})" + ) for i, rel in enumerate(relationships): rel_id = ( @@ -377,18 +427,37 @@ class ConflictDetector: rel_groups[rel_id] = [] rel_groups[rel_id].append(rel) - # Update progress periodically - if (i + 1) % update_interval == 0 or (i + 1) == total_rels: + remaining = total_rels - (i + 1) + # Update progress: always update for small datasets, or at intervals for large ones + should_update = ( + (i + 1) % update_interval == 0 or + (i + 1) == total_rels or + i == 0 or + total_rels <= 10 # Always update for small datasets + ) + if should_update: self.progress_tracker.update_progress( tracking_id, processed=i + 1, total=total_rels, - message=f"Grouping relationships... {i + 1}/{total_rels}" + message=f"Grouping relationships... {i + 1}/{total_rels} (remaining: {remaining})" ) # Check for conflicts total_groups = len(rel_groups) - group_update_interval = max(1, total_groups // 20) # Update every 5% + if total_groups <= 10: + group_update_interval = 1 # Update every item for small datasets + else: + group_update_interval = max(1, min(10, total_groups // 100)) + + # Initial progress update for group checking + remaining_groups = total_groups + self.progress_tracker.update_progress( + tracking_id, + processed=0, + total=total_groups, + message=f"Checking relationship groups... 0/{total_groups} (remaining: {remaining_groups})" + ) for j, (rel_id, rel_list) in enumerate(rel_groups.items()): if len(rel_list) < 2: @@ -413,13 +482,20 @@ class ConflictDetector: conflicts.append(conflict) self.detected_conflicts[conflict.conflict_id] = conflict - # Update progress periodically - if (j + 1) % group_update_interval == 0 or (j + 1) == total_groups: + remaining_groups = total_groups - (j + 1) + # Update progress: always update for small datasets, or at intervals for large ones + should_update = ( + (j + 1) % group_update_interval == 0 or + (j + 1) == total_groups or + j == 0 or + total_groups <= 10 # Always update for small datasets + ) + if should_update: self.progress_tracker.update_progress( tracking_id, processed=j + 1, total=total_groups, - message=f"Checking relationship groups... {j + 1}/{total_groups}" + message=f"Checking relationship groups... {j + 1}/{total_groups} (remaining: {remaining_groups})" ) self.progress_tracker.stop_tracking( @@ -472,20 +548,39 @@ class ConflictDetector: ) total_fields = len(fields_to_check) - update_interval = max(1, total_fields // 20) # Update every 5% + if total_fields <= 10: + update_interval = 1 # Update every item for small datasets + else: + update_interval = max(1, min(10, total_fields // 100)) + + # Initial progress update + remaining = total_fields + self.progress_tracker.update_progress( + tracking_id, + processed=0, + total=total_fields, + message=f"Checking fields for conflicts... 0/{total_fields} (remaining: {remaining})" + ) # Check each field for i, field_name in enumerate(fields_to_check): conflicts = self.detect_value_conflicts(entities, field_name, entity_type) all_conflicts.extend(conflicts) - # Update progress periodically - if (i + 1) % update_interval == 0 or (i + 1) == total_fields: + remaining = total_fields - (i + 1) + # Update progress: always update for small datasets, or at intervals for large ones + should_update = ( + (i + 1) % update_interval == 0 or + (i + 1) == total_fields or + i == 0 or + total_fields <= 10 # Always update for small datasets + ) + if should_update: self.progress_tracker.update_progress( tracking_id, processed=i + 1, total=total_fields, - message=f"Checking fields for conflicts... {i + 1}/{total_fields}" + message=f"Checking fields for conflicts... {i + 1}/{total_fields} (remaining: {remaining})" ) self.progress_tracker.stop_tracking( @@ -626,7 +721,19 @@ class ConflictDetector: # Group entities by ID entity_groups: Dict[str, List[Dict[str, Any]]] = {} total_entities = len(entities) - update_interval = max(1, total_entities // 20) # Update every 5% + if total_entities <= 10: + update_interval = 1 # Update every item for small datasets + else: + update_interval = max(1, min(10, total_entities // 100)) + + # Initial progress update + remaining = total_entities + self.progress_tracker.update_progress( + tracking_id, + processed=0, + total=total_entities, + message=f"Grouping entities... 0/{total_entities} (remaining: {remaining})" + ) for i, entity in enumerate(entities): entity_id = entity.get("id") or entity.get("entity_id") @@ -637,18 +744,37 @@ class ConflictDetector: entity_groups[entity_id] = [] entity_groups[entity_id].append(entity) - # Update progress periodically - if (i + 1) % update_interval == 0 or (i + 1) == total_entities: + remaining = total_entities - (i + 1) + # Update progress: always update for small datasets, or at intervals for large ones + should_update = ( + (i + 1) % update_interval == 0 or + (i + 1) == total_entities or + i == 0 or + total_entities <= 10 # Always update for small datasets + ) + if should_update: self.progress_tracker.update_progress( tracking_id, processed=i + 1, total=total_entities, - message=f"Grouping entities... {i + 1}/{total_entities}" + message=f"Grouping entities... {i + 1}/{total_entities} (remaining: {remaining})" ) # Check each entity group for type conflicts total_groups = len(entity_groups) - group_update_interval = max(1, total_groups // 20) # Update every 5% + if total_groups <= 10: + group_update_interval = 1 # Update every item for small datasets + else: + group_update_interval = max(1, min(10, total_groups // 100)) + + # Initial progress update for group checking + remaining_groups = total_groups + self.progress_tracker.update_progress( + tracking_id, + processed=0, + total=total_groups, + message=f"Checking entity groups for type conflicts... 0/{total_groups} (remaining: {remaining_groups})" + ) for j, (entity_id, entity_list) in enumerate(entity_groups.items()): if len(entity_list) < 2: @@ -706,13 +832,20 @@ class ConflictDetector: f"{unique_types}" ) - # Update progress periodically for group checking - if (j + 1) % group_update_interval == 0 or (j + 1) == total_groups: + remaining_groups = total_groups - (j + 1) + # Update progress: always update for small datasets, or at intervals for large ones + should_update = ( + (j + 1) % group_update_interval == 0 or + (j + 1) == total_groups or + j == 0 or + total_groups <= 10 # Always update for small datasets + ) + if should_update: self.progress_tracker.update_progress( tracking_id, processed=j + 1, total=total_groups, - message=f"Checking entity groups for type conflicts... {j + 1}/{total_groups}" + message=f"Checking entity groups for type conflicts... {j + 1}/{total_groups} (remaining: {remaining_groups})" ) self.progress_tracker.stop_tracking( @@ -765,7 +898,19 @@ class ConflictDetector: # Group entities by ID entity_groups: Dict[str, List[Dict[str, Any]]] = {} total_entities = len(entities) - update_interval = max(1, total_entities // 20) # Update every 5% + if total_entities <= 10: + update_interval = 1 # Update every item for small datasets + else: + update_interval = max(1, min(10, total_entities // 100)) + + # Initial progress update + remaining = total_entities + self.progress_tracker.update_progress( + tracking_id, + processed=0, + total=total_entities, + message=f"Grouping entities... 0/{total_entities} (remaining: {remaining})" + ) for i, entity in enumerate(entities): entity_id = entity.get("id") or entity.get("entity_id") @@ -776,18 +921,37 @@ class ConflictDetector: entity_groups[entity_id] = [] entity_groups[entity_id].append(entity) - # Update progress periodically - if (i + 1) % update_interval == 0 or (i + 1) == total_entities: + remaining = total_entities - (i + 1) + # Update progress: always update for small datasets, or at intervals for large ones + should_update = ( + (i + 1) % update_interval == 0 or + (i + 1) == total_entities or + i == 0 or + total_entities <= 10 # Always update for small datasets + ) + if should_update: self.progress_tracker.update_progress( tracking_id, processed=i + 1, total=total_entities, - message=f"Grouping entities... {i + 1}/{total_entities}" + message=f"Grouping entities... {i + 1}/{total_entities} (remaining: {remaining})" ) # Check each entity group for temporal conflicts total_groups = len(entity_groups) - group_update_interval = max(1, total_groups // 20) # Update every 5% + if total_groups <= 10: + group_update_interval = 1 # Update every item for small datasets + else: + group_update_interval = max(1, min(10, total_groups // 100)) + + # Initial progress update for group checking + remaining_groups = total_groups + self.progress_tracker.update_progress( + tracking_id, + processed=0, + total=total_groups, + message=f"Checking entity groups for temporal conflicts... 0/{total_groups} (remaining: {remaining_groups})" + ) for j, (entity_id, entity_list) in enumerate(entity_groups.items()): if len(entity_list) < 2: @@ -875,13 +1039,20 @@ class ConflictDetector: f"has conflicting values: {unique_values}" ) - # Update progress periodically for group checking - if (j + 1) % group_update_interval == 0 or (j + 1) == total_groups: + remaining_groups = total_groups - (j + 1) + # Update progress: always update for small datasets, or at intervals for large ones + should_update = ( + (j + 1) % group_update_interval == 0 or + (j + 1) == total_groups or + j == 0 or + total_groups <= 10 # Always update for small datasets + ) + if should_update: self.progress_tracker.update_progress( tracking_id, processed=j + 1, total=total_groups, - message=f"Checking entity groups for temporal conflicts... {j + 1}/{total_groups}" + message=f"Checking entity groups for temporal conflicts... {j + 1}/{total_groups} (remaining: {remaining_groups})" ) self.progress_tracker.stop_tracking( @@ -930,7 +1101,19 @@ class ConflictDetector: # Group entities by ID entity_groups: Dict[str, List[Dict[str, Any]]] = {} total_entities = len(entities) - update_interval = max(1, total_entities // 20) # Update every 5% + if total_entities <= 10: + update_interval = 1 # Update every item for small datasets + else: + update_interval = max(1, min(10, total_entities // 100)) + + # Initial progress update + remaining = total_entities + self.progress_tracker.update_progress( + tracking_id, + processed=0, + total=total_entities, + message=f"Grouping entities... 0/{total_entities} (remaining: {remaining})" + ) for i, entity in enumerate(entities): entity_id = entity.get("id") or entity.get("entity_id") @@ -941,18 +1124,37 @@ class ConflictDetector: entity_groups[entity_id] = [] entity_groups[entity_id].append(entity) - # Update progress periodically - if (i + 1) % update_interval == 0 or (i + 1) == total_entities: + remaining = total_entities - (i + 1) + # Update progress: always update for small datasets, or at intervals for large ones + should_update = ( + (i + 1) % update_interval == 0 or + (i + 1) == total_entities or + i == 0 or + total_entities <= 10 # Always update for small datasets + ) + if should_update: self.progress_tracker.update_progress( tracking_id, processed=i + 1, total=total_entities, - message=f"Grouping entities... {i + 1}/{total_entities}" + message=f"Grouping entities... {i + 1}/{total_entities} (remaining: {remaining})" ) # Check each entity group for logical conflicts total_groups = len(entity_groups) - group_update_interval = max(1, total_groups // 20) # Update every 5% + if total_groups <= 10: + group_update_interval = 1 # Update every item for small datasets + else: + group_update_interval = max(1, min(10, total_groups // 100)) + + # Initial progress update for group checking + remaining_groups = total_groups + self.progress_tracker.update_progress( + tracking_id, + processed=0, + total=total_groups, + message=f"Checking entity groups for logical conflicts... 0/{total_groups} (remaining: {remaining_groups})" + ) for j, (entity_id, entity_list) in enumerate(entity_groups.items()): if len(entity_list) < 2: @@ -1017,13 +1219,20 @@ class ConflictDetector: ) break - # Update progress periodically for group checking - if (j + 1) % group_update_interval == 0 or (j + 1) == total_groups: + remaining_groups = total_groups - (j + 1) + # Update progress: always update for small datasets, or at intervals for large ones + should_update = ( + (j + 1) % group_update_interval == 0 or + (j + 1) == total_groups or + j == 0 or + total_groups <= 10 # Always update for small datasets + ) + if should_update: self.progress_tracker.update_progress( tracking_id, processed=j + 1, total=total_groups, - message=f"Checking entity groups for logical conflicts... {j + 1}/{total_groups}" + message=f"Checking entity groups for logical conflicts... {j + 1}/{total_groups} (remaining: {remaining_groups})" ) self.progress_tracker.stop_tracking( @@ -1165,8 +1374,23 @@ class ConflictDetector: resolved_count = 0 unresolved_count = 0 + + total_conflicts = len(conflicts) + if total_conflicts <= 10: + update_interval = 1 # Update every item for small datasets + else: + update_interval = max(1, min(10, total_conflicts // 100)) + + # Initial progress update + remaining = total_conflicts + self.progress_tracker.update_progress( + tracking_id, + processed=0, + total=total_conflicts, + message=f"Resolving conflicts... 0/{total_conflicts} (remaining: {remaining})" + ) - for conflict in conflicts: + for i, conflict in enumerate(conflicts): if self.auto_resolve: # Simple resolution logic: pick value with highest confidence # This is a placeholder for more complex logic @@ -1177,6 +1401,22 @@ class ConflictDetector: unresolved_count += 1 else: unresolved_count += 1 + + remaining = total_conflicts - (i + 1) + # Update progress: always update for small datasets, or at intervals for large ones + should_update = ( + (i + 1) % update_interval == 0 or + (i + 1) == total_conflicts or + i == 0 or + total_conflicts <= 10 # Always update for small datasets + ) + if should_update: + self.progress_tracker.update_progress( + tracking_id, + processed=i + 1, + total=total_conflicts, + message=f"Resolving conflicts... {i + 1}/{total_conflicts} (remaining: {remaining})" + ) self.progress_tracker.stop_tracking( tracking_id, diff --git a/semantica/conflicts/conflict_resolver.py b/semantica/conflicts/conflict_resolver.py index 4a3ce7b3..122c7da7 100644 --- a/semantica/conflicts/conflict_resolver.py +++ b/semantica/conflicts/conflict_resolver.py @@ -304,19 +304,38 @@ class ConflictResolver: try: results = [] total_conflicts = len(conflicts) - update_interval = max(1, total_conflicts // 20) # Update every 5% + if total_conflicts <= 10: + update_interval = 1 # Update every item for small datasets + else: + update_interval = max(1, min(10, total_conflicts // 100)) + + # Initial progress update + remaining = total_conflicts + self.progress_tracker.update_progress( + tracking_id, + processed=0, + total=total_conflicts, + message=f"Resolving conflicts... 0/{total_conflicts} (remaining: {remaining})" + ) for i, conflict in enumerate(conflicts): result = self.resolve_conflict(conflict, strategy) results.append(result) - # Update progress periodically - if (i + 1) % update_interval == 0 or (i + 1) == total_conflicts: + remaining = total_conflicts - (i + 1) + # Update progress: always update for small datasets, or at intervals for large ones + should_update = ( + (i + 1) % update_interval == 0 or + (i + 1) == total_conflicts or + i == 0 or + total_conflicts <= 10 # Always update for small datasets + ) + if should_update: self.progress_tracker.update_progress( tracking_id, processed=i + 1, total=total_conflicts, - message=f"Resolving conflicts... {i + 1}/{total_conflicts}" + message=f"Resolving conflicts... {i + 1}/{total_conflicts} (remaining: {remaining})" ) self.progress_tracker.stop_tracking( diff --git a/semantica/deduplication/cluster_builder.py b/semantica/deduplication/cluster_builder.py index b83d9a82..2d4312ff 100644 --- a/semantica/deduplication/cluster_builder.py +++ b/semantica/deduplication/cluster_builder.py @@ -138,8 +138,10 @@ class ClusterBuilder: self.max_cluster_size = max_cluster_size self.use_hierarchical = use_hierarchical - # Initialize progress tracker + # Initialize progress tracker and ensure it's enabled self.progress_tracker = get_progress_tracker() + if not self.progress_tracker.enabled: + self.progress_tracker.enabled = True self.logger.debug( f"Cluster builder initialized: threshold={similarity_threshold}, " @@ -171,8 +173,17 @@ class ClusterBuilder: try: threshold = options.get("threshold", self.similarity_threshold) - self.progress_tracker.update_tracking( - tracking_id, message=f"Clustering {len(entities)} entities..." + total_steps = 4 # Clustering, filtering, finding unclustered, quality metrics + current_step = 0 + + # Step 1: Clustering + current_step += 1 + remaining_steps = total_steps - current_step + self.progress_tracker.update_progress( + tracking_id, + processed=current_step, + total=total_steps, + message=f"Clustering {len(entities)} entities... ({current_step}/{total_steps}, remaining: {remaining_steps} steps)" ) if self.use_hierarchical: @@ -180,9 +191,14 @@ class ClusterBuilder: else: clusters = self._graph_based_clustering(entities, threshold, tracking_id) - # Filter clusters by size - self.progress_tracker.update_tracking( - tracking_id, message="Filtering clusters by size..." + # Step 2: Filter clusters by size + current_step += 1 + remaining_steps = total_steps - current_step + self.progress_tracker.update_progress( + tracking_id, + processed=current_step, + total=total_steps, + message=f"Filtering clusters by size... ({current_step}/{total_steps}, {len(clusters)} clusters, remaining: {remaining_steps} steps)" ) valid_clusters = [ c @@ -190,9 +206,14 @@ class ClusterBuilder: if self.min_cluster_size <= len(c.entities) <= self.max_cluster_size ] - # Find unclustered entities - self.progress_tracker.update_tracking( - tracking_id, message="Finding unclustered entities..." + # Step 3: Find unclustered entities + current_step += 1 + remaining_steps = total_steps - current_step + self.progress_tracker.update_progress( + tracking_id, + processed=current_step, + total=total_steps, + message=f"Finding unclustered entities... ({current_step}/{total_steps}, remaining: {remaining_steps} steps)" ) clustered_entity_ids = set() for cluster in valid_clusters: @@ -204,9 +225,14 @@ class ClusterBuilder: e for e in entities if (e.get("id") or id(e)) not in clustered_entity_ids ] - # Calculate quality metrics - self.progress_tracker.update_tracking( - tracking_id, message="Calculating cluster quality metrics..." + # Step 4: Calculate quality metrics + current_step += 1 + remaining_steps = total_steps - current_step + self.progress_tracker.update_progress( + tracking_id, + processed=current_step, + total=total_steps, + message=f"Calculating cluster quality metrics... ({current_step}/{total_steps}, {len(valid_clusters)} clusters, remaining: {remaining_steps} steps)" ) quality_metrics = self._calculate_cluster_quality(valid_clusters) @@ -246,6 +272,23 @@ class ClusterBuilder: clusters_dict = {} cluster_id_counter = 0 + total_pairs = len(similarity_pairs) + processed_pairs = 0 + if total_pairs <= 10: + update_interval = 1 # Update every item for small datasets + else: + update_interval = max(1, min(10, total_pairs // 100)) + + # Initial progress update + if tracking_id and total_pairs > 0: + remaining = total_pairs + self.progress_tracker.update_progress( + tracking_id, + processed=0, + total=total_pairs, + message=f"Building clusters from similarity pairs... 0/{total_pairs} (remaining: {remaining})" + ) + for entity1, entity2, score in similarity_pairs: entity1_id = entity1.get("id") or id(entity1) entity2_id = entity2.get("id") or id(entity2) @@ -293,6 +336,24 @@ class ClusterBuilder: entity_to_cluster[entity_id] = cluster1 del clusters_dict[cluster2] + + processed_pairs += 1 + remaining = total_pairs - processed_pairs + # Update progress: always update for small datasets, or at intervals for large ones + if tracking_id: + should_update = ( + processed_pairs % update_interval == 0 or + processed_pairs == total_pairs or + processed_pairs == 1 or + total_pairs <= 10 # Always update for small datasets + ) + if should_update: + self.progress_tracker.update_progress( + tracking_id, + processed=processed_pairs, + total=total_pairs, + message=f"Building clusters from similarity pairs... {processed_pairs}/{total_pairs} (remaining: {remaining})" + ) return list(clusters_dict.values()) @@ -311,7 +372,10 @@ class ClusterBuilder: merged = True iteration = 0 total_iterations = len(entities) # Maximum iterations - update_interval = max(1, total_iterations // 20) # Update every 5% + if total_iterations <= 10: + update_interval = 1 # Update every iteration for small datasets + else: + update_interval = max(1, total_iterations // 20) # Update every 5% while merged: merged = False @@ -320,7 +384,20 @@ class ClusterBuilder: total_comparisons = len(clusters) * (len(clusters) - 1) // 2 processed_comparisons = 0 - comparison_update_interval = max(1, total_comparisons // 20) if total_comparisons > 0 else 1 + if total_comparisons <= 10: + comparison_update_interval = 1 # Update every comparison for small datasets + else: + comparison_update_interval = max(1, total_comparisons // 20) if total_comparisons > 0 else 1 + + # Initial progress update for comparisons + if tracking_id and total_comparisons > 0: + remaining_comparisons = total_comparisons + self.progress_tracker.update_progress( + tracking_id, + processed=0, + total=total_comparisons, + message=f"Comparing clusters... 0/{total_comparisons} (remaining: {remaining_comparisons})" + ) for i in range(len(clusters)): for j in range(i + 1, len(clusters)): @@ -332,13 +409,22 @@ class ClusterBuilder: best_merge = (i, j) processed_comparisons += 1 - if tracking_id and (processed_comparisons % comparison_update_interval == 0 or processed_comparisons == total_comparisons): - self.progress_tracker.update_progress( - tracking_id, - processed=processed_comparisons, - total=total_comparisons, - message=f"Comparing clusters... {processed_comparisons}/{total_comparisons}" + remaining_comparisons = total_comparisons - processed_comparisons + # Update progress: always update for small datasets, or at intervals for large ones + if tracking_id: + should_update = ( + processed_comparisons % comparison_update_interval == 0 or + processed_comparisons == total_comparisons or + processed_comparisons == 1 or + total_comparisons <= 10 # Always update for small datasets ) + if should_update: + self.progress_tracker.update_progress( + tracking_id, + processed=processed_comparisons, + total=total_comparisons, + message=f"Comparing clusters... {processed_comparisons}/{total_comparisons} (remaining: {remaining_comparisons})" + ) if best_merge: i, j = best_merge @@ -353,13 +439,22 @@ class ClusterBuilder: merged = True iteration += 1 - if tracking_id and (iteration % update_interval == 0): - self.progress_tracker.update_progress( - tracking_id, - processed=iteration, - total=total_iterations, - message=f"Hierarchical clustering iteration {iteration}... {len(clusters)} clusters remaining" + remaining_iterations = total_iterations - iteration + # Update progress: always update for small datasets, or at intervals for large ones + if tracking_id: + should_update = ( + iteration % update_interval == 0 or + iteration == total_iterations or + iteration == 1 or + total_iterations <= 10 # Always update for small datasets ) + if should_update: + self.progress_tracker.update_progress( + tracking_id, + processed=iteration, + total=total_iterations, + message=f"Hierarchical clustering iteration {iteration}/{total_iterations}... {len(clusters)} clusters remaining (remaining: {remaining_iterations} iterations)" + ) return clusters diff --git a/semantica/deduplication/duplicate_detector.py b/semantica/deduplication/duplicate_detector.py index 6550f92a..11e6ceec 100644 --- a/semantica/deduplication/duplicate_detector.py +++ b/semantica/deduplication/duplicate_detector.py @@ -217,8 +217,20 @@ class DuplicateDetector: # Create duplicate candidates from similar pairs candidates = [] total_similarities = len(similarities) - # Update more frequently: every 1% or at least every 10 items - update_interval = max(1, min(10, total_similarities // 100)) + # Update more frequently: every 1% or at least every 10 items, but always update for small datasets + if total_similarities <= 10: + update_interval = 1 # Update every item for small datasets + else: + update_interval = max(1, min(10, total_similarities // 100)) + + # Initial progress update - ALWAYS show this + remaining = total_similarities + self.progress_tracker.update_progress( + tracking_id, + processed=0, + total=total_similarities, + message=f"Creating duplicate candidates... 0/{total_similarities} (remaining: {remaining})" + ) for i, (entity1, entity2, score) in enumerate(similarities): candidate = self._create_duplicate_candidate(entity1, entity2, score) @@ -227,13 +239,20 @@ class DuplicateDetector: if candidate.confidence >= self.confidence_threshold: candidates.append(candidate) - # Update progress more frequently - if (i + 1) % update_interval == 0 or (i + 1) == total_similarities or i == 0: + remaining = total_similarities - (i + 1) + # Update progress: always update for small datasets, or at intervals for large ones + should_update = ( + (i + 1) % update_interval == 0 or + (i + 1) == total_similarities or + i == 0 or + total_similarities <= 10 # Always update for small datasets + ) + if should_update: self.progress_tracker.update_progress( tracking_id, processed=i + 1, total=total_similarities, - message=f"Creating duplicate candidates... {i + 1}/{total_similarities}" + message=f"Creating duplicate candidates... {i + 1}/{total_similarities} (remaining: {remaining})" ) # Sort by confidence (highest first) @@ -335,19 +354,38 @@ class DuplicateDetector: # Calculate group metrics for each group total_groups = len(groups) # Update more frequently: every item if small, or every 1% if large - update_interval = max(1, min(5, total_groups // 100)) + if total_groups <= 10: + update_interval = 1 # Update every item for small datasets + else: + update_interval = max(1, min(5, total_groups // 100)) + + # Initial progress update - ALWAYS show this + remaining = total_groups + self.progress_tracker.update_progress( + tracking_id, + processed=0, + total=total_groups, + message=f"Calculating group metrics... 0/{total_groups} (remaining: {remaining})" + ) for i, group in enumerate(groups): group.confidence = self._calculate_group_confidence(group) group.representative = self._select_representative(group) - # Update progress more frequently - if (i + 1) % update_interval == 0 or (i + 1) == total_groups or i == 0: + remaining = total_groups - (i + 1) + # Update progress: always update for small datasets, or at intervals for large ones + should_update = ( + (i + 1) % update_interval == 0 or + (i + 1) == total_groups or + i == 0 or + total_groups <= 10 # Always update for small datasets + ) + if should_update: self.progress_tracker.update_progress( tracking_id, processed=i + 1, total=total_groups, - message=f"Calculating group metrics... {i + 1}/{total_groups}" + message=f"Calculating group metrics... {i + 1}/{total_groups} (remaining: {remaining})" ) self.logger.info( @@ -445,17 +483,20 @@ class DuplicateDetector: candidates = [] total_comparisons = len(new_entities) * len(existing_entities) processed = 0 - # Update more frequently: every 1% or at least every 50 items - update_interval = max(1, min(50, total_comparisons // 100)) + # Update more frequently: every 1% or at least every 10 items, but always update for small datasets + if total_comparisons <= 10: + update_interval = 1 # Update every item for small datasets + else: + update_interval = max(1, min(10, total_comparisons // 100)) - # Initial progress update - if total_comparisons > 0: - self.progress_tracker.update_progress( - tracking_id, - processed=0, - total=total_comparisons, - message=f"Starting incremental detection... 0/{total_comparisons}" - ) + # Initial progress update - ALWAYS show this + remaining = total_comparisons + self.progress_tracker.update_progress( + tracking_id, + processed=0, + total=total_comparisons, + message=f"Starting incremental detection... 0/{total_comparisons} (remaining: {remaining})" + ) # Compare each new entity with all existing entities for new_entity in new_entities: @@ -476,13 +517,20 @@ class DuplicateDetector: candidates.append(candidate) processed += 1 - # Update progress more frequently - if processed % update_interval == 0 or processed == total_comparisons or processed == 1: + remaining = total_comparisons - processed + # Update progress: always update for small datasets, or at intervals for large ones + should_update = ( + processed % update_interval == 0 or + processed == total_comparisons or + processed == 1 or + total_comparisons <= 10 # Always update for small datasets + ) + if should_update: self.progress_tracker.update_progress( tracking_id, processed=processed, total=total_comparisons, - message=f"Comparing entities... {processed}/{total_comparisons}" + message=f"Comparing entities... {processed}/{total_comparisons} (remaining: {remaining})" ) # Sort by confidence (highest first) diff --git a/semantica/deduplication/entity_merger.py b/semantica/deduplication/entity_merger.py index 3782a60c..fd0f9dd0 100644 --- a/semantica/deduplication/entity_merger.py +++ b/semantica/deduplication/entity_merger.py @@ -218,16 +218,19 @@ class EntityMerger: mergeable_groups = [g for g in duplicate_groups if len(g.entities) >= 2] total_groups = len(mergeable_groups) # Update more frequently: every item if small, or every 1% if large - update_interval = max(1, min(5, total_groups // 100)) + if total_groups <= 10: + update_interval = 1 # Update every item for small datasets + else: + update_interval = max(1, min(5, total_groups // 100)) - # Initial progress update - if total_groups > 0: - self.progress_tracker.update_progress( - tracking_id, - processed=0, - total=total_groups, - message=f"Starting merge operations... 0/{total_groups}" - ) + # Initial progress update - ALWAYS show this + remaining = total_groups + self.progress_tracker.update_progress( + tracking_id, + processed=0, + total=total_groups, + message=f"Starting merge operations... 0/{total_groups} (remaining: {remaining})" + ) # Merge each duplicate group for i, group in enumerate(mergeable_groups): @@ -262,13 +265,20 @@ class EntityMerger: merge_operations.append(operation) self.merge_history.append(operation) - # Update progress more frequently - if (i + 1) % update_interval == 0 or (i + 1) == total_groups or i == 0: + remaining = total_groups - (i + 1) + # Update progress: always update for small datasets, or at intervals for large ones + should_update = ( + (i + 1) % update_interval == 0 or + (i + 1) == total_groups or + i == 0 or + total_groups <= 10 # Always update for small datasets + ) + if should_update: self.progress_tracker.update_progress( tracking_id, processed=i + 1, total=total_groups, - message=f"Merging groups... {i + 1}/{total_groups}" + message=f"Merging groups... {i + 1}/{total_groups} (remaining: {remaining})" ) self.logger.info( diff --git a/semantica/deduplication/merge_strategy.py b/semantica/deduplication/merge_strategy.py index c3930a97..ca26d68f 100644 --- a/semantica/deduplication/merge_strategy.py +++ b/semantica/deduplication/merge_strategy.py @@ -144,8 +144,10 @@ class MergeStrategyManager: # Custom merge strategies (callable functions) self.custom_strategies: Dict[str, Callable] = {} - # Initialize progress tracker + # Initialize progress tracker and ensure it's enabled self.progress_tracker = get_progress_tracker() + if not self.progress_tracker.enabled: + self.progress_tracker.enabled = True self.logger.debug( f"Merge strategy manager initialized (default: {self.default_strategy.value})" @@ -238,28 +240,54 @@ class MergeStrategyManager: if not isinstance(strategy, MergeStrategy): strategy = self.default_strategy - self.progress_tracker.update_tracking( - tracking_id, message=f"Selecting base entity using {strategy.value}..." + total_steps = 4 # Select base, merge properties, merge relationships, build entity + current_step = 0 + + # Step 1: Select base entity + current_step += 1 + remaining_steps = total_steps - current_step + self.progress_tracker.update_progress( + tracking_id, + processed=current_step, + total=total_steps, + message=f"Selecting base entity using {strategy.value}... ({current_step}/{total_steps}, remaining: {remaining_steps} steps)" ) - # Select base entity base_entity = self._select_base_entity(entities, strategy) - self.progress_tracker.update_tracking( - tracking_id, message="Merging properties..." + # Step 2: Merge properties + current_step += 1 + remaining_steps = total_steps - current_step + total_properties = sum(len(e.get("properties", {})) for e in entities) + self.progress_tracker.update_progress( + tracking_id, + processed=current_step, + total=total_steps, + message=f"Merging properties... ({current_step}/{total_steps}, {total_properties} properties, remaining: {remaining_steps} steps)" ) - # Merge properties merged_properties, property_conflicts = self._merge_properties( entities, base_entity, strategy ) - self.progress_tracker.update_tracking( - tracking_id, message="Merging relationships..." + # Step 3: Merge relationships + current_step += 1 + remaining_steps = total_steps - current_step + total_relationships = sum(len(e.get("relationships", [])) for e in entities) + self.progress_tracker.update_progress( + tracking_id, + processed=current_step, + total=total_steps, + message=f"Merging relationships... ({current_step}/{total_steps}, {total_relationships} relationships, remaining: {remaining_steps} steps)" ) - # Merge relationships merged_relationships = self._merge_relationships(entities, base_entity) - self.progress_tracker.update_tracking( - tracking_id, message="Building merged entity..." + # Step 4: Build merged entity + current_step += 1 + remaining_steps = total_steps - current_step + self.progress_tracker.update_progress( + tracking_id, + processed=current_step, + total=total_steps, + message=f"Building merged entity... ({current_step}/{total_steps}, remaining: {remaining_steps} steps)" ) # Build merged entity merged_entity = { diff --git a/semantica/deduplication/similarity_calculator.py b/semantica/deduplication/similarity_calculator.py index a96c5aff..39e8895e 100644 --- a/semantica/deduplication/similarity_calculator.py +++ b/semantica/deduplication/similarity_calculator.py @@ -560,23 +560,27 @@ class SimilarityCalculator: # Calculate total pairs: n*(n-1)/2 total_pairs = len(entities) * (len(entities) - 1) // 2 processed = 0 - # Update more frequently: every 1% or at least every 50 items - update_interval = max(1, min(50, total_pairs // 100)) + # Update more frequently: every 1% or at least every 10 items, but always update for small datasets + if total_pairs <= 10: + update_interval = 1 # Update every item for small datasets + else: + update_interval = max(1, min(10, total_pairs // 100)) - # Initial progress update to show tracking started + # Initial progress update to show tracking started - ALWAYS show this self.progress_tracker.update_tracking( tracking_id, status="running", message=f"Calculating similarity for {len(entities)} entities ({total_pairs} pairs)..." ) - if total_pairs > 0: - self.progress_tracker.update_progress( - tracking_id, - processed=0, - total=total_pairs, - message=f"Starting similarity calculation... 0/{total_pairs}" - ) + # Always show initial progress, even if total_pairs is 0 + remaining = total_pairs - processed + self.progress_tracker.update_progress( + tracking_id, + processed=0, + total=total_pairs, + message=f"Starting similarity calculation... 0/{total_pairs} (remaining: {remaining})" + ) for i in range(len(entities)): for j in range(i + 1, len(entities)): @@ -586,13 +590,20 @@ class SimilarityCalculator: results.append((entities[i], entities[j], similarity.score)) processed += 1 - # Update progress more frequently - if processed % update_interval == 0 or processed == total_pairs or processed == 1: + remaining = total_pairs - processed + # Update progress: always update for small datasets, or at intervals for large ones + should_update = ( + processed % update_interval == 0 or + processed == total_pairs or + processed == 1 or + total_pairs <= 10 # Always update for small datasets + ) + if should_update: self.progress_tracker.update_progress( tracking_id, processed=processed, total=total_pairs, - message=f"Comparing entity pairs... {processed}/{total_pairs}" + message=f"Comparing entity pairs... {processed}/{total_pairs} (remaining: {remaining})" ) self.progress_tracker.stop_tracking( diff --git a/semantica/semantic_extract/coreference_resolver.py b/semantica/semantic_extract/coreference_resolver.py index 0b2934d6..2ca1ba34 100644 --- a/semantica/semantic_extract/coreference_resolver.py +++ b/semantica/semantic_extract/coreference_resolver.py @@ -139,31 +139,54 @@ class CoreferenceResolver: ) try: - # Extract mentions - self.progress_tracker.update_tracking( - tracking_id, message="Extracting mentions..." + total_steps = 4 # Extract mentions, resolve pronouns, detect coreferences, build chains + current_step = 0 + + # Step 1: Extract mentions + current_step += 1 + remaining_steps = total_steps - current_step + self.progress_tracker.update_progress( + tracking_id, + processed=current_step, + total=total_steps, + message=f"Extracting mentions... ({current_step}/{total_steps}, remaining: {remaining_steps} steps)" ) mentions = self._extract_mentions(text) - # Resolve pronouns - self.progress_tracker.update_tracking( - tracking_id, message="Resolving pronouns..." + # Step 2: Resolve pronouns + current_step += 1 + remaining_steps = total_steps - current_step + self.progress_tracker.update_progress( + tracking_id, + processed=current_step, + total=total_steps, + message=f"Resolving pronouns... ({current_step}/{total_steps}, {len(mentions)} mentions, remaining: {remaining_steps} steps)" ) pronoun_resolutions = self.pronoun_resolver.resolve_pronouns( text, mentions, **options ) - # Detect entity coreferences - self.progress_tracker.update_tracking( - tracking_id, message="Detecting entity coreferences..." + # Step 3: Detect entity coreferences + current_step += 1 + remaining_steps = total_steps - current_step + self.progress_tracker.update_progress( + tracking_id, + processed=current_step, + total=total_steps, + message=f"Detecting entity coreferences... ({current_step}/{total_steps}, remaining: {remaining_steps} steps)" ) entity_corefs = self.entity_detector.detect_entity_coreferences( text, mentions, **options ) - # Build chains - self.progress_tracker.update_tracking( - tracking_id, message="Building coreference chains..." + # Step 4: Build chains + current_step += 1 + remaining_steps = total_steps - current_step + self.progress_tracker.update_progress( + tracking_id, + processed=current_step, + total=total_steps, + message=f"Building coreference chains... ({current_step}/{total_steps}, remaining: {remaining_steps} steps)" ) chains = self.chain_builder.build_coreference_chains(mentions, **options) @@ -207,8 +230,18 @@ class CoreferenceResolver: "their": r"\btheir\b", } - for pronoun, pattern in pronoun_patterns.items(): - for match in re.finditer(pattern, text, re.IGNORECASE): + total_patterns = len(pronoun_patterns) + if total_patterns <= 10: + pattern_update_interval = 1 # Update every pattern for small datasets + else: + pattern_update_interval = max(1, min(5, total_patterns // 20)) + + for pattern_idx, (pronoun, pattern) in enumerate(pronoun_patterns.items(), 1): + # Count matches first + matches = list(re.finditer(pattern, text, re.IGNORECASE)) + total_matches = len(matches) + + for match in matches: mentions.append( Mention( text=match.group(0), diff --git a/semantica/semantic_extract/event_detector.py b/semantica/semantic_extract/event_detector.py index b4e50116..5d1d4270 100644 --- a/semantica/semantic_extract/event_detector.py +++ b/semantica/semantic_extract/event_detector.py @@ -176,11 +176,46 @@ class EventDetector: } # Detect events using patterns - self.progress_tracker.update_tracking( - tracking_id, message="Scanning text for event patterns..." + total_event_types = len(event_patterns_to_use) + if total_event_types <= 10: + event_type_update_interval = 1 # Update every type for small datasets + else: + event_type_update_interval = max(1, min(5, total_event_types // 20)) + + # Initial progress update + remaining_types = total_event_types + self.progress_tracker.update_progress( + tracking_id, + processed=0, + total=total_event_types, + message=f"Scanning text for event patterns... 0/{total_event_types} event types (remaining: {remaining_types})" ) + + event_type_idx = 0 for event_type, pattern in event_patterns_to_use.items(): - for match in re.finditer(pattern, text, re.IGNORECASE): + event_type_idx += 1 + remaining_types = total_event_types - event_type_idx + + # Count matches first to show progress + matches = list(re.finditer(pattern, text, re.IGNORECASE)) + total_matches = len(matches) + + # Initialize match update interval + if total_matches <= 10: + match_update_interval = 1 + else: + match_update_interval = max(1, min(10, total_matches // 100)) + + if tracking_id and total_matches > 0: + remaining_matches = total_matches + self.progress_tracker.update_progress( + tracking_id, + processed=0, + total=total_matches, + message=f"Processing {event_type} events... 0/{total_matches} matches (remaining: {remaining_matches})" + ) + + for match_idx, match in enumerate(matches, 1): # Extract surrounding context start = max(0, match.start() - 50) end = min(len(text), match.end() + 50) @@ -213,6 +248,39 @@ class EventDetector: metadata={"context": context}, ) events.append(event) + + # Update progress for matches + if tracking_id and total_matches > 0: + remaining_matches = total_matches - match_idx + should_update = ( + match_idx % match_update_interval == 0 or + match_idx == total_matches or + match_idx == 1 or + total_matches <= 10 # Always update for small datasets + ) + if should_update: + self.progress_tracker.update_progress( + tracking_id, + processed=match_idx, + total=total_matches, + message=f"Processing {event_type} events... {match_idx}/{total_matches} matches (remaining: {remaining_matches})" + ) + + # Update progress for event types + if tracking_id: + should_update = ( + event_type_idx % event_type_update_interval == 0 or + event_type_idx == total_event_types or + event_type_idx == 1 or + total_event_types <= 10 # Always update for small datasets + ) + if should_update: + self.progress_tracker.update_progress( + tracking_id, + processed=event_type_idx, + total=total_event_types, + message=f"Scanning text for event patterns... {event_type_idx}/{total_event_types} event types (remaining: {remaining_types})" + ) self.progress_tracker.stop_tracking( tracking_id, diff --git a/semantica/semantic_extract/ner_extractor.py b/semantica/semantic_extract/ner_extractor.py index 14c48ebf..9e561063 100644 --- a/semantica/semantic_extract/ner_extractor.py +++ b/semantica/semantic_extract/ner_extractor.py @@ -168,15 +168,19 @@ class NERExtractor: try: results = [] total_items = len(text) - # Update more frequently: every 1% or at least every 10 items - update_interval = max(1, min(10, total_items // 100)) + # Update more frequently: every 1% or at least every 10 items, but always update for small datasets + if total_items <= 10: + update_interval = 1 # Update every item for small datasets + else: + update_interval = max(1, min(10, total_items // 100)) - # Initial progress update + # Initial progress update - ALWAYS show this + remaining = total_items self.progress_tracker.update_progress( tracking_id, processed=0, total=total_items, - message=f"Starting batch extraction... 0/{total_items}" + message=f"Starting batch extraction... 0/{total_items} (remaining: {remaining})" ) for idx, item in enumerate(text, 1): @@ -194,13 +198,20 @@ class NERExtractor: except Exception: results.append([]) - # Update progress more frequently - if idx % update_interval == 0 or idx == total_items or idx == 1: + remaining = total_items - idx + # Update progress: always update for small datasets, or at intervals for large ones + should_update = ( + idx % update_interval == 0 or + idx == total_items or + idx == 1 or + total_items <= 10 # Always update for small datasets + ) + if should_update: self.progress_tracker.update_progress( tracking_id, processed=idx, total=total_items, - message=f"Processing documents... {idx}/{total_items}" + message=f"Processing documents... {idx}/{total_items} (remaining: {remaining})" ) self.progress_tracker.stop_tracking( diff --git a/semantica/semantic_extract/relation_extractor.py b/semantica/semantic_extract/relation_extractor.py index b971a1d2..fb2961dc 100644 --- a/semantica/semantic_extract/relation_extractor.py +++ b/semantica/semantic_extract/relation_extractor.py @@ -194,15 +194,19 @@ class RelationExtractor: results = [] # Ensure lists are same length min_len = min(len(text), len(entities)) - # Update more frequently: every 1% or at least every 10 items - update_interval = max(1, min(10, min_len // 100)) + # Update more frequently: every 1% or at least every 10 items, but always update for small datasets + if min_len <= 10: + update_interval = 1 # Update every item for small datasets + else: + update_interval = max(1, min(10, min_len // 100)) - # Initial progress update + # Initial progress update - ALWAYS show this + remaining = min_len self.progress_tracker.update_progress( tracking_id, processed=0, total=min_len, - message=f"Starting batch extraction... 0/{min_len}" + message=f"Starting batch extraction... 0/{min_len} (remaining: {remaining})" ) for i in range(min_len): @@ -223,13 +227,20 @@ class RelationExtractor: results.append(self.extract_relations(doc_text, ent_item, **kwargs)) - # Update progress more frequently - if (i + 1) % update_interval == 0 or (i + 1) == min_len or i == 0: + remaining = min_len - (i + 1) + # Update progress: always update for small datasets, or at intervals for large ones + should_update = ( + (i + 1) % update_interval == 0 or + (i + 1) == min_len or + i == 0 or + min_len <= 10 # Always update for small datasets + ) + if should_update: self.progress_tracker.update_progress( tracking_id, processed=i + 1, total=min_len, - message=f"Processing documents... {i + 1}/{min_len}" + message=f"Processing documents... {i + 1}/{min_len} (remaining: {remaining})" ) self.progress_tracker.stop_tracking( diff --git a/semantica/semantic_extract/semantic_analyzer.py b/semantica/semantic_extract/semantic_analyzer.py index 294a4cbc..1a2b4e28 100644 --- a/semantica/semantic_extract/semantic_analyzer.py +++ b/semantica/semantic_extract/semantic_analyzer.py @@ -367,6 +367,10 @@ class SemanticClusterer: """Initialize semantic clusterer.""" self.logger = get_logger("semantic_clusterer") self.config = config + # Initialize progress tracker and ensure it's enabled + self.progress_tracker = get_progress_tracker() + if not self.progress_tracker.enabled: + self.progress_tracker.enabled = True def cluster(self, texts: List[str], **options) -> List[SemanticCluster]: """ @@ -384,38 +388,89 @@ class SemanticClusterer: if not texts: return [] - similarity_threshold = options.get("similarity_threshold", 0.5) - similarity_analyzer = SimilarityAnalyzer() + # Track clustering + tracking_id = self.progress_tracker.start_tracking( + module="semantic_extract", + submodule="SemanticClusterer", + message=f"Clustering {len(texts)} texts", + ) - clusters = [] - assigned = set() + try: + similarity_threshold = options.get("similarity_threshold", 0.5) + similarity_analyzer = SimilarityAnalyzer() - cluster_id = 0 - for i, text1 in enumerate(texts): - if i in assigned: - continue + clusters = [] + assigned = set() - cluster_texts = [text1] - assigned.add(i) + total_texts = len(texts) + if total_texts <= 10: + update_interval = 1 # Update every item for small datasets + else: + update_interval = max(1, min(10, total_texts // 100)) + + # Initial progress update + remaining = total_texts + self.progress_tracker.update_progress( + tracking_id, + processed=0, + total=total_texts, + message=f"Clustering texts... 0/{total_texts} (remaining: {remaining})" + ) - # Find similar texts - for j, text2 in enumerate(texts[i + 1 :], start=i + 1): - if j in assigned: + cluster_id = 0 + for i, text1 in enumerate(texts): + if i in assigned: continue - similarity = similarity_analyzer.calculate_similarity(text1, text2) - if similarity >= similarity_threshold: - cluster_texts.append(text2) - assigned.add(j) + cluster_texts = [text1] + assigned.add(i) - # Create cluster - cluster = SemanticCluster( - texts=cluster_texts, - cluster_id=cluster_id, - centroid=cluster_texts[0], # Use first as centroid - similarity_score=similarity_threshold, + # Find similar texts + remaining_texts = len(texts) - (i + 1) + for j, text2 in enumerate(texts[i + 1 :], start=i + 1): + if j in assigned: + continue + + similarity = similarity_analyzer.calculate_similarity(text1, text2) + if similarity >= similarity_threshold: + cluster_texts.append(text2) + assigned.add(j) + + # Create cluster + cluster = SemanticCluster( + texts=cluster_texts, + cluster_id=cluster_id, + centroid=cluster_texts[0], # Use first as centroid + similarity_score=similarity_threshold, + ) + clusters.append(cluster) + cluster_id += 1 + + remaining = total_texts - (i + 1) + # Update progress: always update for small datasets, or at intervals for large ones + should_update = ( + (i + 1) % update_interval == 0 or + (i + 1) == total_texts or + i == 0 or + total_texts <= 10 # Always update for small datasets + ) + if should_update: + self.progress_tracker.update_progress( + tracking_id, + processed=i + 1, + total=total_texts, + message=f"Clustering texts... {i + 1}/{total_texts} (remaining: {remaining})" + ) + + self.progress_tracker.stop_tracking( + tracking_id, + status="completed", + message=f"Created {len(clusters)} clusters", ) - clusters.append(cluster) - cluster_id += 1 + return clusters - return clusters + except Exception as e: + self.progress_tracker.stop_tracking( + tracking_id, status="failed", message=str(e) + ) + raise diff --git a/semantica/semantic_extract/semantic_network_extractor.py b/semantica/semantic_extract/semantic_network_extractor.py index a2eb7643..335157da 100644 --- a/semantica/semantic_extract/semantic_network_extractor.py +++ b/semantica/semantic_extract/semantic_network_extractor.py @@ -217,10 +217,18 @@ class SemanticNetworkExtractor: relations = rel_extractor.extract_relations(text, entities, **options) # Build network - self.progress_tracker.update_tracking( - tracking_id, message="Building semantic network..." + total_steps = 2 # Create nodes, create edges + current_step = 0 + + current_step += 1 + remaining_steps = total_steps - current_step + self.progress_tracker.update_progress( + tracking_id, + processed=current_step, + total=total_steps, + message=f"Building semantic network... Creating nodes from {len(entities)} entities ({current_step}/{total_steps}, remaining: {remaining_steps} steps)" ) - network = self._build_network(entities, relations) + network = self._build_network(entities, relations, tracking_id, total_steps, current_step) self.progress_tracker.stop_tracking( tracking_id, @@ -236,7 +244,7 @@ class SemanticNetworkExtractor: raise def _build_network( - self, entities: List[Entity], relations: List[Relation] + self, entities: List[Entity], relations: List[Relation], tracking_id: str = None, total_steps: int = 2, current_step: int = 1 ) -> SemanticNetwork: """Build semantic network from entities and relations.""" nodes = [] @@ -244,7 +252,23 @@ class SemanticNetworkExtractor: node_map = {} # Create nodes from entities - for entity in entities: + total_entities = len(entities) + if total_entities <= 10: + entity_update_interval = 1 # Update every item for small datasets + else: + entity_update_interval = max(1, min(10, total_entities // 100)) + + # Initial progress update for entities + if tracking_id and total_entities > 0: + remaining_entities = total_entities + self.progress_tracker.update_progress( + tracking_id, + processed=0, + total=total_entities, + message=f"Creating nodes from entities... 0/{total_entities} (remaining: {remaining_entities})" + ) + + for i, entity in enumerate(entities): node_id = f"entity_{len(nodes)}" node_map[entity.text] = node_id @@ -260,9 +284,42 @@ class SemanticNetworkExtractor: metadata=entity.metadata, ) nodes.append(node) + + remaining_entities = total_entities - (i + 1) + # Update progress: always update for small datasets, or at intervals for large ones + if tracking_id: + should_update = ( + (i + 1) % entity_update_interval == 0 or + (i + 1) == total_entities or + i == 0 or + total_entities <= 10 # Always update for small datasets + ) + if should_update: + self.progress_tracker.update_progress( + tracking_id, + processed=i + 1, + total=total_entities, + message=f"Creating nodes from entities... {i + 1}/{total_entities} (remaining: {remaining_entities})" + ) # Create edges from relations - for relation in relations: + total_relations = len(relations) + if total_relations <= 10: + relation_update_interval = 1 # Update every item for small datasets + else: + relation_update_interval = max(1, min(10, total_relations // 100)) + + if tracking_id and total_relations > 0: + # Initial progress update for relations + remaining_relations = total_relations + self.progress_tracker.update_progress( + tracking_id, + processed=0, + total=total_relations, + message=f"Creating edges from relations... 0/{total_relations} (remaining: {remaining_relations})" + ) + + for j, relation in enumerate(relations): subject_id = node_map.get(relation.subject.text) object_id = node_map.get(relation.object.text) @@ -278,6 +335,23 @@ class SemanticNetworkExtractor: metadata=relation.metadata, ) edges.append(edge) + + remaining_relations = len(relations) - (j + 1) + # Update progress: always update for small datasets, or at intervals for large ones + if tracking_id: + should_update = ( + (j + 1) % relation_update_interval == 0 or + (j + 1) == len(relations) or + j == 0 or + len(relations) <= 10 # Always update for small datasets + ) + if should_update: + self.progress_tracker.update_progress( + tracking_id, + processed=j + 1, + total=len(relations), + message=f"Creating edges from relations... {j + 1}/{len(relations)} (remaining: {remaining_relations})" + ) return SemanticNetwork( nodes=nodes, diff --git a/semantica/semantic_extract/triplet_extractor.py b/semantica/semantic_extract/triplet_extractor.py index f4f93813..3658e6d3 100644 --- a/semantica/semantic_extract/triplet_extractor.py +++ b/semantica/semantic_extract/triplet_extractor.py @@ -206,11 +206,29 @@ class TripletExtractor: # Try each method in order (fallback chain) all_triplets = [] - for method_name in methods: + total_methods = len(methods) + if total_methods <= 10: + method_update_interval = 1 # Update every method for small datasets + else: + method_update_interval = max(1, min(5, total_methods // 20)) + + # Initial progress update for methods + remaining_methods = total_methods + self.progress_tracker.update_progress( + tracking_id, + processed=0, + total=total_methods, + message=f"Starting triplet extraction... 0/{total_methods} methods (remaining: {remaining_methods})" + ) + + for method_idx, method_name in enumerate(methods, 1): try: - self.progress_tracker.update_tracking( + remaining_methods = total_methods - method_idx + self.progress_tracker.update_progress( tracking_id, - message=f"Extracting triplets using {method_name}...", + processed=method_idx, + total=total_methods, + message=f"Extracting triplets using {method_name}... ({method_idx}/{total_methods}, remaining: {remaining_methods} methods)" ) method_func = get_triplet_method(method_name) diff --git a/semantica/utils/progress_tracker.py b/semantica/utils/progress_tracker.py index 95ce92ff..2b5a66a7 100644 --- a/semantica/utils/progress_tracker.py +++ b/semantica/utils/progress_tracker.py @@ -116,7 +116,8 @@ class ConsoleProgressDisplay(ProgressDisplay): def _should_update(self) -> bool: """Check if enough time has passed for update.""" now = time.time() - if now - self.last_update >= self.update_interval: + # If last_update is 0.0 or very old, always update (forced update) + if self.last_update <= 0.0 or (now - self.last_update) >= self.update_interval: self.last_update = now return True return False @@ -546,7 +547,7 @@ class JupyterProgressDisplay(ProgressDisplay): return "".join(html_parts) def update(self, item: ProgressItem) -> None: - """Update Jupyter progress display.""" + """Update Jupyter progress display (works in Jupyter and Google Colab).""" # Add or update item existing = None for i, existing_item in enumerate(self.items): @@ -563,13 +564,51 @@ class JupyterProgressDisplay(ProgressDisplay): else: self.items.append(item) - # Update display + # Update display - always update immediately in Jupyter/Colab if IPYTHON_AVAILABLE: html = self._build_html(self.items) - if self.display_handle is None: + try: + # Check if we're in Google Colab (Colab sometimes needs fresh displays) + is_colab = False + try: + import sys + import os + is_colab = ('google.colab' in sys.modules or + os.environ.get("COLAB_GPU") is not None) + except Exception: + pass + + if self.display_handle is None: + # First time - create display + self.display_handle = display(HTML(html), display_id=True) + else: + # Try to update existing display + try: + # In Colab, sometimes update() doesn't work, so we recreate + if is_colab: + # Clear and recreate for Colab compatibility + try: + clear_output(wait=False) + except Exception: + pass + self.display_handle = display(HTML(html), display_id=True) + else: + # Regular Jupyter - try update first + self.display_handle.update(HTML(html)) + except (AttributeError, TypeError, Exception): + # If update fails, create new display (works in both Jupyter and Colab) + try: + clear_output(wait=False) + except Exception: + pass + self.display_handle = display(HTML(html), display_id=True) + except Exception: + # Fallback: always create new display if update fails + try: + clear_output(wait=False) + except Exception: + pass self.display_handle = display(HTML(html), display_id=True) - else: - self.display_handle.update(HTML(html)) def show_summary(self, items: List[ProgressItem]) -> None: """Show final summary in Jupyter.""" @@ -823,22 +862,33 @@ class ProgressTracker: Initialize progress tracker. Args: - enabled: Enable progress tracking + enabled: Enable progress tracking (default: True, always enabled) use_emoji: Use emoji indicators update_interval: Minimum time between updates (seconds) """ - self.enabled = enabled + # Always enable progress tracking by default - cannot be disabled via constructor + # This ensures progress is always shown automatically + self.enabled = True # Force enabled, ignore parameter self.use_emoji = use_emoji self.update_interval = update_interval - # Detect environment + # Detect environment - will be checked dynamically self.is_jupyter = self._detect_jupyter() # Create displays self.displays: List[ProgressDisplay] = [] - if self.is_jupyter and IPYTHON_AVAILABLE: - self.displays.append(JupyterProgressDisplay(use_emoji=use_emoji)) + # Always try Jupyter first if available, fallback to console + if IPYTHON_AVAILABLE: + # Try to detect Jupyter - if available, use it + if self.is_jupyter: + self.displays.append(JupyterProgressDisplay(use_emoji=use_emoji)) + # Also add console as fallback for immediate feedback + self.displays.append( + ConsoleProgressDisplay( + use_emoji=use_emoji, update_interval=update_interval + ) + ) else: self.displays.append( ConsoleProgressDisplay( @@ -855,12 +905,48 @@ class ProgressTracker: self.lock = threading.Lock() def _detect_jupyter(self) -> bool: - """Detect if running in Jupyter notebook.""" + """Detect if running in Jupyter notebook or Google Colab.""" if not IPYTHON_AVAILABLE: return False try: ipython = get_ipython() - return ipython is not None and hasattr(ipython, "kernel") + if ipython is None: + return False + + # Method 1: Check for Google Colab + # Colab has 'google.colab' in sys.modules or environment variables + try: + import sys + import os + if 'google.colab' in sys.modules: + return True + # Check environment variables (Colab sets these) + if os.environ.get("COLAB_GPU") is not None or os.environ.get("COLAB_JUPYTER_TRANSPORT") is not None: + return True + # Check IPython config for Colab + if hasattr(ipython, 'config') and hasattr(ipython.config, 'IPKernelApp'): + config_str = str(ipython.config.IPKernelApp) + if 'google.colab' in config_str or 'colab' in config_str.lower(): + return True + except Exception: + pass + + # Method 2: Check for kernel attribute (Jupyter/Colab) + if hasattr(ipython, "kernel"): + return True + + # Method 3: Check for IPython shell class name + if hasattr(ipython, "__class__"): + class_name = ipython.__class__.__name__ + # Check for Jupyter, Colab, or ZMQ shell types + if any(name in class_name for name in ["ZMQInteractiveShell", "Jupyter", "Colab", "InteractiveShell"]): + return True + + # Method 4: Check for IPython display capability + if hasattr(ipython, "display_pub"): + return True + + return False except Exception: return False @@ -871,6 +957,11 @@ class ProgressTracker: with cls._lock: if cls._instance is None: cls._instance = cls() + # Ensure it's always enabled + cls._instance.enabled = True + else: + # Always ensure enabled when getting instance + cls._instance.enabled = True return cls._instance def start_tracking( @@ -895,6 +986,15 @@ class ProgressTracker: if not self.enabled: return "" + # Re-detect Jupyter environment in case it wasn't detected at init + # This helps if the tracker was created before Jupyter was fully initialized + if IPYTHON_AVAILABLE and not self.is_jupyter: + self.is_jupyter = self._detect_jupyter() + # If Jupyter is now detected and we don't have a Jupyter display, add it + if self.is_jupyter and not any(isinstance(d, JupyterProgressDisplay) for d in self.displays): + # Insert Jupyter display at the beginning for priority + self.displays.insert(0, JupyterProgressDisplay(use_emoji=self.use_emoji)) + # Auto-detect if not provided if not module or not submodule: detected_module, detected_submodule = ModuleDetector.detect_from_call_stack( @@ -986,6 +1086,14 @@ class ProgressTracker: if not self.enabled or not tracking_id: return + # Re-detect Jupyter environment in case it wasn't detected at init + if IPYTHON_AVAILABLE and not self.is_jupyter: + self.is_jupyter = self._detect_jupyter() + # If Jupyter is now detected and we don't have a Jupyter display, add it + if self.is_jupyter and not any(isinstance(d, JupyterProgressDisplay) for d in self.displays): + # Insert Jupyter display at the beginning for priority + self.displays.insert(0, JupyterProgressDisplay(use_emoji=self.use_emoji)) + with self.lock: if tracking_id in self.active_items: item = self.active_items[tracking_id] @@ -998,9 +1106,21 @@ class ProgressTracker: if message: item.message = message - # Update displays + # Update displays - force immediate update for progress for display in self.displays: - display.update(item) + # For Jupyter, always update immediately + if isinstance(display, JupyterProgressDisplay): + display.update(item) + # For console, force update by temporarily bypassing interval check + elif isinstance(display, ConsoleProgressDisplay): + # Force update by setting last_update far in the past + original_last_update = display.last_update + display.last_update = 0.0 # This will make _should_update return True + display.update(item) + # Restore original value (update() will set it to current time anyway) + display.last_update = original_last_update + else: + display.update(item) def _calculate_eta(self, item: ProgressItem) -> Optional[float]: """ @@ -1132,6 +1252,18 @@ def get_progress_tracker() -> ProgressTracker: global _global_tracker if _global_tracker is None: _global_tracker = ProgressTracker.get_instance() + + # Always ensure progress tracker is enabled automatically + _global_tracker.enabled = True + + # Re-detect Jupyter environment dynamically (in case it wasn't ready at init) + if IPYTHON_AVAILABLE and not _global_tracker.is_jupyter: + _global_tracker.is_jupyter = _global_tracker._detect_jupyter() + # If Jupyter is now detected and we don't have a Jupyter display, add it + if _global_tracker.is_jupyter and not any(isinstance(d, JupyterProgressDisplay) for d in _global_tracker.displays): + # Insert Jupyter display at the beginning for priority + _global_tracker.displays.insert(0, JupyterProgressDisplay(use_emoji=_global_tracker.use_emoji)) + return _global_tracker