"""Analyzer for language distribution and mismatch detection.""" import re from collections import Counter, defaultdict from typing import List, Dict, Any import logging from .base_analyzer import BaseAnalyzer logger = logging.getLogger(__name__) class LanguageAnalyzer(BaseAnalyzer): """Analyze language distribution and detect mismatches.""" def __init__(self, output_dir): super().__init__("language", output_dir) # Character range patterns for language detection self.lang_patterns = { "zh": re.compile(r"[\u4e00-\u9fff]"), # Chinese "ja": re.compile(r"[\u3040-\u309f\u30a0-\u30ff]"), # Japanese Hiragana/Katakana "ko": re.compile(r"[\uac00-\ud7af\u1100-\u11ff]"), # Korean "ar": re.compile(r"[\u0600-\u06ff\u0750-\u077f]"), # Arabic "he": re.compile(r"[\u0590-\u05ff]"), # Hebrew "ru": re.compile(r"[\u0400-\u04ff]"), # Cyrillic "th": re.compile(r"[\u0e00-\u0e7f]"), # Thai "hi": re.compile(r"[\u0900-\u097f]"), # Devanagari (Hindi) "el": re.compile(r"[\u0370-\u03ff\u1f00-\u1fff]"), # Greek } def detect_language(self, text: str) -> Dict[str, float]: """Detect languages present in text based on character patterns. Args: text: Text to analyze Returns: Dictionary of detected languages with character ratios """ if not text: return {} total_chars = len(text) lang_chars = defaultdict(int) for lang, pattern in self.lang_patterns.items(): matches = pattern.findall(text) if matches: lang_chars[lang] = len(matches) # Check for Latin characters (English, Spanish, etc.) latin_matches = re.findall(r"[a-zA-Z]", text) if latin_matches: lang_chars["latin"] = len(latin_matches) # Calculate ratios lang_ratios = {} for lang, count in lang_chars.items(): lang_ratios[lang] = count / total_chars return lang_ratios def process_chunk(self, records: List[Dict[str, Any]]) -> Dict[str, Any]: """Process a chunk of records for language analysis. Args: records: List of JSON records Returns: Intermediate results for this chunk """ lang_counter = Counter() mismatches = [] missing_lang_count = 0 empty_lang_count = 0 lang_text_combinations = Counter() for record in records: lang_field = record.get("lang", "") text = record.get("text", "") if "lang" not in record: missing_lang_count += 1 continue if not lang_field: empty_lang_count += 1 continue lang_counter[lang_field] += 1 # Detect actual language in text if text: detected_langs = self.detect_language(text) # Check for mismatches if detected_langs: primary_detected = max(detected_langs.items(), key=lambda x: x[1]) primary_lang = primary_detected[0] # Map detected language to expected format lang_mapping = { "latin": ["en", "es", "fr", "de", "pt", "it", "nl"], "zh": ["zh", "zh-CN", "zh-TW"], "ja": ["ja"], "ko": ["ko"], "ru": ["ru"], "ar": ["ar"], "he": ["he"], "th": ["th"], "hi": ["hi"], "el": ["el"], } # Check if detected language matches declared language is_match = False for detected_key, expected_langs in lang_mapping.items(): if primary_lang == detected_key and lang_field in expected_langs: is_match = True break elif primary_lang in expected_langs and lang_field == primary_lang: is_match = True break if not is_match and primary_detected[1] > 0.3: # At least 30% of text mismatch_info = { "id": record.get("id", "unknown"), "declared_lang": lang_field, "detected_lang": primary_lang, "confidence": primary_detected[1], "text_sample": text[:200], } mismatches.append(mismatch_info) # Track language-text combinations lang_text_combinations[f"{lang_field}_{primary_lang}"] += 1 return { "lang_counts": dict(lang_counter), "mismatches": mismatches, "missing_lang_count": missing_lang_count, "empty_lang_count": empty_lang_count, "lang_text_combinations": dict(lang_text_combinations), "total_records": len(records), } def aggregate(self, chunk_results: List[Dict[str, Any]]) -> Dict[str, Any]: """Aggregate results from all chunks. Args: chunk_results: List of results from each chunk Returns: Final aggregated analysis """ all_lang_counts = [] all_mismatches = [] all_combinations = [] total_missing = 0 total_empty = 0 total_records = 0 for chunk in chunk_results: all_lang_counts.append(chunk.get("lang_counts", {})) all_mismatches.extend(chunk.get("mismatches", [])) all_combinations.append(chunk.get("lang_text_combinations", {})) total_missing += chunk.get("missing_lang_count", 0) total_empty += chunk.get("empty_lang_count", 0) total_records += chunk.get("total_records", 0) # Aggregate counters from ..utils.stats_utils import aggregate_counters total_lang_counts = aggregate_counters(all_lang_counts) total_combinations = aggregate_counters(all_combinations) # Sort languages by frequency lang_distribution = sorted(total_lang_counts.items(), key=lambda x: x[1], reverse=True) # Group mismatches by type mismatch_types = Counter() for mismatch in all_mismatches: key = f"{mismatch['declared_lang']}_vs_{mismatch['detected_lang']}" mismatch_types[key] += 1 # Sample mismatches (first 100) mismatch_sample = all_mismatches[:100] return { "summary": { "total_records": total_records, "records_with_lang": total_records - total_missing, "records_without_lang": total_missing, "empty_lang": total_empty, "unique_languages": len(total_lang_counts), "total_mismatches": len(all_mismatches), "mismatch_rate": len(all_mismatches) / (total_records - total_missing) if (total_records - total_missing) > 0 else 0, }, "language_distribution": lang_distribution, "language_counts": total_lang_counts, "mismatch_types": dict(mismatch_types.most_common()), "mismatch_sample": mismatch_sample, "lang_text_combinations": dict( sorted(total_combinations.items(), key=lambda x: x[1], reverse=True)[:50] ), }