# -*- coding: utf-8 -*- """ Created on Tue Aug 11 16:52:40 2020 @author: luol2 """ import logging import regex import sys import io """ A Python 3 refactoring of Vincent Van Asch's Python 2 code at http://www.cnts.ua.ac.be/~vincent/scripts/abbreviations.py Based on A Simple Algorithm for Identifying Abbreviations Definitions in Biomedical Text A. Schwartz and M. Hearst Biocomputing, 2003, pp 451-462. """ logging.basicConfig(format='%(asctime)s : %(levelname)s : %(message)s', level=logging.INFO) log = logging.getLogger('Abbre') class Candidate(str): def __init__(self, value): super().__init__() self.start = 0 self.stop = 0 def set_position(self, start, stop): self.start = start self.stop = stop def yield_lines_from_file(file_path): with open(file_path, 'rb') as f: for line in f: try: line = line.decode('utf-8') except UnicodeDecodeError: line = line.decode('latin-1').encode('utf-8').decode('utf-8') line = line.strip() yield line f.close() def yield_lines_from_doc(doc_text): for line in doc_text.split("\n"): yield line.strip() def best_candidates(sentence): """ :param sentence: line read from input file :return: a Candidate iterator """ if '(' in sentence: # Check some things first if sentence.count('(') != sentence.count(')'): raise ValueError("Unbalanced parentheses: {}".format(sentence)) if sentence.find('(') > sentence.find(')'): raise ValueError("First parentheses is right: {}".format(sentence)) closeindex = -1 while 1: # Look for open parenthesis openindex = sentence.find('(', closeindex + 1) if openindex == -1: break # Look for closing parentheses closeindex = openindex + 1 open = 1 skip = False while open: try: char = sentence[closeindex] except IndexError: # We found an opening bracket but no associated closing bracket # Skip the opening bracket skip = True break if char == '(': open += 1 elif char in [')', ';', ':']: open -= 1 closeindex += 1 if skip: closeindex = openindex + 1 continue # Output if conditions are met start = openindex + 1 stop = closeindex - 1 candidate = sentence[start:stop] # Take into account whitespace that should be removed start = start + len(candidate) - len(candidate.lstrip()) stop = stop - len(candidate) + len(candidate.rstrip()) candidate = sentence[start:stop] if conditions(candidate): new_candidate = Candidate(candidate) new_candidate.set_position(start, stop) yield new_candidate def conditions(candidate): """ Based on Schwartz&Hearst 2 <= len(str) <= 10 len(tokens) <= 2 re.search('\p{L}', str) str[0].isalnum() and extra: if it matches (\p{L}\.?\s?){2,} it is a good candidate. :param candidate: candidate abbreviation :return: True if this is a good candidate """ viable = True if regex.match('(\p{L}\.?\s?){2,}', candidate.lstrip()): viable = True if len(candidate) < 2 or len(candidate) > 10: viable = False if len(candidate.split()) > 2: viable = False if not regex.search('\p{L}', candidate): viable = False if not candidate[0].isalnum(): viable = False return viable def get_definition(candidate, sentence): """ Takes a candidate and a sentence and returns the definition candidate. The definintion candidate is the set of tokens (in front of the candidate) that starts with a token starting with the first character of the candidate :param candidate: candidate abbreviation :param sentence: current sentence (single line from input file) :return: candidate definition for this abbreviation """ # Take the tokens in front of the candidate tokens = regex.split(r'[\s\-]+', sentence[:candidate.start - 2].lower()) #print(tokens) # the char that we are looking for key = candidate[0].lower() # Count the number of tokens that start with the same character as the candidate # print(tokens) firstchars = [t[0] for t in tokens] # print(firstchars) definition_freq = firstchars.count(key) candidate_freq = candidate.lower().count(key) # Look for the list of tokens in front of candidate that # have a sufficient number of tokens starting with key if candidate_freq <= definition_freq: # we should at least have a good number of starts count = 0 start = 0 startindex = len(firstchars) - 1 while count < candidate_freq: if abs(start) > len(firstchars): raise ValueError("candiate {} not found".format(candidate)) start -= 1 # Look up key in the definition try: startindex = firstchars.index(key, len(firstchars) + start) except ValueError: pass # Count the number of keys in definition count = firstchars[startindex:].count(key) # We found enough keys in the definition so return the definition as a definition candidate start = len(' '.join(tokens[:startindex])) stop = candidate.start - 1 candidate = sentence[start:stop] # Remove whitespace start = start + len(candidate) - len(candidate.lstrip()) stop = stop - len(candidate) + len(candidate.rstrip()) candidate = sentence[start:stop] new_candidate = Candidate(candidate) new_candidate.set_position(start, stop) #print('new_candidate:') #print(new_candidate,start,stop) return new_candidate else: raise ValueError('There are less keys in the tokens in front of candidate than there are in the candidate') def select_definition(definition, abbrev): """ Takes a definition candidate and an abbreviation candidate and returns True if the chars in the abbreviation occur in the definition Based on A simple algorithm for identifying abbreviation definitions in biomedical texts, Schwartz & Hearst :param definition: candidate definition :param abbrev: candidate abbreviation :return: """ if len(definition) < len(abbrev): raise ValueError('Abbreviation is longer than definition') if abbrev in definition.split(): raise ValueError('Abbreviation is full word of definition') sindex = -1 lindex = -1 while 1: try: longchar = definition[lindex].lower() except IndexError: raise shortchar = abbrev[sindex].lower() if not shortchar.isalnum(): sindex -= 1 if sindex == -1 * len(abbrev): if shortchar == longchar: if lindex == -1 * len(definition) or not definition[lindex - 1].isalnum(): break else: lindex -= 1 else: lindex -= 1 if lindex == -1 * (len(definition) + 1): raise ValueError("definition {} was not found in {}".format(abbrev, definition)) else: if shortchar == longchar: sindex -= 1 lindex -= 1 else: lindex -= 1 # print('lindex:',lindex,len(definition),definition[lindex:len(definition)]) new_candidate = Candidate(definition[lindex:len(definition)]) new_candidate.set_position(definition.start+lindex+len(definition), definition.stop) definition = new_candidate tokens = len(definition.split()) length = len(abbrev) if tokens > min([length + 5, length * 2]): raise ValueError("did not meet min(|A|+5, |A|*2) constraint") # Do not return definitions that contain unbalanced parentheses if definition.count('(') != definition.count(')'): raise ValueError("Unbalanced parentheses not allowed in a definition") # print('select:') # print(definition,definition.start, definition.stop) new_definition_dict={'definition':definition,'start':definition.start,'stop':definition.stop} return new_definition_dict def extract_abbreviation_definition_pairs(file_path=None, doc_text=None): abbrev_map = [] omit = 0 written = 0 if file_path: sentence_iterator = enumerate(yield_lines_from_file(file_path)) elif doc_text: sentence_iterator = enumerate(yield_lines_from_doc(doc_text)) else: return abbrev_map for i, sentence in sentence_iterator: #print(sentence) try: for candidate in best_candidates(sentence): #print(candidate) try: #print('begin get definition') definition = get_definition(candidate, sentence) #print('get_definition:') #print(definition) except (ValueError, IndexError) as e: #log.debug("{} Omitting candidate {}. Reason: {}".format(i, candidate, e.args[0])) omit += 1 else: try: definition_dict = select_definition(definition, candidate) except (ValueError, IndexError) as e: #log.debug("{} Omitting definition {} for candidate {}. Reason: {}".format(i, definition_dict, candidate, e.args[0])) omit += 1 else: definition_dict['abbre']=candidate abbrev_map.append(definition_dict) written += 1 except (ValueError, IndexError) as e: log.debug("{} Error processing sentence {}: {}".format(i, sentence, e.args[0])) log.debug("{} abbreviations detected and kept ({} omitted)".format(written, omit)) return abbrev_map def postprocess_abbr(ner_result,ori_text): final_result={} if len(ner_result)==0: return [] # abbr recognition abbr_result=extract_abbreviation_definition_pairs(doc_text=ori_text) # read ner results nor_loc_list={} #{entity_name_location:entity_information} for ele in ner_result: nor_loc_list[str(ele[0])+' '+str(ele[1])]=ele final_result['\t'.join(ele)]=[int(ele[0]),int(ele[1])] #abbr matching for abbr in abbr_result: abbr_index=str(abbr['start'])+' '+str(abbr['stop']) if abbr_index in nor_loc_list.keys(): line=ori_text abbr_text=abbr['abbre'] abbr_eid=0 while line.find(abbr_text)>=0: abbr_sid=line.find(abbr_text)+abbr_eid abbr_eid=abbr_sid+len(abbr_text) # print(abbr_sid,abbr_eid) if abbr_sid>0 and abbr_eid0 and abbr_eid==len(ori_text): if ori_text[abbr_sid-1].isalnum()==False : final_result[str(abbr_sid)+'\t'+str(abbr_eid)+'\t'+nor_loc_list[abbr_index][2]+'\t'+nor_loc_list[abbr_index][3]]=[abbr_sid,abbr_eid] line=ori_text[abbr_eid:] # print(final_result) sorted_final_result=sorted(final_result.items(), key=lambda kv:(kv[1]), reverse=False) final_result=[] for ele in sorted_final_result: final_result.append(ele[0].split('\t')) return final_result def ner_abbr(ner_result,abbr_result,ori_text): # read ner results nor_name_list={} #{entity_name:entity_information} nor_loc_list={} #{entity_name_location:entity_information} final_result={} #{entity_information:location} use to sort for ele in ner_result: temp_seg=ele.split('\t') nor_loc_list[temp_seg[0]+' '+temp_seg[1]]=temp_seg nor_name_list[temp_seg[2].lower()]=temp_seg final_result['\t'.join(temp_seg[0:4])]=[int(temp_seg[0]),int(temp_seg[1])] #abbr matching for abbr in abbr_result: abbr_index=str(abbr['start'])+' '+str(abbr['stop']) if abbr_index in nor_loc_list.keys(): line=ori_text abbr_text=abbr['abbre'] abbr_eid=0 while line.find(abbr_text)>=0: abbr_sid=line.find(abbr_text)+abbr_eid abbr_eid=abbr_sid+len(abbr_text) # print(abbr_sid,abbr_eid) if abbr_sid>0 and abbr_eid0 and abbr_eid==len(ori_text): if ori_text[abbr_sid-1].isalnum()==False : final_result[str(abbr_sid)+'\t'+str(abbr_eid)+'\t'+abbr_text+'\t'+nor_loc_list[abbr_index][3]]=[abbr_sid,abbr_eid] line=ori_text[abbr_eid:] # print(final_result) final_result=sorted(final_result.items(), key=lambda kv:(kv[1]), reverse=False) return final_result if __name__ == '__main__': path='//panfs/pan1/bionlp/lulab/luoling/HPO_project/diseaseTag/data/test/results/' fin=open(path+'NCBI_test_phecr_95.tsv','r',encoding='utf-8') context=fin.read().strip().split('\n\n') fin.close() fout=open(path+'NCBI_test_phecr_abbre_95.tsv','w',encoding='utf-8') for doc in context: lines=doc.split('\n') ori_text=lines[1] # print(ori_text) fout.write(lines[0]+'\n'+lines[1]+'\n') if len(lines)>2: abbr_result=extract_abbreviation_definition_pairs(doc_text=ori_text) print(abbr_result) abbr_out=ner_abbr(lines[2:],abbr_result,ori_text) else: abbr_out=[] # print('final:',abbr_out) for ele in abbr_out: fout.write(ele[0]+'\n') fout.write('\n') # sys.exit() fout.close() #last_out=combine_ml_dict_fn(abbr_out,infile) #print(last_out)