BIDI_ML_INFO_EXTRACTION/BiddingKG/maxcompute/evaluates.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



# 配置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"))


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

    def __init__(self):

        # self.out = init_env(["BiddingKG.z01","BiddingKG.z02"],"local_package")
        import uuid
        global uuid

        import logging
        import datetime
        logging.basicConfig(level = logging.INFO,format = '%(asctime)s - %(name)s - %(levelname)s - %(message)s')

        logging.info("time1"+str(datetime.datetime.now().strftime('%y-%m-%d %H:%M:%S')))

        self.out = init_env(["BiddingKG.zip.env"],str(uuid.uuid4()))
        logging.info("time2"+str(datetime.datetime.now().strftime('%y-%m-%d %H:%M:%S')))
        self.out = init_env(["wiki_128_word_embedding_new.vector.env"],".")
        logging.info("time3"+str(datetime.datetime.now().strftime('%y-%m-%d %H:%M:%S')))
        # self.out = init_env(["envs_py37.zip.env"],str(uuid.uuid4()))
        self.out = include_package_path("envs_py37.env.zip")
        logging.info("time4"+str(datetime.datetime.now().strftime('%y-%m-%d %H:%M:%S')))
        self.out = init_env(["so.env"],".")
        logging.info("time5"+str(datetime.datetime.now().strftime('%y-%m-%d %H:%M:%S')))
        import BiddingKG.dl.interface.predictor as predictor
        logging.info("time6"+str(datetime.datetime.now().strftime('%y-%m-%d %H:%M:%S')))
        import BiddingKG.dl.interface.Entitys as Entitys
        logging.info("time6.1"+str(datetime.datetime.now().strftime('%y-%m-%d %H:%M:%S')))
        import BiddingKG.dl.interface.getAttributes as getAttributes
        logging.info("time6.2"+str(datetime.datetime.now().strftime('%y-%m-%d %H:%M:%S')))
        import BiddingKG.dl.entityLink.entityLink as entityLink
        logging.info("time6.2"+str(datetime.datetime.now().strftime('%y-%m-%d %H:%M:%S')))
        import BiddingKG.dl.interface.Preprocessing as Preprocessing
        logging.info("time6.3"+str(datetime.datetime.now().strftime('%y-%m-%d %H:%M:%S')))
        import json
        import time

        from BiddingKG.dl.common.Utils import log
        import numpy as np



        global predictor,Entitys,getAttributes,entityLink,json,MyEncoder,Preprocessing,time,log,MyEncoder,np
        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)

    def process(self,content,_doc_id,_title,page_time):
        if content is not None and _doc_id not in [105677700,126694044,126795572,126951461]:
            k = str(uuid.uuid4())
            cost_time = dict()
            start_time = time.time()
            log("start process doc %s"%(str(_doc_id)))
            try:
                list_articles,list_sentences,list_entitys,_cost_time = Preprocessing.get_preprocessed([[k,content,"",str(_doc_id),str(_title)]],useselffool=True)
                log("get preprocessed done of doc_id%s"%(_doc_id))
                cost_time["preprocess"] = time.time()-start_time
                cost_time.update(_cost_time)
                '''
                for articles in list_articles:
                    print(articles.content)
                    
                '''
                start_time = time.time()
                codeName = predictor.getPredictor("codeName").predict(list_sentences,MAX_AREA=2000,list_entitys=list_entitys)
                log("get codename done of doc_id%s"%(_doc_id))
                cost_time["codename"] = time.time()-start_time

                start_time = time.time()
                predictor.getPredictor("prem").predict(list_sentences,list_entitys)
                log("get prem done of doc_id%s"%(_doc_id))
                cost_time["prem"] = time.time()-start_time
                start_time = time.time()
                predictor.getPredictor("roleRule").predict(list_articles,list_sentences, list_entitys,codeName)
                cost_time["rule"] = time.time()-start_time
                start_time = time.time()
                predictor.getPredictor("epc").predict(list_sentences,list_entitys)
                log("get epc done of doc_id%s"%(_doc_id))
                cost_time["person"] = time.time()-start_time
                start_time = time.time()
                entityLink.link_entitys(list_entitys)
                '''
                for list_entity in list_entitys:
                    for _entity in list_entity:
                        for _ent in _entity.linked_entitys:
                            print(_entity.entity_text,_ent.entity_text)
                '''
                prem = getAttributes.getPREMs(list_sentences,list_entitys,list_articles)
                log("get attributes done of doc_id%s"%(_doc_id))
                cost_time["attrs"] = time.time()-start_time

                #print(prem)
                data_res = Preprocessing.union_result(codeName, prem)[0][1]
                data_res["cost_time"] = cost_time
                data_res["success"] = True

                _article = list_articles[0]
                self.forward(page_time,int(_article.doc_id),json.dumps(data_res,cls=MyEncoder,ensure_ascii=False))
            except Exception as e:
                log("%s===error docid:%s"%(str(e),str(_doc_id)))