BIDI_ML_INFO_EXTRACTION/BiddingKG/maxcompute/extract_check.py

#coding=utf-8
# evaluate为该方法的入口函数，必须用这个名字

from odps.udf import annotate
from odps.distcache import get_cache_archive
from odps.distcache import get_cache_file
from odps.udf import BaseUDTF
from odps.udf import BaseUDAF



# 配置pandas依赖包
def include_package_path(res_name):
    import os, sys
    archive_files = get_cache_archive(res_name)
    dir_names = sorted([os.path.dirname(os.path.normpath(f.name)) for f in archive_files
                        if '.dist_info' not in f.name], key=lambda v: len(v))
    sys.path.append(dir_names[0])

    return os.path.dirname(dir_names[0])

# 可能出现类似RuntimeError: xxx has been blocked by sandbox
# 这是因为包含C的库，会被沙盘block，可设置set odps.isolation.session.enable = true
def include_file(file_name):
    import os, sys
    so_file = get_cache_file(file_name)
    sys.path.append(os.path.dirname(os.path.abspath(so_file.name)))

def include_so(file_name):
    import os, sys
    so_file = get_cache_file(file_name)

    with open(so_file.name, 'rb') as fp:
        content=fp.read()
        so = open(file_name, "wb")
        so.write(content)
        so.flush()
        so.close()

#初始化业务数据包，由于上传限制,python版本以及archive解压包不统一等各种问题，需要手动导入
def init_env(list_files,package_name):
    import os,sys

    if len(list_files)==1:
        so_file = get_cache_file(list_files[0])
        cmd_line = os.path.abspath(so_file.name)
        os.system("unzip -o %s -d %s"%(cmd_line,package_name))
    elif len(list_files)>1:
        cmd_line = "cat"
        for _file in list_files:
            so_file = get_cache_file(_file)
            cmd_line += " "+os.path.abspath(so_file.name)
        cmd_line += " > temp.zip"
        os.system(cmd_line)
        os.system("unzip -o temp.zip -d %s"%(package_name))
    # os.system("rm -rf %s/*.dist-info"%(package_name))
    # return os.listdir(os.path.abspath("local_package"))
    # os.system("echo export LD_LIBRARY_PATH=%s >> ~/.bashrc"%(os.path.abspath("local_package")))
    # os.system("source ~/.bashrc")
    sys.path.insert(0,os.path.abspath(package_name))

    # sys.path.append(os.path.join(os.path.abspath("local_package"),"interface_real"))

import json
class MyEncoder(json.JSONEncoder):

    def default(self, obj):
        if isinstance(obj, np.ndarray):
            return obj.tolist()
        elif isinstance(obj, bytes):
            return str(obj, encoding='utf-8')
        elif isinstance(obj, (np.float_, np.float16, np.float32,
                              np.float64)):
            return float(obj)
        elif isinstance(obj,(np.int64)):
            return int(obj)
        return json.JSONEncoder.default(self, obj)


@annotate("string,string,string,string,string,string,string,string,string,string,string,string,string,string,string,string,string,string,string,string->string")
class f_json_extract_online(BaseUDTF):

    def __init__(self):

        import uuid
        global uuid

        import logging
        import datetime
        import numpy as np



        global json,MyEncoder,time,log,MyEncoder,np


    def process(self,page_time,doctitle,
                tenderee,tenderee_contact,tenderee_phone,agency,
                agency_contact,agency_phone,sub_docs_json,project_code,
                project_name,product,time_bidclose,time_bidopen,time_release,
                moneysource,person_review,bidway,punish,serviceTime):
        _dict = {}
        _dict["code"] = project_code if project_code is not None else ""
        _dict["name"] = project_name if project_name is not None else ""
        if product is not None and product!="":
            _dict["product"] = product.split(",")
        else:
            _dict["product"] = []
        _dict["time_bidclose"] = time_bidclose if time_bidclose is not None else ""
        _dict["time_bidopen"] = time_bidopen if time_bidopen is not None else ""
        _dict["time_release"] = time_release if time_release is not None else ""
        _dict["moneysource"] = moneysource if moneysource is not None else ""
        if person_review not in (None,''):
            _dict["person_review"] = person_review.split(",")
        else:
            _dict["person_review"] = []
        _dict["bidway"] = bidway if bidway is not None else ""
        _dict["serviceTime"] = serviceTime if serviceTime is not None else ""
        if punish not in (None,''):
            _punish = json.loads(punish)
        else:
            _punish = {}
        for k,v in _punish.items():
            _dict[k] = v

        if sub_docs_json not in (None,''):
            _docs = json.loads(sub_docs_json)
        else:
            _docs = [{}]
        set_comp_contact = set()
        if tenderee not in (None,"") and tenderee_contact not in (None,""):
            set_comp_contact.add("%s-%s-%s-%s"%("tenderee",tenderee,tenderee_contact,tenderee_phone))
        if agency not in (None,"") and agency_contact not in (None,""):
            set_comp_contact.add("%s-%s-%s-%s"%("agency",agency,agency_contact,agency_phone))
        set_pack_comp = set()
        if tenderee not in (None,""):
            set_pack_comp.add("%s-%s-%s"%("Project","tenderee",tenderee))
        if agency not in (None,""):
            set_pack_comp.add("%s-%s-%s"%("Project","agency",agency))
        set_pack_money = set()
        for _d in _docs:
            if len(_d.keys())>0:
                sub_project_name = _d.get("sub_project_name","Project")
                bidding_budget = float(_d.get("bidding_budget",0))
                win_tenderer = _d.get("win_tenderer","")
                win_bid_price = float(_d.get("win_bid_price",0))
                win_tenderer_manager = _d.get("win_tenderer_manager","")
                win_tenderer_phone = _d.get("win_tenderer_phone","")
                second_tenderer = _d.get("second_tenderer","")
                second_bid_price = float(_d.get("second_bid_price",0))
                second_tenderer_manager = _d.get("second_tenderer_manager","")
                second_tenderer_phone = _d.get("second_tenderer_phone","")
                third_tenderer = _d.get("third_tenderer","")
                third_bid_price = float(_d.get("third_bid_price",0))
                third_tenderer_manager = _d.get("third_tenderer_manager","")
                third_tenderer_phone = _d.get("third_tenderer_phone","")
                if win_tenderer not in (None,"") and win_tenderer_manager not in (None,""):
                    set_comp_contact.add("%s-%s-%s-%s"%("win_tenderee",win_tenderer,win_tenderer_manager,win_tenderer_phone))
                if second_tenderer not in (None,"") and second_tenderer_manager not in (None,""):
                    set_comp_contact.add("%s-%s-%s-%s"%("second_tenderer",second_tenderer,second_tenderer_manager,second_tenderer_phone))
                if third_tenderer not in (None,"") and third_tenderer_manager not in (None,""):
                    set_comp_contact.add("%s-%s-%s-%s"%("third_tenderer",third_tenderer,third_tenderer_manager,third_tenderer_phone))

                if win_tenderer not in (None,""):
                    set_pack_comp.add("%s-%s-%s"%(sub_project_name,"win_tenderer",win_tenderer))
                if second_tenderer not in (None,""):
                    set_pack_comp.add("%s-%s-%s"%(sub_project_name,"second_tenderer",second_tenderer))
                if third_tenderer not in (None,""):
                    set_pack_comp.add("%s-%s-%s"%(sub_project_name,"third_tenderer",third_tenderer))

                if bidding_budget>0:
                    set_pack_money.add("%s-%s-%2f"%(sub_project_name,"bidding_budget",bidding_budget))
                if win_bid_price>0:
                    set_pack_money.add("%s-%s-%2f"%(sub_project_name,"win_tenderer",win_bid_price))
                if second_bid_price>0:
                    set_pack_money.add("%s-%s-%2f"%(sub_project_name,"second_tenderer",second_bid_price))
                if third_bid_price>0:
                    set_pack_money.add("%s-%s-%2f"%(sub_project_name,"third_tenderer",third_bid_price))
        _dict["set_comp_contact"] = list(set_comp_contact)
        _dict["set_pack_comp"] = list(set_pack_comp)
        _dict["set_pack_money"] = list(set_pack_money)
        self.forward(json.dumps(_dict,cls=MyEncoder,sort_keys=True,indent=4,ensure_ascii=False))

@annotate("string,string->string")
class f_compair_extract(object):

    def __init__(self):
        import logging
        import re
        import json
        global logging,re,json
        logging.basicConfig(level = logging.INFO,format = '%(asctime)s - %(name)s - %(levelname)s - %(message)s')

    def evaluate(self, json_online,json_result):
        dict_online = json.loads(json_online)
        dict_result = json.loads(json_result)

        logging.info(json_online)

        dict_test = {}
        set_comp_contact = set()
        set_pack_comp = set()
        set_pack_money = set()
        logging.info("1")
        for k,v in dict_result.items():
            if k in ["bidway","moneysource","time_bidclose","serviceTime","time_bidopen","time_release","name"]:
                dict_test[k] = v
            elif k in ["code"]:
                if len(v)>0:
                    dict_test["code"] = v[0]
                else:
                    dict_test["code"] = ""
            elif k in ["person_review","product"]:
                list_temp = v
                list_temp.sort(key=lambda x:x)
                dict_test[k] = list_temp
            elif k in ["punish"]:
                for k1,v1 in v.items():
                    dict_test[k1] = v1
            elif k in ["prem"]:
                for _pack,_prem in v.items():
                    bidding_budget = float(_prem.get("tendereeMoney",0))
                    role_lists = _prem.get("roleList",[])
                    if bidding_budget>0:
                        set_pack_money.add("%s-%s-%2f"%(_pack,"bidding_budget",bidding_budget))
                    for _role in role_lists:
                        role_type = _role[0]
                        role_name = _role[1]
                        role_money = 0 if _role[2]=="" else float(_role[2])
                        contact_list = _role[3]
                        for _person,_phone in contact_list:
                            set_comp_contact.add("%s-%s-%s-%s"%(role_type,role_name,_person,_phone))
                        set_pack_comp.add("%s-%s-%s"%(_pack,role_type,role_name))
                        if role_money >0:
                            set_pack_money.add("%s-%s-%2f"%(_pack,role_type,role_money))
        dict_test["set_comp_contact"] = list(set_comp_contact)
        dict_test["set_pack_comp"] = list(set_pack_comp)
        dict_test["set_pack_money"] = list(set_pack_money)

        logging.info(dict_test)
        logging.info("2")
        dict_compair = {}
        set_keys_online = set(dict_online.keys())
        set_keys_test = set(dict_test.keys())
        union_keys = list(set_keys_online|set_keys_test)
        logging.info(str(union_keys))
        for _key in union_keys:
            logging.info(_key)
            v_online = dict_online.get(_key,"")
            v_test = dict_test.get(_key,"")
            logging.info(v_online)
            logging.info(v_test)
            if isinstance(v_online,list) or isinstance(v_test,list):
                logging.info("3")
                if v_online=="":
                    v_online = []
                if v_test=="":
                    v_test = []
                v_online.sort(key=lambda x:x)
                v_test.sort(key=lambda x:x)
                s_online = set(v_online)
                s_test = set(v_test)
                diff_count = len(s_online-s_test)+len(s_test-s_online)
                dict_compair[_key+"_diff"] = diff_count
                dict_compair[_key+"_online"] = v_online
                dict_compair[_key+"_test"] = v_test
            elif isinstance(v_online,str):
                logging.info("4")
                if v_online==v_test:
                    diff_count = 0
                else:
                    diff_count = 1
                dict_compair[_key+"_diff"] = diff_count
                dict_compair[_key+"_online"] = v_online
                dict_compair[_key+"_test"] = v_test

        return json.dumps(dict_compair,sort_keys=True,indent=4,ensure_ascii=False)

import hashlib
def getMD5(sourceHtml):
    if sourceHtml is not None and len(sourceHtml)>0:
        if isinstance(sourceHtml,str):
            bs = sourceHtml.encode()
        elif isinstance(sourceHtml,bytes):
            bs = sourceHtml
        else:
            return ""
        md5 = hashlib.md5()
        md5.update(bs)
        return md5.hexdigest()
    return ""

def getFingerprint(sourceHtml):
    md5 = getMD5(sourceHtml)
    if md5!="":
        _fingerprint = "md5=%s"%(md5)
    else:
        _fingerprint = ""
    return _fingerprint

@annotate("string,string->string")
class f_getFingerprint(object):

    def __init__(self):
        import logging
        import re
        import json
        global logging,re,json
        logging.basicConfig(level = logging.INFO,format = '%(asctime)s - %(name)s - %(levelname)s - %(message)s')

    def evaluate(self, doctitle,dochtmlcon):
        fingerprint = getFingerprint(doctitle+dochtmlcon)
        return fingerprint

@annotate('bigint,string,string,string,string,string,string,string,string->string')
class f_check_dumplicate(BaseUDAF):
    '''
    去重合并后重新判断，组内个数大于5时，dottitle、tenderee、win_tenderer、bidding_budget组内只能有一个取值
    组内个数小于等于5时，tenderee、win_tenderer、bidding_budget组内只能有一个取值
    '''
    def __init__(self):
        import logging
        import json,re
        global json,logging,re
        logging.basicConfig(level = logging.INFO,format = '%(asctime)s - %(name)s - %(levelname)s - %(message)s')

    def new_buffer(self):
        return [list()]

    def iterate(self, buffer,docid,doctitle,project_code,project_name,tenderee,agency,win_tenderer,bidding_budget,win_bid_price):
        buffer[0].append({"docid":docid,"doctitle":doctitle,"project_code":project_code,"project_name":project_name,
                          "tenderee":tenderee,"agency":agency,"win_tenderer":win_tenderer,"bidding_budget":bidding_budget,"win_bid_price":win_bid_price})

    def merge(self, buffer, pbuffer):
        buffer[0].extend(pbuffer[0])

    def terminate(self, buffer):
        list_group = []
        list_group.append(buffer[0])
        return json.dumps(list_group,ensure_ascii=False)


@annotate('string -> bigint,bigint,string,string,string,string,string,string,string,string')
class f_check_dumplicate_group(BaseUDTF):
    '''
    从最后的结果中获取组
    '''

    def __init__(self):
        import logging
        import json
        global json,logging
        logging.basicConfig(level = logging.INFO,format = '%(asctime)s - %(name)s - %(levelname)s - %(message)s')

    def process(self,list_group):
        if list_group is not None:
            final_group = json.loads(list_group)
            logging.info(list_group)
            for _groups in final_group:
                for _group in _groups:
                    self.forward(_groups[0]["docid"],_group["docid"],_group["doctitle"],_group["project_code"],_group["project_name"],_group["tenderee"],_group["agency"],_group["win_tenderer"],_group["bidding_budget"],_group["win_bid_price"])

@annotate('string->bigint')
class f_is_contain(BaseUDAF):
    '''
    去重合并后重新判断，组内个数大于5时，dottitle、tenderee、win_tenderer、bidding_budget组内只能有一个取值
    组内个数小于等于5时，tenderee、win_tenderer、bidding_budget组内只能有一个取值
    '''
    def __init__(self):
        import logging
        import json,re
        global json,logging,re
        logging.basicConfig(level = logging.INFO,format = '%(asctime)s - %(name)s - %(levelname)s - %(message)s')

    def new_buffer(self):
        return [list()]

    def iterate(self, buffer,doctitle):
        buffer[0].append(doctitle)

    def merge(self, buffer, pbuffer):
        buffer[0].extend(pbuffer[0])

    def terminate(self, buffer):
        is_contain = 1
        list_doctitle = buffer[0]
        main_doctitle = ""
        for _doctitle in list_doctitle:
            if _doctitle in main_doctitle or main_doctitle in _doctitle:
                if len(_doctitle)>len(main_doctitle):
                    main_doctitle = _doctitle
            else:
                is_contain = 0
                break
        return is_contain



def getSet(list_dict,key):
    _set = set()
    for item in list_dict:
        if key in item:
            if item[key]!='' and item[key] is not None:
                if re.search("^[\d\.]+$",item[key]) is not None:
                    _set.add(str(float(item[key])))
                else:
                    _set.add(str(item[key]))
    return _set

def split_with_time(list_dict,sort_key,timedelta=86400*2):
    if len(list_dict)>0:
        if sort_key in list_dict[0]:
            list_dict.sort(key=lambda x:x[sort_key])
            list_group = []
            _begin = 0
            for i in range(len(list_dict)-1):
                if abs(list_dict[i][sort_key]-list_dict[i+1][sort_key])<timedelta:
                    continue
                else:
                    _group = []
                    for j in range(_begin,i+1):
                        _group.append(list_dict[j])
                    if len(_group)>1:
                        list_group.append(_group)
                    _begin = i + 1
            if len(list_dict)>1:
                _group = []
                for j in range(_begin,len(list_dict)):
                    _group.append(list_dict[j])
                if len(_group)>1:
                    list_group.append(_group)
            return list_group
    return [list_dict]

@annotate('bigint,bigint,string,string,string,string,string,string,string,string,string,string,string,string,string->string')
class f_check_dumplicate_1(BaseUDAF):
    '''
    项目编号、中标单位、len(项目编号)>7、中标单位<> ""、合并后非空招标单位数<2、合并后同公告类型非空金额相同
    '''
    def __init__(self):
        import logging
        import json,re
        global json,logging,re
        logging.basicConfig(level = logging.INFO,format = '%(asctime)s - %(name)s - %(levelname)s - %(message)s')

    def new_buffer(self):
        return [list()]

    def iterate(self, buffer,docid,page_time_stamp,set_limit_column1,set_limit_column2,set_limit_column3,set_limit_column4,contain_column,doctitle,project_code,project_name,tenderee,agency,win_tenderer,bidding_budget,win_bid_price):
        buffer[0].append({"docid":docid,"page_time_stamp":page_time_stamp,"set_limit_column1":set_limit_column1,
                          "set_limit_column2":set_limit_column2,"set_limit_column3":set_limit_column3,"set_limit_column4":set_limit_column4,
                          "contain_column":contain_column,"doctitle":doctitle,"project_code":project_code,"project_name":project_name,
                          "tenderee":tenderee,"agency":agency,"win_tenderer":win_tenderer,"bidding_budget":bidding_budget,"win_bid_price":win_bid_price})

    def merge(self, buffer, pbuffer):
        buffer[0].extend(pbuffer[0])

    def terminate(self, buffer):
        list_split = split_with_time(buffer[0],"page_time_stamp")
        list_group = []
        for _split in list_split:
            flag = True
            keys = ["set_limit_column1","set_limit_column2","set_limit_column3","set_limit_column4"]
            for _key in keys:
                logging.info(_key+str(getSet(_split,_key)))
                if len(getSet(_split,_key))>1:
                    flag = False
                    break

            MAX_CONTAIN_COLUMN = None
            #判断组内每条公告是否包含
            if flag:
                for _d in _split:
                    contain_column = _d["contain_column"]
                    if contain_column is not None and contain_column !="":
                        if MAX_CONTAIN_COLUMN is None:
                            MAX_CONTAIN_COLUMN = contain_column
                        else:
                            if len(MAX_CONTAIN_COLUMN)<len(contain_column):
                                if contain_column.find(MAX_CONTAIN_COLUMN)==-1:
                                    flag = False
                                    break
                                MAX_CONTAIN_COLUMN = contain_column
                            else:
                                if MAX_CONTAIN_COLUMN.find(contain_column)==-1:
                                    flag = False
                                    break
            if flag:
                if len(_split)>1:
                    list_group.append(_split)
        return json.dumps(list_group)