File size: 4,063 Bytes
eca5751 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 | """Text Cleaner - Làm sạch text data."""
from __future__ import annotations
import re
import html
from typing import Dict, Any, List
from dataclasses import dataclass
@dataclass
class CleanerConfig:
"""Config cho TextCleaner."""
remove_html: bool = True
remove_urls: bool = False
remove_emojis: bool = False
normalize_whitespace: bool = True
normalize_unicode: bool = True
remove_control_chars: bool = True
min_length: int = 50
max_length: int = 100000
fix_encoding: bool = True
class TextCleaner:
"""Làm sạch text data cho training.
Usage:
cleaner = TextCleaner()
cleaned = cleaner.clean("some messy text...")
"""
# Common patterns
HTML_TAG_RE = re.compile(r"<[^>]+>")
URL_RE = re.compile(r"https?://\S+|www\.\S+")
MULTI_SPACE_RE = re.compile(r"[ \t]+")
MULTI_NEWLINE_RE = re.compile(r"\n{3,}")
CONTROL_CHARS_RE = re.compile(r"[\x00-\x08\x0b\x0c\x0e-\x1f\x7f-\x9f]")
EMOJI_RE = re.compile(
"["
"\U0001F600-\U0001F64F"
"\U0001F300-\U0001F5FF"
"\U0001F680-\U0001F6FF"
"\U0001F1E0-\U0001F1FF"
"\U00002702-\U000027B0"
"\U000024C2-\U0001F251"
"]+",
flags=re.UNICODE,
)
def __init__(self, config: CleanerConfig = None):
self.config = config or CleanerConfig()
def clean(self, text: str) -> str:
"""Clean a single text."""
if not text or not isinstance(text, str):
return ""
cfg = self.config
# Fix encoding issues
if cfg.fix_encoding:
text = text.replace("\ufeff", "").replace("\u200b", "")
# Normalize unicode
if cfg.normalize_unicode:
import unicodedata
text = unicodedata.normalize("NFC", text)
# Remove control characters
if cfg.remove_control_chars:
text = self.CONTROL_CHARS_RE.sub("", text)
# Decode HTML entities
text = html.unescape(text)
# Remove HTML tags
if cfg.remove_html:
text = self.HTML_TAG_RE.sub(" ", text)
# Remove URLs
if cfg.remove_urls:
text = self.URL_RE.sub("[URL]", text)
# Remove emojis
if cfg.remove_emojis:
text = self.EMOJI_RE.sub("", text)
# Normalize whitespace
if cfg.normalize_whitespace:
text = self.MULTI_SPACE_RE.sub(" ", text)
text = self.MULTI_NEWLINE_RE.sub("\n\n", text)
text = text.strip()
return text
def clean_batch(self, texts: List[str]) -> List[str]:
"""Clean multiple texts."""
return [self.clean(t) for t in texts]
def filter(self, text: str) -> bool:
"""Return True if text passes quality filters."""
if not text:
return False
if len(text) < self.config.min_length:
return False
if len(text) > self.config.max_length:
return False
# Check ratio of printable chars
non_print = sum(1 for c in text if not c.isprintable() and c not in "\n\r\t")
if non_print / len(text) > 0.05:
return False
# Check word repetition (low diversity)
words = text.split()
if len(words) > 20:
unique_ratio = len(set(words)) / len(words)
if unique_ratio < 0.3:
return False
return True
def process(self, sample: Dict[str, Any]) -> Dict[str, Any]:
"""Process a sample dict (in-place safe)."""
sample = dict(sample)
if "text" in sample:
cleaned = self.clean(sample["text"])
if not self.filter(cleaned):
return None # Filter out
sample["text"] = cleaned
sample["metadata"] = sample.get("metadata", {})
sample["metadata"]["cleaned"] = True
sample["metadata"]["cleaned_length"] = len(cleaned)
return sample
|