| """Text normalization, quality filtering and exact deduplication.""" |
|
|
| from __future__ import annotations |
|
|
| import hashlib |
| import re |
| import unicodedata |
| from collections.abc import Iterable, Iterator |
| from html.parser import HTMLParser |
|
|
| from zenith.data.sources.base import Record |
|
|
| _BAD_CHARS = re.compile(r"[\ud800-\udfff\x00-\x08\x0b\x0c\x0e-\x1f\x7f]") |
| _NON_WORD = re.compile(r"[\W_]+") |
| _MANY_BLANK_LINES = re.compile(r"\n(?:[ \t]*\n){3,}") |
|
|
|
|
| class _TextExtractor(HTMLParser): |
| def __init__(self): |
| super().__init__(convert_charrefs=True) |
| self.parts: list[str] = [] |
| self._skip = 0 |
|
|
| def handle_starttag(self, tag, attrs): |
| if tag in ("script", "style"): |
| self._skip += 1 |
| elif tag in ("p", "br", "div", "li", "tr", "h1", "h2", "h3", "h4", "pre"): |
| self.parts.append("\n") |
|
|
| def handle_endtag(self, tag): |
| if tag in ("script", "style") and self._skip: |
| self._skip -= 1 |
|
|
| def handle_data(self, data): |
| if not self._skip: |
| self.parts.append(data) |
|
|
|
|
| def _strip_html(text: str) -> str: |
| parser = _TextExtractor() |
| parser.feed(text) |
| parser.close() |
| return "".join(parser.parts).strip() |
|
|
|
|
| def normalize_text(text: str, html: bool = False) -> str: |
| text = _BAD_CHARS.sub("�", text) |
| if html: |
| text = _strip_html(text) |
| text = unicodedata.normalize("NFC", text) |
| return _MANY_BLANK_LINES.sub("\n\n\n", text) |
|
|
|
|
| def keep(record: Record) -> bool: |
| text = record.text |
| if len(text) < 200: |
| return False |
| letters = [c for c in text if c.isalpha()] |
| if not letters or sum(c.isascii() for c in letters) / len(letters) < 0.8: |
| return False |
| lines = [l for l in text.splitlines() if l.strip()] |
| return len(set(lines)) / max(len(lines), 1) >= 0.3 |
|
|
|
|
| def dedup_key(text: str) -> bytes: |
| """Hash of the text with case, whitespace and punctuation removed, so trivial variants collide.""" |
| return hashlib.sha256(_NON_WORD.sub("", text.lower()).encode("utf-8", errors="surrogatepass")).digest() |
|
|
|
|
| def dedupe(records: Iterable[Record]) -> Iterator[Record]: |
| """Drop exact and near-exact (case/whitespace/punctuation) duplicates, keeping the first.""" |
| seen: set[bytes] = set() |
| for r in records: |
| h = dedup_key(r.text) |
| if h not in seen: |
| seen.add(h) |
| yield r |
|
|
