Spaces:

Ndg07
/

heatmap

Running

heatmap / backend /data_sources /base /data_validator.py

Initial commit: Enhanced Fake News Detection System

3f0cc2f 12 months ago

14.7 kB

#!/usr/bin/env python3
"""
Data validation utilities for ensuring quality of ingested events.
Implements multi-layer filtering and validation logic.
"""

import logging
import re
from typing import Dict, List, Optional, Set
from datetime import datetime, timedelta

from .base_connector import RawEvent

logger = logging.getLogger(__name__)


class DataValidator:
    """Multi-layer data validation for raw events."""
    
    def __init__(self, config: Dict = None):
        """Initialize validator with configuration.
        
        Args:
            config: Validation configuration dictionary
        """
        self.config = config or {}
        
        # Content quality thresholds
        self.min_content_length = self.config.get('min_content_length', 10)
        self.max_content_length = self.config.get('max_content_length', 50000)
        self.min_word_count = self.config.get('min_word_count', 3)
        
        # Language settings
        self.allowed_languages = set(self.config.get('allowed_languages', [
            'en', 'hi', 'bn', 'te', 'ta', 'mr', 'gu', 'kn', 'ml', 'or', 'pa', 'as'
        ]))
        
        # Geographic relevance
        self.require_india_relevance = self.config.get('require_india_relevance', True)
        
        # Spam and quality filters
        self.spam_keywords = set(self.config.get('spam_keywords', [
            'buy now', 'click here', 'limited offer', 'act now', 'free money',
            'earn money', 'work from home', 'get rich quick', 'miracle cure',
            'lose weight fast', 'enlargement', 'viagra', 'casino', 'lottery'
        ]))
        
        # Indian context keywords for relevance checking
        self.india_keywords = set([
            'india', 'indian', 'bharat', 'bharatiya', 'delhi', 'mumbai', 'bangalore',
            'chennai', 'kolkata', 'hyderabad', 'pune', 'ahmedabad', 'surat', 'jaipur',
            'lucknow', 'kanpur', 'nagpur', 'indore', 'bhopal', 'visakhapatnam', 'patna',
            'vadodara', 'ludhiana', 'agra', 'nashik', 'faridabad', 'meerut', 'rajkot',
            'kalyan', 'vasai', 'varanasi', 'srinagar', 'aurangabad', 'dhanbad', 'amritsar',
            'navi mumbai', 'allahabad', 'ranchi', 'howrah', 'coimbatore', 'jabalpur',
            'gwalior', 'vijayawada', 'jodhpur', 'madurai', 'raipur', 'kota', 'guwahati',
            'chandigarh', 'solapur', 'hubli', 'tiruchirappalli', 'bareilly', 'mysore',
            'tiruppur', 'gurgaon', 'aligarh', 'jalandhar', 'bhubaneswar', 'salem',
            'warangal', 'mira', 'bhiwandi', 'thiruvananthapuram', 'bhavnagar', 'dehradun',
            'durgapur', 'asansol', 'nanded', 'kolhapur', 'ajmer', 'gulbarga', 'jamnagar',
            'ujjain', 'loni', 'siliguri', 'jhansi', 'ulhasnagar', 'nellore', 'jammu',
            'sangli', 'belgaum', 'mangalore', 'ambattur', 'tirunelveli', 'malegaon',
            'gaya', 'jalgaon', 'udaipur', 'maheshtala'
        ])
        
        # Indian states and union territories
        self.indian_states = set([
            'andhra pradesh', 'arunachal pradesh', 'assam', 'bihar', 'chhattisgarh',
            'goa', 'gujarat', 'haryana', 'himachal pradesh', 'jharkhand', 'karnataka',
            'kerala', 'madhya pradesh', 'maharashtra', 'manipur', 'meghalaya', 'mizoram',
            'nagaland', 'odisha', 'punjab', 'rajasthan', 'sikkim', 'tamil nadu',
            'telangana', 'tripura', 'uttar pradesh', 'uttarakhand', 'west bengal',
            'andaman and nicobar islands', 'chandigarh', 'dadra and nagar haveli',
            'daman and diu', 'delhi', 'lakshadweep', 'puducherry', 'jammu and kashmir',
            'ladakh'
        ])
        
        # Statistics
        self.stats = {
            'total_validated': 0,
            'passed': 0,
            'failed': 0,
            'failure_reasons': {}
        }
    
    def validate_event(self, event: RawEvent) -> tuple[bool, Optional[str]]:
        """Validate a raw event through multiple layers.
        
        Args:
            event: RawEvent to validate
            
        Returns:
            Tuple of (is_valid, failure_reason)
        """
        self.stats['total_validated'] += 1
        
        try:
            # Layer 1: Basic validation
            is_valid, reason = self._validate_basic(event)
            if not is_valid:
                self._record_failure(reason)
                return False, reason
            
            # Layer 2: Content quality
            is_valid, reason = self._validate_content_quality(event)
            if not is_valid:
                self._record_failure(reason)
                return False, reason
            
            # Layer 3: Language validation
            is_valid, reason = self._validate_language(event)
            if not is_valid:
                self._record_failure(reason)
                return False, reason
            
            # Layer 4: Geographic relevance
            if self.require_india_relevance:
                is_valid, reason = self._validate_india_relevance(event)
                if not is_valid:
                    self._record_failure(reason)
                    return False, reason
            
            # Layer 5: Spam and quality filtering
            is_valid, reason = self._validate_not_spam(event)
            if not is_valid:
                self._record_failure(reason)
                return False, reason
            
            # All validations passed
            self.stats['passed'] += 1
            return True, None
            
        except Exception as e:
            logger.error(f"Validation error for event {event.event_id}: {e}")
            self._record_failure(f"validation_error: {str(e)}")
            return False, f"validation_error: {str(e)}"
    
    def _validate_basic(self, event: RawEvent) -> tuple[bool, Optional[str]]:
        """Basic validation checks."""
        
        # Check required fields
        if not event.content:
            return False, "empty_content"
        
        if not event.source_id:
            return False, "missing_source_id"
        
        if not event.timestamp:
            return False, "missing_timestamp"
        
        # Check content length
        content_length = len(event.content)
        if content_length < self.min_content_length:
            return False, f"content_too_short: {content_length} < {self.min_content_length}"
        
        if content_length > self.max_content_length:
            return False, f"content_too_long: {content_length} > {self.max_content_length}"
        
        # Check word count
        word_count = len(event.content.split())
        if word_count < self.min_word_count:
            return False, f"insufficient_words: {word_count} < {self.min_word_count}"
        
        # Check timestamp is reasonable (not too old or in future)
        now = datetime.now(event.timestamp.tzinfo or None)
        age = now - event.timestamp
        
        if age.days > 30:  # Older than 30 days
            return False, f"content_too_old: {age.days} days"
        
        if age.total_seconds() < -3600:  # More than 1 hour in future
            return False, "content_from_future"
        
        return True, None
    
    def _validate_content_quality(self, event: RawEvent) -> tuple[bool, Optional[str]]:
        """Validate content quality and structure."""
        
        content = event.content.lower()
        
        # Check for excessive repetition
        words = content.split()
        if len(words) > 10:
            unique_words = set(words)
            repetition_ratio = len(unique_words) / len(words)
            if repetition_ratio < 0.3:  # Less than 30% unique words
                return False, f"excessive_repetition: {repetition_ratio:.2f}"
        
        # Check for excessive capitalization
        if len(event.content) > 50:
            caps_ratio = sum(1 for c in event.content if c.isupper()) / len(event.content)
            if caps_ratio > 0.5:  # More than 50% uppercase
                return False, f"excessive_caps: {caps_ratio:.2f}"
        
        # Check for excessive punctuation
        punct_count = sum(1 for c in event.content if c in '!?.,;:')
        if len(words) > 0:
            punct_ratio = punct_count / len(words)
            if punct_ratio > 0.3:  # More than 30% punctuation to words ratio
                return False, f"excessive_punctuation: {punct_ratio:.2f}"
        
        # Check for minimum sentence structure
        sentences = re.split(r'[.!?]+', event.content)
        valid_sentences = [s.strip() for s in sentences if len(s.strip()) > 5]
        if len(valid_sentences) == 0 and len(words) > 20:
            return False, "no_sentence_structure"
        
        return True, None
    
    def _validate_language(self, event: RawEvent) -> tuple[bool, Optional[str]]:
        """Validate language is in allowed set."""
        
        if not self.allowed_languages:
            return True, None  # No language restrictions
        
        # Use detected language or try to detect
        language = event.language
        if not language:
            language = self._detect_language_simple(event.content)
        
        if language and language not in self.allowed_languages:
            return False, f"unsupported_language: {language}"
        
        return True, None
    
    def _validate_india_relevance(self, event: RawEvent) -> tuple[bool, Optional[str]]:
        """Validate content is relevant to India."""
        
        content_lower = event.content.lower()
        title_lower = (event.title or "").lower()
        
        # Check for India-related keywords
        text_to_check = f"{content_lower} {title_lower}"
        
        # Direct keyword matching
        for keyword in self.india_keywords:
            if keyword in text_to_check:
                return True, None
        
        # Check for Indian states
        for state in self.indian_states:
            if state in text_to_check:
                return True, None
        
        # Check location hint
        if event.location_hint:
            location_lower = event.location_hint.lower()
            if any(keyword in location_lower for keyword in self.india_keywords):
                return True, None
            if any(state in location_lower for state in self.indian_states):
                return True, None
        
        # Check metadata for India relevance
        if event.metadata:
            metadata_str = str(event.metadata).lower()
            if any(keyword in metadata_str for keyword in self.india_keywords):
                return True, None
        
        return False, "not_india_relevant"
    
    def _validate_not_spam(self, event: RawEvent) -> tuple[bool, Optional[str]]:
        """Validate content is not spam or low quality."""
        
        content_lower = event.content.lower()
        
        # Check for spam keywords
        for spam_keyword in self.spam_keywords:
            if spam_keyword in content_lower:
                return False, f"spam_keyword: {spam_keyword}"
        
        # Check for excessive URLs
        url_count = len(re.findall(r'http[s]?://(?:[a-zA-Z]|[0-9]|[$-_@.&+]|[!*\\(\\),]|(?:%[0-9a-fA-F][0-9a-fA-F]))+', event.content))
        word_count = len(event.content.split())
        if word_count > 0 and url_count / word_count > 0.2:  # More than 20% URLs
            return False, f"excessive_urls: {url_count}/{word_count}"
        
        # Check for excessive numbers (might indicate spam/ads)
        number_count = len(re.findall(r'\b\d+\b', event.content))
        if word_count > 0 and number_count / word_count > 0.3:  # More than 30% numbers
            return False, f"excessive_numbers: {number_count}/{word_count}"
        
        # Check for promotional language patterns
        promo_patterns = [
            r'\b(call|contact|visit)\s+(?:us\s+)?(?:at\s+)?\+?\d{10,}',  # Phone numbers
            r'\b(?:only|just)\s+(?:rs\.?|₹)\s*\d+',  # Price mentions
            r'\b(?:free|discount|offer|sale)\b.*\b(?:today|now|limited)\b',  # Promotional offers
        ]
        
        for pattern in promo_patterns:
            if re.search(pattern, content_lower):
                return False, f"promotional_content: {pattern}"
        
        return True, None
    
    def _detect_language_simple(self, content: str) -> Optional[str]:
        """Simple language detection based on script."""
        
        if not content:
            return None
        
        # Count characters by script
        latin_count = sum(1 for c in content if ord(c) < 256 and c.isalpha())
        devanagari_count = sum(1 for c in content if 0x0900 <= ord(c) <= 0x097F)
        bengali_count = sum(1 for c in content if 0x0980 <= ord(c) <= 0x09FF)
        tamil_count = sum(1 for c in content if 0x0B80 <= ord(c) <= 0x0BFF)
        telugu_count = sum(1 for c in content if 0x0C00 <= ord(c) <= 0x0C7F)
        gujarati_count = sum(1 for c in content if 0x0A80 <= ord(c) <= 0x0AFF)
        
        total_alpha = sum([latin_count, devanagari_count, bengali_count, tamil_count, telugu_count, gujarati_count])
        
        if total_alpha == 0:
            return None
        
        # Determine dominant script
        if latin_count / total_alpha > 0.7:
            return 'en'
        elif devanagari_count / total_alpha > 0.5:
            return 'hi'  # Could also be Marathi, but Hindi is more common
        elif bengali_count / total_alpha > 0.5:
            return 'bn'
        elif tamil_count / total_alpha > 0.5:
            return 'ta'
        elif telugu_count / total_alpha > 0.5:
            return 'te'
        elif gujarati_count / total_alpha > 0.5:
            return 'gu'
        
        return 'en'  # Default to English
    
    def _record_failure(self, reason: str):
        """Record validation failure statistics."""
        self.stats['failed'] += 1
        if reason not in self.stats['failure_reasons']:
            self.stats['failure_reasons'][reason] = 0
        self.stats['failure_reasons'][reason] += 1
    
    def get_stats(self) -> Dict:
        """Get validation statistics."""
        stats = self.stats.copy()
        if stats['total_validated'] > 0:
            stats['pass_rate'] = stats['passed'] / stats['total_validated']
            stats['fail_rate'] = stats['failed'] / stats['total_validated']
        else:
            stats['pass_rate'] = 0.0
            stats['fail_rate'] = 0.0
        
        return stats
    
    def reset_stats(self):
        """Reset validation statistics."""
        self.stats = {
            'total_validated': 0,
            'passed': 0,
            'failed': 0,
            'failure_reasons': {}
        }

	#!/usr/bin/env python3
	"""
	Data validation utilities for ensuring quality of ingested events.
	Implements multi-layer filtering and validation logic.
	"""

	import logging
	import re
	from typing import Dict, List, Optional, Set
	from datetime import datetime, timedelta

	from .base_connector import RawEvent

	logger = logging.getLogger(__name__)


	class DataValidator:
	"""Multi-layer data validation for raw events."""

	def __init__(self, config: Dict = None):
	"""Initialize validator with configuration.

	Args:
	config: Validation configuration dictionary
	"""
	self.config = config or {}

	# Content quality thresholds
	self.min_content_length = self.config.get('min_content_length', 10)
	self.max_content_length = self.config.get('max_content_length', 50000)
	self.min_word_count = self.config.get('min_word_count', 3)

	# Language settings
	self.allowed_languages = set(self.config.get('allowed_languages', [
	'en', 'hi', 'bn', 'te', 'ta', 'mr', 'gu', 'kn', 'ml', 'or', 'pa', 'as'
	]))

	# Geographic relevance
	self.require_india_relevance = self.config.get('require_india_relevance', True)

	# Spam and quality filters
	self.spam_keywords = set(self.config.get('spam_keywords', [
	'buy now', 'click here', 'limited offer', 'act now', 'free money',
	'earn money', 'work from home', 'get rich quick', 'miracle cure',
	'lose weight fast', 'enlargement', 'viagra', 'casino', 'lottery'
	]))

	# Indian context keywords for relevance checking
	self.india_keywords = set([
	'india', 'indian', 'bharat', 'bharatiya', 'delhi', 'mumbai', 'bangalore',
	'chennai', 'kolkata', 'hyderabad', 'pune', 'ahmedabad', 'surat', 'jaipur',
	'lucknow', 'kanpur', 'nagpur', 'indore', 'bhopal', 'visakhapatnam', 'patna',
	'vadodara', 'ludhiana', 'agra', 'nashik', 'faridabad', 'meerut', 'rajkot',
	'kalyan', 'vasai', 'varanasi', 'srinagar', 'aurangabad', 'dhanbad', 'amritsar',
	'navi mumbai', 'allahabad', 'ranchi', 'howrah', 'coimbatore', 'jabalpur',
	'gwalior', 'vijayawada', 'jodhpur', 'madurai', 'raipur', 'kota', 'guwahati',
	'chandigarh', 'solapur', 'hubli', 'tiruchirappalli', 'bareilly', 'mysore',
	'tiruppur', 'gurgaon', 'aligarh', 'jalandhar', 'bhubaneswar', 'salem',
	'warangal', 'mira', 'bhiwandi', 'thiruvananthapuram', 'bhavnagar', 'dehradun',
	'durgapur', 'asansol', 'nanded', 'kolhapur', 'ajmer', 'gulbarga', 'jamnagar',
	'ujjain', 'loni', 'siliguri', 'jhansi', 'ulhasnagar', 'nellore', 'jammu',
	'sangli', 'belgaum', 'mangalore', 'ambattur', 'tirunelveli', 'malegaon',
	'gaya', 'jalgaon', 'udaipur', 'maheshtala'
	])

	# Indian states and union territories
	self.indian_states = set([
	'andhra pradesh', 'arunachal pradesh', 'assam', 'bihar', 'chhattisgarh',
	'goa', 'gujarat', 'haryana', 'himachal pradesh', 'jharkhand', 'karnataka',
	'kerala', 'madhya pradesh', 'maharashtra', 'manipur', 'meghalaya', 'mizoram',
	'nagaland', 'odisha', 'punjab', 'rajasthan', 'sikkim', 'tamil nadu',
	'telangana', 'tripura', 'uttar pradesh', 'uttarakhand', 'west bengal',
	'andaman and nicobar islands', 'chandigarh', 'dadra and nagar haveli',
	'daman and diu', 'delhi', 'lakshadweep', 'puducherry', 'jammu and kashmir',
	'ladakh'
	])

	# Statistics
	self.stats = {
	'total_validated': 0,
	'passed': 0,
	'failed': 0,
	'failure_reasons': {}
	}

	def validate_event(self, event: RawEvent) -> tuple[bool, Optional[str]]:
	"""Validate a raw event through multiple layers.

	Args:
	event: RawEvent to validate

	Returns:
	Tuple of (is_valid, failure_reason)
	"""
	self.stats['total_validated'] += 1

	try:
	# Layer 1: Basic validation
	is_valid, reason = self._validate_basic(event)
	if not is_valid:
	self._record_failure(reason)
	return False, reason

	# Layer 2: Content quality
	is_valid, reason = self._validate_content_quality(event)
	if not is_valid:
	self._record_failure(reason)
	return False, reason

	# Layer 3: Language validation
	is_valid, reason = self._validate_language(event)
	if not is_valid:
	self._record_failure(reason)
	return False, reason

	# Layer 4: Geographic relevance
	if self.require_india_relevance:
	is_valid, reason = self._validate_india_relevance(event)
	if not is_valid:
	self._record_failure(reason)
	return False, reason

	# Layer 5: Spam and quality filtering
	is_valid, reason = self._validate_not_spam(event)
	if not is_valid:
	self._record_failure(reason)
	return False, reason

	# All validations passed
	self.stats['passed'] += 1
	return True, None

	except Exception as e:
	logger.error(f"Validation error for event {event.event_id}: {e}")
	self._record_failure(f"validation_error: {str(e)}")
	return False, f"validation_error: {str(e)}"

	def _validate_basic(self, event: RawEvent) -> tuple[bool, Optional[str]]:
	"""Basic validation checks."""

	# Check required fields
	if not event.content:
	return False, "empty_content"

	if not event.source_id:
	return False, "missing_source_id"

	if not event.timestamp:
	return False, "missing_timestamp"

	# Check content length
	content_length = len(event.content)
	if content_length < self.min_content_length:
	return False, f"content_too_short: {content_length} < {self.min_content_length}"

	if content_length > self.max_content_length:
	return False, f"content_too_long: {content_length} > {self.max_content_length}"

	# Check word count
	word_count = len(event.content.split())
	if word_count < self.min_word_count:
	return False, f"insufficient_words: {word_count} < {self.min_word_count}"

	# Check timestamp is reasonable (not too old or in future)
	now = datetime.now(event.timestamp.tzinfo or None)
	age = now - event.timestamp

	if age.days > 30: # Older than 30 days
	return False, f"content_too_old: {age.days} days"

	if age.total_seconds() < -3600: # More than 1 hour in future
	return False, "content_from_future"

	return True, None

	def _validate_content_quality(self, event: RawEvent) -> tuple[bool, Optional[str]]:
	"""Validate content quality and structure."""

	content = event.content.lower()

	# Check for excessive repetition
	words = content.split()
	if len(words) > 10:
	unique_words = set(words)
	repetition_ratio = len(unique_words) / len(words)
	if repetition_ratio < 0.3: # Less than 30% unique words
	return False, f"excessive_repetition: {repetition_ratio:.2f}"

	# Check for excessive capitalization
	if len(event.content) > 50:
	caps_ratio = sum(1 for c in event.content if c.isupper()) / len(event.content)
	if caps_ratio > 0.5: # More than 50% uppercase
	return False, f"excessive_caps: {caps_ratio:.2f}"

	# Check for excessive punctuation
	punct_count = sum(1 for c in event.content if c in '!?.,;:')
	if len(words) > 0:
	punct_ratio = punct_count / len(words)
	if punct_ratio > 0.3: # More than 30% punctuation to words ratio
	return False, f"excessive_punctuation: {punct_ratio:.2f}"

	# Check for minimum sentence structure
	sentences = re.split(r'[.!?]+', event.content)
	valid_sentences = [s.strip() for s in sentences if len(s.strip()) > 5]
	if len(valid_sentences) == 0 and len(words) > 20:
	return False, "no_sentence_structure"

	return True, None

	def _validate_language(self, event: RawEvent) -> tuple[bool, Optional[str]]:
	"""Validate language is in allowed set."""

	if not self.allowed_languages:
	return True, None # No language restrictions

	# Use detected language or try to detect
	language = event.language
	if not language:
	language = self._detect_language_simple(event.content)

	if language and language not in self.allowed_languages:
	return False, f"unsupported_language: {language}"

	return True, None

	def _validate_india_relevance(self, event: RawEvent) -> tuple[bool, Optional[str]]:
	"""Validate content is relevant to India."""

	content_lower = event.content.lower()
	title_lower = (event.title or "").lower()

	# Check for India-related keywords
	text_to_check = f"{content_lower} {title_lower}"

	# Direct keyword matching
	for keyword in self.india_keywords:
	if keyword in text_to_check:
	return True, None

	# Check for Indian states
	for state in self.indian_states:
	if state in text_to_check:
	return True, None

	# Check location hint
	if event.location_hint:
	location_lower = event.location_hint.lower()
	if any(keyword in location_lower for keyword in self.india_keywords):
	return True, None
	if any(state in location_lower for state in self.indian_states):
	return True, None

	# Check metadata for India relevance
	if event.metadata:
	metadata_str = str(event.metadata).lower()
	if any(keyword in metadata_str for keyword in self.india_keywords):
	return True, None

	return False, "not_india_relevant"

	def _validate_not_spam(self, event: RawEvent) -> tuple[bool, Optional[str]]:
	"""Validate content is not spam or low quality."""

	content_lower = event.content.lower()

	# Check for spam keywords
	for spam_keyword in self.spam_keywords:
	if spam_keyword in content_lower:
	return False, f"spam_keyword: {spam_keyword}"

	# Check for excessive URLs
	url_count = len(re.findall(r'http[s]?://(?:[a-zA-Z]\|[0-9]\|[$-_@.&+]\|[!*\\(\\),]\|(?:%[0-9a-fA-F][0-9a-fA-F]))+', event.content))
	word_count = len(event.content.split())
	if word_count > 0 and url_count / word_count > 0.2: # More than 20% URLs
	return False, f"excessive_urls: {url_count}/{word_count}"

	# Check for excessive numbers (might indicate spam/ads)
	number_count = len(re.findall(r'\b\d+\b', event.content))
	if word_count > 0 and number_count / word_count > 0.3: # More than 30% numbers
	return False, f"excessive_numbers: {number_count}/{word_count}"

	# Check for promotional language patterns
	promo_patterns = [
	r'\b(call\|contact\|visit)\s+(?:us\s+)?(?:at\s+)?\+?\d{10,}', # Phone numbers
	r'\b(?:only\|just)\s+(?:rs\.?\|₹)\s*\d+', # Price mentions
	r'\b(?:free\|discount\|offer\|sale)\b.*\b(?:today\|now\|limited)\b', # Promotional offers
	]

	for pattern in promo_patterns:
	if re.search(pattern, content_lower):
	return False, f"promotional_content: {pattern}"

	return True, None

	def _detect_language_simple(self, content: str) -> Optional[str]:
	"""Simple language detection based on script."""

	if not content:
	return None

	# Count characters by script
	latin_count = sum(1 for c in content if ord(c) < 256 and c.isalpha())
	devanagari_count = sum(1 for c in content if 0x0900 <= ord(c) <= 0x097F)
	bengali_count = sum(1 for c in content if 0x0980 <= ord(c) <= 0x09FF)
	tamil_count = sum(1 for c in content if 0x0B80 <= ord(c) <= 0x0BFF)
	telugu_count = sum(1 for c in content if 0x0C00 <= ord(c) <= 0x0C7F)
	gujarati_count = sum(1 for c in content if 0x0A80 <= ord(c) <= 0x0AFF)

	total_alpha = sum([latin_count, devanagari_count, bengali_count, tamil_count, telugu_count, gujarati_count])

	if total_alpha == 0:
	return None

	# Determine dominant script
	if latin_count / total_alpha > 0.7:
	return 'en'
	elif devanagari_count / total_alpha > 0.5:
	return 'hi' # Could also be Marathi, but Hindi is more common
	elif bengali_count / total_alpha > 0.5:
	return 'bn'
	elif tamil_count / total_alpha > 0.5:
	return 'ta'
	elif telugu_count / total_alpha > 0.5:
	return 'te'
	elif gujarati_count / total_alpha > 0.5:
	return 'gu'

	return 'en' # Default to English

	def _record_failure(self, reason: str):
	"""Record validation failure statistics."""
	self.stats['failed'] += 1
	if reason not in self.stats['failure_reasons']:
	self.stats['failure_reasons'][reason] = 0
	self.stats['failure_reasons'][reason] += 1

	def get_stats(self) -> Dict:
	"""Get validation statistics."""
	stats = self.stats.copy()
	if stats['total_validated'] > 0:
	stats['pass_rate'] = stats['passed'] / stats['total_validated']
	stats['fail_rate'] = stats['failed'] / stats['total_validated']
	else:
	stats['pass_rate'] = 0.0
	stats['fail_rate'] = 0.0

	return stats

	def reset_stats(self):
	"""Reset validation statistics."""
	self.stats = {
	'total_validated': 0,
	'passed': 0,
	'failed': 0,
	'failure_reasons': {}
	}