File size: 3,913 Bytes
6b73a07 | 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 | """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)
|