| """Text cleaning and filtering utilities for translation corpora.""" | |
| from __future__ import annotations | |
| import re | |
| import unicodedata | |
| from typing import Callable, Iterable, Sequence | |
| CONTROL_OR_ZERO_WIDTH_RE = re.compile(r"[\u0000-\u001f\u007f-\u009f\u200b\u200c\u200d\ufeff]") | |
| SPACE_RE = re.compile(r"\s+") | |
| PUNCT_RE = re.compile(r"[^\w\s\u4e00-\u9fff]", flags=re.UNICODE) | |
| def clean_text(text: str, lowercase: bool = False, remove_punctuation: bool = False) -> str: | |
| """Normalize a single sentence without changing its meaning aggressively.""" | |
| if text is None: | |
| return "" | |
| text = unicodedata.normalize("NFKC", str(text)) | |
| text = CONTROL_OR_ZERO_WIDTH_RE.sub("", text) | |
| text = SPACE_RE.sub(" ", text).strip() | |
| if lowercase: | |
| text = text.lower() | |
| if remove_punctuation: | |
| text = PUNCT_RE.sub("", text) | |
| text = SPACE_RE.sub(" ", text).strip() | |
| return text | |
| def _default_length(text: str) -> int: | |
| """Use whitespace tokens for Latin text and character count for CJK-heavy text.""" | |
| cjk_chars = sum(1 for ch in text if "\u4e00" <= ch <= "\u9fff") | |
| if cjk_chars >= max(1, len(text) // 3): | |
| return len(text.replace(" ", "")) | |
| return len(text.split()) | |
| def filter_by_length( | |
| src: str, | |
| tgt: str, | |
| min_src_len: int = 1, | |
| min_tgt_len: int = 1, | |
| max_src_len: int = 256, | |
| max_tgt_len: int = 256, | |
| length_ratio_threshold: float = 3.0, | |
| length_fn: Callable[[str], int] | None = None, | |
| ) -> bool: | |
| """Return True when a sentence pair passes basic length and ratio checks.""" | |
| length_fn = length_fn or _default_length | |
| src_len = length_fn(src) | |
| tgt_len = length_fn(tgt) | |
| if src_len < min_src_len or tgt_len < min_tgt_len: | |
| return False | |
| if src_len > max_src_len or tgt_len > max_tgt_len: | |
| return False | |
| shorter = max(1, min(src_len, tgt_len)) | |
| longer = max(src_len, tgt_len) | |
| return longer / shorter <= length_ratio_threshold | |
| def deduplicate_pairs(pairs: Iterable[tuple[str, str]]) -> list[tuple[str, str]]: | |
| """Deduplicate by exact cleaned source-target pair while preserving order.""" | |
| seen: set[tuple[str, str]] = set() | |
| result: list[tuple[str, str]] = [] | |
| for src, tgt in pairs: | |
| key = (src, tgt) | |
| if key in seen: | |
| continue | |
| seen.add(key) | |
| result.append(key) | |
| return result | |
| def preprocess_pipeline( | |
| src_texts: Sequence[str], | |
| tgt_texts: Sequence[str], | |
| lowercase_src: bool = False, | |
| lowercase_tgt: bool = False, | |
| remove_punctuation: bool = False, | |
| max_src_len: int = 256, | |
| max_tgt_len: int = 256, | |
| min_src_len: int = 1, | |
| min_tgt_len: int = 1, | |
| filter_by_length_enabled: bool = True, | |
| length_ratio_threshold: float = 3.0, | |
| deduplicate: bool = True, | |
| ) -> tuple[list[str], list[str]]: | |
| """Clean, filter, and optionally deduplicate parallel source-target texts.""" | |
| if len(src_texts) != len(tgt_texts): | |
| raise ValueError("src_texts and tgt_texts must have the same length") | |
| pairs: list[tuple[str, str]] = [] | |
| for raw_src, raw_tgt in zip(src_texts, tgt_texts): | |
| src = clean_text(raw_src, lowercase=lowercase_src, remove_punctuation=remove_punctuation) | |
| tgt = clean_text(raw_tgt, lowercase=lowercase_tgt, remove_punctuation=remove_punctuation) | |
| if not src or not tgt: | |
| continue | |
| if filter_by_length_enabled and not filter_by_length( | |
| src, | |
| tgt, | |
| min_src_len=min_src_len, | |
| min_tgt_len=min_tgt_len, | |
| max_src_len=max_src_len, | |
| max_tgt_len=max_tgt_len, | |
| length_ratio_threshold=length_ratio_threshold, | |
| ): | |
| continue | |
| pairs.append((src, tgt)) | |
| if deduplicate: | |
| pairs = deduplicate_pairs(pairs) | |
| if not pairs: | |
| return [], [] | |
| src_clean, tgt_clean = zip(*pairs) | |
| return list(src_clean), list(tgt_clean) | |