Spaces:
Sleeping
Sleeping
import re | |
from collections import OrderedDict | |
from html import escape | |
from pathlib import Path | |
import dateparser | |
import grobid_tei_xml | |
from bs4 import BeautifulSoup | |
from tqdm import tqdm | |
def get_span_start(type, title=None): | |
title_ = ' title="' + title + '"' if title is not None else "" | |
return '<span class="label ' + type + '"' + title_ + '>' | |
def get_span_end(): | |
return '</span>' | |
def get_rs_start(type): | |
return '<rs type="' + type + '">' | |
def get_rs_end(): | |
return '</rs>' | |
def has_space_between_value_and_unit(quantity): | |
return quantity['offsetEnd'] < quantity['rawUnit']['offsetStart'] | |
def decorate_text_with_annotations(text, spans, tag="span"): | |
""" | |
Decorate a text using spans, using two style defined by the tag: | |
- "span" generated HTML like annotated text | |
- "rs" generate XML like annotated text (format SuperMat) | |
""" | |
sorted_spans = list(sorted(spans, key=lambda item: item['offset_start'])) | |
annotated_text = "" | |
start = 0 | |
for span in sorted_spans: | |
type = span['type'].replace("<", "").replace(">", "") | |
if 'unit_type' in span and span['unit_type'] is not None: | |
type = span['unit_type'].replace(" ", "_") | |
annotated_text += escape(text[start: span['offset_start']]) | |
title = span['quantified'] if 'quantified' in span else None | |
annotated_text += get_span_start(type, title) if tag == "span" else get_rs_start(type) | |
annotated_text += escape(text[span['offset_start']: span['offset_end']]) | |
annotated_text += get_span_end() if tag == "span" else get_rs_end() | |
start = span['offset_end'] | |
annotated_text += escape(text[start: len(text)]) | |
return annotated_text | |
def extract_quantities(client, x_all, column_text_index): | |
# relevant_items = ['magnetic field strength', 'magnetic induction', 'maximum energy product', | |
# "magnetic flux density", "magnetic flux"] | |
# property_keywords = ['coercivity', 'remanence'] | |
output_data = [] | |
for idx, example in tqdm(enumerate(x_all), desc="extract quantities"): | |
text = example[column_text_index] | |
spans = GrobidQuantitiesProcessor(client).extract_quantities(text) | |
data_record = { | |
"id": example[0], | |
"filename": example[1], | |
"passage_id": example[2], | |
"text": text, | |
"spans": spans | |
} | |
output_data.append(data_record) | |
return output_data | |
def extract_materials(client, x_all, column_text_index): | |
output_data = [] | |
for idx, example in tqdm(enumerate(x_all), desc="extract materials"): | |
text = example[column_text_index] | |
spans = GrobidMaterialsProcessor(client).extract_materials(text) | |
data_record = { | |
"id": example[0], | |
"filename": example[1], | |
"passage_id": example[2], | |
"text": text, | |
"spans": spans | |
} | |
output_data.append(data_record) | |
return output_data | |
def get_parsed_value_type(quantity): | |
if 'parsedValue' in quantity and 'structure' in quantity['parsedValue']: | |
return quantity['parsedValue']['structure']['type'] | |
class BaseProcessor(object): | |
# def __init__(self, grobid_superconductors_client=None, grobid_quantities_client=None): | |
# self.grobid_superconductors_client = grobid_superconductors_client | |
# self.grobid_quantities_client = grobid_quantities_client | |
patterns = [ | |
r'\d+e\d+' | |
] | |
def post_process(self, text): | |
output = text.replace('À', '-') | |
output = output.replace('¼', '=') | |
output = output.replace('þ', '+') | |
output = output.replace('Â', 'x') | |
output = output.replace('$', '~') | |
output = output.replace('−', '-') | |
output = output.replace('–', '-') | |
for pattern in self.patterns: | |
output = re.sub(pattern, lambda match: match.group().replace('e', '-'), output) | |
return output | |
class GrobidProcessor(BaseProcessor): | |
def __init__(self, grobid_client): | |
# super().__init__() | |
self.grobid_client = grobid_client | |
def process_structure(self, input_path): | |
pdf_file, status, text = self.grobid_client.process_pdf("processFulltextDocument", | |
input_path, | |
consolidate_header=True, | |
consolidate_citations=False, | |
segment_sentences=False, | |
tei_coordinates=False, | |
include_raw_citations=False, | |
include_raw_affiliations=False, | |
generateIDs=True) | |
if status != 200: | |
return | |
output_data = self.parse_grobid_xml(text) | |
output_data['filename'] = Path(pdf_file).stem.replace(".tei", "") | |
return output_data | |
def process_single(self, input_file): | |
doc = self.process_structure(input_file) | |
for paragraph in doc['passages']: | |
entities = self.process_single_text(paragraph['text']) | |
paragraph['spans'] = entities | |
return doc | |
def parse_grobid_xml(self, text): | |
output_data = OrderedDict() | |
doc_biblio = grobid_tei_xml.parse_document_xml(text) | |
biblio = { | |
"doi": doc_biblio.header.doi if doc_biblio.header.doi is not None else "", | |
"authors": ", ".join([author.full_name for author in doc_biblio.header.authors]), | |
"title": doc_biblio.header.title, | |
"hash": doc_biblio.pdf_md5 | |
} | |
try: | |
year = dateparser.parse(doc_biblio.header.date).year | |
biblio["publication_year"] = year | |
except: | |
pass | |
output_data['biblio'] = biblio | |
passages = [] | |
output_data['passages'] = passages | |
# if biblio['title'] is not None and len(biblio['title']) > 0: | |
# passages.append({ | |
# "text": self.post_process(biblio['title']), | |
# "type": "paragraph", | |
# "section": "<header>", | |
# "subSection": "<title>", | |
# "passage_id": "title0" | |
# }) | |
if doc_biblio.abstract is not None and len(doc_biblio.abstract) > 0: | |
passages.append({ | |
"text": self.post_process(doc_biblio.abstract), | |
"type": "paragraph", | |
"section": "<header>", | |
"subSection": "<abstract>", | |
"passage_id": "abstract0" | |
}) | |
soup = BeautifulSoup(text, 'xml') | |
text_blocks_body = get_children_body(soup, verbose=False) | |
passages.extend([ | |
{ | |
"text": self.post_process(''.join(text for text in sentence.find_all(text=True) if | |
text.parent.name != "ref" or ( | |
text.parent.name == "ref" and text.parent.attrs[ | |
'type'] != 'bibr'))), | |
"type": "paragraph", | |
"section": "<body>", | |
"subSection": "<paragraph>", | |
"passage_id": str(paragraph_id) + str(sentence_id) | |
} | |
for paragraph_id, paragraph in enumerate(text_blocks_body) for | |
sentence_id, sentence in enumerate(paragraph) | |
]) | |
text_blocks_figures = get_children_figures(soup, verbose=False) | |
passages.extend([ | |
{ | |
"text": self.post_process(''.join(text for text in sentence.find_all(text=True) if | |
text.parent.name != "ref" or ( | |
text.parent.name == "ref" and text.parent.attrs[ | |
'type'] != 'bibr'))), | |
"type": "paragraph", | |
"section": "<body>", | |
"subSection": "<figure>", | |
"passage_id": str(paragraph_id) + str(sentence_id) | |
} | |
for paragraph_id, paragraph in enumerate(text_blocks_figures) for | |
sentence_id, sentence in enumerate(paragraph) | |
]) | |
return output_data | |
class GrobidQuantitiesProcessor(BaseProcessor): | |
def __init__(self, grobid_quantities_client): | |
self.grobid_quantities_client = grobid_quantities_client | |
def extract_quantities(self, text): | |
status, result = self.grobid_quantities_client.process_text(text.strip()) | |
if status != 200: | |
result = {} | |
spans = [] | |
if 'measurements' in result: | |
found_measurements = self.parse_measurements_output(result) | |
for m in found_measurements: | |
item = { | |
"text": text[m['offset_start']:m['offset_end']], | |
'offset_start': m['offset_start'], | |
'offset_end': m['offset_end'] | |
} | |
if 'raw' in m and m['raw'] != item['text']: | |
item['text'] = m['raw'] | |
if 'quantified_substance' in m: | |
item['quantified'] = m['quantified_substance'] | |
if 'type' in m: | |
item["unit_type"] = m['type'] | |
item['type'] = 'property' | |
# if 'raw_value' in m: | |
# item['raw_value'] = m['raw_value'] | |
spans.append(item) | |
return spans | |
def parse_measurements_output(result): | |
measurements_output = [] | |
for measurement in result['measurements']: | |
type = measurement['type'] | |
measurement_output_object = {} | |
quantity_type = None | |
has_unit = False | |
parsed_value_type = None | |
if 'quantified' in measurement: | |
if 'normalizedName' in measurement['quantified']: | |
quantified_substance = measurement['quantified']['normalizedName'] | |
measurement_output_object["quantified_substance"] = quantified_substance | |
if 'measurementOffsets' in measurement: | |
measurement_output_object["offset_start"] = measurement["measurementOffsets"]['start'] | |
measurement_output_object["offset_end"] = measurement["measurementOffsets"]['end'] | |
else: | |
# If there are no offsets we skip the measurement | |
continue | |
# if 'measurementRaw' in measurement: | |
# measurement_output_object['raw_value'] = measurement['measurementRaw'] | |
if type == 'value': | |
quantity = measurement['quantity'] | |
parsed_value = GrobidQuantitiesProcessor.get_parsed(quantity) | |
if parsed_value: | |
measurement_output_object['parsed'] = parsed_value | |
normalized_value = GrobidQuantitiesProcessor.get_normalized(quantity) | |
if normalized_value: | |
measurement_output_object['normalized'] = normalized_value | |
raw_value = GrobidQuantitiesProcessor.get_raw(quantity) | |
if raw_value: | |
measurement_output_object['raw'] = raw_value | |
if 'type' in quantity: | |
quantity_type = quantity['type'] | |
if 'rawUnit' in quantity: | |
has_unit = True | |
parsed_value_type = get_parsed_value_type(quantity) | |
elif type == 'interval': | |
if 'quantityMost' in measurement: | |
quantityMost = measurement['quantityMost'] | |
if 'type' in quantityMost: | |
quantity_type = quantityMost['type'] | |
if 'rawUnit' in quantityMost: | |
has_unit = True | |
parsed_value_type = get_parsed_value_type(quantityMost) | |
if 'quantityLeast' in measurement: | |
quantityLeast = measurement['quantityLeast'] | |
if 'type' in quantityLeast: | |
quantity_type = quantityLeast['type'] | |
if 'rawUnit' in quantityLeast: | |
has_unit = True | |
parsed_value_type = get_parsed_value_type(quantityLeast) | |
elif type == 'listc': | |
quantities = measurement['quantities'] | |
if 'type' in quantities[0]: | |
quantity_type = quantities[0]['type'] | |
if 'rawUnit' in quantities[0]: | |
has_unit = True | |
parsed_value_type = get_parsed_value_type(quantities[0]) | |
if quantity_type is not None or has_unit: | |
measurement_output_object['type'] = quantity_type | |
if parsed_value_type is None or parsed_value_type not in ['ALPHABETIC', 'TIME']: | |
measurements_output.append(measurement_output_object) | |
return measurements_output | |
def get_parsed(quantity): | |
parsed_value = parsed_unit = None | |
if 'parsedValue' in quantity and 'parsed' in quantity['parsedValue']: | |
parsed_value = quantity['parsedValue']['parsed'] | |
if 'parsedUnit' in quantity and 'name' in quantity['parsedUnit']: | |
parsed_unit = quantity['parsedUnit']['name'] | |
if parsed_value and parsed_unit: | |
if has_space_between_value_and_unit(quantity): | |
return str(parsed_value) + str(parsed_unit) | |
else: | |
return str(parsed_value) + " " + str(parsed_unit) | |
def get_normalized(quantity): | |
normalized_value = normalized_unit = None | |
if 'normalizedQuantity' in quantity: | |
normalized_value = quantity['normalizedQuantity'] | |
if 'normalizedUnit' in quantity and 'name' in quantity['normalizedUnit']: | |
normalized_unit = quantity['normalizedUnit']['name'] | |
if normalized_value and normalized_unit: | |
if has_space_between_value_and_unit(quantity): | |
return str(normalized_value) + " " + str(normalized_unit) | |
else: | |
return str(normalized_value) + str(normalized_unit) | |
def get_raw(quantity): | |
raw_value = raw_unit = None | |
if 'rawValue' in quantity: | |
raw_value = quantity['rawValue'] | |
if 'rawUnit' in quantity and 'name' in quantity['rawUnit']: | |
raw_unit = quantity['rawUnit']['name'] | |
if raw_value and raw_unit: | |
if has_space_between_value_and_unit(quantity): | |
return str(raw_value) + " " + str(raw_unit) | |
else: | |
return str(raw_value) + str(raw_unit) | |
class GrobidMaterialsProcessor(BaseProcessor): | |
def __init__(self, grobid_superconductors_client): | |
self.grobid_superconductors_client = grobid_superconductors_client | |
def extract_materials(self, text): | |
preprocessed_text = text.strip() | |
status, result = self.grobid_superconductors_client.process_text(preprocessed_text, | |
"processText_disable_linking") | |
if status != 200: | |
result = {} | |
spans = [] | |
if 'passages' in result: | |
materials = self.parse_superconductors_output(result, preprocessed_text) | |
for m in materials: | |
item = {"text": preprocessed_text[m['offset_start']:m['offset_end']]} | |
item['offset_start'] = m['offset_start'] | |
item['offset_end'] = m['offset_end'] | |
if 'formula' in m: | |
item["formula"] = m['formula'] | |
item['type'] = 'material' | |
item['raw_value'] = m['text'] | |
spans.append(item) | |
return spans | |
def parse_materials(self, text): | |
status, result = self.grobid_superconductors_client.process_texts(text.strip(), "parseMaterials") | |
if status != 200: | |
result = [] | |
results = [] | |
for position_material in result: | |
compositions = [] | |
for material in position_material: | |
if 'resolvedFormulas' in material: | |
for resolved_formula in material['resolvedFormulas']: | |
if 'formulaComposition' in resolved_formula: | |
compositions.append(resolved_formula['formulaComposition']) | |
elif 'formula' in material: | |
if 'formulaComposition' in material['formula']: | |
compositions.append(material['formula']['formulaComposition']) | |
results.append(compositions) | |
return results | |
def parse_material(self, text): | |
status, result = self.grobid_superconductors_client.process_text(text.strip(), "parseMaterial") | |
if status != 200: | |
result = [] | |
compositions = self.output_info(result) | |
return compositions | |
def output_info(self, result): | |
compositions = [] | |
for material in result: | |
if 'resolvedFormulas' in material: | |
for resolved_formula in material['resolvedFormulas']: | |
if 'formulaComposition' in resolved_formula: | |
compositions.append(resolved_formula['formulaComposition']) | |
elif 'formula' in material: | |
if 'formulaComposition' in material['formula']: | |
compositions.append(material['formula']['formulaComposition']) | |
if 'name' in material: | |
compositions.append(material['name']) | |
return compositions | |
def parse_superconductors_output(result, original_text): | |
materials = [] | |
for passage in result['passages']: | |
sentence_offset = original_text.index(passage['text']) | |
if 'spans' in passage: | |
spans = passage['spans'] | |
for material_span in filter(lambda s: s['type'] == '<material>', spans): | |
text_ = material_span['text'] | |
base_material_information = { | |
"text": text_, | |
"offset_start": sentence_offset + material_span['offset_start'], | |
'offset_end': sentence_offset + material_span['offset_end'] | |
} | |
materials.append(base_material_information) | |
return materials | |
class GrobidAggregationProcessor(GrobidProcessor, GrobidQuantitiesProcessor, GrobidMaterialsProcessor): | |
def __init__(self, grobid_client, grobid_quantities_client=None, grobid_superconductors_client=None): | |
GrobidProcessor.__init__(self, grobid_client) | |
self.gqp = GrobidQuantitiesProcessor(grobid_quantities_client) | |
self.gmp = GrobidMaterialsProcessor(grobid_superconductors_client) | |
def process_single_text(self, text): | |
extracted_quantities_spans = self.gqp.extract_quantities(text) | |
extracted_materials_spans = self.gmp.extract_materials(text) | |
all_entities = extracted_quantities_spans + extracted_materials_spans | |
entities = self.prune_overlapping_annotations(all_entities) | |
return entities | |
def extract_quantities(self, text): | |
return self.gqp.extract_quantities(text) | |
def extract_materials(self, text): | |
return self.gmp.extract_materials(text) | |
def prune_overlapping_annotations(entities: list) -> list: | |
# Sorting by offsets | |
sorted_entities = sorted(entities, key=lambda d: d['offset_start']) | |
if len(entities) <= 1: | |
return sorted_entities | |
to_be_removed = [] | |
previous = None | |
first = True | |
for current in sorted_entities: | |
if first: | |
first = False | |
previous = current | |
continue | |
if previous['offset_start'] < current['offset_start'] \ | |
and previous['offset_end'] < current['offset_end'] \ | |
and (previous['offset_end'] < current['offset_start'] \ | |
and not (previous['text'] == "-" and current['text'][0].isdigit())): | |
previous = current | |
continue | |
if previous['offset_end'] < current['offset_end']: | |
if current['type'] == previous['type']: | |
# Type is the same | |
if current['offset_start'] == previous['offset_end']: | |
if current['type'] == 'property': | |
if current['text'].startswith("."): | |
print( | |
f"Merging. {current['text']} <{current['type']}> with {previous['text']} <{previous['type']}>") | |
# current entity starts with a ".", suspiciously look like a truncated value | |
to_be_removed.append(previous) | |
current['text'] = previous['text'] + current['text'] | |
current['raw_value'] = current['text'] | |
current['offset_start'] = previous['offset_start'] | |
elif previous['text'].endswith(".") and current['text'][0].isdigit(): | |
print( | |
f"Merging. {current['text']} <{current['type']}> with {previous['text']} <{previous['type']}>") | |
# previous entity ends with ".", current entity starts with a number | |
to_be_removed.append(previous) | |
current['text'] = previous['text'] + current['text'] | |
current['raw_value'] = current['text'] | |
current['offset_start'] = previous['offset_start'] | |
elif previous['text'].startswith("-"): | |
print( | |
f"Merging. {current['text']} <{current['type']}> with {previous['text']} <{previous['type']}>") | |
# previous starts with a `-`, sherlock this is another truncated value | |
current['text'] = previous['text'] + current['text'] | |
current['raw_value'] = current['text'] | |
current['offset_start'] = previous['offset_start'] | |
to_be_removed.append(previous) | |
else: | |
print("Other cases to be considered: ", previous, current) | |
else: | |
if current['text'].startswith("-"): | |
print( | |
f"Merging. {current['text']} <{current['type']}> with {previous['text']} <{previous['type']}>") | |
# previous starts with a `-`, sherlock this is another truncated value | |
current['text'] = previous['text'] + current['text'] | |
current['raw_value'] = current['text'] | |
current['offset_start'] = previous['offset_start'] | |
to_be_removed.append(previous) | |
else: | |
print("Other cases to be considered: ", previous, current) | |
elif previous['text'] == "-" and current['text'][0].isdigit(): | |
print( | |
f"Merging. {current['text']} <{current['type']}> with {previous['text']} <{previous['type']}>") | |
# previous starts with a `-`, sherlock this is another truncated value | |
current['text'] = previous['text'] + " " * (current['offset_start'] - previous['offset_end']) + \ | |
current['text'] | |
current['raw_value'] = current['text'] | |
current['offset_start'] = previous['offset_start'] | |
to_be_removed.append(previous) | |
else: | |
print( | |
f"Overlapping. {current['text']} <{current['type']}> with {previous['text']} <{previous['type']}>") | |
# take the largest one | |
if len(previous['text']) > len(current['text']): | |
to_be_removed.append(current) | |
elif len(previous['text']) < len(current['text']): | |
to_be_removed.append(previous) | |
else: | |
to_be_removed.append(previous) | |
elif current['type'] != previous['type']: | |
print( | |
f"Overlapping. {current['text']} <{current['type']}> with {previous['text']} <{previous['type']}>") | |
if len(previous['text']) > len(current['text']): | |
to_be_removed.append(current) | |
elif len(previous['text']) < len(current['text']): | |
to_be_removed.append(previous) | |
else: | |
if current['type'] == "material": | |
to_be_removed.append(previous) | |
else: | |
to_be_removed.append(current) | |
previous = current | |
elif previous['offset_end'] > current['offset_end']: | |
to_be_removed.append(current) | |
# the previous goes after the current, so we keep the previous and we discard the current | |
else: | |
if current['type'] == "material": | |
to_be_removed.append(previous) | |
else: | |
to_be_removed.append(current) | |
previous = current | |
new_sorted_entities = [e for e in sorted_entities if e not in to_be_removed] | |
return new_sorted_entities | |
class XmlProcessor(BaseProcessor): | |
def __init__(self, grobid_superconductors_client, grobid_quantities_client): | |
super().__init__(grobid_superconductors_client, grobid_quantities_client) | |
def process_structure(self, input_file): | |
text = "" | |
with open(input_file, encoding='utf-8') as fi: | |
text = fi.read() | |
output_data = self.parse_xml(text) | |
output_data['filename'] = Path(input_file).stem.replace(".tei", "") | |
return output_data | |
def process_single(self, input_file): | |
doc = self.process_structure(input_file) | |
for paragraph in doc['passages']: | |
entities = self.process_single_text(paragraph['text']) | |
paragraph['spans'] = entities | |
return doc | |
def parse_xml(self, text): | |
output_data = OrderedDict() | |
soup = BeautifulSoup(text, 'xml') | |
text_blocks_children = get_children_list_supermat(soup, verbose=False) | |
passages = [] | |
output_data['passages'] = passages | |
passages.extend([ | |
{ | |
"text": self.post_process(''.join(text for text in sentence.find_all(text=True) if | |
text.parent.name != "ref" or ( | |
text.parent.name == "ref" and text.parent.attrs[ | |
'type'] != 'bibr'))), | |
"type": "paragraph", | |
"section": "<body>", | |
"subSection": "<paragraph>", | |
"passage_id": str(paragraph_id) + str(sentence_id) | |
} | |
for paragraph_id, paragraph in enumerate(text_blocks_children) for | |
sentence_id, sentence in enumerate(paragraph) | |
]) | |
return output_data | |
def get_children_list_supermat(soup, use_paragraphs=False, verbose=False): | |
children = [] | |
child_name = "p" if use_paragraphs else "s" | |
for child in soup.tei.children: | |
if child.name == 'teiHeader': | |
pass | |
children.append(child.find_all("title")) | |
children.extend([subchild.find_all(child_name) for subchild in child.find_all("abstract")]) | |
children.extend([subchild.find_all(child_name) for subchild in child.find_all("ab", {"type": "keywords"})]) | |
elif child.name == 'text': | |
children.extend([subchild.find_all(child_name) for subchild in child.find_all("body")]) | |
if verbose: | |
print(str(children)) | |
return children | |
def get_children_list_grobid(soup: object, use_paragraphs: object = True, verbose: object = False) -> object: | |
children = [] | |
child_name = "p" if use_paragraphs else "s" | |
for child in soup.TEI.children: | |
if child.name == 'teiHeader': | |
pass | |
# children.extend(child.find_all("title", attrs={"level": "a"}, limit=1)) | |
# children.extend([subchild.find_all(child_name) for subchild in child.find_all("abstract")]) | |
elif child.name == 'text': | |
children.extend([subchild.find_all(child_name) for subchild in child.find_all("body")]) | |
children.extend([subchild.find_all("figDesc") for subchild in child.find_all("body")]) | |
if verbose: | |
print(str(children)) | |
return children | |
def get_children_body(soup: object, use_paragraphs: object = True, verbose: object = False) -> object: | |
children = [] | |
child_name = "p" if use_paragraphs else "s" | |
for child in soup.TEI.children: | |
if child.name == 'text': | |
children.extend([subchild.find_all(child_name) for subchild in child.find_all("body")]) | |
if verbose: | |
print(str(children)) | |
return children | |
def get_children_figures(soup: object, use_paragraphs: object = True, verbose: object = False) -> object: | |
children = [] | |
child_name = "p" if use_paragraphs else "s" | |
for child in soup.TEI.children: | |
if child.name == 'text': | |
children.extend([subchild.find_all("figDesc") for subchild in child.find_all("body")]) | |
if verbose: | |
print(str(children)) | |
return children | |