services/normalization_utils.py

import re
import time
import numpy as np
from jsonschema import validate
from nltk.tokenize import sent_tokenize
import json
class DocumentNormalizer:
    def __init__(self, json_schema_path ):
        self.json_schema = json.load(open(json_schema_path,"r"))
        self.cit_marker_matcher = re.compile("(^[^A-Za-z\d]*)([0-9]+)(?=[^A-Za-z\d]*$)")
        self.sentence_boundary_matcher = re.compile("\.\s")
    
    def normalize( self, paper, requires_validation = True ):
        ##### Author #####
        parsed_authors = self.parse_author( paper )
        ##### Title #####
        parsed_title = self.parse_title( paper )
        ##### Venue #####
        parsed_venue = self.parse_venue( paper )
        ##### DOI #####
        parsed_doi = self.parse_doi(paper)
        ##### URL #####
        parsed_url = self.parse_url(paper)
        ##### PublicationDate #####
        parsed_pub_date = self.parse_pub_date(paper)
        ##### Reference #####
        parsed_reference, bib_entry_key_to_row_id_mapper = self.parse_reference(paper)
        
        ##### Content #####
        parsed_content = self.parse_content(paper, bib_entry_key_to_row_id_mapper)
        ##### Abstract (The abstract text stored in the metadata) #####
        abstract_text = (" ".join(self.get_sentence_list_from_parsed_sections( parsed_content["Abstract_Parsed"] ))).strip()
        
        ##### Last_update_unixtime ######
        Last_update_unixtime = int(time.time())
        ##### Others #####
        Abstract_in_metadata = abstract_text != ""
        isDuplicated = False
        
        normalized_paper = {
            "Author":parsed_authors,
            "Title":parsed_title,
            "Abstract":abstract_text,
            "Venue":parsed_venue,
            "DOI":parsed_doi,
            "URL":parsed_url,
            "PublicationDate":parsed_pub_date,
            "Content":parsed_content,
            "Reference":parsed_reference,
            "Last_update_unixtime":Last_update_unixtime,
            "Abstract_in_metadata":Abstract_in_metadata,
            "isDuplicated":isDuplicated
        }
        
        ##### Additional IDs, this is only added for S2ORC dataset #####
        additional_ids = self.parse_additional_ids(paper)
        normalized_paper.update( additional_ids )
        
        if requires_validation:
            try:
                validate(instance=normalized_paper, schema=self.json_schema)
            except:
                return None
        return normalized_paper
    
    def get_sentence_list_from_parsed_sections(self, parsed_sections ):
        sentence_list = []
        for section in parsed_sections:
            sentence_list.append(str(section.get( "section_title", "" )))
            for para in section.get("section_text",[]):
                for sen in para.get("paragraph_text", []):
                    sentence_list.append( str(sen.get("sentence_text","")) )
        return sentence_list
    
    
    def parse_author(self, paper ):
        try:
            parsed_authors = []
            authors = paper.get("authors", [] )
            for author in authors:
                parsed_authors.append(
                    {
                        "GivenName":str( author.get( "first", "" ).replace("None","") ),
                        "FamilyName":str( author.get( "last", "" ).replace("None","") )
                    }
                )
        except:
            parsed_authors = []
        return parsed_authors
    
    def parse_title(self, paper ):
        try:
            parsed_title = str(paper.get("title", "")).replace("None","").lstrip("[").rstrip("]")
        except:
            parsed_title = ""
        return parsed_title
    
    def parse_venue(self, paper):
        try:
            parsed_venue = str(paper.get("venue", "")).replace("None","")
        except:
            parsed_venue = ""
        
        if parsed_venue.strip() == "":
            try:
                parsed_venue = str(paper.get("journal","")).replace("None","")
            except:
                parsed_venue = ""
        return parsed_venue
    
    def parse_doi(self, paper):
        try:
            parsed_doi = str( paper.get("doi","") ).replace("None","")
        except:
            parsed_doi = ""
        return parsed_doi
    
    def parse_url(self, paper):
        try:
            parsed_doi = str(paper.get("doi","")).strip().replace("%", "%25").replace('"', "%22").replace("#", "%23").replace(" ", "%20").replace("?", "%3F").replace("None","")
            if parsed_doi.strip() != "":
                parsed_url = "https://doi.org/" + parsed_doi
            else:
                parsed_url = str(paper.get("s2_url", ""))
        except:
            parsed_url = ""
        return parsed_url
        
    
    def parse_pub_date( self, paper ):
        try:
            year = str(int(paper.get("year", ""))).replace("None","")
        except:
            year = ""
        return {
            "Year":year
        }

    def parse_para( self, para, bib_entry_key_to_row_id_mapper ):
        paragraph_text = [{ "sentence_id":str(sen_id), "sentence_text": str(sen), "cite_spans":[] }  
                          for sen_id, sen in  enumerate(self.sent_tok( str(para.get("text",""))) )]
        para_cite_spans = para.get( "cite_spans", [] )
        for cite_span in para_cite_spans:
            start, end = cite_span["start"], cite_span["end"]
            for sen in paragraph_text:
                if start < len( sen["sentence_text"] ):
                    end = min( end, len( sen["sentence_text"] ) )
                    sen["cite_spans"].append(
                        {
                            "start":start,
                            "end":end,
                            "text":sen["sentence_text"][start:end],
                            "ref_id":cite_span["ref_id"]
                        }
                    )
                    break
                else:
                    start -= len( sen["sentence_text"] )
                    end -= len( sen["sentence_text"] )
        cleaned_paragraph_text = []
        for sen in paragraph_text:
            sentence_text = sen["sentence_text"]
            cite_spans = sen["cite_spans"]
            
            sentence_text = sentence_text.rstrip()
            
            cite_spans.sort( key= lambda x:x["start"] )
            
            cleaned_cite_spans = []
            for sen_cite_span in cite_spans:
                if sen_cite_span["ref_id"] not in bib_entry_key_to_row_id_mapper:
                    continue
                
                start, end = sen_cite_span["start"], sen_cite_span["end"]
                ## make sure ther is no overlapping between multiple citation markers
                if len(cleaned_cite_spans) > 0 and start < int(cleaned_cite_spans[-1]["end"]):
                    continue
                
                if start >= len(sentence_text):
                    continue
                end = min( end, len(sentence_text) )
                
                sen_cite_span["start"] = str(start)
                sen_cite_span["end"] = str(end)
                sen_cite_span["text"] = sentence_text[start:end]
                sen_cite_span["ref_id"] = str(bib_entry_key_to_row_id_mapper[ sen_cite_span["ref_id"] ])
                
                cleaned_cite_spans.append( sen_cite_span )
            
            sentence_id = str(len(cleaned_paragraph_text))
            cleaned_paragraph_text.append(
                {
                    "sentence_id":sentence_id,
                    "sentence_text":sentence_text,
                    "cite_spans":cleaned_cite_spans
                }
            )
            
        return cleaned_paragraph_text
    
    
    def parse_para_list( self, para_list, bib_entry_key_to_row_id_mapper ):
        section_list = []
        current_section = None
        
        for para in para_list:            
            paragraph_text = self.parse_para( para, bib_entry_key_to_row_id_mapper )
            
            para_section = str(para.get("section",""))
            
            if current_section is None or (para_section != "" and para_section != current_section["section_title"]):
                if current_section is not None:
                    section_list.append(current_section)
                current_section = {
                    "section_id":str(len(section_list)),
                    "section_title":para_section,
                    "section_text":[
                        {
                            "paragraph_id":"0",
                            "paragraph_text":paragraph_text
                        }
                    ]
                }
            else:
                next_para_id = len(current_section["section_text"])
                current_section["section_text"].append(
                    {
                        "paragraph_id":str(next_para_id),
                        "paragraph_text":paragraph_text
                    }
                )
        if current_section is not None:
            section_list.append(current_section)
            
        if (" ".join(self.get_sentence_list_from_parsed_sections( section_list ))).strip() == "":
            section_list = []
            
        return section_list
                
    def parse_content( self, paper, bib_entry_key_to_row_id_mapper ):
        ### Abstract 
        abstract = ""
        ### Abstract_Parsed
        try:
            pdf_parsed_abstract = paper.get("pdf_parse",{}).get("abstract",[])
            if len( pdf_parsed_abstract ) == 0:
                abstract_text = str(paper.get("abstract",""))
                if abstract_text != "None" and abstract_text != "":
                    pdf_parsed_abstract = [ { "section":"Abstract", "text":abstract_text } ]
            assert len(pdf_parsed_abstract) > 0
            
            abstract_parsed = self.parse_para_list( pdf_parsed_abstract, bib_entry_key_to_row_id_mapper )
        except:
            abstract_parsed = []
        
        ### Fullbody
        fullbody = ""
        
        ### Fullbody_Parsed
        try:
            fullbody_parsed = self.parse_para_list( paper.get( "pdf_parse", {} ).get("body_text", []), bib_entry_key_to_row_id_mapper )
        except:
            fullbody_parsed = []
        return {
            "Abstract":abstract,
            "Abstract_Parsed":abstract_parsed,
            "Fullbody":fullbody,
            "Fullbody_Parsed":fullbody_parsed
        }
    
    def parse_reference(self, paper):
        try:
            bibref_text = {}
            body_text = paper.get("pdf_parse",{}).get("body_text", [])
            for para in body_text:
                for cit in para.get("cite_spans", []):
                    if isinstance(cit, dict):
                        ref_id, ref_text = cit.get("ref_id",""), cit.get("text","")
                        if ref_id != "":
                            bibref_text[ref_id] = ref_text
                            
            for ref_id in bibref_text:
                ref_text = bibref_text[ref_id]
                matched_texts = self.cit_marker_matcher.findall(ref_text)
                if len(matched_texts) > 0:
                    ref_text = matched_texts[0][1]+"."
                else:
                    ref_text = ""
                bibref_text[ref_id] = ref_text
                
        except:
            bibref_text = {}
            
        try:
            reference = []
            bib_entry_key_to_row_id_mapper = {}
            
            bib_entries = paper.get("pdf_parse",{}).get("bib_entries",{})
            bib_entry_keys = list(bib_entries.keys())
            try:
                bib_entry_keys.sort( key = lambda x : int(x[6:]) )
            except:
                pass    

            for bib_entry_key in bib_entry_keys:
                try:
                    parsed_entry = self.convert_bibentry_to_metadata( bib_entries[bib_entry_key] )
                    reference_text = self.get_citation_from_paper_metadata(parsed_entry)
                    if bibref_text.get(bib_entry_key,"").strip() != "":
                        reference_text = bibref_text[bib_entry_key] + " "+ reference_text
                    parsed_entry["ReferenceText"] = reference_text
                    
                    bib_entry_key_to_row_id_mapper[bib_entry_key] = len(reference)
                    reference.append(parsed_entry)
                except:
                    continue
        except:
            reference = []
            bib_entry_key_to_row_id_mapper = {}

        return reference, bib_entry_key_to_row_id_mapper
    
    def parse_additional_ids(self, paper):
        try:
            S2CID = str(paper.get("paper_id", "")).replace("None","")
            PMID = str(paper.get("pubmed_id", "")).replace("None","")
            PMCID = str(paper.get("pmc_id", "")).replace("None","")
            ArxivId = str(paper.get("arxiv_id", "")).replace("None","")
            ACLId = str(paper.get("acl_id","")).replace("None","")
            MAGId = str(paper.get("mag_id","")).replace("None","")
        except:
            S2CID = ""
            PMID = ""
            PMCID = ""
            ArxivId = ""
            ACLId = ""
            MAGId = ""
        return {
            "S2CID":S2CID,
            "PMID":PMID,
            "PMCID":PMCID,
            "ArxivId":ArxivId,
            "ACLId":ACLId,
            "MAGId":MAGId
        }

    
    def sent_tok(self, text, min_sen_len = 10 ):

        sens = self.sentence_boundary_matcher.split( text )
        for pos in range( len(sens)-1 ):
            sens[pos] += ". "
        
        return self.merge_sens( sens, min_sen_len = min_sen_len )

    def merge_sens(self, sens, min_sen_len = 10 ):
        out_sens =[]
        current_sen = None
    
        for sen in sens:
            sen_len = len(sen.split())
            if sen_len >= min_sen_len:
                if current_sen is not None:
                    out_sens.append( current_sen )
                current_sen = sen
            else:
                if current_sen is not None: 
                    current_sen += sen
                else:
                    current_sen = sen
        if current_sen is not None:
            if len( current_sen.split() ) < min_sen_len and len( out_sens ) > 0:
                out_sens[-1] += current_sen
            else:
                out_sens.append(current_sen)
        return out_sens
    
    def convert_bibentry_to_metadata(self, bibentry):
        metadata = {}
        metadata["Title"] = bibentry["title"]
        metadata["Author"] = []
        for author in bibentry.get("authors",[]):
            metadata["Author"].append({
                "GivenName":author.get("first",""),
                "FamilyName": author.get("last", "")
            })
        metadata["Venue"] = bibentry.get("venue","")
        metadata["PublicationDate"] = {"Year":str( bibentry.get("year","") )}
        return metadata


    def get_citation_from_paper_metadata(self,  paper_metadata ):
        author = paper_metadata.get("Author",[])
        title = paper_metadata.get("Title","")
        venue = paper_metadata.get("Venue","")
        year = paper_metadata.get("PublicationDate",{}).get("Year","")   
            
        author_list = []
        for pos,author_item in enumerate(author):
            if pos == 0:
                author_list.append( "%s, %s"%(  author_item.get("FamilyName",""), author_item.get("GivenName","") ) )
            else:
                author_list.append( "%s %s"%(  author_item.get("GivenName",""), author_item.get("FamilyName","") ) )    

        if len(author_list)>3:
            author_info = author_list[0] + " et al"
        elif len(author_list)>1:
            author_info = ", ".join( author_list[:-1] ) + ", and " + author_list[-1]
        elif len(author_list)==1:
            author_info = author_list[0]
        else:
            author_info = ""
        author_info += "."  

        title_info = "“"+title.rstrip(".")+".”"
        journal_info = venue
        if year.strip() != "":
            year_info = "(%s)"%(year)
        else:
            year_info = ""  

        citation_text = " ".join(" ".join( [author_info, title_info, journal_info, year_info  ] ).split()) +"."

        return citation_text