metrics/linguistic.py

# DEPENDENCIES
import re
import numpy as np
from typing import Any
from typing import Dict
from typing import List
from typing import Tuple
from loguru import logger
from collections import Counter
from config.threshold_config import Domain
from metrics.base_metric import BaseMetric
from metrics.base_metric import MetricResult
from models.model_manager import get_model_manager
from config.threshold_config import get_threshold_for_domain


class LinguisticMetric(BaseMetric):
    """
    Linguistic analysis using POS tagging, syntactic complexity, and grammatical patterns
    
    Measures (Aligned with Documentation):
    - POS tag diversity and patterns
    - Syntactic complexity and sentence structure
    - Grammatical patterns and usage
    - Writing style analysis
    """
    def __init__(self):
        super().__init__(name        = "linguistic",
                         description = "POS tag diversity, syntactic complexity, and grammatical pattern analysis",
                        )
        self.nlp = None
    

    def initialize(self) -> bool:
        """
        Initialize the linguistic metric
        """
        try:
            logger.info("Initializing linguistic metric...")
            
            # Load spaCy model for linguistic analysis
            model_manager = get_model_manager()
            self.nlp      = model_manager.load_model("linguistic_spacy")
            
            self.is_initialized = True
            logger.success("Linguistic metric initialized successfully")
            
            return True
            
        except Exception as e:
            logger.error(f"Failed to initialize linguistic metric: {repr(e)}")
            return False
    

    def compute(self, text: str, **kwargs) -> MetricResult:
        """
        Compute linguistic analysis with FULL DOMAIN THRESHOLD INTEGRATION
        """
        try:
            if ((not text) or (len(text.strip()) < 50)):
                return MetricResult(metric_name       = self.name,
                                    ai_probability    = 0.5,
                                    human_probability = 0.5,
                                    mixed_probability = 0.0,
                                    confidence        = 0.1,
                                    error             = "Text too short for linguistic analysis",
                                   )
            
            # Get domain-specific thresholds
            domain                           = kwargs.get('domain', Domain.GENERAL)
            domain_thresholds                = get_threshold_for_domain(domain)
            linguistic_thresholds            = domain_thresholds.linguistic
            
            # Calculate comprehensive linguistic features
            features                         = self._calculate_linguistic_features(text)
            
            # Calculate raw linguistic score (0-1 scale)
            raw_linguistic_score, confidence = self._analyze_linguistic_patterns(features)
            
            # Apply domain-specific thresholds to convert raw score to probabilities
            ai_prob, human_prob, mixed_prob  = self._apply_domain_thresholds(raw_linguistic_score, linguistic_thresholds, features)
            
            # Apply confidence multiplier from domain thresholds
            confidence                      *= linguistic_thresholds.confidence_multiplier
            confidence                       = max(0.0, min(1.0, confidence))
            
            return MetricResult(metric_name       = self.name,
                                ai_probability    = ai_prob,
                                human_probability = human_prob,
                                mixed_probability = mixed_prob,
                                confidence        = confidence,
                                details           = {**features, 
                                                     'domain_used'     : domain.value,
                                                     'ai_threshold'    : linguistic_thresholds.ai_threshold,
                                                     'human_threshold' : linguistic_thresholds.human_threshold,
                                                     'raw_score'       : raw_linguistic_score,
                                                    },
                               )
            
        except Exception as e:
            logger.error(f"Error in linguistic computation: {repr(e)}")
            return MetricResult(metric_name       = self.name,
                                ai_probability    = 0.5,
                                human_probability = 0.5,
                                mixed_probability = 0.0,
                                confidence        = 0.0,
                                error             = str(e),
                               )
    

    def _apply_domain_thresholds(self, raw_score: float, thresholds: Any, features: Dict[str, Any]) -> tuple:
        """
        Apply domain-specific thresholds to convert raw score to probabilities
        """
        ai_threshold    = thresholds.ai_threshold
        human_threshold = thresholds.human_threshold

        # Calculate probabilities based on threshold distances
        if (raw_score >= ai_threshold):
            # Above AI threshold - strongly AI
            distance_from_threshold = raw_score - ai_threshold
            ai_prob                 = 0.7 + (distance_from_threshold * 0.3)  # 0.7 to 1.0
            human_prob              = 0.3 - (distance_from_threshold * 0.3)  # 0.3 to 0.0

        elif (raw_score <= human_threshold):
            # Below human threshold - strongly human
            distance_from_threshold = human_threshold - raw_score
            ai_prob                 = 0.3 - (distance_from_threshold * 0.3)  # 0.3 to 0.0
            human_prob              = 0.7 + (distance_from_threshold * 0.3)  # 0.7 to 1.0
        
        else:
            # Between thresholds - uncertain zone
            range_width             = ai_threshold - human_threshold
            if (range_width > 0):
                position_in_range = (raw_score - human_threshold) / range_width
                ai_prob           = 0.3 + (position_in_range * 0.4)  # 0.3 to 0.7
                human_prob        = 0.7 - (position_in_range * 0.4)  # 0.7 to 0.3
            
            else:
                ai_prob    = 0.5
                human_prob = 0.5
        
        # Ensure probabilities are valid
        ai_prob    = max(0.0, min(1.0, ai_prob))
        human_prob = max(0.0, min(1.0, human_prob))
        
        # Calculate mixed probability based on linguistic variance
        mixed_prob = self._calculate_mixed_probability(features)
        
        # Normalize to sum to 1.0
        total      = ai_prob + human_prob + mixed_prob
        if (total > 0):
            ai_prob    /= total
            human_prob /= total
            mixed_prob /= total
        
        return ai_prob, human_prob, mixed_prob
    

    def _calculate_linguistic_features(self, text: str) -> Dict[str, Any]:
        """
        Calculate comprehensive linguistic analysis features
        """
        if not self.nlp:
            return self._get_default_features()
        
        try:
            # Process text with spaCy
            doc                     = self.nlp(text)
            
            # Extract POS tags and dependencies
            pos_tags                = [token.pos_ for token in doc]
            dependencies            = [token.dep_ for token in doc]
            
            # Calculate POS diversity and patterns
            pos_diversity           = self._calculate_pos_diversity(pos_tags = pos_tags)
            pos_entropy             = self._calculate_pos_entropy(pos_tags = pos_tags)
            
            # Calculate syntactic complexity
            syntactic_complexity    = self._calculate_syntactic_complexity(doc = doc)
            avg_sentence_complexity = self._calculate_sentence_complexity(doc = doc)
            
            # Analyze grammatical patterns
            grammatical_patterns    = self._analyze_grammatical_patterns(doc = doc)
            writing_style_score     = self._analyze_writing_style(doc = doc)
            
            # Chunk-based analysis for whole-text understanding
            chunk_features          = self._calculate_chunk_linguistics(text       = text, 
                                                                        chunk_size = 200,
                                                                       )
            
            # Calculate specific AI linguistic patterns
            ai_pattern_score        = self._detect_ai_linguistic_patterns(doc = doc)
            
            return {"pos_diversity"           : round(pos_diversity, 4),
                    "pos_entropy"             : round(pos_entropy, 4),
                    "syntactic_complexity"    : round(syntactic_complexity, 4),
                    "avg_sentence_complexity" : round(avg_sentence_complexity, 4),
                    "grammatical_consistency" : round(grammatical_patterns['consistency'], 4),
                    "transition_word_usage"   : round(grammatical_patterns['transition_usage'], 4),
                    "passive_voice_ratio"     : round(grammatical_patterns['passive_ratio'], 4),
                    "writing_style_score"     : round(writing_style_score, 4),
                    "ai_pattern_score"        : round(ai_pattern_score, 4),
                    "avg_chunk_complexity"    : round(np.mean(chunk_features['complexities']) if chunk_features['complexities'] else 0.0, 4),
                    "complexity_variance"     : round(np.var(chunk_features['complexities']) if chunk_features['complexities'] else 0.0, 4),
                    "num_sentences"           : len(list(doc.sents)),
                    "num_chunks_analyzed"     : len(chunk_features['complexities']),
                   }
            
        except Exception as e:
            logger.warning(f"Linguistic analysis failed: {repr(e)}")
            return self._get_default_features()
    

    def _calculate_pos_diversity(self, pos_tags: List[str]) -> float:
        """
        Calculate POS tag diversity : Higher diversity = more varied sentence structures
        """
        if not pos_tags:
            return 0.0
        
        unique_pos = len(set(pos_tags))
        total_pos  = len(pos_tags)
        
        diversity  = unique_pos / total_pos
        return diversity
    

    def _calculate_pos_entropy(self, pos_tags: List[str]) -> float:
        """
        Calculate entropy of POS tag distribution
        """
        if not pos_tags:
            return 0.0
        
        pos_counts = Counter(pos_tags)
        total_tags = len(pos_tags)
        
        entropy = 0.0
        for count in pos_counts.values():
            probability = count / total_tags
            entropy    -= probability * np.log2(probability)
        
        return entropy
    

    def _calculate_syntactic_complexity(self, doc) -> float:
        """
        Calculate overall syntactic complexity : based on dependency tree depth and structure
        """
        complexities = list()
        
        for sent in doc.sents:
            # Calculate dependency tree depth
            depths = list()
            for token in sent:
                depth = self._calculate_dependency_depth(token)
                depths.append(depth)
            
            if depths:
                avg_depth  = np.mean(depths)
                max_depth  = np.max(depths)
                complexity = (avg_depth + max_depth) / 2.0
                complexities.append(complexity)
        
        return np.mean(complexities) if complexities else 0.0
    

    def _calculate_dependency_depth(self, token, depth: int = 0) -> int:
        """
        Calculate dependency tree depth for a token
        """
        if not list(token.children):
            return depth
        
        child_depths = [self._calculate_dependency_depth(child, depth + 1) for child in token.children]

        return max(child_depths) if child_depths else depth
    

    def _calculate_sentence_complexity(self, doc) -> float:
        """
        Calculate average sentence complexity
        """
        complexities = list()
        
        for sent in doc.sents:
            # Simple complexity measure based on sentence length and structure
            words       = [token for token in sent if not token.is_punct]
            num_clauses = len([token for token in sent if token.dep_ in ['cc', 'mark']])
            
            if (len(words) > 0):
                complexity = (len(words) / 10.0) + (num_clauses * 0.5)

                complexities.append(complexity)
        
        return np.mean(complexities) if complexities else 0.0
    

    def _analyze_grammatical_patterns(self, doc) -> Dict[str, float]:
        """
        Analyze grammatical patterns and consistency
        """
        # Count different grammatical constructions
        passive_voice        = 0
        active_voice         = 0
        transition_words     = 0
        total_sentences      = 0
        
        transition_words_set = {'however', 'therefore', 'moreover', 'furthermore', 'consequently', 'additionally', 'nevertheless', 'nonetheless', 'thus', 'hence'}
        
        for sent in doc.sents:
            total_sentences += 1
            sent_text        = sent.text.lower()
            
            # Check for passive voice patterns
            if (any(token.dep_ == 'nsubjpass' for token in sent)):
                passive_voice += 1
            
            else:
                active_voice += 1
            
            # Count transition words
            for word in transition_words_set:
                if word in sent_text:
                    transition_words += 1
                    break
        
        # Calculate ratios
        passive_ratio    = passive_voice / total_sentences if total_sentences > 0 else 0.0
        transition_usage = transition_words / total_sentences if total_sentences > 0 else 0.0
        
        # Calculate consistency (lower variance in patterns)
        consistency      = 1.0 - min(1.0, abs(passive_ratio - 0.3) + abs(transition_usage - 0.2))
        
        return {'consistency'      : max(0.0, consistency),
                'passive_ratio'    : passive_ratio,
                'transition_usage' : transition_usage,
               }
    

    def _analyze_writing_style(self, doc) -> float:
        """
        Analyze writing style characteristics
        """
        style_indicators = list()
        
        # Sentence length variation
        sent_lengths = [len([token for token in sent if not token.is_punct]) for sent in doc.sents]
        
        if sent_lengths:
            length_variation = np.std(sent_lengths) / np.mean(sent_lengths) if np.mean(sent_lengths) > 0 else 0.0
            # Moderate variation is more human-like
            style_score      = 1.0 - min(1.0, abs(length_variation - 0.5))

            style_indicators.append(style_score)
        
        # Punctuation usage
        punct_ratio = len([token for token in doc if token.is_punct]) / len(doc) if len(doc) > 0 else 0.0
        # Balanced punctuation is more human-like
        punct_score = 1.0 - min(1.0, abs(punct_ratio - 0.1))

        style_indicators.append(punct_score)
        
        return np.mean(style_indicators) if style_indicators else 0.5
    

    def _detect_ai_linguistic_patterns(self, doc) -> float:
        """
        Detect AI-specific linguistic patterns
        """ 
        patterns_detected     = 0
        total_patterns        = 5
        
        # Pattern 1: Overuse of certain transition words
        transition_overuse    = self._check_transition_overuse(doc)
        
        if transition_overuse:
            patterns_detected += 1
        
        # Pattern 2: Unnatural POS sequences
        pos_sequences         = self._check_unnatural_pos_sequences(doc)
        
        if pos_sequences:
            patterns_detected += 1
        
        # Pattern 3: Overly consistent sentence structures
        structure_consistency = self._check_structure_consistency(doc)
        
        if structure_consistency:
            patterns_detected += 1
        
        # Pattern 4: Unusual grammatical constructions
        unusual_grammar       = self._check_unusual_grammar(doc)
        
        if unusual_grammar:
            patterns_detected += 1
        
        # Pattern 5: Repetitive phrasing
        repetitive_phrasing   = self._check_repetitive_phrasing(doc)
       
        if repetitive_phrasing:
            patterns_detected += 1
        
        return patterns_detected / total_patterns
    

    def _check_transition_overuse(self, doc) -> bool:
        """
        Check for overuse of transition words (common AI pattern)
        """
        transition_words = {'however', 'therefore', 'moreover', 'furthermore', 'additionally'}
        transition_count = sum(1 for token in doc if token.lemma_.lower() in transition_words)
        
        # More than 5% of words being transitions is suspicious
        return transition_count / len(doc) > 0.05 if len(doc) > 0 else False
    

    def _check_unnatural_pos_sequences(self, doc) -> bool:
        """
        Check for unnatural POS tag sequences
        """
        pos_sequences = list()

        for sent in doc.sents:
            sent_pos = [token.pos_ for token in sent]
            pos_sequences.extend([(sent_pos[i], sent_pos[i+1]) for i in range(len(sent_pos)-1)])
        
        # Look for repetitive or unnatural sequences
        if not pos_sequences:
            return False
        
        sequence_counts  = Counter(pos_sequences)
        most_common_freq = max(sequence_counts.values()) / len(pos_sequences) if pos_sequences else 0
        
        # High frequency of specific sequences suggests AI
        return (most_common_freq > 0.1)
    

    def _check_structure_consistency(self, doc) -> bool:
        """
        Check for overly consistent sentence structures
        """
        sent_structures = list()
        
        for sent in doc.sents:
            # Simple structure representation
            structure = tuple(token.dep_ for token in sent if token.dep_ not in ['punct', 'det'])
            sent_structures.append(structure)
        
        if (len(sent_structures) < 3):
            return False
        
        # Calculate structure similarity
        unique_structures = len(set(sent_structures))
        similarity_ratio  = unique_structures / len(sent_structures)
        
        # Low diversity suggests AI
        return (similarity_ratio < 0.5)
    

    def _check_unusual_grammar(self, doc) -> bool:
        """
        Check for unusual grammatical constructions
        """
        unusual_constructions = 0
        
        for token in doc:
            # Check for unusual dependency relations i.e. less common relations
            if token.dep_ in ['attr', 'oprd']:  
                unusual_constructions += 1
        
        # More than 2% unusual constructions is suspicious
        return (unusual_constructions / len(doc) > 0.02) if (len(doc) > 0) else False
    

    def _check_repetitive_phrasing(self, doc) -> bool:
        """
        Check for repetitive phrasing patterns
        """
        phrases = list()

        for sent in doc.sents:
            # Extract noun phrases
            noun_phrases = [chunk.text.lower() for chunk in sent.noun_chunks]
            phrases.extend(noun_phrases)
        
        if not phrases:
            return False
        
        phrase_counts    = Counter(phrases)
        repeated_phrases = sum(1 for count in phrase_counts.values() if count > 1)
        
        # High repetition suggests AI
        return (repeated_phrases / len(phrases) > 0.3)
    

    def _calculate_chunk_linguistics(self, text: str, chunk_size: int = 200) -> Dict[str, List[float]]:
        """
        Calculate linguistic features across text chunks
        """
        complexities = list()
        words        = text.split()
        
        for i in range(0, len(words), chunk_size // 2):
            chunk = ' '.join(words[i:i + chunk_size])
            
            if (len(chunk) > 50):
                try:
                    chunk_doc = self.nlp(chunk)
                    
                    # Check if processing was successful
                    if (chunk_doc and (len(list(chunk_doc.sents)) > 0)):
                        complexity = self._calculate_syntactic_complexity(chunk_doc)
                        complexities.append(complexity)
                
                except Exception as e:
                    logger.debug(f"Chunk linguistic analysis failed: {e}")
                    continue
        
        return {'complexities': complexities}
    

    def _analyze_linguistic_patterns(self, features: Dict[str, Any]) -> tuple:
        """
        Analyze linguistic patterns to determine RAW linguistic score (0-1 scale) : Higher score = more AI-like
        """
        # Check feature validity first
        required_features = ['pos_diversity', 'syntactic_complexity', 'grammatical_consistency', 'transition_word_usage', 'ai_pattern_score', 'complexity_variance']
        
        valid_features    = [features.get(feat, 0) for feat in required_features if features.get(feat, 0) > 0]
        
        if (len(valid_features) < 4):
            # Low confidence if insufficient features
            return 0.5, 0.3  

        # Initialize ai_indicator list 
        ai_indicators = list()
        
        # Low POS diversity suggests AI
        if (features['pos_diversity'] < 0.3):
            ai_indicators.append(0.8)

        elif (features['pos_diversity'] < 0.5):
            ai_indicators.append(0.6)

        else:
            ai_indicators.append(0.2)
        
        # Low syntactic complexity suggests AI
        if (features['syntactic_complexity'] < 2.0):
            ai_indicators.append(0.7)

        elif (features['syntactic_complexity'] < 3.0):
            ai_indicators.append(0.4)

        else:
            ai_indicators.append(0.2)
        
        # High grammatical consistency suggests AI (unnaturally consistent)
        if (features['grammatical_consistency'] > 0.8):
            ai_indicators.append(0.9)

        elif (features['grammatical_consistency'] > 0.6):
            ai_indicators.append(0.5)

        else:
            ai_indicators.append(0.3)
        
        # High transition word usage suggests AI
        if (features['transition_word_usage'] > 0.3):
            ai_indicators.append(0.7)
        
        elif (features['transition_word_usage'] > 0.15):
            ai_indicators.append(0.4)

        else:
            ai_indicators.append(0.2)
        
        # High AI pattern score suggests AI
        if (features['ai_pattern_score'] > 0.6):
            ai_indicators.append(0.8)

        elif (features['ai_pattern_score'] > 0.3):
            ai_indicators.append(0.5)

        else:
            ai_indicators.append(0.2)
        
        # Low complexity variance suggests AI
        if (features['complexity_variance'] < 0.1):
            ai_indicators.append(0.7)

        elif (features['complexity_variance'] < 0.3):
            ai_indicators.append(0.4)

        else:
            ai_indicators.append(0.2)
        
        # Calculate raw score and confidence
        raw_score  = np.mean(ai_indicators) if ai_indicators else 0.5
        confidence = 1.0 - (np.std(ai_indicators) / 0.5) if ai_indicators else 0.5
        confidence = max(0.1, min(0.9, confidence))
        
        return raw_score, confidence
    

    def _calculate_mixed_probability(self, features: Dict[str, Any]) -> float:
        """
        Calculate probability of mixed AI/Human content
        """
        mixed_indicators = list()
        
        # Moderate POS diversity might indicate mixing
        if (0.35 <= features['pos_diversity'] <= 0.55):
            mixed_indicators.append(0.3)

        else:
            mixed_indicators.append(0.0)
        
        # High complexity variance suggests mixed content
        if (features['complexity_variance'] > 0.5):
            mixed_indicators.append(0.4)

        elif (features['complexity_variance'] > 0.3):
            mixed_indicators.append(0.2)

        else:
            mixed_indicators.append(0.0)
        
        # Inconsistent AI pattern detection
        if (0.2 <= features['ai_pattern_score'] <= 0.6):
            mixed_indicators.append(0.3)

        else:
            mixed_indicators.append(0.0)
        
        return min(0.3, np.mean(mixed_indicators)) if mixed_indicators else 0.0
    

    def _get_default_features(self) -> Dict[str, Any]:
        """
        Return default features when analysis is not possible
        """
        return {"pos_diversity"           : 0.5,
                "pos_entropy"             : 2.5,
                "syntactic_complexity"    : 2.5,
                "avg_sentence_complexity" : 2.0,
                "grammatical_consistency" : 0.5,
                "transition_word_usage"   : 0.1,
                "passive_voice_ratio"     : 0.2,
                "writing_style_score"     : 0.5,
                "ai_pattern_score"        : 0.3,
                "avg_chunk_complexity"    : 2.5,
                "complexity_variance"     : 0.2,
                "num_sentences"           : 0,
                "num_chunks_analyzed"     : 0,
               }
    

    def cleanup(self):
        """
        Clean up resources
        """
        self.nlp = None
        super().cleanup()



# Export
__all__ = ["LinguisticMetric"]