from __future__ import annotations
import quopri
import re
import sys
import unicodedata
from typing import Optional, Tuple
import numpy as np
from unstructured.file_utils.encoding import (
format_encoding_str,
)
from unstructured.nlp.patterns import (
DOUBLE_PARAGRAPH_PATTERN_RE,
E_BULLET_PATTERN,
LINE_BREAK_RE,
PARAGRAPH_PATTERN,
PARAGRAPH_PATTERN_RE,
UNICODE_BULLETS_RE,
UNICODE_BULLETS_RE_0W,
)
def clean_non_ascii_chars(text) -> str:
"""Cleans non-ascii characters from unicode string.
Example
-------
\x88This text contains non-ascii characters!\x88
-> This text contains non-ascii characters!
"""
en = text.encode("ascii", "ignore")
return en.decode()
def clean_bullets(text: str) -> str:
"""Cleans unicode bullets from a section of text.
Example
-------
● This is an excellent point! -> This is an excellent point!
"""
search = UNICODE_BULLETS_RE.match(text)
if search is None:
return text
cleaned_text = UNICODE_BULLETS_RE.sub("", text, 1)
return cleaned_text.strip()
def clean_ordered_bullets(text) -> str:
"""Cleans the start of bulleted text sections up to three “sub-section”
bullets accounting numeric and alphanumeric types.
Example
-------
1.1 This is a very important point -> This is a very important point
a.b This is a very important point -> This is a very important point
"""
text_sp = text.split()
text_cl = " ".join(text_sp[1:])
if any(["." not in text_sp[0], ".." in text_sp[0]]):
return text
bullet = re.split(pattern=r"[\.]", string=text_sp[0])
if not bullet[-1]:
del bullet[-1]
if len(bullet[0]) > 2:
return text
return text_cl
def clean_ligatures(text) -> str:
"""Replaces ligatures with their most likely equivalent characters.
Example
-------
The benefits -> The benefits
High quality financial -> High quality financial
"""
ligatures_map = {
"æ": "ae",
"Æ": "AE",
"ff": "ff",
"fi": "fi",
"fl": "fl",
"ffi": "ffi",
"ffl": "ffl",
"ſt": "ft",
"ʪ": "ls",
"œ": "oe",
"Œ": "OE",
"ȹ": "qp",
"st": "st",
"ʦ": "ts",
}
cleaned_text: str = text
for k, v in ligatures_map.items():
cleaned_text = cleaned_text.replace(k, v)
return cleaned_text
def group_bullet_paragraph(paragraph: str) -> list:
"""Groups paragraphs with bullets that have line breaks for visual/formatting purposes.
For example:
'''○ The big red fox
is walking down the lane.
○ At the end of the lane
the fox met a friendly bear.'''
Gets converted to
'''○ The big red fox is walking down the lane.
○ At the end of the land the fox met a bear.'''
"""
clean_paragraphs = []
# pytesseract converts some bullet points to standalone "e" characters.
# Substitute "e" with bullets since they are later used in partition_text
# to determine list element type.
paragraph = (re.sub(E_BULLET_PATTERN, "·", paragraph)).strip()
bullet_paras = re.split(UNICODE_BULLETS_RE_0W, paragraph)
for bullet in bullet_paras:
if bullet:
clean_paragraphs.append(re.sub(PARAGRAPH_PATTERN, " ", bullet))
return clean_paragraphs
def group_broken_paragraphs(
text: str,
line_split: re.Pattern[str] = PARAGRAPH_PATTERN_RE,
paragraph_split: re.Pattern[str] = DOUBLE_PARAGRAPH_PATTERN_RE,
) -> str:
"""Groups paragraphs that have line breaks for visual/formatting purposes.
For example:
'''The big red fox
is walking down the lane.
At the end of the lane
the fox met a bear.'''
Gets converted to
'''The big red fox is walking down the lane.
At the end of the land the fox met a bear.'''
"""
paragraphs = paragraph_split.split(text)
clean_paragraphs = []
for paragraph in paragraphs:
if not paragraph.strip():
continue
# NOTE(robinson) - This block is to account for lines like the following that shouldn't be
# grouped together, but aren't separated by a double line break.
# Apache License
# Version 2.0, January 2004
# http://www.apache.org/licenses/
para_split = line_split.split(paragraph)
all_lines_short = all(len(line.strip().split(" ")) < 5 for line in para_split)
# pytesseract converts some bullet points to standalone "e" characters
if UNICODE_BULLETS_RE.match(paragraph.strip()) or E_BULLET_PATTERN.match(paragraph.strip()):
clean_paragraphs.extend(group_bullet_paragraph(paragraph))
elif all_lines_short:
clean_paragraphs.extend([line for line in para_split if line.strip()])
else:
clean_paragraphs.append(re.sub(PARAGRAPH_PATTERN, " ", paragraph))
return "\n\n".join(clean_paragraphs)
def new_line_grouper(
text: str,
paragraph_split: re.Pattern[str] = LINE_BREAK_RE,
) -> str:
"""
Concatenates text document that has one-line paragraph break pattern
For example,
Iwan Roberts
Roberts celebrating after scoring a goal for Norwich City
in 2004
Will be returned as:
Iwan Roberts\n\nRoberts celebrating after scoring a goal for Norwich City\n\nin 2004
"""
paragraphs = paragraph_split.split(text)
clean_paragraphs = []
for paragraph in paragraphs:
if not paragraph.strip():
continue
clean_paragraphs.append(paragraph)
return "\n\n".join(clean_paragraphs)
def blank_line_grouper(
text: str,
paragraph_split: re.Pattern = DOUBLE_PARAGRAPH_PATTERN_RE,
) -> str:
"""
Concatenates text document that has blank-line paragraph break pattern
For example,
Vestibulum auctor dapibus neque.
Nunc dignissim risus id metus.
Will be returned as:
Vestibulum auctor dapibus neque.\n\nNunc dignissim risus id metus.\n\n
"""
return group_broken_paragraphs(text)
def auto_paragraph_grouper(
text: str,
line_split: re.Pattern[str] = LINE_BREAK_RE,
max_line_count: int = 2000,
threshold: float = 0.1,
) -> str:
"""
Checks the ratio of new line (\n) over the total max_line_count
If the ratio of new line is less than the threshold,
the document is considered a new-line grouping type
and return the original text
If the ratio of new line is greater than or equal to the threshold,
the document is considered a blank-line grouping type
and passed on to blank_line_grouper function
"""
lines = line_split.split(text)
max_line_count = min(len(lines), max_line_count)
line_count, empty_line_count = 0, 0
for line in lines[:max_line_count]:
line_count += 1
if not line.strip():
empty_line_count += 1
ratio = empty_line_count / line_count
# NOTE(klaijan) - for ratio < threshold, we pass to new-line grouper,
# otherwise to blank-line grouper
if ratio < threshold:
return new_line_grouper(text)
else:
return blank_line_grouper(text)
# TODO(robinson) - There's likely a cleaner was to accomplish this and get all of the
# unicode characters instead of just the quotes. Doing this for now since quotes are
# an issue that are popping up in the SEC filings tests
def replace_unicode_quotes(text: str) -> str:
"""Replaces unicode bullets in text with the expected character
Example
-------
\x93What a lovely quote!\x94 -> “What a lovely quote!”
"""
# NOTE(robinson) - We should probably make this something more sane like a regex
# instead of a whole big series of replaces
text = text.replace("\x91", "‘")
text = text.replace("\x92", "’")
text = text.replace("\x93", "“")
text = text.replace("\x94", "”")
text = text.replace("'", "'")
text = text.replace("â\x80\x99", "'")
text = text.replace("â\x80“", "—")
text = text.replace("â\x80”", "–")
text = text.replace("â\x80˜", "‘")
text = text.replace("â\x80¦", "…")
text = text.replace("â\x80™", "’")
text = text.replace("â\x80œ", "“")
text = text.replace("â\x80?", "”")
text = text.replace("â\x80ť", "”")
text = text.replace("â\x80ś", "“")
text = text.replace("â\x80¨", "—")
text = text.replace("â\x80ł", "″")
text = text.replace("â\x80Ž", "")
text = text.replace("â\x80‚", "")
text = text.replace("â\x80‰", "")
text = text.replace("â\x80‹", "")
text = text.replace("â\x80", "")
text = text.replace("â\x80s'", "")
return text
tbl = dict.fromkeys(
i for i in range(sys.maxunicode) if unicodedata.category(chr(i)).startswith("P")
)
def remove_punctuation(s: str) -> str:
"""Removes punctuation from a given string."""
s = s.translate(tbl)
return s
def remove_sentence_punctuation(s: str, exclude_punctuation: Optional[list]) -> str:
tbl_new = tbl.copy()
if exclude_punctuation:
for punct in exclude_punctuation:
del tbl_new[ord(punct)]
s = s.translate(tbl_new)
return s
def clean_extra_whitespace(text: str) -> str:
"""Cleans extra whitespace characters that appear between words.
Example
-------
ITEM 1. BUSINESS -> ITEM 1. BUSINESS
"""
cleaned_text = re.sub(r"[\xa0\n]", " ", text)
cleaned_text = re.sub(r"([ ]{2,})", " ", cleaned_text)
return cleaned_text.strip()
def clean_dashes(text: str) -> str:
"""Cleans dash characters in text.
Example
-------
ITEM 1. -BUSINESS -> ITEM 1. BUSINESS
"""
# NOTE(Yuming): '\u2013' is the unicode string of 'EN DASH', a variation of "-"
return re.sub(r"[-\u2013]", " ", text).strip()
def clean_trailing_punctuation(text: str) -> str:
"""Clean all trailing punctuation in text
Example
-------
ITEM 1. BUSINESS. -> ITEM 1. BUSINESS
"""
return text.strip().rstrip(".,:;")
def replace_mime_encodings(text: str, encoding: str = "utf-8") -> str:
"""Replaces MIME encodings with their equivalent characters in the specified encoding.
Example
-------
5 w=E2=80-99s -> 5 w’s
"""
formatted_encoding = format_encoding_str(encoding)
return quopri.decodestring(text.encode(formatted_encoding)).decode(formatted_encoding)
def clean_prefix(text: str, pattern: str, ignore_case: bool = False, strip: bool = True) -> str:
"""Removes prefixes from a string according to the specified pattern. Strips leading
whitespace if the strip parameter is set to True.
Input
-----
text: The text to clean
pattern: The pattern for the prefix. Can be a simple string or a regex pattern
ignore_case: If True, ignores case in the pattern
strip: If True, removes leading whitespace from the cleaned string.
"""
flags = re.IGNORECASE if ignore_case else 0
clean_text = re.sub(rf"^{pattern}", "", text, flags=flags)
clean_text = clean_text.lstrip() if strip else clean_text
return clean_text
def clean_postfix(text: str, pattern: str, ignore_case: bool = False, strip: bool = True) -> str:
"""Removes postfixes from a string according to the specified pattern. Strips trailing
whitespace if the strip parameters is set to True.
Input
-----
text: The text to clean
pattern: The pattern for the postfix. Can be a simple string or a regex pattern
ignore_case: If True, ignores case in the pattern
strip: If True, removes trailing whitespace from the cleaned string.
"""
flags = re.IGNORECASE if ignore_case else 0
clean_text = re.sub(rf"{pattern}$", "", text, flags=flags)
clean_text = clean_text.rstrip() if strip else clean_text
return clean_text
def clean(
text: str,
extra_whitespace: bool = False,
dashes: bool = False,
bullets: bool = False,
trailing_punctuation: bool = False,
lowercase: bool = False,
) -> str:
"""Cleans text.
Input
-----
extra_whitespace: Whether to clean extra whitespace characters in text.
dashes: Whether to clean dash characters in text.
bullets: Whether to clean unicode bullets from a section of text.
trailing_punctuation: Whether to clean all trailing punctuation in text.
lowercase: Whether to return lowercase text.
"""
cleaned_text = text.lower() if lowercase else text
cleaned_text = (
clean_trailing_punctuation(cleaned_text) if trailing_punctuation else cleaned_text
)
cleaned_text = clean_dashes(cleaned_text) if dashes else cleaned_text
cleaned_text = clean_extra_whitespace(cleaned_text) if extra_whitespace else cleaned_text
cleaned_text = clean_bullets(cleaned_text) if bullets else cleaned_text
return cleaned_text.strip()
def bytes_string_to_string(text: str, encoding: str = "utf-8"):
"""Converts a string representation of a byte string to a regular string using the
specified encoding."""
text_bytes = bytes([ord(char) for char in text])
formatted_encoding = format_encoding_str(encoding)
return text_bytes.decode(formatted_encoding)
def clean_extra_whitespace_with_index_run(text: str) -> Tuple[str, np.ndarray]:
"""Cleans extra whitespace characters that appear between words.
Calculate distance between characters of original text and cleaned text.
Returns cleaned text along with array of indices it has moved from original.
Example
-------
ITEM 1. BUSINESS -> ITEM 1. BUSINESS
array([0., 0., 0., 0., 0., 0., 0., 0., 4., 4., 4., 4., 4., 4., 4., 4., 4., 4., 4., 4.]))
"""
cleaned_text = re.sub(r"[\xa0\n]", " ", text)
cleaned_text = re.sub(r"([ ]{2,})", " ", cleaned_text)
cleaned_text = cleaned_text.strip()
moved_indices = np.zeros(len(text))
distance, original_index, cleaned_index = 0, 0, 0
while cleaned_index < len(cleaned_text):
if text[original_index] == cleaned_text[cleaned_index] or (
bool(re.match("[\xa0\n]", text[original_index]))
and bool(re.match(" ", cleaned_text[cleaned_index]))
):
moved_indices[cleaned_index] = distance
original_index += 1
cleaned_index += 1
continue
distance += 1
moved_indices[cleaned_index] = distance
original_index += 1
moved_indices[cleaned_index:] = distance
return cleaned_text, moved_indices
def index_adjustment_after_clean_extra_whitespace(index, moved_indices) -> int:
return int(index - moved_indices[index])