text_pattern_analyzer.py

#!/usr/bin/env python3
"""
Text Pattern Analyzer
A comprehensive tool for detecting and extracting repeating patterns, keywords,
and structural similarities in text data.
"""

import re
import csv
import json
import math
import itertools
from collections import Counter, defaultdict
from typing import List, Dict, Tuple, Optional, Union, Generator
from pathlib import Path
import statistics

class TextPreprocessor:
    """Handles text preprocessing and normalization."""
    
    def __init__(self, preserve_case: bool = False, preserve_punctuation: bool = False):
        self.preserve_case = preserve_case
        self.preserve_punctuation = preserve_punctuation
    
    def clean_text(self, text: str) -> str:
        """Clean and normalize text."""
        if not self.preserve_case:
            text = text.lower()
        
        if not self.preserve_punctuation:
            # Replace punctuation with spaces but preserve sentence boundaries
            text = re.sub(r'[^\w\s]', ' ', text)
        
        # Normalize whitespace
        text = re.sub(r'\s+', ' ', text).strip()
        return text
    
    def tokenize_sentences(self, text: str) -> List[str]:
        """Split text into sentences."""
        # Simple sentence splitting on common terminators
        sentences = re.split(r'[.!?]+', text)
        return [s.strip() for s in sentences if s.strip()]
    
    def tokenize_words(self, text: str) -> List[str]:
        """Split text into words."""
        return text.split()
    
    def extract_paragraphs(self, text: str) -> List[str]:
        """Extract paragraphs from text."""
        paragraphs = re.split(r'\n\s*\n', text)
        return [p.strip() for p in paragraphs if p.strip()]

class PatternDetector:
    """Core pattern detection and analysis engine."""
    
    def __init__(self, preprocessor: TextPreprocessor):
        self.preprocessor = preprocessor
    
    def generate_ngrams(self, tokens: List[str], n: int) -> Generator[Tuple[str, ...], None, None]:
        """Generate n-grams from tokens."""
        for i in range(len(tokens) - n + 1):
            yield tuple(tokens[i:i + n])
    
    def find_repeated_ngrams(self, text: str, n: int, min_frequency: int = 2) -> Dict[Tuple[str, ...], int]:
        """Find repeated n-grams with their frequencies."""
        cleaned_text = self.preprocessor.clean_text(text)
        tokens = self.preprocessor.tokenize_words(cleaned_text)
        
        ngram_counts = Counter(self.generate_ngrams(tokens, n))
        return {ngram: count for ngram, count in ngram_counts.items() if count >= min_frequency}
    
    def find_phrase_patterns(self, text: str, min_length: int = 3, max_length: int = 10) -> Dict[str, Dict]:
        """Find repeating phrase patterns of varying lengths."""
        results = {}
        
        for n in range(min_length, max_length + 1):
            ngrams = self.find_repeated_ngrams(text, n)
            for ngram, count in ngrams.items():
                phrase = ' '.join(ngram)
                results[phrase] = {
                    'frequency': count,
                    'length': n,
                    'words': ngram
                }
        
        return results
    
    def find_sentence_structure_patterns(self, text: str) -> Dict[str, Dict]:
        """Identify recurring sentence structures using POS-like patterns."""
        sentences = self.preprocessor.tokenize_sentences(text)
        structure_patterns = defaultdict(list)
        
        for i, sentence in enumerate(sentences):
            cleaned_sentence = self.preprocessor.clean_text(sentence)
            words = self.preprocessor.tokenize_words(cleaned_sentence)
            
            # Create a simple structure pattern based on word characteristics
            structure = []
            for word in words:
                if word.isdigit():
                    structure.append('NUM')
                elif len(word) <= 2:
                    structure.append('SHORT')
                elif len(word) >= 8:
                    structure.append('LONG')
                elif word.isupper():
                    structure.append('UPPER')
                else:
                    structure.append('WORD')
            
            pattern = '-'.join(structure)
            structure_patterns[pattern].append({
                'sentence': sentence.strip(),
                'position': i,
                'word_count': len(words)
            })
        
        # Filter patterns that appear more than once
        return {pattern: data for pattern, data in structure_patterns.items() if len(data) > 1}
    
    def detect_statistical_anomalies(self, word_frequencies: Dict[str, int], 
                                   threshold_std: float = 2.0) -> Dict[str, Dict]:
        """Detect words with statistically anomalous frequencies."""
        if not word_frequencies:
            return {}
        
        frequencies = list(word_frequencies.values())
        mean_freq = statistics.mean(frequencies)
        
        try:
            std_freq = statistics.stdev(frequencies)
        except statistics.StatisticsError:
            return {}
        
        anomalies = {}
        for word, freq in word_frequencies.items():
            if std_freq > 0:
                z_score = (freq - mean_freq) / std_freq
                if abs(z_score) > threshold_std:
                    anomalies[word] = {
                        'frequency': freq,
                        'z_score': z_score,
                        'type': 'high_frequency' if z_score > 0 else 'low_frequency'
                    }
        
        return anomalies

class LocationTracker:
    """Tracks contextual locations of patterns within text."""
    
    def __init__(self, preprocessor: TextPreprocessor):
        self.preprocessor = preprocessor
    
    def find_pattern_locations(self, original_text: str, pattern: str) -> List[Dict]:
        """Find all locations of a pattern in the original text."""
        locations = []
        cleaned_pattern = self.preprocessor.clean_text(pattern)
        
        # Split text into lines for context
        lines = original_text.split('\n')
        
        for line_num, line in enumerate(lines, 1):
            cleaned_line = self.preprocessor.clean_text(line)
            
            # Find all occurrences in this line
            start_pos = 0
            while True:
                pos = cleaned_line.find(cleaned_pattern, start_pos)
                if pos == -1:
                    break
                
                # Calculate character position in original text
                char_pos = sum(len(lines[i]) + 1 for i in range(line_num - 1)) + pos
                
                locations.append({
                    'line': line_num,
                    'char_position': char_pos,
                    'context': line.strip(),
                    'pattern_start': pos,
                    'pattern_end': pos + len(cleaned_pattern)
                })
                
                start_pos = pos + 1
        
        return locations
    
    def highlight_patterns_in_text(self, text: str, patterns: List[str], 
                                 highlight_format: str = "**{}**") -> str:
        """Highlight patterns in text using specified format."""
        highlighted_text = text
        
        # Sort patterns by length (longest first) to avoid partial replacements
        sorted_patterns = sorted(patterns, key=len, reverse=True)
        
        for pattern in sorted_patterns:
            # Create case-insensitive pattern if not preserving case
            if not self.preprocessor.preserve_case:
                pattern_regex = re.compile(re.escape(pattern), re.IGNORECASE)
                highlighted_text = pattern_regex.sub(
                    lambda m: highlight_format.format(m.group()), 
                    highlighted_text
                )
            else:
                highlighted_text = highlighted_text.replace(
                    pattern, highlight_format.format(pattern)
                )
        
        return highlighted_text

class InputHandler:
    """Handles different input sources and formats."""
    
    @staticmethod
    def read_text_file(file_path: str, encoding: str = 'utf-8') -> str:
        """Read text from a plain text file."""
        with open(file_path, 'r', encoding=encoding) as f:
            return f.read()
    
    @staticmethod
    def read_csv_file(file_path: str, text_column: str, encoding: str = 'utf-8') -> str:
        """Read and concatenate text from a specific column in CSV."""
        texts = []
        with open(file_path, 'r', encoding=encoding, newline='') as f:
            reader = csv.DictReader(f)
            for row in reader:
                if text_column in row and row[text_column]:
                    texts.append(row[text_column])
        return '\n'.join(texts)
    
    @staticmethod
    def process_string_input(text: str) -> str:
        """Process direct string input."""
        return text

class OutputFormatter:
    """Formats and exports analysis results in various formats."""
    
    @staticmethod
    def format_json(results: Dict) -> str:
        """Format results as JSON."""
        return json.dumps(results, indent=2, ensure_ascii=False)
    
    @staticmethod
    def format_csv(results: Dict) -> str:
        """Format results as CSV."""
        output_lines = []
        
        # Handle different result types
        if 'phrase_patterns' in results:
            output_lines.append('Pattern Type,Pattern,Frequency,Length,Locations')
            for pattern, data in results['phrase_patterns'].items():
                locations_count = len(data.get('locations', []))
                output_lines.append(f"Phrase,\"{pattern}\",{data['frequency']},{data['length']},{locations_count}")
        
        if 'anomalies' in results:
            output_lines.append('\nAnomaly Type,Word,Frequency,Z-Score')
            for word, data in results['anomalies'].items():
                output_lines.append(f"{data['type']},\"{word}\",{data['frequency']},{data['z_score']:.2f}")
        
        return '\n'.join(output_lines)
    
    @staticmethod
    def format_plain_text(results: Dict) -> str:
        """Format results as plain text report."""
        report_lines = ['TEXT PATTERN ANALYSIS REPORT', '=' * 50, '']
        
        if 'phrase_patterns' in results:
            report_lines.extend(['REPEATING PHRASE PATTERNS:', '-' * 30])
            for pattern, data in results['phrase_patterns'].items():
                report_lines.append(f"Pattern: {pattern}")
                report_lines.append(f"  Frequency: {data['frequency']}")
                report_lines.append(f"  Length: {data['length']} words")
                if 'locations' in data:
                    report_lines.append(f"  Locations: {len(data['locations'])} occurrences")
                report_lines.append('')
        
        if 'sentence_structures' in results:
            report_lines.extend(['SENTENCE STRUCTURE PATTERNS:', '-' * 35])
            for structure, sentences in results['sentence_structures'].items():
                report_lines.append(f"Structure: {structure}")
                report_lines.append(f"  Occurrences: {len(sentences)}")
                report_lines.append(f"  Example: {sentences[0]['sentence'][:100]}...")
                report_lines.append('')
        
        if 'anomalies' in results:
            report_lines.extend(['STATISTICAL ANOMALIES:', '-' * 25])
            for word, data in results['anomalies'].items():
                report_lines.append(f"Word: {word}")
                report_lines.append(f"  Frequency: {data['frequency']}")
                report_lines.append(f"  Z-Score: {data['z_score']:.2f}")
                report_lines.append(f"  Type: {data['type']}")
                report_lines.append('')
        
        return '\n'.join(report_lines)

class TextPatternAnalyzer:
    """Main analyzer class that orchestrates the analysis process."""
    
    def __init__(self, preserve_case: bool = False, preserve_punctuation: bool = False):
        self.preprocessor = TextPreprocessor(preserve_case, preserve_punctuation)
        self.pattern_detector = PatternDetector(self.preprocessor)
        self.location_tracker = LocationTracker(self.preprocessor)
        self.output_formatter = OutputFormatter()
    
    def analyze_text(self, text: str, 
                    min_ngram: int = 2, 
                    max_ngram: int = 5,
                    min_frequency: int = 2,
                    detect_anomalies: bool = True,
                    anomaly_threshold: float = 2.0,
                    track_locations: bool = True) -> Dict:
        """Perform comprehensive text analysis."""
        
        results = {
            'text_stats': {
                'total_characters': len(text),
                'total_words': len(self.preprocessor.tokenize_words(text)),
                'total_sentences': len(self.preprocessor.tokenize_sentences(text))
            }
        }
        
        # Find phrase patterns
        phrase_patterns = self.pattern_detector.find_phrase_patterns(
            text, min_ngram, max_ngram
        )
        
        # Filter by minimum frequency
        phrase_patterns = {
            pattern: data for pattern, data in phrase_patterns.items() 
            if data['frequency'] >= min_frequency
        }
        
        # Add location information if requested
        if track_locations:
            for pattern in phrase_patterns:
                locations = self.location_tracker.find_pattern_locations(text, pattern)
                phrase_patterns[pattern]['locations'] = locations
        
        results['phrase_patterns'] = phrase_patterns
        
        # Find sentence structure patterns
        sentence_structures = self.pattern_detector.find_sentence_structure_patterns(text)
        results['sentence_structures'] = sentence_structures
        
        # Detect statistical anomalies
        if detect_anomalies:
            cleaned_text = self.preprocessor.clean_text(text)
            words = self.preprocessor.tokenize_words(cleaned_text)
            word_frequencies = Counter(words)
            
            anomalies = self.pattern_detector.detect_statistical_anomalies(
                word_frequencies, anomaly_threshold
            )
            results['anomalies'] = anomalies
        
        return results
    
    def analyze_file(self, file_path: str, **kwargs) -> Dict:
        """Analyze text from a file."""
        path = Path(file_path)
        
        if path.suffix.lower() == '.csv':
            text_column = kwargs.pop('text_column', 'text')
            text = InputHandler.read_csv_file(file_path, text_column)
        else:
            text = InputHandler.read_text_file(file_path)
        
        return self.analyze_text(text, **kwargs)
    
    def highlight_text(self, text: str, patterns: List[str], 
                      highlight_format: str = "**{}**") -> str:
        """Highlight patterns in the original text."""
        return self.location_tracker.highlight_patterns_in_text(
            text, patterns, highlight_format
        )
    
    def export_results(self, results: Dict, output_format: str = 'json') -> str:
        """Export results in specified format."""
        if output_format.lower() == 'json':
            return self.output_formatter.format_json(results)
        elif output_format.lower() == 'csv':
            return self.output_formatter.format_csv(results)
        elif output_format.lower() == 'txt':
            return self.output_formatter.format_plain_text(results)
        else:
            raise ValueError(f"Unsupported output format: {output_format}")
    
    def save_results(self, results: Dict, output_file: str, output_format: str = 'json'):
        """Save results to a file."""
        formatted_results = self.export_results(results, output_format)
        
        with open(output_file, 'w', encoding='utf-8') as f:
            f.write(formatted_results)

# Example usage and demonstration
if __name__ == "__main__":
    # Initialize analyzer
    analyzer = TextPatternAnalyzer(preserve_case=False, preserve_punctuation=False)
    
    # Sample text for analysis
    sample_text = """
    The quick brown fox jumps over the lazy dog. The quick brown fox is very agile.
    Machine learning algorithms are powerful tools. Machine learning has revolutionized data science.
    Data science is an interdisciplinary field. Data science combines statistics and programming.
    The quick brown fox appears frequently in typing exercises. Quick brown animals are common in nature.
    Statistical analysis reveals patterns in data. Statistical methods help identify trends.
    """
    
    # Perform analysis
    results = analyzer.analyze_text(
        sample_text,
        min_ngram=2,
        max_ngram=4,
        min_frequency=2,
        detect_anomalies=True,
        track_locations=True
    )
    
    # Print results in different formats
    print("=== JSON FORMAT ===")
    print(analyzer.export_results(results, 'json'))
    
    print("\n=== PLAIN TEXT REPORT ===")
    print(analyzer.export_results(results, 'txt'))
    
    # Demonstrate text highlighting
    top_patterns = list(results['phrase_patterns'].keys())[:3]
    highlighted_text = analyzer.highlight_text(sample_text, top_patterns)
    print("\n=== HIGHLIGHTED TEXT ===")
    print(highlighted_text)