data_quality_checker.py

"""
数据质量检测器主模块
根据YAML配置文件对输入数据进行质量检测和清洗
"""

import re
import yaml
import json
import logging
from typing import Dict, List, Any, Tuple, Optional
from dataclasses import dataclass
from pathlib import Path
import unicodedata


@dataclass
class DetectionResult:
    """检测结果数据类"""
    rule_name: str
    passed: bool
    issues: List[str]
    suggestions: List[str]
    cleaned_text: Optional[str] = None


class DataQualityChecker:
    """数据质量检测器"""
    
    def __init__(self, config_path: str = "config.yaml"):
        """初始化检测器"""
        self.config = self._load_config(config_path)
        self._setup_logging()
        
    def _load_config(self, config_path: str) -> Dict[str, Any]:
        """加载配置文件"""
        with open(config_path, 'r', encoding='utf-8') as f:
            return yaml.safe_load(f)
    
    def _setup_logging(self):
        """设置日志"""
        log_level = self.config.get('global', {}).get('log_level', 'INFO')
        logging.basicConfig(
            level=getattr(logging, log_level),
            format='%(asctime)s - %(levelname)s - %(message)s'
        )
        self.logger = logging.getLogger(__name__)
    
    def check_text_quality(self, text: str) -> Dict[str, DetectionResult]:
        """对文本进行全面的质量检测"""
        results = {}
        
        # 文本完整性检测
        if self.config.get('text_integrity', {}).get('truncation_detection', {}).get('enabled', False):
            results['truncation_detection'] = self._check_truncation(text)
            
        if self.config.get('text_integrity', {}).get('completeness_validation', {}).get('enabled', False):
            results['completeness_validation'] = self._check_completeness(text)
        
        # 文本一致性检测
        if self.config.get('text_consistency', {}).get('traditional_simplified_mix', {}).get('enabled', False):
            results['traditional_simplified_mix'] = self._check_traditional_simplified_mix(text)
            
        if self.config.get('text_consistency', {}).get('chinese_english_format', {}).get('enabled', False):
            results['chinese_english_format'] = self._check_chinese_english_format(text)
        
        # 内容重复性检测
        if self.config.get('content_duplication', {}).get('duplicate_filter', {}).get('enabled', False):
            results['duplicate_filter'] = self._check_content_duplication(text)
        
        # 格式规范性检测
        if self.config.get('format_validation', {}).get('special_characters', {}).get('enabled', False):
            results['special_characters'] = self._check_special_characters(text)
            
        if self.config['format_validation']['json_format_validation']['enabled']:
            results['json_format_validation'] = self._check_json_format(text)
            
        if self.config.get('format_validation', {}).get('garbled_characters', {}).get('enabled', False):
            results['garbled_characters'] = self._check_garbled_characters(text)
        
        return results
    
    def _check_truncation(self, text: str) -> DetectionResult:
        """检测长文本截断"""
        config = self.config['text_integrity']['truncation_detection']
        issues = []
        suggestions = []
        
        # 检查文本长度
        max_length = config.get('max_length', 4096)
        if len(text) >= max_length:
            issues.append(f"文本长度({len(text)})接近或超过最大长度({max_length})")
            suggestions.append("检查是否存在截断问题")
        
        # 检查截断指示符
        indicators = config.get('truncation_indicators', [])
        for indicator in indicators:
            if text.endswith(indicator):
                issues.append(f"发现截断指示符: {indicator}")
                suggestions.append("文本可能被截断，需要获取完整内容")
        
        # 检查句子结尾
        if config.get('check_sentence_endings', True):
            if not re.search(r'[。！？.!?]$', text.strip()):
                issues.append("文本没有正常的句子结尾")
                suggestions.append("检查文本是否完整")
        
        return DetectionResult(
            rule_name="truncation_detection",
            passed=len(issues) == 0,
            issues=issues,
            suggestions=suggestions
        )
    
    def _check_completeness(self, text: str) -> DetectionResult:
        """检查内容完整性"""
        config = self.config['text_integrity']['completeness_validation']
        issues = []
        suggestions = []
        
        # 检查模板匹配
        templates = config.get('templates', [])
        template_matched = False
        for template in templates:
            pattern = template['pattern']
            if re.search(pattern, text, re.MULTILINE | re.DOTALL):
                template_matched = True
                break
        
        if templates and not template_matched:
            issues.append("内容不符合预期模板格式")
            suggestions.append("检查内容格式是否完整")
        
        # 检查不完整指示符
        incomplete_indicators = config.get('incomplete_indicators', [])
        for indicator in incomplete_indicators:
            if indicator in text:
                issues.append(f"发现不完整指示符: {indicator}")
                suggestions.append("内容可能不完整，需要补充")
        
        return DetectionResult(
            rule_name="completeness_validation",
            passed=len(issues) == 0,
            issues=issues,
            suggestions=suggestions
        )
    
    def _check_traditional_simplified_mix(self, text: str) -> DetectionResult:
        """检查繁简体混用"""
        issues = []
        suggestions = []
        
        # 简单的繁简体检测逻辑
        traditional_chars = set()
        simplified_chars = set()
        
        for char in text:
            if self._is_traditional_chinese(char):
                traditional_chars.add(char)
            elif self._is_simplified_chinese(char):
                simplified_chars.add(char)
        
        if traditional_chars and simplified_chars:
            issues.append(f"发现繁简体混用: 繁体字 {list(traditional_chars)[:5]}, 简体字 {list(simplified_chars)[:5]}")
            suggestions.append("统一使用繁体字或简体字")
        
        return DetectionResult(
            rule_name="traditional_simplified_mix",
            passed=len(issues) == 0,
            issues=issues,
            suggestions=suggestions
        )
    
    def _check_chinese_english_format(self, text: str) -> DetectionResult:
        """检查中英文格式"""
        config = self.config['text_consistency']['chinese_english_format']
        issues = []
        suggestions = []
        cleaned_text = text
        
        space_rules = config.get('space_rules', {})
        
        # 检查中英文之间的空格
        if space_rules.get('chinese_english_space', True):
            # 中文后直接跟英文
            pattern1 = r'([\u4e00-\u9fff])([a-zA-Z])'
            matches1 = re.findall(pattern1, text)
            if matches1:
                issues.append(f"中文与英文之间缺少空格: {matches1[:3]}")
                suggestions.append("在中文与英文之间添加空格")
                cleaned_text = re.sub(pattern1, r'\1 \2', cleaned_text)
            
            # 英文后直接跟中文
            pattern2 = r'([a-zA-Z])([\u4e00-\u9fff])'
            matches2 = re.findall(pattern2, text)
            if matches2:
                issues.append(f"英文与中文之间缺少空格: {matches2[:3]}")
                suggestions.append("在英文与中文之间添加空格")
                cleaned_text = re.sub(pattern2, r'\1 \2', cleaned_text)
        
        # 检查中文与数字之间的空格
        if space_rules.get('chinese_number_space', True):
            pattern3 = r'([\u4e00-\u9fff])(\d)'
            matches3 = re.findall(pattern3, text)
            if matches3:
                issues.append(f"中文与数字之间缺少空格: {matches3[:3]}")
                suggestions.append("在中文与数字之间添加空格")
                cleaned_text = re.sub(pattern3, r'\1 \2', cleaned_text)
        
        return DetectionResult(
            rule_name="chinese_english_format",
            passed=len(issues) == 0,
            issues=issues,
            suggestions=suggestions,
            cleaned_text=cleaned_text if cleaned_text != text else None
        )
    
    def _check_content_duplication(self, text: str) -> DetectionResult:
        """检查内容重复"""
        config = self.config['content_duplication']['duplicate_filter']
        issues = []
        suggestions = []
        cleaned_text = text
        
        min_length = config.get('min_duplicate_length', 10)
        
        # 段落级别去重
        if config.get('check_paragraph_level', True):
            # 首先处理连接在一起的段落（如：段落A段落B -> 段落A\n段落B）
            # 寻找可能的段落分隔模式
            title_patterns = [
                r'([。！？])([一二三四五六七八九十]+、)',  # 句号后接标题
                r'([。！？])(\([一二三四五六七八九十]+\))',  # 句号后接条目
                r'([。！？])([A-Z][a-z]+)',  # 句号后接英文
                r'([。！？])([第][一二三四五六七八九十]+[章节条])'  # 句号后接章节
            ]
            
            # 分离连接的段落
            separated_text = cleaned_text
            for pattern in title_patterns:
                separated_text = re.sub(pattern, r'\1\n\2', separated_text)
            
            # 现在按行处理
            lines = separated_text.split('\n')
            seen_paragraphs = set()
            unique_lines = []
            
            for line in lines:
                original_line = line
                line_stripped = line.strip()
                
                if len(line_stripped) >= min_length:
                    # 检查是否与已知段落完全匹配
                    if line_stripped in seen_paragraphs:
                        issues.append(f"发现重复段落: {line_stripped[:50]}...")
                        suggestions.append("删除重复段落")
                        continue
                    
                    # 检查是否包含已知段落作为子串，并处理连接的段落
                    line_processed = False
                    temp_line = line_stripped
                    
                    for seen_para in list(seen_paragraphs):
                        if len(seen_para) >= min_length:
                            # 如果当前行包含已见过的段落
                            if seen_para in temp_line:
                                # 从当前行中移除重复部分
                                temp_line = temp_line.replace(seen_para, '').strip()
                                issues.append(f"发现部分重复段落: {seen_para[:50]}...")
                                suggestions.append("删除重复段落")
                                line_processed = True
                    
                    # 如果处理后还有内容
                    if line_processed and len(temp_line) >= min_length:
                        # 检查处理后的内容是否也是重复的
                        if temp_line not in seen_paragraphs:
                            seen_paragraphs.add(temp_line)
                            # 保留处理后的内容
                            unique_lines.append(temp_line)
                        else:
                            issues.append(f"发现重复段落: {temp_line[:50]}...")
                            suggestions.append("删除重复段落")
                    elif line_processed and len(temp_line) < min_length:
                        # 处理后内容太短，可能全是重复的，跳过这一行
                        pass
                    elif not line_processed:
                        # 没有发现重复，正常添加
                        seen_paragraphs.add(line_stripped)
                        unique_lines.append(original_line)
                else:
                    # 短段落或空行直接保留
                    unique_lines.append(original_line)
            
            # 重新组合文本
            cleaned_text = '\n'.join(unique_lines)
        
        # 句子级别去重（在段落去重基础上进行）
        if config.get('check_sentence_level', True):
            seen_sentences = set()
            
            # 找到所有句子的位置和内容
            sentence_pattern = r'([^。！？.!?]+[。！？.!?]?)'
            matches = list(re.finditer(sentence_pattern, cleaned_text))
            
            # 从后往前处理，避免位置偏移
            for match in reversed(matches):
                sentence_full = match.group(1)
                sentence_content = sentence_full.strip()
                
                # 移除末尾的标点符号来检查重复
                sentence_clean = re.sub(r'[。！？.!?]+$', '', sentence_content).strip()
                
                if len(sentence_clean) >= min_length:
                    if sentence_clean in seen_sentences:
                        issues.append(f"发现重复句子: {sentence_clean[:30]}...")
                        suggestions.append("删除重复句子")
                        # 从文本中删除这个重复句子
                        start_pos = match.start()
                        end_pos = match.end()
                        cleaned_text = cleaned_text[:start_pos] + cleaned_text[end_pos:]
                    else:
                        seen_sentences.add(sentence_clean)
                else:
                    # 记录短句子，避免误删
                    if sentence_clean:
                        seen_sentences.add(sentence_clean)
        
        return DetectionResult(
            rule_name="duplicate_filter",
            passed=len(issues) == 0,
            issues=issues,
            suggestions=suggestions,
            cleaned_text=cleaned_text if len(issues) > 0 else None
        )
    
    def _check_special_characters(self, text: str) -> DetectionResult:
        """检查特殊字符"""
        config = self.config['format_validation']['special_characters']
        issues = []
        suggestions = []
        cleaned_text = text
        
        # 检查禁用字符
        forbidden_chars = config.get('forbidden_chars', [])
        for char in forbidden_chars:
            if char in text:
                issues.append(f"发现禁用字符: {repr(char)}")
                suggestions.append(f"移除或替换字符: {repr(char)}")
        
        # 应用替换规则
        replacement_rules = config.get('replacement_rules', {})
        for old, new in replacement_rules.items():
            if old in cleaned_text:
                cleaned_text = cleaned_text.replace(old, new)
                if old != new:
                    issues.append(f"字符替换: {repr(old)} -> {repr(new)}")
                    suggestions.append("已自动应用字符替换规则")
        
        # 转义字符处理
        escape_config = config.get('escape_characters', {})
        if escape_config.get('enabled', False):
            escape_issues, escape_suggestions, cleaned_text = self._handle_escape_characters(cleaned_text, escape_config)
            issues.extend(escape_issues)
            suggestions.extend(escape_suggestions)
        
        # 表情符号检测
        emoji_config = config.get('emoji_detection', {})
        if emoji_config.get('enabled', False):
            emojis = self._detect_emojis(cleaned_text)
            emoji_count = len(emojis)
            if emoji_count > 0:
                issues.append(f"发现 {emoji_count} 个表情符号: {', '.join(emojis[:5])}{'...' if len(emojis) > 5 else ''}")
                action = emoji_config.get('action', 'mark')
                if action == 'remove':
                    cleaned_text = self._remove_emojis(cleaned_text)
                    suggestions.append("已移除所有表情符号")
                elif action == 'replace':
                    replacement_text = emoji_config.get('replacement_text', '[表情]')
                    cleaned_text = self._replace_emojis(cleaned_text, replacement_text)
                    suggestions.append(f"已将表情符号替换为: {replacement_text}")
                else:
                    suggestions.append("建议移除或替换表情符号以确保文本正式性")
        
        # 特殊符号检测
        symbol_config = config.get('special_symbol_detection', {})
        if symbol_config.get('enabled', False):
            symbols = self._detect_special_symbols(cleaned_text)
            symbol_count = len(symbols)
            if symbol_count > 0:
                issues.append(f"发现 {symbol_count} 个特殊符号: {', '.join(symbols[:5])}{'...' if len(symbols) > 5 else ''}")
                action = symbol_config.get('action', 'mark')
                if action == 'remove':
                    cleaned_text = self._remove_special_symbols(cleaned_text)
                    suggestions.append("已移除所有特殊符号")
                elif action == 'replace':
                    replacement_text = symbol_config.get('replacement_text', '[符号]')
                    cleaned_text = self._replace_special_symbols(cleaned_text, replacement_text)
                    suggestions.append(f"已将特殊符号替换为: {replacement_text}")
                else:
                    suggestions.append("建议检查特殊符号的必要性")
        
        # 异常字符检测
        abnormal_config = config.get('abnormal_chars', {})
        if abnormal_config.get('enabled', False):
            abnormal_chars = self._detect_abnormal_chars(cleaned_text)
            abnormal_count = len(abnormal_chars)
            if abnormal_count > 0:
                issues.append(f"发现 {abnormal_count} 个异常字符: {', '.join([repr(c) for c in abnormal_chars[:5]])}{'...' if len(abnormal_chars) > 5 else ''}")
                action = abnormal_config.get('action', 'remove')
                if action == 'remove':
                    cleaned_text = self._remove_abnormal_chars(cleaned_text)
                    suggestions.append("已移除所有异常字符")
                elif action == 'replace':
                    replacement_text = abnormal_config.get('replacement_text', '[异常字符]')
                    cleaned_text = self._replace_abnormal_chars(cleaned_text, replacement_text)
                    suggestions.append(f"已将异常字符替换为: {replacement_text}")
                else:
                    suggestions.append("建议移除或替换异常字符")
        
        return DetectionResult(
            rule_name="special_characters",
            passed=len(issues) == 0,
            issues=issues,
            suggestions=suggestions,
            cleaned_text=cleaned_text if cleaned_text != text else None
        )
    
    def _check_json_format(self, text: str) -> DetectionResult:
        """检查JSON/JSONL格式完整性"""
        config = self.config['format_validation']['json_format_validation']
        issues = []
        suggestions = []
        cleaned_text = text
        
        # 检测文本格式类型
        format_type = self._detect_format_type(text)
        
        if format_type == "json":
            # JSON格式验证
            json_issues, json_suggestions, json_cleaned = self._validate_json_format(text, config)
            issues.extend(json_issues)
            suggestions.extend(json_suggestions)
            if json_cleaned != text:
                cleaned_text = json_cleaned
                
        elif format_type == "jsonl":
            # JSONL格式验证
            jsonl_issues, jsonl_suggestions, jsonl_cleaned = self._validate_jsonl_format(text, config)
            issues.extend(jsonl_issues)
            suggestions.extend(jsonl_suggestions)
            if jsonl_cleaned != text:
                cleaned_text = jsonl_cleaned
        else:
            # 通用引号和括号检查
            quote_issues, quote_suggestions, quote_cleaned = self._validate_quotes_and_brackets(text, config)
            issues.extend(quote_issues)
            suggestions.extend(quote_suggestions)
            if quote_cleaned != text:
                cleaned_text = quote_cleaned
        
        return DetectionResult(
            rule_name="json_format_validation",
            passed=len(issues) == 0,
            issues=issues,
            suggestions=suggestions,
            cleaned_text=cleaned_text if cleaned_text != text else None
        )
    
    def _detect_format_type(self, text: str) -> str:
        """检测文本格式类型"""
        text = text.strip()
        
        # 检查是否为JSON格式
        if (text.startswith('{') and text.endswith('}')) or (text.startswith('[') and text.endswith(']')):
            return "json"
        
        # 检查是否为JSONL格式（每行都是JSON对象）
        lines = text.strip().split('\n')
        if len(lines) > 1:
            json_line_count = 0
            for line in lines:
                line = line.strip()
                if line and (line.startswith('{') and line.endswith('}')):
                    json_line_count += 1
            
            if json_line_count >= len(lines) * 0.7:  # 70%以上的行是JSON格式
                return "jsonl"
        
        return "text"
    
    def _validate_json_format(self, text: str, config: dict) -> tuple:
        """验证JSON格式"""
        import json
        issues = []
        suggestions = []
        cleaned_text = text
        
        try:
            # 尝试解析JSON
            json.loads(text)
            return issues, suggestions, cleaned_text
        except json.JSONDecodeError as e:
            issues.append(f"JSON格式错误: {str(e)}")
            
            # 尝试修复常见的JSON问题
            if config.get('auto_fix', False):
                fixed_text = self._fix_json_issues(text)
                try:
                    json.loads(fixed_text)
                    cleaned_text = fixed_text
                    suggestions.append("已自动修复JSON格式问题")
                except:
                    suggestions.append("无法自动修复JSON格式，请手动检查")
            else:
                suggestions.append("建议检查JSON格式，可能存在引号缺失、括号不匹配等问题")
        
        return issues, suggestions, cleaned_text
    
    def _validate_jsonl_format(self, text: str, config: dict) -> tuple:
        """验证JSONL格式"""
        import json
        issues = []
        suggestions = []
        cleaned_lines = []
        
        lines = text.strip().split('\n')
        has_fixes = False
        
        for i, line in enumerate(lines):
            line = line.strip()
            if not line:
                cleaned_lines.append(line)
                continue
                
            try:
                json.loads(line)
                cleaned_lines.append(line)
            except json.JSONDecodeError as e:
                issues.append(f"第{i+1}行JSON格式错误: {str(e)}")
                
                if config.get('auto_fix', False):
                    fixed_line = self._fix_json_issues(line)
                    try:
                        json.loads(fixed_line)
                        cleaned_lines.append(fixed_line)
                        has_fixes = True
                    except:
                        cleaned_lines.append(line)  # 保留原行
                        suggestions.append(f"第{i+1}行无法自动修复")
                else:
                    cleaned_lines.append(line)
                    suggestions.append(f"建议检查第{i+1}行的JSON格式")
        
        if has_fixes:
            suggestions.append("已自动修复部分JSONL格式问题")
        
        cleaned_text = '\n'.join(cleaned_lines)
        return issues, suggestions, cleaned_text
    
    def _validate_quotes_and_brackets(self, text: str, config: dict) -> tuple:
        """验证引号和括号配对"""
        issues = []
        suggestions = []
        cleaned_text = text
        
        # 检查双引号配对
        quote_issues = self._check_quote_pairing(text, '"')
        if quote_issues:
            issues.extend(quote_issues)
            suggestions.append("检查双引号配对")
        
        # 检查单引号配对
        single_quote_issues = self._check_quote_pairing(text, "'")
        if single_quote_issues:
            issues.extend(single_quote_issues)
            suggestions.append("检查单引号配对")
        
        # 检查括号配对
        bracket_pairs = config.get('bracket_pairs', [])
        for left, right in bracket_pairs:
            bracket_issues = self._check_bracket_pairing(text, left, right)
            if bracket_issues:
                issues.extend(bracket_issues)
                suggestions.append(f"检查括号 '{left}{right}' 的配对")
        
        # 如果启用自动修复
        if config.get('auto_fix', False) and issues:
            fixed_text = self._fix_quote_and_bracket_issues(text)
            if fixed_text != text:
                cleaned_text = fixed_text
                suggestions.append("已尝试自动修复引号和括号问题")
        
        return issues, suggestions, cleaned_text
    
    def _check_quote_pairing(self, text: str, quote_char: str) -> list:
        """检查引号配对"""
        issues = []
        quote_count = 0
        in_escape = False
        
        for i, char in enumerate(text):
            if char == '\\' and not in_escape:
                in_escape = True
                continue
            elif in_escape:
                in_escape = False
                continue
            elif char == quote_char:
                quote_count += 1
        
        if quote_count % 2 != 0:
            issues.append(f"引号 '{quote_char}' 数量不匹配，共发现 {quote_count} 个")
        
        return issues
    
    def _check_bracket_pairing(self, text: str, left: str, right: str) -> list:
        """检查括号配对"""
        issues = []
        stack = []
        
        for i, char in enumerate(text):
            if char == left:
                stack.append(i)
            elif char == right:
                if not stack:
                    issues.append(f"位置 {i}: 多余的右括号 '{right}'")
                else:
                    stack.pop()
        
        if stack:
            issues.append(f"未匹配的左括号 '{left}' 在位置: {stack}")
        
        return issues
    
    def _fix_json_issues(self, text: str) -> str:
        """尝试修复常见的JSON问题"""
        import re
        
        # 移除末尾的逗号（在}或]前）
        text = re.sub(r',(\s*[}\]])', r'\1', text)
        
        # 修复单引号为双引号（但要避免在字符串内部）
        text = re.sub(r"'([^']*)'(?=\s*:)", r'"\1"', text)  # 键名
        text = re.sub(r":\s*'([^']*)'", r': "\1"', text)    # 字符串值
        
        # 为没有引号的键添加引号
        text = re.sub(r'(\w+)(\s*:)', r'"\1"\2', text)
        
        return text
    
    def _fix_quote_and_bracket_issues(self, text: str) -> str:
        """尝试修复引号和括号问题"""
        # 这里可以实现一些简单的修复逻辑
        # 比如在文本末尾添加缺失的引号或括号
        
        # 检查是否缺少末尾的双引号
        quote_count = text.count('"') - text.count('\\"')
        if quote_count % 2 != 0:
            text += '"'
        
        # 检查常见的括号不匹配并尝试修复
        open_braces = text.count('{') - text.count('\\{')
        close_braces = text.count('}') - text.count('\\}')
        if open_braces > close_braces:
            text += '}' * (open_braces - close_braces)
        
        open_brackets = text.count('[') - text.count('\\[')
        close_brackets = text.count(']') - text.count('\\]')
        if open_brackets > close_brackets:
            text += ']' * (open_brackets - close_brackets)
        
        return text

    def _detect_emojis(self, text: str) -> list:
        """检测表情符号"""
        import re
        # 表情符号的Unicode范围（更精确的范围）
        emoji_pattern = re.compile(
            "["
            "\U0001F600-\U0001F64F"  # emoticons
            "\U0001F300-\U0001F5FF"  # symbols & pictographs
            "\U0001F680-\U0001F6FF"  # transport & map symbols
            "\U0001F1E0-\U0001F1FF"  # flags (iOS)
            "\U0001f926-\U0001f937"  # additional faces
            "\U0001F900-\U0001F9FF"  # supplemental symbols
            "]", re.UNICODE
        )
        return emoji_pattern.findall(text)

    def _remove_emojis(self, text: str) -> str:
        """移除表情符号"""
        import re
        emoji_pattern = re.compile(
            "["
            "\U0001F600-\U0001F64F"
            "\U0001F300-\U0001F5FF"
            "\U0001F680-\U0001F6FF"
            "\U0001F1E0-\U0001F1FF"
            "\U0001f926-\U0001f937"
            "\U0001F900-\U0001F9FF"
            "]", re.UNICODE
        )
        return emoji_pattern.sub('', text)

    def _replace_emojis(self, text: str, replacement: str = "[表情]") -> str:
        """替换表情符号"""
        import re
        emoji_pattern = re.compile(
            "["
            "\U0001F600-\U0001F64F"
            "\U0001F300-\U0001F5FF"
            "\U0001F680-\U0001F6FF"
            "\U0001F1E0-\U0001F1FF"
            "\U0001f926-\U0001f937"
            "\U0001F900-\U0001F9FF"
            "]", re.UNICODE
        )
        return emoji_pattern.sub(replacement, text)

    def _detect_special_symbols(self, text: str) -> list:
        """检测特殊符号"""
        import re
        # 特殊符号模式
        special_symbols = [
            r'[★☆♦♣♠♥]',              # 星星和扑克符号
            r'[♀♂♪♫♬]',                # 性别和音乐符号
            r'[↑↓←→↖↗↘↙]',            # 箭头符号
            r'[①②③④⑤⑥⑦⑧⑨⑩]',        # 圆圈数字
            r'[ⅠⅡⅢⅣⅤⅥⅦⅧⅨⅩ]',       # 罗马数字
            r'[※○●◎◇◆□■△▲▽▼]',     # 几何符号
            r'[℃℉°′″‰‱]',             # 度量符号
            r'[™®©§¶†‡•‹›«»]',        # 商标和引用符号
        ]
        
        found_symbols = []
        for pattern in special_symbols:
            matches = re.findall(pattern, text)
            found_symbols.extend(matches)
        
        return found_symbols

    def _remove_special_symbols(self, text: str) -> str:
        """移除特殊符号"""
        import re
        special_symbols_pattern = r'[★☆♦♣♠♥♀♂♪♫♬↑↓←→↖↗↘↙①②③④⑤⑥⑦⑧⑨⑩ⅠⅡⅢⅣⅤⅥⅦⅧⅨⅩ※○●◎◇◆□■△▲▽▼℃℉°′″‰‱™®©§¶†‡•‹›«»]'
        return re.sub(special_symbols_pattern, '', text)

    def _replace_special_symbols(self, text: str, replacement: str = "[符号]") -> str:
        """替换特殊符号"""
        import re
        special_symbols_pattern = r'[★☆♦♣♠♥♀♂♪♫♬↑↓←→↖↗↘↙①②③④⑤⑥⑦⑧⑨⑩ⅠⅡⅢⅣⅤⅥⅦⅧⅨⅩ※○●◎◇◆□■△▲▽▼℃℉°′″‰‱™®©§¶†‡•‹›«»]'
        return re.sub(special_symbols_pattern, replacement, text)

    def _detect_abnormal_chars(self, text: str) -> list:
        """检测异常字符"""
        import re
        abnormal_patterns = [
            r'[\x00-\x08\x0B\x0C\x0E-\x1F\x7F-\x9F]',  # 控制字符（排除换行符\x0A和回车符\x0D）
            r'[�]',                                      # 替换字符（乱码）
            r'[\uFFFE\uFFFF]',                          # 非字符
            r'[\u200B-\u200D\uFEFF]',                   # 零宽字符
            r'[\uE000-\uF8FF]',                         # 私用区字符
        ]
        
        found_chars = []
        for pattern in abnormal_patterns:
            matches = re.findall(pattern, text)
            found_chars.extend(matches)
        
        return found_chars

    def _remove_abnormal_chars(self, text: str) -> str:
        """移除异常字符"""
        import re
        abnormal_pattern = r'[\x00-\x08\x0B\x0C\x0E-\x1F\x7F-\x9F�\uFFFE\uFFFF\u200B-\u200D\uFEFF\uE000-\uF8FF]'
        return re.sub(abnormal_pattern, '', text)

    def _replace_abnormal_chars(self, text: str, replacement: str = "[异常字符]") -> str:
        """替换异常字符"""
        import re
        abnormal_pattern = r'[\x00-\x08\x0B\x0C\x0E-\x1F\x7F-\x9F�\uFFFE\uFFFF\u200B-\u200D\uFEFF\uE000-\uF8FF]'
        return re.sub(abnormal_pattern, replacement, text)
    
    def _handle_escape_characters(self, text: str, config: dict) -> Tuple[List[str], List[str], str]:
        """处理转义字符"""
        issues = []
        suggestions = []
        cleaned_text = text
        
        # 转义字符映射表
        escape_mappings = {
            '\\n': '\n',    # 换行符
            '\\r': '\r',    # 回车符
            '\\t': '\t',    # 制表符
            '\\b': '\b',    # 退格符
            '\\f': '\f',    # 换页符
            '\\v': '\v',    # 垂直制表符
            '\\0': '\0',    # 空字符
            '\\"': '"',     # 双引号
            "\\'": "'",     # 单引号
            '\\\\': '\\',   # 反斜杠
        }
        
        action = config.get('action', 'convert')
        
        if action == 'convert':
            # 转换转义序列为实际字符
            original_text = cleaned_text
            for escape_seq, actual_char in escape_mappings.items():
                if escape_seq in cleaned_text:
                    cleaned_text = cleaned_text.replace(escape_seq, actual_char)
                    issues.append(f"转义字符转换: {repr(escape_seq)} -> {repr(actual_char)}")
            
            if cleaned_text != original_text:
                suggestions.append("已将转义字符转换为实际字符")
                
        elif action == 'normalize':
            # 标准化转义字符格式
            import re
            # 检测并标准化各种换行符
            if re.search(r'\\r\\n|\\n\\r', cleaned_text):
                cleaned_text = re.sub(r'\\r\\n|\\n\\r', '\\n', cleaned_text)
                issues.append("发现混合换行符格式")
                suggestions.append("已标准化换行符为 \\n")
            
            # 检测连续的转义空白字符
            if re.search(r'(\\t){2,}', cleaned_text):
                issues.append("发现连续的转义制表符")
                suggestions.append("建议简化连续的制表符")
                
        elif action == 'mark':
            # 仅标记转义字符，不做转换
            for escape_seq in escape_mappings.keys():
                if escape_seq in cleaned_text:
                    issues.append(f"发现转义字符: {repr(escape_seq)}")
            
            if issues:
                suggestions.append("发现转义字符，请确认是否需要转换")
        
        # 检测无效的转义序列
        import re
        invalid_escapes = re.findall(r'\\[^nrtbfv0"\'\\]', cleaned_text)
        if invalid_escapes:
            issues.append(f"发现无效转义序列: {', '.join(set(invalid_escapes))}")
            suggestions.append("请检查并修正无效的转义序列")
        
        return issues, suggestions, cleaned_text
    
    def _check_garbled_characters(self, text: str) -> DetectionResult:
        """检查乱码字符"""
        config = self.config['format_validation']['garbled_characters']
        issues = []
        suggestions = []
        cleaned_text = text
        
        garbled_patterns = config.get('garbled_patterns', [])
        action = config.get('action', 'remove')
        
        for pattern in garbled_patterns:
            matches = re.findall(pattern, text)
            if matches:
                issues.append(f"发现乱码字符: {matches[:5]}")
                suggestions.append(f"建议{action}乱码字符")
                
                if action == 'remove':
                    cleaned_text = re.sub(pattern, '', cleaned_text)
                elif action == 'replace':
                    cleaned_text = re.sub(pattern, '[?]', cleaned_text)
        
        return DetectionResult(
            rule_name="garbled_characters",
            passed=len(issues) == 0,
            issues=issues,
            suggestions=suggestions,
            cleaned_text=cleaned_text if cleaned_text != text else None
        )
    
    def _is_traditional_chinese(self, char: str) -> bool:
        """判断是否为繁体中文字符（简化判断）"""
        # 这里使用一些常见的繁体字进行判断
        traditional_chars = "學習種類動機經驗開發環境時間標準確認問題機會應該並且選擇過程結果優勢測試數據處理"
        return char in traditional_chars
    
    def _is_simplified_chinese(self, char: str) -> bool:
        """判断是否为简体中文字符（简化判断）"""
        # 对应的简体字
        simplified_chars = "学习种类动机经验开发环境时间标准确认问题机会应该并且选择过程结果优势测试数据处理"
        return char in simplified_chars
    
    def generate_report(self, results: Dict[str, DetectionResult], format_type: str = "json") -> str:
        """生成检测报告"""
        report_data = {
            "summary": {
                "total_rules": len(results),
                "rules_with_issues": sum(1 for r in results.values() if not r.passed),
                "rules_checked": len(results)
            },
            "details": {}
        }
        
        for rule_name, result in results.items():
            report_data["details"][rule_name] = {
                "has_issues": not result.passed,
                "issues": result.issues,
                "suggestions": result.suggestions,
                "has_cleaned_text": result.cleaned_text is not None
            }
        
        if format_type == "json":
            return json.dumps(report_data, ensure_ascii=False, indent=2)
        elif format_type == "yaml":
            return yaml.dump(report_data, allow_unicode=True, default_flow_style=False)
        else:
            # 文本格式
            lines = ["=== 数据质量检测报告 ===", ""]
            lines.append(f"检测规则数: {report_data['summary']['total_rules']}")
            lines.append(f"发现问题的规则: {report_data['summary']['rules_with_issues']}")
            lines.append("")
            
            for rule_name, details in report_data["details"].items():
                lines.append(f"【{rule_name}】")
                if details['issues']:
                    lines.append("检测到的问题:")
                    for issue in details['issues']:
                        lines.append(f"  • {issue}")
                    if details['suggestions']:
                        lines.append("改进建议:")
                        for suggestion in details['suggestions']:
                            lines.append(f"  ➤ {suggestion}")
                else:
                    lines.append("  ✓ 未发现问题")
                lines.append("")
            
            return "\n".join(lines)


def main():
    """主函数示例"""
    checker = DataQualityChecker()
    
    # 示例文本
    test_text = """
    这是一个测试文本，包含了一些問題。
    
    这里有English单词没有空格。还有123数字也没有空格。
    
    这是重复的内容。
    这是重复的内容。
    
    还有一些特殊字符　和((不匹配的括号。
    
    可能包含乱码字符�和其他问题...
    """
    
    # 执行检测
    results = checker.check_text_quality(test_text)
    
    # 生成报告
    report = checker.generate_report(results, "txt")
    print(report)
    
    # 获取清洗后的文本
    cleaned_texts = []
    for result in results.values():
        if result.cleaned_text:
            cleaned_texts.append(result.cleaned_text)
    
    if cleaned_texts:
        print("\n=== 清洗后的文本 ===")
        # 这里可以根据需要合并或选择最佳的清洗结果
        print(cleaned_texts[-1])


if __name__ == "__main__":
    main()