"""Phase 6 migration from interface/getAttributes.py. This module hosts the package/scope-related functions migrated verbatim from ``BiddingKG.dl.interface.getAttributes`` as part of the Phase 6 assembly re-organization. It provides: - The ``dict_role_id`` and ``role2id_dict`` role-id mapping constants. - ``getPackageScopePattern`` and the module-level ``pattern_packageScope`` regex used to identify package-scope keywords. - ``getPackagesFromArticle`` which extracts package/标段 information from an article's sentences and injects the resulting package entities into the entity list. - ``getPackage`` which resolves the package a given entity belongs to based on the package scope. """ from __future__ import absolute_import import re import os import copy import pandas as pd from BiddingKG.dl.common.Utils import uniform_package_name, find_package from BiddingKG.dl.interface.Entitys import Entity dict_role_id = {"0":"tenderee", "1":"agency", "2":"win_tenderer", "3":"second_tenderer", "4":"third_tenderer"} role2id_dict = {"tenderee":0, "agency":1, "win_tenderer":2, "second_tenderer":3, "third_tenderer":4} def getPackageScopePattern(): ''' @summary: 获取包的作用域关键词 ''' df = pd.read_excel(os.path.join(os.path.dirname(__file__), "..", "interface", "end.xls")) pattern = "(" for item in df["list_word"]: item = str(item).replace("(","\(").replace(")","\)").replace(".","\.").replace("[","\[").replace("]","\]").replace("-","\-") pattern += item+"|" pattern = pattern[:-1]+")[:：是为]|业绩.{,30}标段[0-9A-Za-z一二三四五六七八九十]{0,3}|##attachment##" return pattern pattern_packageScope = getPackageScopePattern() def getPackagesFromArticle(list_sentence, list_entity): ''' @param: list_sentence:文章的句子list @summary: 将包的信息插入list_entity中 @return: type:list if [包号,句子index,词偏移,标段号] meaning:文章的包/标段信息 ''' if len(list_sentence) == 0: return None list_sentence.sort(key=lambda x: x.sentence_index) PackageList = [] PackageList_scope = [] PackageSet = set() dict_packageCode = dict() main_body_pack = set() # 2024/04/28 保存正文包号 # package_number_pattern = re.compile( # '((施工|监理|监测|勘察|设计|劳务)(标段)?[0-9一二三四五六七八九十ⅠⅡⅢⅣⅤⅥⅦa-zA-Z]{,4}(标段?|包))|(([a-zA-Z]包[：）]?)?第?[0-9一二三四五六七八九十ⅠⅡⅢⅣⅤⅥⅦa-zA-Z]{1,4}标[段包]?)|((标[段号的包项]|([标分子]|合同|项目|采购|（)包|包[组件号])[0-9一二三四五六七八九十ⅠⅡⅢⅣⅤⅥⅦA-Za-z]{1,4})|(([，；。、：（]|第)[0-9一二三四五六七八九十ⅠⅡⅢⅣⅤⅥⅦ]{1,4}分?包)|([a-zA-Z][0-9]{,3}分?[包标])|.{,1}((包组|包件|包号|分?包|标[段号的包]|子项目)编?号?[:：]?[a-zA-Z0-9一二三四五六七八九十ⅠⅡⅢⅣⅤⅥⅦ-]+)|[，；。、：（]包[0-9一二三四五六七八九十ⅠⅡⅢⅣⅤⅥⅦ]{1，4}[^\w]') # 标号 # package_number_pattern = re.compile( # '((施工|监理|监测|勘察|设计|劳务)(标段)?：?第?([一二三四五六七八九十]{1,3}|[ⅠⅡⅢⅣⅤⅥⅦ]{1,3}|[a-zA-Z0-9]{1,9}\-?[a-zA-Z0-9-]{,9})?[分子]?(标[段包项]?|包[组件标]?|合同[包段]))\ # |(([a-zA-Z]包[：（）]?)?第?([一二三四五六七八九十]{1,3}|[ⅠⅡⅢⅣⅤⅥⅦ]{1,3}|[a-zA-Z0-9]{1,9}\-?[a-zA-Z0-9-]{,9})[分子]?(标[段包项]?|合同[包段]))\ # |(([，；。、：（]|第)?([一二三四五六七八九十]{1,3}|[ⅠⅡⅢⅣⅤⅥⅦ]{1,3}|[a-zA-Z0-9]{1,9}\-?[a-zA-Z0-9-]{,9})[分子]?(标[段包项]?|包[组件标]?|合同[包段]))\ # |((标[段包项]|标段（包）|包[组件标]|[标分子（]包)(\[|【)?：?([一二三四五六七八九十]{1,3}|[ⅠⅡⅢⅣⅤⅥⅦ]{1,3}|[a-zA-Z0-9]{1,9}\-?[a-zA-Z0-9-]{,9}))\ # |[，；。、：（](标的?|项目|子项目?)(\[|【)?：?([一二三四五六七八九十]+|[0-9]{1,9})\ # |((([标分子（]|合同|项目|采购)包|[，。]标的|子项目|[分子]标|标[段包项]|包[组件标]?)编?号[:：]?[a-zA-Z0-9一二三四五六七八九十ⅠⅡⅢⅣⅤⅥⅦ-]{1,9})\ # |[，；。、：（]?(合同|分|子)?包：?([一二三四五六七八九十]{1,3}|[ⅠⅡⅢⅣⅤⅥⅦ]{1,3}|[a-zA-Z0-9]{1,9}\-?[a-zA-Z0-9-]{,9})') other_package_pattern = re.compile( '((项目|物资|设备|场次|标段|标的|产品)(名称)?)[:：]([^，。]{2,50}?)[，。]') # # 2020/11/23 大网站规则 调整 package_N_name_pattern， package_N_name_pattern 中的项目 改为 子项目 win_tenderer_pattern = re.compile('(中标候?选?人|供应商)(名称)?[:：](.{2,25})[，。]') # 2020/11/23 大网站规则 调整 model_pattern = re.compile('(型号|序号)[:：]([^，。]{2,20})[，。]') # 2020/11/23 大网站规则 调整 number_pattern = re.compile("[0-9A-Za-z一二三四五六七八九十ⅠⅡⅢⅣⅤⅥⅦ]{1,4}") package_code_pattern = re.compile("(?:编号[:：]?\s*)([-\dA-Za-z\(\)]{1,20})") # 纯数字类型的包号统一，例如：'01','1' re_digital = re.compile("^\d+$") def changeIndexFromWordToWords(tokens, word_index): ''' @summary:转换某个字的字偏移为词偏移 ''' before_index = 0 after_index = 0 for i in range(len(tokens)): after_index = after_index + len(tokens[i]) if before_index <= word_index and after_index >= word_index: return i before_index = after_index package_names = [] def extractPackageCode(tokens, word_index, size=20, pattern=package_code_pattern): ''' @summary:抽取包附近的标段号 @param: tokens:包所在句子的分词 word_index:包所在字偏移 size:左右各取多少个词 pattern:提取标段号的正则 @return: type:string,meaning:标段号 ''' index = changeIndexFromWordToWords(tokens, word_index) if index < size: begin = index else: begin = index - size if index + size > len(tokens): end = len(tokens) else: end = index + size # 拿到左右两边的词语组成短语 text = "".join(tokens[begin:end]) # 在短语中的字偏移 new_word_index = word_index - len("".join(tokens[:begin])) min_distance = len(text) packageCode = None for the_iter in re.finditer(pattern, text): # 算出最小距离 distance = min([abs(new_word_index - the_iter.span()[0]), abs(new_word_index - the_iter.span()[1])]) if distance < min_distance: min_distance = distance packageCode = the_iter.group(1) return packageCode def get_package(): PackageList_scope = [] True_package = set() for i in range(len(list_sentence)): PackageList_item = [] PackageList_item_scope = [] content = list_sentence[i].sentence_text # content = content.replace('号，', '号：').replace(':', '：').replace('(', '（').replace(')', '）') # # .replace('-包',' 包').replace('包-', '包 ').replace('-标', ' 标').replace('标段-', '标段 ').replace('-合同包', ' 合同包') # 72760191 标段：№10 # content = re.sub('[一二三四五六七八九十\d](标[段包项]|包[组件标])编号', ' 标段编号', content) # # for it in re.finditer('CA标|(每个?|所有|相关|个|各|不分)[分子]?(标[段包项]?|包[组件标]?|合同包)|(质量|责任)三包|包[/每]|标段(划分|范围)|(承|压缩|软|皮|书|挂)包\ # |标[识注签贴配]|[商油]标号|第X包|第[一二三四五六七八九十]+至[一二三四五六七八九十]+(标[段包项]?|包[组件标]?|合同[包段])\ # |\.(docx|doc|pdf|xlsx|xls|jpg)|[一二三四五]次|五金|\d+[年月]|[\d.,]+万?元|\d+\.\d+', content): # content = content.replace(it.group(0), ' ' * len(it.group(0))) # tokens = list_sentence[i].tokens # _names = [] # for iter in re.finditer(package_number_pattern, content): # if re.search('(业绩|信誉要求)：', content[:iter.start()]): # 前面有业绩或信誉的标段去掉 # continue # # print('提取到标段：%s， 前后文：%s'%(iter.group(), content[iter.start()-5:iter.end()+5])) # if re.match('\d', iter.group(0)) and re.search('\d\.$', content[:iter.start()]): # 排除2.10标段3 5.4标段划分 这种情况 # # print('过滤掉错误包：', iter.group()) # continue # if re.search('[承每书/]包|XX|xx', iter.group(0)) or re.search('\d包[/每]\w|一包[0-9一二三四五六七八九十]+', content[iter.start():iter.end()+3]) or re.search('[a-zA-Z0-9一二三四五六七八九十ⅠⅡⅢⅣⅤⅥⅦ-]{6,}', iter.group(0)): # # print('过滤掉错误包：', iter.group()) # continue # elif iter.end()+2 < len(content) and re.search('标准|标的物|标志|包装|划分|标书', content[iter.start():iter.end()+2]): # # print('过滤掉错误包：',iter.group()) # continue # elif re.search('同一(标段?|包)', content[max(0, iter.start()-2):iter.end()]): # 不得参加同一标段 # # print('过滤掉错误包：', iter.group()) # continue # elif re.search('三包', content[max(0, iter.start()-2):iter.end()]) and re.search('第三包', content[max(0, iter.start()-2):iter.end()])==None: # 规规章和“三包”规定 # # print('过滤掉错误包：', iter.group()) # continue # elif re.search('[1-9]\d{2,}$|\d{4,}|^[1-9]\d{2,}|合同包[A-Za-z]{2,}', iter.group(0)): # # print('过滤掉错误包号5：', iter.group(0)) # continue tokens = list_sentence[i].tokens _names = [] for iter in find_package(content): temp_package_number = uniform_package_name(iter.group(0)) True_package.add(temp_package_number) PackageList_item.append({"name": temp_package_number, "sentence_index": list_sentence[i].sentence_index, "offsetWords_begin": changeIndexFromWordToWords(tokens, iter.span()[0]), "offsetWord_begin": iter.span()[0], "offsetWord_end": iter.span()[1]}) # PackageList_item.append([temp_package_number,i,changeIndexFromWordToWords(tokens,iter.span()[0]),iter.span()[0],iter.span()[1]]) code = extractPackageCode(tokens, iter.span()[0]) if code is not None: dict_packageCode[temp_package_number] = code PackageSet.add(temp_package_number) if not list_sentence[i].in_attachment: # 保存不在附件的包号 main_body_pack.add(temp_package_number) # 识别packageScope for iter in re.finditer(pattern_packageScope, content): PackageList_item_scope.append({"name": "", "sentence_index": list_sentence[i].sentence_index, "offsetWords_begin": changeIndexFromWordToWords(tokens, iter.span()[0]), "offsetWord_begin": iter.span()[0], "offsetWord_end": iter.span()[1]}) # PackageList_item_scope.append(["",i,changeIndexFromWordToWords(tokens,iter.span()[0]),iter.span()[0],iter.span()[1]]) PackageList_item_scope = PackageList_item + PackageList_item_scope PackageList_item_scope.sort(key=lambda x: x["offsetWord_begin"]) PackageList_scope = PackageList_scope + PackageList_item_scope PackageList_item.sort(key=lambda x: x["sentence_index"]) return PackageList_scope, True_package def get_win_project(): '''获取多个项目多个中标人的项目''' PackageList_scope = [] True_package = set() # 2020/11/23 大网站规则 调整 if len(PackageSet) == 0 and len( set([it.entity_text for it in list_entity if it.entity_type in ['org', 'company'] and it.label == 2])) > 1: for i in range(len(list_sentence)): PackageList_item = [] PackageList_item_scope = [] content = list_sentence[i].sentence_text tokens = list_sentence[i].tokens names = re.findall(other_package_pattern, content) N_names = re.findall(win_tenderer_pattern, content) if len(names) != 1 or len(N_names) != 1: continue for iter in re.finditer(other_package_pattern, content): temp_package_number = iter.group(4) xinghao = re.search(model_pattern, content) if xinghao: temp_package_number = temp_package_number + '+' + xinghao.group(2) # print('新正则采购包名补充',temp_package_number) if re.search(re_digital, temp_package_number): temp_package_number = str(int(temp_package_number)) True_package.add(temp_package_number) PackageList_item.append( {"name": temp_package_number, "sentence_index": list_sentence[i].sentence_index, "offsetWords_begin": changeIndexFromWordToWords(tokens, iter.span()[0]), "offsetWord_begin": iter.span()[0], "offsetWord_end": iter.span()[1]}) # PackageList_item.append([temp_package_number,i,changeIndexFromWordToWords(tokens,iter.span()[0]),iter.span()[0],iter.span()[1]]) code = extractPackageCode(tokens, iter.span()[0]) if code is not None: dict_packageCode[temp_package_number] = code PackageSet.add(temp_package_number) if not list_sentence[i].in_attachment: # 保存不在附件的包号 main_body_pack.add(temp_package_number) # 识别packageScope for iter in re.finditer(pattern_packageScope, content): PackageList_item_scope.append({"name": "", "sentence_index": list_sentence[i].sentence_index, "offsetWords_begin": changeIndexFromWordToWords(tokens, iter.span()[0]), "offsetWord_begin": iter.span()[0], "offsetWord_end": iter.span()[1]}) # PackageList_item_scope.append(["",i,changeIndexFromWordToWords(tokens,iter.span()[0]),iter.span()[0],iter.span()[1]]) PackageList_item_scope = PackageList_item + PackageList_item_scope PackageList_item_scope.sort(key=lambda x: x["offsetWord_begin"]) PackageList_scope = PackageList_scope + PackageList_item_scope PackageList_item.sort(key=lambda x: x["sentence_index"]) return PackageList_scope, True_package def get_package_scope(PackageList_scope): PackageList = [] pattern_punctuation = "[:：（）\(\),，。；;]" # print("===packageList_scope",PackageList_scope) for i in range(len(list_sentence)): for j in range(len(PackageList_scope)): if i == PackageList_scope[j]["sentence_index"] and PackageList_scope[j]["name"] != "": _flag = False left_str = list_sentence[i].sentence_text[ PackageList_scope[j]["offsetWord_begin"] - 30:PackageList_scope[j][ "offsetWord_begin"] + 1] right_str = list_sentence[i].sentence_text[ PackageList_scope[j]["offsetWord_begin"]:PackageList_scope[j]["offsetWord_begin"] + 30] _left_find = re.findall(pattern_punctuation, left_str) _right_find = re.findall(pattern_punctuation, right_str) # print(left_str) if re.search("同", left_str[-1:]) is not None and PackageList_scope[j]["name"] == "一": continue if re.search("划分", right_str[:10]) is not None: continue if len(_left_find) > 0 and _left_find[-1] in [":", "："]: _flag = True if len(_right_find) > 0 and _right_find[0] in [":", "："]: _flag = True if _flag: scope_begin = [PackageList_scope[j]["sentence_index"], PackageList_scope[j]["offsetWords_begin"]] else: scope_begin = [PackageList_scope[j]["sentence_index"], 0] # 2024/10/10 改为包作用域开始位置为包号所在句子开头 # if j == 0: # scope_begin = [0, 0] # else: # scope_begin = [PackageList_scope[j - 1]["sentence_index"], # PackageList_scope[j - 1]["offsetWords_begin"]] if j == len(PackageList_scope) - 1: scope_end = [list_sentence[-1].sentence_index, changeIndexFromWordToWords(list_sentence[-1].tokens, len(list_sentence[ -1].sentence_text))] else: scope_end = [PackageList_scope[j + 1]["sentence_index"], PackageList_scope[j + 1]["offsetWords_begin"]] if j>0 and PackageList_scope[j - 1]["sentence_index"] == PackageList_scope[j]["sentence_index"] and \ PackageList_scope[j - 1]["offsetWord_begin"] <= PackageList_scope[j]["offsetWord_begin"] and \ PackageList_scope[j - 1]["offsetWord_end"] >= PackageList_scope[j]["offsetWord_end"]: continue # add package to entity _pack_entity = Entity(doc_id=list_sentence[0].doc_id, entity_id="%s_%s_%s_%s" % ( list_sentence[0].doc_id, i, PackageList_scope[j]["offsetWord_begin"], PackageList_scope[j]["offsetWord_begin"]), entity_text=PackageList_scope[j]["name"], entity_type="package", sentence_index=PackageList_scope[j]["sentence_index"], begin_index=changeIndexFromWordToWords(list_sentence[i].tokens, PackageList_scope[j][ "offsetWord_begin"]), end_index=changeIndexFromWordToWords(list_sentence[i].tokens, PackageList_scope[j]["offsetWord_end"]), wordOffset_begin=PackageList_scope[j]["offsetWord_begin"], wordOffset_end=PackageList_scope[j]["offsetWord_end"], in_attachment=list_sentence[i].in_attachment) list_entity.append(_pack_entity) copy_pack = copy.copy(PackageList_scope[j]) copy_pack["scope"] = [scope_begin, scope_end] copy_pack["hit"] = set() copy_pack["pointer"] = _pack_entity PackageList.append(copy_pack) return PackageList PackageList_scope, True_package = get_package() # PackageList_scope2, True_package2 = get_win_project() # 20240508 与表格提取重复，去掉 # if len(True_package2) > 2: # 同时包含多标段及多中标人的 # PackageList_scope = PackageList_scope + PackageList_scope2 PackageList = get_package_scope(PackageList_scope) # if len(PackageSet)<2: # 20230922只提取到一个包号的去掉，都放在默认包project 2024/02/02 注释掉，防止多标段每篇公告只公布一个标段的没法提取标段号 # return [], set(), {} return PackageList, PackageSet, dict_packageCode, main_body_pack def getPackage(packageList,sentence_index,begin_index,roleid,MAX_DIS=None,DIRECT=None): ''' @param: packageList:文章的包的信息,包号-sent_index-词偏移-字偏移-[[前作用域句子,句内偏移],[后作用域句子,句内偏移]]-匹配集合 sentence_index:实体所在的句子 begin_index:实体所在句子的起始位置 @return:公司实体所属的包 @summary: 优化多标段，确定标段作用域之后，寻找作用域包含该实体的所有包，从前往后找到一个还没有该roleid的包返回，若找到的包都有roleid,则返回第一个，若没有找到包，返回None ''' ''' if len(packageList)==0: return None before_index = None after_index = None equal_index = None equal_count = 0 for pack_index in range(len(packageList)): if packageList[pack_index][1]>sentence_index and after_index is None: after_index = pack_index if packageList[pack_index][1]int(begin_index): if packageList[i-1][4]: return packageList[i-1][0] else: if packageList[i][4]: return packageList[i-1][0] else: return packageList[i][0] return packageList[end_index-1][0] ''' if len(packageList)==0: return None,False list_legalPack = [] for pack_index in range(len(packageList)): if DIRECT=="L" and (packageList[pack_index]["sentence_index"]>sentence_index or (packageList[pack_index]["sentence_index"]==sentence_index and packageList[pack_index]["offsetWords_begin"]>begin_index)): continue if DIRECT=="R" and (packageList[pack_index]["sentence_index"]sentence_index or (packageList[pack_index]["scope"][1][0]==sentence_index and packageList[pack_index]["scope"][1][1]>=begin_index)): if MAX_DIS is not None: if abs(sentence_index-packageList[pack_index]["sentence_index"])<=MAX_DIS: list_legalPack.append(pack_index) else: list_legalPack.append(pack_index) # if (packageList[pack_index]["scope"][0][0] < sentence_index # or (packageList[pack_index]["scope"][0][0] == sentence_index # and packageList[pack_index]["scope"][0][1] <= begin_index)) # and (packageList[pack_index]["scope"][1][0] > sentence_index # or (packageList[pack_index]["scope"][1][0] == sentence_index # and packageList[pack_index]["scope"][1][1] >= begin_index)): # pass _flag = True for _index in list_legalPack: if roleid in packageList[_index]["hit"]: continue else: _flag = False packageList[_index]["hit"].add(roleid) return packageList[_index]["pointer"],_flag if len(list_legalPack)>0: return packageList[0]["pointer"],_flag return None,False __all__ = ["getPackage", "getPackageScopePattern", "getPackagesFromArticle", "pattern_packageScope", "dict_role_id", "role2id_dict"]