| import unicodedata |
| import re |
| import typing as tp |
|
|
| from functools import lru_cache |
| from typing import List |
|
|
| from sacremoses import MosesPunctNormalizer |
| from preprocessors.ssplit import sent_splitter |
| from preprocessors.tstrip import get_ascii_hashtag_replacer, get_non_printing_char_replacer, get_url_replacer |
|
|
| @lru_cache |
| def get_cleaned_splitter(lang_code: str) -> tp.Callable: |
| """ |
| Return sentence processor |
| """ |
| |
| mpn = MosesPunctNormalizer(lang="en") |
| mpn.substitutions = [(re.compile(pat), sub) for pat, sub in mpn.substitutions] |
|
|
| |
| replace_hashtag = get_ascii_hashtag_replacer(" ") |
| replace_nonprint = get_non_printing_char_replacer(" ") |
| replace_url = get_url_replacer(" ") |
|
|
| def process(text: str) -> List: |
| """ |
| Normalize, split and clean sentences |
| """ |
| sentence_splits = sent_splitter(text, lang_code) |
| cleaned_sents = [] |
| for sentence in sentence_splits: |
| clean = mpn.normalize(sentence) |
| clean = replace_nonprint(replace_hashtag(replace_url(clean))) |
| clean = unicodedata.normalize("NFC", clean) |
| cleaned_sents.append(clean) |
|
|
| return cleaned_sents |
| |
| return process |
|
|
| def splitAndClean(text, lang_code: str) -> List: |
| """Cleans input, splits into sentences""" |
|
|
| splitter = get_cleaned_splitter(lang_code) |
| return splitter(text) |