Session 2 - Module A: Advanced Document Analytics¶

import numpy as np
from typing import Dict, List, Any, Tuple
from dataclasses import dataclass
import re

@dataclass
class DocumentComplexityScore:
    """Comprehensive document complexity assessment."""
    structural_complexity: float
    semantic_complexity: float
    processing_difficulty: float
    overall_score: float
    recommendations: List[str]
    confidence_level: float

class EnterpriseComplexityAnalyzer:
    """Production-grade document complexity analysis."""

    def __init__(self):
        # Domain-specific complexity patterns
        self.complexity_patterns = {
            "legal": {
                "high_complexity": [r"whereas", r"heretofore", r"§\s*\d+", r"article\s+[IVX]+"],
                "structural_markers": [r"section\s+\d+", r"clause\s+\d+", r"paragraph\s+\([a-z]\)"],
                "citation_density": [r"\d+\s+U\.S\.C\.", r"\d+\s+F\.\d+d", r"Fed\.\s*R\.\s*Civ\.\s*P\."]
            },
            "technical": {
                "code_blocks": [r"```[\s\S]*?```", r"(?:^|\n)    .*", r"`[^`]+`"],
                "api_references": [r"[A-Z][a-zA-Z]+\.[a-zA-Z]+\(\)", r"HTTP[S]?://"],
                "version_patterns": [r"v?\d+\.\d+\.\d+", r"version\s+\d+\.\d+"]
            },
            "medical": {
                "dosage_patterns": [r"\d+\s*mg", r"\d+\s*ml", r"\d+\s*units?"],
                "medical_terms": [r"[A-Z][a-z]+itis", r"[A-Z][a-z]+osis", r"[A-Z][a-z]+emia"],
                "procedure_codes": [r"CPT\s*\d+", r"ICD-\d+"]
            }
        }

        # Structural complexity weights
        self.structure_weights = {
            "table_density": 3.0,
            "list_density": 1.5,
            "heading_hierarchy": 2.0,
            "code_block_density": 2.5,
            "citation_density": 2.0
        }

    def analyze_enterprise_complexity(self, document_text: str, 
                                    domain: str = "general") -> DocumentComplexityScore:
        """Comprehensive enterprise document complexity analysis."""

        # Multi-dimensional complexity assessment
        structural_score = self._analyze_structural_complexity(document_text, domain)
        semantic_score = self._analyze_semantic_complexity(document_text, domain)
        processing_score = self._assess_processing_difficulty(document_text, domain)

        # Calculate weighted overall score
        overall_score = (structural_score * 0.4 + 
                        semantic_score * 0.3 + 
                        processing_score * 0.3)

        # Generate actionable recommendations
        recommendations = self._generate_processing_recommendations(
            structural_score, semantic_score, processing_score, domain
        )

        # Calculate confidence based on pattern matches
        confidence = self._calculate_confidence_level(document_text, domain)

        return DocumentComplexityScore(
            structural_complexity=structural_score,
            semantic_complexity=semantic_score,
            processing_difficulty=processing_score,
            overall_score=overall_score,
            recommendations=recommendations,
            confidence_level=confidence
        )

    def _analyze_structural_complexity(self, text: str, domain: str) -> float:
        """Analyze document structural complexity with domain awareness."""
        complexity_score = 0.0
        text_length = len(text)

        if text_length == 0:
            return 0.0

        # Base structural elements
        table_count = text.count('|')
        code_blocks = len(re.findall(r'```[\s\S]*?```', text))
        lists = len(re.findall(r'^\s*[-*+]\s+', text, re.MULTILINE))
        headings = len(re.findall(r'^#+\s+', text, re.MULTILINE))

        # Calculate density scores
        table_density = (table_count / text_length) * 1000
        code_density = (code_blocks / text_length) * 1000
        list_density = (lists / text_length) * 1000
        heading_density = (headings / text_length) * 1000

        # Apply domain-specific weights
        complexity_score += table_density * self.structure_weights["table_density"]
        complexity_score += code_density * self.structure_weights["code_block_density"]
        complexity_score += list_density * self.structure_weights["list_density"]
        complexity_score += heading_density * self.structure_weights["heading_hierarchy"]

        # Domain-specific structural patterns
        if domain in self.complexity_patterns:
            domain_patterns = self.complexity_patterns[domain]
            for pattern_type, patterns in domain_patterns.items():
                pattern_matches = sum(len(re.findall(pattern, text, re.IGNORECASE)) 
                                    for pattern in patterns)
                pattern_density = (pattern_matches / text_length) * 1000
                complexity_score += pattern_density * 1.5

        # Normalize to 0-1 scale
        return min(complexity_score / 100, 1.0)

    def _analyze_semantic_complexity(self, text: str, domain: str) -> float:
        """Analyze semantic complexity including terminology and concepts."""
        words = text.split()

        if not words:
            return 0.0

        # Vocabulary complexity metrics
        unique_words = set(words)
        vocabulary_richness = len(unique_words) / len(words)

        # Technical terminology density
        technical_terms = self._count_technical_terms(text, domain)
        technical_density = technical_terms / len(words)

        # Sentence complexity
        sentences = re.split(r'[.!?]+', text)
        avg_sentence_length = len(words) / len(sentences) if sentences else 0

        # Readability approximation (simplified)
        long_words = len([w for w in words if len(w) > 6])
        long_word_ratio = long_words / len(words)

        # Calculate semantic complexity score
        complexity_factors = [
            vocabulary_richness * 0.3,
            min(technical_density * 10, 1.0) * 0.3,
            min(avg_sentence_length / 25, 1.0) * 0.2,
            long_word_ratio * 0.2
        ]

        return sum(complexity_factors)

    def _assess_processing_difficulty(self, text: str, domain: str) -> float:
        """Assess computational and processing difficulty."""
        # Document size factor
        size_complexity = min(len(text) / 10000, 1.0) * 0.3

        # Mixed content complexity
        has_tables = '|' in text and text.count('|') > 5
        has_code = '```' in text or text.count('    ') > 10
        has_images = any(ext in text.lower() for ext in ['.jpg', '.png', '.gif', '.pdf'])

        mixed_content_score = sum([has_tables, has_code, has_images]) / 3 * 0.4

        # Cross-reference complexity
        cross_refs = len(re.findall(r'see\s+(?:section|chapter|page)\s+\d+', text, re.IGNORECASE))
        cross_ref_complexity = min(cross_refs / 10, 1.0) * 0.3

        return size_complexity + mixed_content_score + cross_ref_complexity

    def _count_technical_terms(self, text: str, domain: str) -> int:
        """Count domain-specific technical terms."""
        technical_count = 0

        # General technical patterns
        acronyms = len(re.findall(r'\b[A-Z]{2,}\b', text))
        technical_count += acronyms

        # Domain-specific terms
        if domain in self.complexity_patterns:
            for pattern_list in self.complexity_patterns[domain].values():
                for pattern in pattern_list:
                    technical_count += len(re.findall(pattern, text, re.IGNORECASE))

        return technical_count

    def _generate_processing_recommendations(self, structural: float, semantic: float, 
                                           processing: float, domain: str) -> List[str]:
        """Generate actionable processing recommendations."""
        recommendations = []

        # Structural recommendations
        if structural > 0.7:
            recommendations.append("Use specialized structural processing for complex document layout")
            recommendations.append("Implement table-aware chunking for structured content")
        elif structural > 0.4:
            recommendations.append("Apply hierarchical chunking to preserve document structure")

        # Semantic recommendations
        if semantic > 0.7:
            recommendations.append("Enable advanced metadata extraction for technical terminology")
            recommendations.append("Use domain-specific preprocessing patterns")
        elif semantic > 0.4:
            recommendations.append("Apply enhanced keyword and entity extraction")

        # Processing recommendations
        if processing > 0.7:
            recommendations.append("Allocate additional computational resources for processing")
            recommendations.append("Consider batch processing with quality checkpoints")

        # Domain-specific recommendations
        if domain == "legal":
            recommendations.append("Preserve citation relationships and statutory references")
        elif domain == "medical":
            recommendations.append("Validate dosage information and maintain treatment correlations")
        elif domain == "technical":
            recommendations.append("Keep code blocks intact and preserve API documentation structure")

        return recommendations

    def _calculate_confidence_level(self, text: str, domain: str) -> float:
        """Calculate confidence in complexity assessment."""
        # Base confidence on document size and pattern matches
        size_confidence = min(len(text) / 1000, 1.0) * 0.5

        # Pattern match confidence
        pattern_matches = 0
        if domain in self.complexity_patterns:
            for pattern_list in self.complexity_patterns[domain].values():
                for pattern in pattern_list:
                    if re.search(pattern, text, re.IGNORECASE):
                        pattern_matches += 1

        pattern_confidence = min(pattern_matches / 5, 1.0) * 0.5

        return size_confidence + pattern_confidence

class ProductionComplexityPipeline:
    """Production pipeline with complexity-based processing selection."""

    def __init__(self):
        self.complexity_analyzer = EnterpriseComplexityAnalyzer()
        self.processing_strategies = {
            "simple": {"max_chunk_size": 500, "overlap": 0.05},
            "standard": {"max_chunk_size": 1000, "overlap": 0.1},
            "complex": {"max_chunk_size": 1500, "overlap": 0.15},
            "enterprise": {"max_chunk_size": 2000, "overlap": 0.2}
        }

    def process_with_adaptive_strategy(self, document: str, domain: str = "general") -> Dict[str, Any]:
        """Process document with complexity-adaptive strategy."""
        # Analyze complexity
        complexity_score = self.complexity_analyzer.analyze_enterprise_complexity(document, domain)

        # Select processing strategy
        if complexity_score.overall_score < 0.3:
            strategy = "simple"
        elif complexity_score.overall_score < 0.6:
            strategy = "standard"
        elif complexity_score.overall_score < 0.8:
            strategy = "complex"
        else:
            strategy = "enterprise"

        # Apply selected strategy
        strategy_config = self.processing_strategies[strategy]

        return {
            "complexity_analysis": complexity_score,
            "selected_strategy": strategy,
            "processing_config": strategy_config,
            "recommendations": complexity_score.recommendations,
            "confidence": complexity_score.confidence_level
        }

class EnterpriseQualityFramework:
    """Comprehensive quality assessment for enterprise document processing."""

    def __init__(self):
        self.quality_dimensions = {
            "coherence": 0.25,      # Topic consistency
            "completeness": 0.25,   # Information preservation
            "accessibility": 0.20,  # Readability and structure
            "accuracy": 0.15,       # Factual correctness
            "efficiency": 0.15      # Processing optimization
        }

        self.domain_quality_criteria = {
            "legal": {
                "citation_preservation": 0.3,
                "clause_integrity": 0.3,
                "terminology_accuracy": 0.4
            },
            "medical": {
                "dosage_accuracy": 0.4,
                "procedure_completeness": 0.3,
                "terminology_precision": 0.3
            },
            "technical": {
                "code_integrity": 0.4,
                "api_documentation": 0.3,
                "version_consistency": 0.3
            }
        }

    def assess_enterprise_quality(self, chunks: List[Document], 
                                domain: str = "general") -> Dict[str, Any]:
        """Comprehensive enterprise quality assessment."""

        # Base quality metrics
        base_quality = self._assess_base_quality(chunks)

        # Domain-specific quality metrics
        domain_quality = self._assess_domain_quality(chunks, domain)

        # Processing efficiency metrics
        efficiency_metrics = self._assess_processing_efficiency(chunks)

        # Calculate weighted overall score
        overall_score = self._calculate_weighted_score(
            base_quality, domain_quality, efficiency_metrics
        )

        # Generate quality improvement recommendations
        recommendations = self._generate_quality_recommendations(
            base_quality, domain_quality, efficiency_metrics, domain
        )

        return {
            "overall_quality": overall_score,
            "base_quality": base_quality,
            "domain_quality": domain_quality,
            "efficiency_metrics": efficiency_metrics,
            "recommendations": recommendations,
            "quality_grade": self._assign_quality_grade(overall_score)
        }

    def _assess_base_quality(self, chunks: List[Document]) -> Dict[str, float]:
        """Assess fundamental quality metrics."""
        if not chunks:
            return {"coherence": 0, "completeness": 0, "accessibility": 0}

        # Coherence assessment
        coherence = self._measure_semantic_coherence(chunks)

        # Completeness assessment
        completeness = self._measure_information_completeness(chunks)

        # Accessibility assessment
        accessibility = self._measure_content_accessibility(chunks)

        return {
            "coherence": coherence,
            "completeness": completeness,
            "accessibility": accessibility
        }

    def _measure_semantic_coherence(self, chunks: List[Document]) -> float:
        """Advanced semantic coherence measurement."""
        if len(chunks) < 2:
            return 1.0

        coherence_scores = []

        for i in range(len(chunks) - 1):
            current_chunk = chunks[i]
            next_chunk = chunks[i + 1]

            # Topic overlap analysis
            current_topics = set(current_chunk.metadata.get("topics", []))
            next_topics = set(next_chunk.metadata.get("topics", []))

            if current_topics and next_topics:
                topic_overlap = len(current_topics & next_topics) / len(current_topics | next_topics)
            else:
                topic_overlap = 0

            # Entity continuity analysis
            current_entities = set(current_chunk.metadata.get("entities", []))
            next_entities = set(next_chunk.metadata.get("entities", []))

            if current_entities and next_entities:
                entity_overlap = len(current_entities & next_entities) / len(current_entities | next_entities)
            else:
                entity_overlap = 0

            # Combined coherence score
            chunk_coherence = (topic_overlap * 0.6 + entity_overlap * 0.4)
            coherence_scores.append(chunk_coherence)

        return np.mean(coherence_scores) if coherence_scores else 0.0

    def _assess_domain_quality(self, chunks: List[Document], domain: str) -> Dict[str, float]:
        """Domain-specific quality assessment."""
        if domain not in self.domain_quality_criteria:
            return {}

        criteria = self.domain_quality_criteria[domain]
        domain_scores = {}

        for criterion, weight in criteria.items():
            if criterion == "citation_preservation" and domain == "legal":
                domain_scores[criterion] = self._assess_citation_preservation(chunks)
            elif criterion == "dosage_accuracy" and domain == "medical":
                domain_scores[criterion] = self._assess_dosage_accuracy(chunks)
            elif criterion == "code_integrity" and domain == "technical":
                domain_scores[criterion] = self._assess_code_integrity(chunks)
            # Add more domain-specific assessments as needed

        return domain_scores

    def _assess_processing_efficiency(self, chunks: List[Document]) -> Dict[str, float]:
        """Assess processing efficiency metrics."""
        if not chunks:
            return {"size_consistency": 0, "metadata_richness": 0, "processing_speed": 0}

        # Size consistency
        chunk_sizes = [len(chunk.page_content) for chunk in chunks]
        avg_size = np.mean(chunk_sizes)
        size_variance = np.var(chunk_sizes)
        size_consistency = 1.0 / (1.0 + size_variance / (avg_size ** 2)) if avg_size > 0 else 0

        # Metadata richness
        expected_metadata = ["topics", "entities", "keywords", "difficulty_level"]
        metadata_scores = []

        for chunk in chunks:
            present_metadata = sum(1 for field in expected_metadata 
                                 if field in chunk.metadata and chunk.metadata[field])
            richness = present_metadata / len(expected_metadata)
            metadata_scores.append(richness)

        metadata_richness = np.mean(metadata_scores) if metadata_scores else 0

        return {
            "size_consistency": size_consistency,
            "metadata_richness": metadata_richness,
            "chunk_count": len(chunks),
            "avg_chunk_size": avg_size
        }

class ProductionQualityMonitor:
    """Real-time quality monitoring for production RAG systems."""

    def __init__(self, quality_thresholds: Dict[str, float] = None):
        self.quality_framework = EnterpriseQualityFramework()
        self.thresholds = quality_thresholds or {
            "overall_quality": 0.7,
            "coherence": 0.6,
            "completeness": 0.7,
            "domain_quality": 0.8
        }
        self.quality_history = []
        self.alerts = []

    def monitor_processing_quality(self, chunks: List[Document], 
                                 document_id: str, domain: str = "general") -> Dict[str, Any]:
        """Real-time quality monitoring with alerting."""

        # Assess quality
        quality_assessment = self.quality_framework.assess_enterprise_quality(chunks, domain)

        # Check against thresholds
        quality_alerts = self._check_quality_thresholds(quality_assessment, document_id)

        # Store quality history
        quality_record = {
            "document_id": document_id,
            "timestamp": datetime.now().isoformat(),
            "domain": domain,
            "quality_scores": quality_assessment,
            "alerts": quality_alerts
        }
        self.quality_history.append(quality_record)

        # Generate quality report
        quality_report = self._generate_quality_report(quality_assessment, quality_alerts)

        return {
            "quality_assessment": quality_assessment,
            "quality_alerts": quality_alerts,
            "quality_report": quality_report,
            "processing_recommendation": self._get_processing_recommendation(quality_assessment)
        }

    def _check_quality_thresholds(self, assessment: Dict[str, Any], 
                                document_id: str) -> List[Dict[str, Any]]:
        """Check quality scores against thresholds and generate alerts."""
        alerts = []

        for metric, threshold in self.thresholds.items():
            if metric in assessment and assessment[metric] < threshold:
                alert = {
                    "document_id": document_id,
                    "metric": metric,
                    "actual_score": assessment[metric],
                    "threshold": threshold,
                    "severity": "high" if assessment[metric] < threshold * 0.8 else "medium",
                    "timestamp": datetime.now().isoformat()
                }
                alerts.append(alert)
                self.alerts.append(alert)

        return alerts

    def get_quality_analytics(self, lookback_hours: int = 24) -> Dict[str, Any]:
        """Generate quality analytics for specified time period."""
        cutoff_time = datetime.now() - timedelta(hours=lookback_hours)

        recent_records = [
            record for record in self.quality_history
            if datetime.fromisoformat(record["timestamp"]) > cutoff_time
        ]

        if not recent_records:
            return {"message": "No quality data available for specified period"}

        # Calculate aggregate metrics
        quality_trends = self._analyze_quality_trends(recent_records)
        alert_summary = self._summarize_alerts(lookback_hours)

        return {
            "period_summary": {
                "documents_processed": len(recent_records),
                "avg_quality_score": quality_trends["avg_overall_quality"],
                "quality_trend": quality_trends["trend_direction"],
                "alert_count": len(alert_summary)
            },
            "quality_trends": quality_trends,
            "alert_summary": alert_summary,
            "recommendations": self._generate_system_recommendations(recent_records)
        }