BaseDataMaintenance/BaseDataMaintenance/maintenance/dataflow.py

# sys.path.append("/data")

from BaseDataMaintenance.dataSource.source import getConnect_activateMQ_ali
from BaseDataMaintenance.common.multiThread import MultiThreadHandler
from BaseDataMaintenance.common.multiProcess import MultiHandler
from queue import Queue
from multiprocessing import Queue as PQueue

from BaseDataMaintenance.model.ots.document_tmp import *
from BaseDataMaintenance.model.ots.attachment import *
from BaseDataMaintenance.model.ots.document_html import *
from BaseDataMaintenance.model.ots.document_extract2 import *
from BaseDataMaintenance.model.ots.project import *
from BaseDataMaintenance.model.ots.project2_tmp import *
from BaseDataMaintenance.model.ots.document import *
from BaseDataMaintenance.model.ots.project_process import *

import base64
from BaseDataMaintenance.dataSource.interface import getAttachDealInterface,sentMsgToDD
from uuid import uuid4
from BaseDataMaintenance.common.ossUtils import *
from BaseDataMaintenance.dataSource.source import is_internal,getAuth

from apscheduler.schedulers.blocking import BlockingScheduler

from BaseDataMaintenance.maintenance.dataflow_settings import *
from threading import Thread
import oss2
from BaseDataMaintenance.maxcompute.documentDumplicate import *
from BaseDataMaintenance.maxcompute.documentMerge import *

from BaseDataMaintenance.common.otsUtils import *
from BaseDataMaintenance.common.activateMQUtils import *

from BaseDataMaintenance.dataMonitor.data_monitor import BaseDataMonitor

from BaseDataMaintenance.dataSource.pool import ConnectorPool

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[\d\.]*$",item[key]) is not None:
                    _set.add(str(float(item[key])))
                else:
                    _set.add(str(item[key]))
    return _set

def getSimilarityOfString(str1,str2):
    _set1 = set()
    _set2 = set()
    if str1 is not None:
        for i in range(1,len(str1)):
            _set1.add(str1[i-1:i+1])
    if str2 is not None:
        for i in range(1,len(str2)):
            _set2.add(str2[i-1:i+1])
    _len = max(1,min(len(_set1),len(_set2)))
    return len(_set1&_set2)/_len

def getDiffIndex(list_dict,key,confidence=100):
    _set = set()
    for _i in range(len(list_dict)):
        item = list_dict[_i]
        if item["confidence"]>=confidence:
            continue
        if key in item:
            if item[key]!='' and item[key] is not None:
                if re.search("^\d+(\.\d+)?$",item[key]) is not None:
                    _set.add(str(float(item[key])))
                else:
                    _set.add(str(item[key]))
        if len(_set)>1:
            return _i
    return len(list_dict)

def transformSWF(bucket,attachment_hub_url,objectPath,localpath,swf_dir):
    swf_urls = []
    try:
        list_files = os.listdir(swf_dir)
        list_files.sort(key=lambda x:x)
        headers = dict()
        headers["x-oss-object-acl"] = oss2.OBJECT_ACL_PUBLIC_READ
        for _file in list_files:
            swf_localpath = "%s/%s"%(swf_dir,_file)
            swf_objectPath = "%s/%s"%(objectPath.split(".")[0],_file)
            uploadFileByPath(bucket,swf_localpath,swf_objectPath,headers)
            _url = "%s/%s"%(attachment_hub_url,swf_objectPath)
            swf_urls.append(_url)
            os.remove(swf_localpath)
    except Exception as e:
        traceback.print_exc()
    return swf_urls



class Dataflow():

    def __init__(self):
        self.ots_client = getConnect_ots()
        self.queue_init = Queue()
        self.queue_attachment = Queue()
        self.queue_attachment_ocr = Queue()
        self.queue_attachment_not_ocr = Queue()
        self.list_attachment_ocr = []
        self.list_attachment_not_ocr = []
        self.queue_extract = Queue()
        self.list_extract = []
        self.queue_dumplicate = PQueue()
        self.dumplicate_set = set()
        self.queue_merge = Queue()
        self.queue_syncho = Queue()
        self.queue_remove = Queue()
        self.queue_remove_project = Queue()

        self.attachment_rec_interface = ""

        self.ots_client_merge = getConnect_ots()

        if is_internal:
            self.bucket_url = "http://oss-cn-hangzhou-internal.aliyuncs.com"
        else:
            self.bucket_url = "http://oss-cn-hangzhou.aliyuncs.com"

        if is_internal:
            self.extract_url = "http://1255640119316927.vpc.cn-hangzhou.pai-eas.aliyuncs.com/api/predict/content_extract"
            self.industy_url = "http://1255640119316927.vpc.cn-hangzhou.pai-eas.aliyuncs.com/api/predict/industry_extract"
            self.other_url = "http://1255640119316927.vpc.cn-hangzhou.pai-eas.aliyuncs.com/api/predict/other_extract"
        else:
            self.extract_url = "http://1255640119316927.cn-hangzhou.pai-eas.aliyuncs.com/api/predict/content_extract"
            self.industy_url = "http://1255640119316927.cn-hangzhou.pai-eas.aliyuncs.com/api/predict/industry_extract"
            self.other_url = "http://1255640119316927.cn-hangzhou.pai-eas.aliyuncs.com/api/predict/other_extract"
        self.header = {'Content-Type': 'application/json',"Authorization":"NzZmOWZlMmU2MGY3YmQ4MDBjM2E5MDAyZjhjNjQ0MzZlMmE0NTMwZg=="}

        self.attachment_hub_url = "https://attachment-hub.oss-cn-hangzhou.aliyuncs.com/"
        self.auth = getAuth()
        oss2.defaults.connection_pool_size = 100
        oss2.defaults.multiget_num_threads = 20
        log("bucket_url:%s"%(self.bucket_url))
        self.attachment_bucket_name = "attachment-hub"
        self.bucket = oss2.Bucket(self.auth,self.bucket_url,self.attachment_bucket_name)
        self.current_path = os.path.dirname(__file__)



    def flow_init(self):
        def producer():
            bool_query = BoolQuery(must_queries=[RangeQuery("crtime",'2022-04-20')])
            rows,next_token,total_count,is_all_succeed = self.ots_client.search("document","document_index",
                                                                                SearchQuery(bool_query,sort=Sort(sorters=[FieldSort("docid")]),limit=100,get_total_count=True),
                                                                                ColumnsToGet(return_type=ColumnReturnType.ALL))
            log("flow_init producer total_count:%d"%total_count)
            list_dict = getRow_ots(rows)
            for _dict in list_dict:
                self.queue_init.put(_dict)
            _count = len(list_dict)
            while next_token:
                rows,next_token,total_count,is_all_succeed = self.ots_client.search("document","document_index",
                                                                                    SearchQuery(bool_query,next_token=next_token,limit=100,get_total_count=True),
                                                                                    ColumnsToGet(return_type=ColumnReturnType.ALL))
                list_dict = getRow_ots(rows)
                for _dict in list_dict:
                    self.queue_init.put(_dict)
                _count += len(list_dict)
        def comsumer():
            mt = MultiThreadHandler(self.queue_init,comsumer_handle,None,30,1,ots_client=self.ots_client)
            mt.run()


        def comsumer_handle(item,result_queue,ots_client):
            _dochtmlcon = item.get(document_tmp_dochtmlcon,"")
            if document_tmp_dochtmlcon in item:
                item.pop(document_tmp_dochtmlcon)
            if document_tmp_doctextcon in item:
                item.pop(document_tmp_doctextcon)
            if document_tmp_attachmenttextcon in item:
                item.pop(document_tmp_attachmenttextcon)

            _status = item.get(document_tmp_status)
            new_status = None
            if _status>=201 and _status<=300:
                item[document_tmp_save] = 1
                new_status = 81
            elif _status>=401 and _status<=450:
                item[document_tmp_save] = 0
                new_status = 81
            else:
                new_status = 1
            # new_status = 1
            item[document_tmp_status] = new_status
            dtmp = Document_tmp(item)
            dhtml = Document_html({document_tmp_partitionkey:item.get(document_tmp_partitionkey),
                                   document_tmp_docid:item.get(document_tmp_docid),
                                   document_tmp_dochtmlcon:_dochtmlcon})


            dtmp.update_row(ots_client)
            dhtml.update_row(ots_client)

        producer()
        comsumer()



    def getTitleFromHtml(self,filemd5,_html):
        _soup = BeautifulSoup(_html,"lxml")

        _find = _soup.find("a",attrs={"data":filemd5})
        _title = ""
        if _find is not None:
            _title = _find.get_text()
        return _title

    def getSourceLinkFromHtml(self,filemd5,_html):
        _soup = BeautifulSoup(_html,"lxml")

        _find = _soup.find("a",attrs={"filelink":filemd5})
        filelink = ""
        if _find is None:
            _find = _soup.find("img",attrs={"filelink":filemd5})
            if _find is not None:
                filelink = _find.attrs.get("src","")
        else:
            filelink = _find.attrs.get("href","")
        return filelink





    def request_attachment_interface(self,attach,_dochtmlcon):

        filemd5 = attach.getProperties().get(attachment_filemd5)
        _status = attach.getProperties().get(attachment_status)
        _filetype = attach.getProperties().get(attachment_filetype)
        _size = attach.getProperties().get(attachment_size)
        _path = attach.getProperties().get(attachment_path)
        _uuid = uuid4()
        objectPath = attach.getProperties().get(attachment_path)
        localpath = os.path.join(self.current_path,"download",_uuid.hex)
        docids = attach.getProperties().get(attachment_docids)

        try:
            if _size>ATTACHMENT_LARGESIZE:
                attach.setValue(attachment_status, ATTACHMENT_TOOLARGE)
                log("attachment :%s of path:%s to large"%(filemd5,_path))
                attach.update_row(self.ots_client)
                return True
            else:
                d_start_time = time.time()
                if downloadFile(self.bucket,objectPath,localpath):
                    time_download = time.time()-d_start_time
                    _data_base64 = base64.b64encode(open(localpath,"rb").read())
                    #调用接口处理结果
                    start_time = time.time()
                    _success,_html,swf_images = getAttachDealInterface(_data_base64,_filetype)
                    if _success:
                        log("process filemd5:%s of type:%s with size:%.3fM download:%ds recognize takes %ds,ret_size:%d"%(filemd5,_filetype,round(_size/1024/1024,4),time_download,time.time()-start_time,len(_html)))
                    else:
                        log("attach interface failed of docid:%s filemd5:%s of type:%s size:%.3fM with result:%s"%(str(docids),filemd5,_filetype,round(_size/1024/1024,4),str(_html)))
                        sentMsgToDD("attach interface failed of docid:%s of filemd5:%s of type:%s size:%.3fM with result:%s"%(str(docids),filemd5,_filetype,round(_size/1024/1024,4),str(_html)))
                        _html = ""
                        return False

                    swf_images = eval(swf_images)
                    if attach.getProperties().get(attachment_filetype)=="swf" and len(swf_images)>0:

                        swf_urls = json.loads(attach.getProperties().get(attachment_swfUrls,"[]"))
                        if len(swf_urls)==0:
                            objectPath = attach.getProperties().get(attachment_path,"")
                            localpath = os.path.join(self.current_path,"download/%s.swf"%(uuid4().hex))
                            swf_dir = os.path.join(self.current_path,"swf_images",uuid4().hex)
                            if not os.path.exists(swf_dir):
                                os.mkdir(swf_dir)
                            for _i in range(len(swf_images)):
                                _base = swf_images[_i]
                                _base = base64.b64decode(_base)
                                filename = "swf_page_%d.png"%(_i)
                                filepath = os.path.join(swf_dir,filename)
                                with open(filepath,"wb") as f:
                                    f.write(_base)

                            swf_urls = transformSWF(self.bucket,self.attachment_hub_url,objectPath,None,swf_dir)
                            if os.path.exists(swf_dir):
                                os.rmdir(swf_dir)
                            attach.setValue(attachment_swfUrls,json.dumps(swf_urls,ensure_ascii=False),True)

                    if re.search("<td",_html) is not None:
                        attach.setValue(attachment_has_table,1,True)

                    _file_title = self.getTitleFromHtml(filemd5,_dochtmlcon)
                    filelink = self.getSourceLinkFromHtml(filemd5,_dochtmlcon)
                    if _file_title!="":
                        attach.setValue(attachment_file_title,_file_title,True)
                    if filelink!="":
                        attach.setValue(attachment_file_link,filelink,True)


                    attach.setValue(attachment_attachmenthtml,_html,True)
                    attach.setValue(attachment_attachmentcon,BeautifulSoup(_html,"lxml").get_text(),True)
                    attach.setValue(attachment_status,ATTACHMENT_PROCESSED,True)
                    attach.setValue(attachment_recsize,len(_html),True)
                    attach.setValue(attachment_process_time,getCurrent_date(format="%Y-%m-%d %H:%M:%S"),True)
                    attach.update_row(self.ots_client) #线上再开放更新
                    return True
                else:
                    return False

        except oss2.exceptions.NotFound as e:
            return True

        except Exception as e:
            traceback.print_exc()
        finally:
            try:
                os.remove(localpath)
            except:
                pass

    def rec_attachments_by_interface(self,list_attach,_dochtmlcon,save=True):
        list_html = []
        swf_urls = []
        for _attach in list_attach:
            #测试全跑

            if _attach.getProperties().get(attachment_status) in (ATTACHMENT_PROCESSED,ATTACHMENT_TOOLARGE):
                _html = _attach.getProperties().get(attachment_attachmenthtml,"")
                if _html is None:
                    _html = ""
                list_html.append(_html)
            else:
                _succeed = self.request_attachment_interface(_attach,_dochtmlcon)
                if not _succeed:
                    return False,"",[]
                _html = _attach.getProperties().get(attachment_attachmenthtml,"")
                if _html is None:
                    _html = ""
                list_html.append(_html)

            if _attach.getProperties().get(attachment_filetype)=="swf":
                swf_urls.extend(json.loads(_attach.getProperties().get(attachment_swfUrls,"[]")))



        return True,list_html,swf_urls




    def generate_dumplicate_query(self,_dict,_dict_must_not,set_match=set(["project_code","project_codes","product"]),set_nested=set(["win_tenderer","bidding_budget","win_bid_price"]),
                                  set_term=set(["project_name","doctitle_refine","docchannel","tenderee","agency","web_source_no","fingerprint","save","docid"]),
                                  set_range=set(["page_time","status"]),set_phrase=set(["doctitle"])):
        list_must_queries = []
        list_must_no_queries = []
        for k,v in _dict.items():
            if k in set_match:
                if isinstance(v,str):
                    l_s = []
                    for s_v in v.split(","):
                        l_s.append(MatchQuery(k,s_v))
                    list_must_queries.append(BoolQuery(should_queries=l_s))
            elif k in set_nested:
                _v = v
                if k!="" and k=="bidding_budget" or k=="win_bid_price":
                    _v = float(_v)
                    list_must_queries.append(NestedQuery("sub_docs_json",TermQuery("sub_docs_json.%s"%k,_v)))
            elif k in set_term:
                list_must_queries.append(TermQuery(k,v))
            elif k in set_phrase:
                list_must_queries.append(MatchPhraseQuery(k,v))
            elif k in set_range:
                if len(v)==1:
                    list_must_queries.append(RangeQuery(k,v[0]))
                elif len(v)==2:
                    list_must_queries.append(RangeQuery(k,v[0],v[1],True,True))
        for k,v in _dict_must_not.items():
            if k in set_match:
                if isinstance(v,str):
                    l_s = []
                    for s_v in v.split(","):
                        l_s.append(MatchQuery(k,s_v))
                    list_must_no_queries.append(BoolQuery(should_queries=l_s))
            elif k in set_nested:
                _v = v
                if k!="" and k=="bidding_budget" or k=="win_bid_price":
                    _v = float(_v)
                    list_must_no_queries.append(NestedQuery("sub_docs_json",TermQuery("sub_docs_json.%s"%k,_v)))
            elif k in set_term:
                list_must_no_queries.append(TermQuery(k,v))
            elif k in set_range:
                if len(v)==1:
                    list_must_no_queries.append(RangeQuery(k,v[0]))
                elif len(v)==2:
                    list_must_no_queries.append(RangeQuery(k,v[0],v[1],True,True))

        return BoolQuery(must_queries=list_must_queries,must_not_queries=list_must_no_queries)

    def f_decode_sub_docs_json(self, project_code,project_name,tenderee,agency,sub_docs_json):
        columns = {"win_tenderer":"","bidding_budget":"","win_bid_price":""}
        extract_count = 0
        if project_code is not None and project_code!="":
            extract_count += 1
        if project_name is not None and project_name!="":
            extract_count += 1
        if tenderee is not None and tenderee!="":
            extract_count += 1
        if agency is not None and agency!="":
            extract_count += 1
        if sub_docs_json is not None:
            sub_docs = json.loads(sub_docs_json)
            sub_docs.sort(key=lambda x:float(x.get("bidding_budget",0)),reverse=True)
            sub_docs.sort(key=lambda x:float(x.get("win_bid_price",0)),reverse=True)
            # log("==%s"%(str(sub_docs)))
            for sub_docs in sub_docs:
                for _key_sub_docs in sub_docs.keys():
                    extract_count += 1
                    if _key_sub_docs in columns:
                        if columns[_key_sub_docs]=="" and str(sub_docs[_key_sub_docs]) not in ["","0"]:
                            if _key_sub_docs in ["bidding_budget","win_bid_price"]:
                                if float(sub_docs[_key_sub_docs])>0:
                                    columns[_key_sub_docs] = str(float(sub_docs[_key_sub_docs]))
                            else:
                                columns[_key_sub_docs] = str(sub_docs[_key_sub_docs])
        return columns["win_tenderer"],columns["bidding_budget"],columns["win_bid_price"],extract_count

    def post_extract(self,_dict):
        win_tenderer,bidding_budget,win_bid_price,extract_count = self.f_decode_sub_docs_json(_dict.get(document_tmp_project_code),_dict.get(document_tmp_project_name),_dict.get(document_tmp_tenderee),_dict.get(document_tmp_agency),_dict.get(document_tmp_sub_docs_json))
        _dict["win_tenderer"] = win_tenderer
        _dict["bidding_budget"] = bidding_budget
        _dict["win_bid_price"] = win_bid_price
        if "extract_count" not in _dict:
            _dict["extract_count"] = extract_count

    def get_dump_columns(self,_dict):
        docchannel = _dict.get(document_tmp_docchannel,0)
        project_code = _dict.get(document_tmp_project_code,"")
        project_name = _dict.get(document_tmp_project_name,"")
        tenderee = _dict.get(document_tmp_tenderee,"")
        agency = _dict.get(document_tmp_agency,"")
        doctitle_refine = _dict.get(document_tmp_doctitle_refine,"")
        win_tenderer = _dict.get("win_tenderer","")
        bidding_budget = _dict.get("bidding_budget","")
        if bidding_budget==0:
            bidding_budget = ""
        win_bid_price = _dict.get("win_bid_price","")
        if win_bid_price==0:
            win_bid_price = ""
        page_time = _dict.get(document_tmp_page_time,"")
        fingerprint = _dict.get(document_tmp_fingerprint,"")
        product = _dict.get(document_tmp_product,"")
        return docchannel,project_code,project_name,tenderee,agency,doctitle_refine,win_tenderer,bidding_budget,win_bid_price,page_time,fingerprint,product

    def f_set_docid_limitNum_contain(self,item, _split,singleNum_keys=["tenderee","win_tenderer"],contain_keys=[],multiNum_keys=[],notlike_keys=["project_code"]):
        flag = True

        for _key in singleNum_keys:
            if len(getSet(_split,_key))>1:
                flag = False
                break
        for _key in multiNum_keys:
            if len(getSet(_split,_key))<=1:
                flag = False
                break

        project_code = item.get("project_code","")
        for _key in notlike_keys:
            if not flag:
                break
            for _d in _split:
                _key_v = _d.get(_key,"")
                _sim = getSimilarityOfString(project_code,_key_v)
                if _sim>0.7 and _sim<1:
                    flag = False
                    break


        #判断组内每条公告是否包含
        if flag:
            if len(contain_keys)>0:
                for _key in contain_keys:
                    MAX_CONTAIN_COLUMN = None
                    for _d in _split:
                        contain_column = _d.get(_key)
                        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:
            return _split
        return []

    def search_data_by_query(self,item,_query,confidence,table_name="document_tmp",table_index="document_tmp_index",sort_column="docid",singleNum_keys=["tenderee","win_tenderer"],contain_keys=[],multiNum_keys=[],notlike_keys=["project_code"],columns=[document_tmp_docchannel,document_tmp_web_source_no,document_tmp_doctitle_refine,document_tmp_project_code,document_tmp_project_name,document_tmp_tenderee,document_tmp_agency,document_tmp_sub_docs_json,document_tmp_extract_count]):

        list_data = []
        if isinstance(_query,list):
            bool_query = BoolQuery(should_queries=_query)
        else:
            bool_query = _query
        rows,next_token,total_count,is_all_succeed = self.ots_client.search(table_name,table_index,
                                                                            SearchQuery(bool_query,sort=Sort(sorters=[FieldSort(sort_column)]),limit=50,get_total_count=True),
                                                                            ColumnsToGet(columns,return_type=ColumnReturnType.SPECIFIED))
        list_dict = getRow_ots(rows)
        for _dict in list_dict:
            self.post_extract(_dict)
            _dict["confidence"] = confidence
            list_data.append(_dict)


        # _count = len(list_dict)
        # while next_token:
        #     rows,next_token,total_count,is_all_succeed = self.ots_client.search(table_name,table_index,
        #                                                                         SearchQuery(bool_query,next_token=next_token,limit=100,get_total_count=True),
        #                                                                         ColumnsToGet(columns,return_type=ColumnReturnType.SPECIFIED))
        #     list_dict = getRow_ots(rows)
        #     for _dict in list_dict:
        #         self.post_extract(_dict)
        #         _dict["confidence"] = confidence
        #         list_data.append(_dict)
        list_dict = self.f_set_docid_limitNum_contain(item,list_dict,singleNum_keys=singleNum_keys,contain_keys=contain_keys,multiNum_keys=multiNum_keys,notlike_keys=notlike_keys)
        return list_dict

    def add_data_by_query(self,item,base_list,set_docid,_query,confidence,singleNum_keys=["tenderee","win_tenderer"],contain_keys=[],multiNum_keys=[],notlike_keys=["project_code"],columns=[document_tmp_save,document_tmp_status,document_tmp_product,document_tmp_docchannel,document_tmp_web_source_no,document_tmp_doctitle_refine,document_tmp_project_code,document_tmp_project_name,document_tmp_tenderee,document_tmp_agency,document_tmp_sub_docs_json,document_tmp_extract_count]):

        list_dict = self.search_data_by_query(item,_query,confidence,singleNum_keys=singleNum_keys,contain_keys=contain_keys,multiNum_keys=multiNum_keys,notlike_keys=notlike_keys,columns=columns)
        for _dict in list_dict:
            self.post_extract(_dict)
            _docid = _dict.get(document_tmp_docid)
            if _docid not in set_docid:
                base_list.append(_dict)
                set_docid.add(_docid)

    def translate_dumplicate_rules(self,status_from,item):
        docchannel,project_code,project_name,tenderee,agency,doctitle_refine,win_tenderer,bidding_budget,win_bid_price,page_time,fingerprint,product = self.get_dump_columns(item)
        if page_time=='':
            page_time = getCurrent_date("%Y-%m-%d")
        base_dict = {
            "status":[status_from[0]],
            "page_time":[timeAdd(page_time,-2),timeAdd(page_time,2)]
        }

        must_not_dict = {"save":0}

        list_rules = []
        singleNum_keys = ["tenderee","win_tenderer"]
        if fingerprint!="":
            _dict = {}
            confidence = 100
            _dict[document_tmp_fingerprint] = fingerprint
            _dict.update(base_dict)
            _query = self.generate_dumplicate_query(_dict,must_not_dict)
            _rule = {"confidence":confidence,
                     "item":item,
                     "query":_query,
                     "singleNum_keys":[],
                     "contain_keys":[],
                     "multiNum_keys":[]}
            list_rules.append(_rule)

        if docchannel in (52,118):

            if bidding_budget!="" and tenderee!="" and project_code!="":
                confidence = 90
                _dict = {document_tmp_docchannel:docchannel,
                         "bidding_budget":item.get("bidding_budget"),
                         document_tmp_tenderee:item.get(document_tmp_tenderee,""),
                         document_tmp_project_code:item.get(document_tmp_project_code,"")
                         }
                _dict.update(base_dict)
                _query = self.generate_dumplicate_query(_dict,must_not_dict)
                _rule = {"confidence":confidence,
                         "query":_query,
                         "singleNum_keys":singleNum_keys,
                         "contain_keys":[],
                         "multiNum_keys":[document_tmp_web_source_no]}
                list_rules.append(_rule)

            if doctitle_refine!="" and tenderee!="" and bidding_budget!="":
                confidence = 80
                _dict = {document_tmp_docchannel:docchannel,
                         "doctitle_refine":doctitle_refine,
                         "tenderee":tenderee,
                         bidding_budget:"bidding_budget"
                         }
                _dict.update(base_dict)
                _query = self.generate_dumplicate_query(_dict,must_not_dict)
                _rule = {"confidence":confidence,
                         "query":_query,
                         "singleNum_keys":singleNum_keys,
                         "contain_keys":[],
                         "multiNum_keys":[document_tmp_web_source_no]}
                list_rules.append(_rule)

            if project_code!="" and doctitle_refine!="" and agency!="" and bidding_budget!="":
                confidence = 90
                _dict = {document_tmp_docchannel:docchannel,
                         "project_code":project_code,
                         "doctitle_refine":doctitle_refine,
                         "agency":agency,
                         "bidding_budget":bidding_budget
                         }
                _dict.update(base_dict)
                _query = self.generate_dumplicate_query(_dict,must_not_dict)
                _rule = {"confidence":confidence,
                         "query":_query,
                         "singleNum_keys":singleNum_keys,
                         "contain_keys":[],
                         "multiNum_keys":[document_tmp_web_source_no]}
                list_rules.append(_rule)

            if project_code!="" and tenderee!="" and bidding_budget!="":
                confidence = 91
                _dict = {document_tmp_docchannel:docchannel,
                         "project_code":project_code,
                         "tenderee":tenderee,
                         "bidding_budget":bidding_budget
                         }
                _dict.update(base_dict)
                _query = self.generate_dumplicate_query(_dict,must_not_dict)
                _rule = {"confidence":confidence,
                         "query":_query,
                         "singleNum_keys":singleNum_keys,
                         "contain_keys":[],
                         "multiNum_keys":[document_tmp_web_source_no]}
                list_rules.append(_rule)

            if doctitle_refine!="" and agency!="" and bidding_budget!="":
                confidence = 71
                _dict = {document_tmp_docchannel:docchannel,
                         "doctitle_refine":doctitle_refine,
                         "agency":agency,
                         "bidding_budget":bidding_budget
                         }
                _dict.update(base_dict)
                _query = self.generate_dumplicate_query(_dict,must_not_dict)
                _rule = {"confidence":confidence,
                         "query":_query,
                         "singleNum_keys":singleNum_keys,
                         "contain_keys":[],
                         "multiNum_keys":[document_tmp_web_source_no]}
                list_rules.append(_rule)

            if project_code!="" and project_name!="" and agency!="" and bidding_budget!="":
                confidence = 91
                _dict = {document_tmp_docchannel:docchannel,
                         "project_code":project_code,
                         "project_name":project_name,
                         "agency":agency,
                         "bidding_budget":bidding_budget
                         }
                _dict.update(base_dict)
                _query = self.generate_dumplicate_query(_dict,must_not_dict)
                n_singleKeys = [i for i in singleNum_keys]
                n_singleKeys.append(document_tmp_web_source_no)
                _rule = {"confidence":confidence,
                         "query":_query,
                         "singleNum_keys":n_singleKeys,
                         "contain_keys":[],
                         "multiNum_keys":[]}
                list_rules.append(_rule)
            ##-- 5. 招标公告 - 同项目编号- 同[项目名称、标题] - 同[招标人、代理公司] - 同预算(!=0) - 同信息源=1
            if project_code!="" and project_name!="" and tenderee!="" and bidding_budget!="":
                confidence = 91
                _dict = {document_tmp_docchannel:docchannel,
                         "project_code":project_code,
                         "project_name":project_name,
                         "tenderee":tenderee,
                         "bidding_budget":bidding_budget
                         }
                _dict.update(base_dict)
                _query = self.generate_dumplicate_query(_dict,must_not_dict)
                n_singleKeys = [i for i in singleNum_keys]
                n_singleKeys.append(document_tmp_web_source_no)
                _rule = {"confidence":confidence,
                         "query":_query,
                         "singleNum_keys":n_singleKeys,
                         "contain_keys":[],
                         "multiNum_keys":[]}
                list_rules.append(_rule)


            if project_code!="" and doctitle_refine!="" and tenderee!="" and bidding_budget!="":
                confidence = 71
                _dict = {document_tmp_docchannel:docchannel,
                         "project_code":project_code,
                         "doctitle_refine":doctitle_refine,
                         "tenderee":tenderee,
                         "bidding_budget":bidding_budget
                         }
                _dict.update(base_dict)
                _query = self.generate_dumplicate_query(_dict,must_not_dict)
                _rule = {"confidence":confidence,
                         "query":_query,
                         "singleNum_keys":singleNum_keys,
                         "contain_keys":[],
                         "multiNum_keys":[document_tmp_web_source_no]}
                list_rules.append(_rule)

            #-- 4. 招标公告 - 同[标题 、项目编号、项目名称]- 同[招标人、代理公司] - 同预算 - 信息源>1
            if project_name!="" and agency!="":
                tmp_bidding = 0
                if bidding_budget!="":
                    tmp_bidding = bidding_budget
                confidence = 51
                _dict = {document_tmp_docchannel:docchannel,
                         "project_name":project_name,
                         "agency":agency,
                         "bidding_budget":tmp_bidding
                         }
                _dict.update(base_dict)
                _query = self.generate_dumplicate_query(_dict,must_not_dict)
                _rule = {"confidence":confidence,
                         "query":_query,
                         "singleNum_keys":singleNum_keys,
                         "contain_keys":[],
                         "multiNum_keys":[document_tmp_web_source_no]}
                list_rules.append(_rule)

            #-- 4. 招标公告 - 同[标题 、项目编号、项目名称]- 同[招标人、代理公司] - 同预算 - 信息源>1
            if project_code!="" and agency!="":
                tmp_bidding = 0
                if bidding_budget!="":
                    tmp_bidding = bidding_budget
                confidence = 51
                _dict = {document_tmp_docchannel:docchannel,
                         "project_code":project_code,
                         "agency":agency,
                         "bidding_budget":tmp_bidding
                         }
                _dict.update(base_dict)
                _query = self.generate_dumplicate_query(_dict,must_not_dict)
                _rule = {"confidence":confidence,
                         "query":_query,
                         "singleNum_keys":singleNum_keys,
                         "contain_keys":[],
                         "multiNum_keys":[document_tmp_web_source_no]}
                list_rules.append(_rule)

        if docchannel not in (101,119,120):

            #-- 7. 非中标公告 - 同项目名称 - 同发布日期 - 同招标人 - 同预算 -  同类型 - 信息源>1 - 同项目编号
            if project_name!="" and tenderee!="" and project_code!="":
                tmp_bidding = 0
                if bidding_budget!="":
                    tmp_bidding = bidding_budget
                confidence = 51
                _dict = {document_tmp_docchannel:docchannel,
                         "project_name":project_name,
                         "tenderee":tenderee,
                         "project_code":project_code
                         }
                _dict.update(base_dict)
                _query = self.generate_dumplicate_query(_dict,must_not_dict)
                _rule = {"confidence":confidence,
                         "query":_query,
                         "singleNum_keys":singleNum_keys,
                         "contain_keys":[],
                         "multiNum_keys":[document_tmp_web_source_no]}
                list_rules.append(_rule)


        if docchannel in (101,119,120):

            #-- 3. 中标公告 - 同项目编号- 同[项目名称、标题] - 同中标人 - 同中标价(==0)
            if project_code!="" and project_name!="" and win_tenderer!="":
                tmp_win = 0
                if win_bid_price!="":
                    tmp_win = win_bid_price
                confidence = 61
                _dict = {document_tmp_docchannel:docchannel,
                         "project_code":project_code,
                         "project_name":project_name,
                         "win_tenderer":win_tenderer,
                         "win_bid_price":tmp_win
                         }
                _dict.update(base_dict)
                _query = self.generate_dumplicate_query(_dict,must_not_dict)
                _rule = {"confidence":confidence,
                         "query":_query,
                         "singleNum_keys":singleNum_keys,
                         "contain_keys":[],
                         "multiNum_keys":[]}
                list_rules.append(_rule)

            if project_code!="" and project_name!="" and bidding_budget!="" and product!="":
                confidence = 72
                _dict = {document_tmp_docchannel:docchannel,
                         "project_code":project_code,
                         "project_name":project_name,
                         "bidding_budget":bidding_budget,
                         "product":product
                         }
                _dict.update(base_dict)
                _query = self.generate_dumplicate_query(_dict,must_not_dict)
                n_singleKeys = [i for i in singleNum_keys]
                n_singleKeys.append(document_tmp_web_source_no)
                _rule = {"confidence":confidence,
                         "query":_query,
                         "singleNum_keys":n_singleKeys,
                         "contain_keys":[],
                         "multiNum_keys":[]}
                list_rules.append(_rule)

            if project_code!='' and doctitle_refine!="" and win_tenderer!="" and win_bid_price!="":
                confidence = 91
                _dict = {document_tmp_docchannel:docchannel,
                         "project_code":project_code,
                         "doctitle_refine":doctitle_refine,
                         "win_tenderer":win_tenderer,
                         "win_bid_price":win_bid_price
                         }
                _dict.update(base_dict)
                _query = self.generate_dumplicate_query(_dict,must_not_dict)
                n_singleKeys = [i for i in singleNum_keys]
                n_singleKeys.append(document_tmp_web_source_no)
                _rule = {"confidence":confidence,
                         "query":_query,
                         "singleNum_keys":n_singleKeys,
                         "contain_keys":[],
                         "multiNum_keys":[]}
                list_rules.append(_rule)

            ##-- 2. 中标公告 - 同项目编号- 同[项目名称、标题] - 同中标人 - 同中标价(!=0) - 同信息源=1
            if project_code!="" and project_name!="" and win_tenderer!="" and win_bid_price!="":
                confidence = 91
                _dict = {document_tmp_docchannel:docchannel,
                         "project_code":project_code,
                         "project_name":project_name,
                         "win_tenderer":win_tenderer,
                         "win_bid_price":win_bid_price
                         }
                _dict.update(base_dict)
                _query = self.generate_dumplicate_query(_dict,must_not_dict)
                n_singleKeys = [i for i in singleNum_keys]
                n_singleKeys.append(document_tmp_web_source_no)
                _rule = {"confidence":confidence,
                         "query":_query,
                         "singleNum_keys":n_singleKeys,
                         "contain_keys":[],
                         "multiNum_keys":[]}
                list_rules.append(_rule)

            if project_name!="" and win_tenderer!="" and win_bid_price!="":
                confidence = 91
                _dict = {document_tmp_docchannel:docchannel,
                         "project_name":project_name,
                         "win_tenderer":win_tenderer,
                         "win_bid_price":win_bid_price,
                         }
                _dict.update(base_dict)
                _query = self.generate_dumplicate_query(_dict,must_not_dict)
                _rule = {"confidence":confidence,
                         "query":_query,
                         "singleNum_keys":singleNum_keys,
                         "contain_keys":[],
                         "multiNum_keys":[document_tmp_web_source_no]}
                list_rules.append(_rule)
            if project_code!="" and win_tenderer!="" and win_bid_price!="":
                confidence = 91
                _dict = {document_tmp_docchannel:docchannel,
                         "project_code":project_code,
                         "win_tenderer":win_tenderer,
                         "win_bid_price":win_bid_price,
                         }
                _dict.update(base_dict)
                _query = self.generate_dumplicate_query(_dict,must_not_dict)
                _rule = {"confidence":confidence,
                         "query":_query,
                         "singleNum_keys":singleNum_keys,
                         "contain_keys":[],
                         "multiNum_keys":[document_tmp_web_source_no]}
                list_rules.append(_rule)

            if project_code!="" and doctitle_refine!="" and win_tenderer!="" and win_bid_price!="":
                confidence = 91
                _dict = {document_tmp_docchannel:docchannel,
                         "project_code":project_code,
                         "doctitle_refine":doctitle_refine,
                         "win_tenderer":win_tenderer,
                         "win_bid_price":win_bid_price
                         }
                _dict.update(base_dict)
                _query = self.generate_dumplicate_query(_dict,must_not_dict)
                n_singleKeys = [i for i in singleNum_keys]
                n_singleKeys.append(document_tmp_web_source_no)
                _rule = {"confidence":confidence,
                         "query":_query,
                         "singleNum_keys":n_singleKeys,
                         "contain_keys":[],
                         "multiNum_keys":[]}
                list_rules.append(_rule)

            if doctitle_refine!="" and win_tenderer!="" and win_bid_price!="":
                confidence=90
                _dict = {document_tmp_docchannel:docchannel,
                         "doctitle_refine":doctitle_refine,
                         "win_tenderer":win_tenderer,
                         "win_bid_price":win_bid_price
                         }
                _dict.update(base_dict)
                _query = self.generate_dumplicate_query(_dict,must_not_dict)
                _rule = {"confidence":confidence,
                         "query":_query,
                         "singleNum_keys":singleNum_keys,
                         "contain_keys":[],
                         "multiNum_keys":[document_tmp_web_source_no]}
                list_rules.append(_rule)

            if project_name!="" and win_tenderer!="" and win_bid_price!="" and project_code!="":
                confidence=95
                _dict = {document_tmp_docchannel:docchannel,
                         "project_name":project_name,
                         "win_tenderer":win_tenderer,
                         "win_bid_price":win_bid_price,
                         "project_code":project_code
                         }
                _dict.update(base_dict)
                _query = self.generate_dumplicate_query(_dict,must_not_dict)
                _rule = {"confidence":confidence,
                         "query":_query,
                         "singleNum_keys":singleNum_keys,
                         "contain_keys":[],
                         "multiNum_keys":[document_tmp_web_source_no]}
                list_rules.append(_rule)

        if docchannel in (51,103,115,116):
            #9.同['公告变更','拍卖出让','土地矿产','招标答疑']- 同[标题 、项目编号、项目名称]- 同[招标人、代理公司] - 同预算 - 同一天 - 不同数据源
            if doctitle_refine!="" and tenderee!="":
                tmp_budget = 0
                if bidding_budget!="":
                    tmp_budget = bidding_budget
                confidence=81
                _dict = {document_tmp_docchannel:docchannel,
                         "doctitle_refine":doctitle_refine,
                         "tenderee":tenderee,
                         "bidding_budget":tmp_budget,
                         }
                _dict.update(base_dict)
                _dict["page_time"] = [page_time,page_time]
                _query = self.generate_dumplicate_query(_dict,must_not_dict)
                _rule = {"confidence":confidence,
                         "query":_query,
                         "singleNum_keys":singleNum_keys,
                         "contain_keys":[],
                         "multiNum_keys":[document_tmp_web_source_no]}
                list_rules.append(_rule)

            #-- 9.同['公告变更','拍卖出让','土地矿产','招标答疑']- 同[标题 、项目编号、项目名称]- 同[招标人、代理公司] - 同预算 - 同一天 - 不同数据源
            if project_code!="" and tenderee!="":
                confidence=81
                tmp_budget = 0
                if bidding_budget!="":
                    tmp_budget = bidding_budget
                _dict = {document_tmp_docchannel:docchannel,
                         "project_code":project_code,
                         "tenderee":tenderee,
                         "bidding_budget":tmp_budget,
                         }
                _dict.update(base_dict)
                _dict["page_time"] = [page_time,page_time]
                _query = self.generate_dumplicate_query(_dict,must_not_dict)
                _rule = {"confidence":confidence,
                         "query":_query,
                         "singleNum_keys":singleNum_keys,
                         "contain_keys":[],
                         "multiNum_keys":[document_tmp_web_source_no]}
                list_rules.append(_rule)
            if project_name!="" and tenderee!="":
                confidence=81
                tmp_budget = 0
                if bidding_budget!="":
                    tmp_budget = bidding_budget
                _dict = {document_tmp_docchannel:docchannel,
                         "project_name":project_name,
                         "tenderee":tenderee,
                         "bidding_budget":tmp_budget,
                         }
                _dict.update(base_dict)
                _dict["page_time"] = [page_time,page_time]
                _query = self.generate_dumplicate_query(_dict,must_not_dict)
                _rule = {"confidence":confidence,
                         "query":_query,
                         "singleNum_keys":singleNum_keys,
                         "contain_keys":[],
                         "multiNum_keys":[document_tmp_web_source_no]}
                list_rules.append(_rule)
            if agency!="" and tenderee!="":
                confidence=81
                tmp_budget = 0
                if bidding_budget!="":
                    tmp_budget = bidding_budget
                _dict = {document_tmp_docchannel:docchannel,
                         "agency":agency,
                         "tenderee":tenderee,
                         "bidding_budget":tmp_budget,
                         "product":product
                         }
                _dict.update(base_dict)
                _dict["page_time"] = [page_time,page_time]
                _query = self.generate_dumplicate_query(_dict,must_not_dict)
                _rule = {"confidence":confidence,
                         "query":_query,
                         "singleNum_keys":singleNum_keys,
                         "contain_keys":[],
                         "multiNum_keys":[document_tmp_web_source_no]}
                list_rules.append(_rule)

            if agency!="" and project_code!="":
                confidence=81
                tmp_budget = 0
                if bidding_budget!="":
                    tmp_budget = bidding_budget
                _dict = {document_tmp_docchannel:docchannel,
                         "agency":agency,
                         "project_code":project_code,
                         "bidding_budget":tmp_budget,
                         "product":product
                         }
                _dict.update(base_dict)
                _dict["page_time"] = [page_time,page_time]
                _query = self.generate_dumplicate_query(_dict,must_not_dict)
                _rule = {"confidence":confidence,
                         "query":_query,
                         "singleNum_keys":singleNum_keys,
                         "contain_keys":[],
                         "multiNum_keys":[document_tmp_web_source_no]}
                list_rules.append(_rule)
            if agency!="" and project_name!="":
                confidence=81
                tmp_budget = 0
                if bidding_budget!="":
                    tmp_budget = bidding_budget
                _dict = {document_tmp_docchannel:docchannel,
                         "agency":agency,
                         "project_name":project_name,
                         "bidding_budget":tmp_budget,
                         "product":product
                         }
                _dict.update(base_dict)
                _dict["page_time"] = [page_time,page_time]
                _query = self.generate_dumplicate_query(_dict,must_not_dict)
                _rule = {"confidence":confidence,
                         "query":_query,
                         "singleNum_keys":singleNum_keys,
                         "contain_keys":[],
                         "multiNum_keys":[document_tmp_web_source_no]}
                list_rules.append(_rule)


        #五选二
        if tenderee!="" and bidding_budget!="" and product!="":
            confidence=80
            _dict = {document_tmp_docchannel:docchannel,
                     "tenderee":tenderee,
                     "bidding_budget":bidding_budget,
                     "product":product,
                     }
            _dict.update(base_dict)
            _dict["page_time"] = [page_time,page_time]
            _query = self.generate_dumplicate_query(_dict,must_not_dict)
            _rule = {"confidence":confidence,
                     "query":_query,
                     "singleNum_keys":singleNum_keys,
                     "contain_keys":[],
                     "multiNum_keys":[]}
            list_rules.append(_rule)

        if tenderee!="" and win_tenderer!="" and product!="":
            confidence=80
            _dict = {document_tmp_docchannel:docchannel,
                     "tenderee":tenderee,
                     "win_tenderer":win_tenderer,
                     "product":product,
                     }
            _dict.update(base_dict)
            _dict["page_time"] = [page_time,page_time]
            _query = self.generate_dumplicate_query(_dict,must_not_dict)
            _rule = {"confidence":confidence,
                     "query":_query,
                     "singleNum_keys":singleNum_keys,
                     "contain_keys":[],
                     "multiNum_keys":[]}
            list_rules.append(_rule)

        if tenderee!="" and win_bid_price!="":
            confidence=80
            _dict = {document_tmp_docchannel:docchannel,
                     "tenderee":tenderee,
                     "win_bid_price":win_bid_price,
                     "product":product,
                     }
            _dict.update(base_dict)
            _dict["page_time"] = [page_time,page_time]
            _query = self.generate_dumplicate_query(_dict,must_not_dict)
            _rule = {"confidence":confidence,
                     "query":_query,
                     "singleNum_keys":singleNum_keys,
                     "contain_keys":[],
                     "multiNum_keys":[]}
            list_rules.append(_rule)

        if tenderee!="" and agency!="":
            confidence=80
            _dict = {document_tmp_docchannel:docchannel,
                     "tenderee":tenderee,
                     "agency":agency,
                     "product":product,
                     }
            _dict.update(base_dict)
            _dict["page_time"] = [page_time,page_time]
            _query = self.generate_dumplicate_query(_dict,must_not_dict)
            _rule = {"confidence":confidence,
                     "query":_query,
                     "singleNum_keys":singleNum_keys,
                     "contain_keys":[],
                     "multiNum_keys":[]}
            list_rules.append(_rule)


        if win_tenderer!="" and bidding_budget!="":
            confidence=80
            _dict = {document_tmp_docchannel:docchannel,
                     "win_tenderer":win_tenderer,
                     "bidding_budget":bidding_budget,
                     "product":product,
                     }
            _dict.update(base_dict)
            _dict["page_time"] = [page_time,page_time]
            _query = self.generate_dumplicate_query(_dict,must_not_dict)
            _rule = {"confidence":confidence,
                     "query":_query,
                     "singleNum_keys":singleNum_keys,
                     "contain_keys":[],
                     "multiNum_keys":[]}
            list_rules.append(_rule)

        if win_bid_price!="" and bidding_budget!="":
            confidence=80
            _dict = {document_tmp_docchannel:docchannel,
                     "win_bid_price":win_bid_price,
                     "bidding_budget":bidding_budget,
                     "product":product,
                     }
            _dict.update(base_dict)
            _dict["page_time"] = [page_time,page_time]
            _query = self.generate_dumplicate_query(_dict,must_not_dict)
            _rule = {"confidence":confidence,
                     "query":_query,
                     "singleNum_keys":singleNum_keys,
                     "contain_keys":[],
                     "multiNum_keys":[]}
            list_rules.append(_rule)

        if agency!="" and bidding_budget!="":
            confidence=80
            _dict = {document_tmp_docchannel:docchannel,
                     "agency":agency,
                     "bidding_budget":bidding_budget,
                     "product":product,
                     }
            _dict.update(base_dict)
            _dict["page_time"] = [page_time,page_time]
            _query = self.generate_dumplicate_query(_dict,must_not_dict)
            _rule = {"confidence":confidence,
                     "query":_query,
                     "singleNum_keys":singleNum_keys,
                     "contain_keys":[],
                     "multiNum_keys":[]}
            list_rules.append(_rule)


        if win_tenderer!="" and win_bid_price!="":
            confidence=80
            _dict = {document_tmp_docchannel:docchannel,
                     "win_tenderer":win_tenderer,
                     "win_bid_price":win_bid_price,
                     "product":product,
                     }
            _dict.update(base_dict)
            _dict["page_time"] = [page_time,page_time]
            _query = self.generate_dumplicate_query(_dict,must_not_dict)
            _rule = {"confidence":confidence,
                     "query":_query,
                     "singleNum_keys":singleNum_keys,
                     "contain_keys":[],
                     "multiNum_keys":[]}
            list_rules.append(_rule)

        if win_tenderer!="" and agency!="":
            confidence=80
            _dict = {document_tmp_docchannel:docchannel,
                     "win_tenderer":win_tenderer,
                     "agency":agency,
                     "product":product,
                     }
            _dict.update(base_dict)
            _dict["page_time"] = [page_time,page_time]
            _query = self.generate_dumplicate_query(_dict,must_not_dict)
            _rule = {"confidence":confidence,
                     "query":_query,
                     "singleNum_keys":singleNum_keys,
                     "contain_keys":[],
                     "multiNum_keys":[]}
            list_rules.append(_rule)

        if doctitle_refine!="" and product!="" and len(doctitle_refine)>7:
            confidence=80
            _dict = {document_tmp_docchannel:docchannel,
                     "doctitle_refine":doctitle_refine,
                     "product":product,
                     }
            _dict.update(base_dict)
            _query = self.generate_dumplicate_query(_dict,must_not_dict)
            _rule = {"confidence":confidence,
                     "query":_query,
                     "singleNum_keys":singleNum_keys,
                     "contain_keys":[],
                     "multiNum_keys":[]}
            list_rules.append(_rule)

        return list_rules

    def dumplicate_fianl_check(self,base_list):
        the_group = base_list
        the_group.sort(key=lambda x:x["confidence"],reverse=True)
        if len(the_group)>10:
            keys = ["tenderee","win_tenderer","win_bid_price","bidding_budget","doctitle_refine"]
        else:
            keys = ["tenderee","win_tenderer","win_bid_price","bidding_budget"]


        #置信度
        list_key_index = []
        for _k in keys:
            if _k=="doctitle":
                list_key_index.append(getDiffIndex(the_group,_k,confidence=30))
            else:
                list_key_index.append(getDiffIndex(the_group,_k))

        _index = min(list_key_index)
        if _index>1:
            return the_group[:_index]
        return []

    def get_best_docid(self,base_list):
        to_reverse = False
        dict_source_count = {}
        for _item in base_list:
            _web_source = _item.get(document_tmp_web_source_no)
            _fingerprint = _item.get(document_tmp_fingerprint)
            if _web_source is not None:
                if _web_source not in dict_source_count:
                    dict_source_count[_web_source] = set()
                dict_source_count[_web_source].add(_fingerprint)
                if len(dict_source_count[_web_source])>=2:
                    to_reverse=True
        if len(base_list)>0:
            base_list.sort(key=lambda x:x["docid"],reverse=to_reverse)
            base_list.sort(key=lambda x:x.get(document_attachment_extract_status,0),reverse=True)
            base_list.sort(key=lambda x:x["extract_count"],reverse=True)
            return base_list[0]["docid"]

    def save_dumplicate(self,base_list,best_docid,status_from,status_to):
        #best_docid need check while others can save directly
        list_dict = []
        for item in base_list:
            docid = item["docid"]
            _dict = {"partitionkey":item["partitionkey"],
                     "docid":item["docid"]}
            if docid==best_docid:
                if item.get("save",1)!=0:
                    _dict["save"] = 1
            else:
                _dict["save"] = 0
            if item.get("status")>=status_from[0] and item.get("status")<=status_from[1]:
                _dict["status"] = random.randint(status_to[0],status_to[1])
            list_dict.append(_dict)
        for _dict in list_dict:
            dtmp = Document_tmp(_dict)
            dtmp.update_row(self.ots_client)


    def flow_test(self,status_to=[1,10]):
        def producer():
            bool_query = BoolQuery(must_queries=[
                # ExistsQuery("docid"),
                # RangeQuery("crtime",range_to='2022-04-10'),
                #                                 RangeQuery("status",61),
                                                 NestedQuery("page_attachments",WildcardQuery("page_attachments.fileMd5","*")),
                                                 ],
                                   must_not_queries=[
                                       # NestedQuery("page_attachments",WildcardQuery("page_attachments.fileMd5","*")),
                                       TermQuery("attachment_extract_status",1),
                                       RangeQuery("status",1,11)
                                                     ]
                                   )
            rows,next_token,total_count,is_all_succeed = self.ots_client.search("document_tmp","document_tmp_index",
                                                                                SearchQuery(bool_query,sort=Sort(sorters=[FieldSort("docid")]),limit=100,get_total_count=True),
                                                                                ColumnsToGet(["docid"],return_type=ColumnReturnType.SPECIFIED))
            log("flow_init producer total_count:%d"%total_count)
            list_dict = getRow_ots(rows)
            for _dict in list_dict:
                self.queue_init.put(_dict)
            _count = len(list_dict)
            while next_token and _count<1000000:
                rows,next_token,total_count,is_all_succeed = self.ots_client.search("document_tmp","document_tmp_index",
                                                                                    SearchQuery(bool_query,next_token=next_token,limit=100,get_total_count=True),
                                                                                    ColumnsToGet(["docid"],return_type=ColumnReturnType.SPECIFIED))
                list_dict = getRow_ots(rows)
                for _dict in list_dict:
                    self.queue_init.put(_dict)
                _count += len(list_dict)
                print("%d/%d"%(_count,total_count))
        def comsumer():
            mt = MultiThreadHandler(self.queue_init,comsumer_handle,None,30,1,ots_client=self.ots_client)
            mt.run()


        def comsumer_handle(item,result_queue,ots_client):
            # print(item)
            dtmp = Document_tmp(item)

            dtmp.setValue(document_tmp_status,random.randint(*status_to),True)
            dtmp.update_row(ots_client)
            # dhtml = Document_html(item)
            # dhtml.update_row(ots_client)


            # dtmp.delete_row(ots_client)
            # dhtml.delete_row(ots_client)


        producer()
        comsumer()


    def flow_dumplicate(self,process_count=flow_process_count,status_from=flow_dumplicate_status_from):
        def producer(columns=[document_tmp_status,document_tmp_page_time,document_tmp_docchannel,document_tmp_product,document_tmp_fingerprint,document_tmp_tenderee,document_tmp_agency,document_tmp_project_code,document_tmp_project_name,document_tmp_doctitle_refine,document_tmp_doctitle,document_tmp_sub_docs_json]):
            bool_query = BoolQuery(must_queries=[RangeQuery(document_tmp_status,*status_from,True,True)])
            rows,next_token,total_count,is_all_succeed = self.ots_client.search("document_tmp","document_tmp_index",
                                                                                SearchQuery(bool_query,sort=Sort(sorters=[FieldSort("docid")]),limit=100,get_total_count=True),
                                                                                ColumnsToGet(columns,return_type=ColumnReturnType.SPECIFIED))
            log("flow_dumplicate producer total_count:%d"%total_count)
            list_dict = getRow_ots(rows)
            for _dict in list_dict:
                self.queue_dumplicate.put(_dict)
            _count = len(list_dict)
            while next_token and _count<flow_process_count:
                rows,next_token,total_count,is_all_succeed = self.ots_client.search("document_tmp","document_tmp_index",
                                                                                    SearchQuery(bool_query,next_token=next_token,limit=100,get_total_count=True),
                                                                                    ColumnsToGet(columns,return_type=ColumnReturnType.SPECIFIED))
                list_dict = getRow_ots(rows)
                for _dict in list_dict:
                    self.queue_dumplicate.put(_dict)
                _count += len(list_dict)
        def comsumer():
            mt = MultiThreadHandler(self.queue_dumplicate,comsumer_handle,None,10,1,ots_client=self.ots_client)
            mt.run()


        def comsumer_handle(item,result_queue,ots_client):
            self.post_extract(item)

            base_list = []
            set_docid = set()

            list_rules = self.translate_dumplicate_rules(flow_dumplicate_status_from,item)

            list_rules.sort(key=lambda x:x["confidence"],reverse=True)
            # print(item,"len_rules",len(list_rules))
            for _rule in list_rules:
                _query = _rule["query"]
                confidence = _rule["confidence"]
                singleNum_keys = _rule["singleNum_keys"]
                contain_keys = _rule["contain_keys"]
                multiNum_keys = _rule["multiNum_keys"]
                self.add_data_by_query(item,base_list,set_docid,_query,confidence,singleNum_keys=singleNum_keys,contain_keys=contain_keys,multiNum_keys=multiNum_keys)

            item["confidence"] = 999
            if item.get(document_tmp_docid) not in set_docid:
                base_list.append(item)

            final_list = self.dumplicate_fianl_check(base_list)
            best_docid = self.get_best_docid(final_list)
            # log(str(final_list))

            _d = {"partitionkey":item["partitionkey"],
                  "docid":item["docid"],
                  "status":random.randint(*flow_dumplicate_status_to),
                  document_tmp_opertime:getCurrent_date(format="%Y-%m-%d %H:%M:%S")
                  }
            dtmp = Document_tmp(_d)

            dup_docid = set()
            for _dict in final_list:
                dup_docid.add(_dict.get(document_tmp_docid))
            if item.get(document_tmp_docid) in dup_docid:
                dup_docid.remove(item.get(document_tmp_docid))

            if len(final_list)==0 or best_docid==item.get(document_tmp_docid):
                dtmp.setValue(document_tmp_save,1,True)
                dtmp.setValue(document_tmp_merge_uuid,self.merge_document(item,flow_dumplicate_status_to),True)
                dmp_docid = ",".join([str(a) for a in list(dup_docid)])
            else:
                dtmp.setValue(document_tmp_save,0,True)
                if best_docid in dup_docid:
                    dup_docid.remove(best_docid)
                    dmp_docid = ",".join([str(a) for a in list(dup_docid)])
                    dmp_docid = "%d,%s"%(best_docid,dmp_docid)
                else:
                    dmp_docid = ",".join([str(a) for a in list(dup_docid)])


            dtmp.setValue(document_tmp_dup_docid,dmp_docid,True)
            dtmp.update_row(self.ots_client)

            #只保留当前公告
            # self.save_dumplicate(final_list,best_docid,status_from,status_to)
            #
            # print("=base=",item)
            # if len(final_list)>=1:
            #     print("==================")
            #     for _dict in final_list:
            #         print(_dict)
            #     print("========>>>>>>>>>>")

        producer()
        comsumer()


    def merge_document(self,item,status_to=None):
        self.post_extract(item)
        docchannel,project_code,project_name,tenderee,agency,doctitle_refine,win_tenderer,bidding_budget,win_bid_price,page_time,fingerprint,product = self.get_dump_columns(item)


        _d = {"partitionkey":item["partitionkey"],
              "docid":item["docid"],
              }
        dtmp = Document_tmp(_d)
        if item.get(document_tmp_save,1)==1:
            list_should_q = []
            if project_code!="" and tenderee!="":
                _q = BoolQuery(must_queries=[MatchQuery("project_code",project_code),
                                             TermQuery("tenderee",tenderee)])
                list_should_q.append(_q)

            if project_name!="" and project_code!="":
                _q = BoolQuery(must_queries=[MatchQuery("project_code",project_code),
                                             TermQuery("project_name",project_name)])
                list_should_q.append(_q)
            if len(list_should_q)>0:
                list_data = self.search_data_by_query(item,list_should_q,100,merge=True,table_name="project2",table_index="project2_index_formerge",sort_column="tenderee",singleNum_keys=["tenderee","win_tenderer"],contain_keys=[],multiNum_keys=[],notlike_keys=["project_code"],columns=["tenderee","win_tenderer"])

                if len(list_data)==1:
                    dtmp.setValue("merge_uuid",list_data[0]["uuid"],True)
                    print(item["docid"],list_data[0]["uuid"])
            else:
                list_should_q = []
                if bidding_budget!="" and project_code!="":
                    _q = BoolQuery(must_queries=[MatchQuery("project_code",project_code),
                                                 TermQuery("bidding_budget",float(bidding_budget))])
                    list_should_q.append(_q)
                if tenderee!="" and bidding_budget!="" and project_name!="":
                    _q = BoolQuery(must_queries=[MatchQuery("tenderee",tenderee),
                                                 TermQuery("bidding_budget",float(bidding_budget)),
                                                 TermQuery("project_name",project_name)])
                    list_should_q.append(_q)
                if tenderee!="" and win_bid_price!="" and project_name!="":
                    _q = BoolQuery(must_queries=[MatchQuery("tenderee",tenderee),
                                                 TermQuery("win_bid_price",float(win_bid_price)),
                                                 TermQuery("project_name",project_name)])
                    list_should_q.append(_q)
                    if len(list_should_q)>0:
                        list_data = self.search_data_by_query(item,list_should_q,100,table_name="project2",table_index="project2_index_formerge",sort_column="tenderee",singleNum_keys=["tenderee","win_tenderer"],contain_keys=[],multiNum_keys=[],notlike_keys=["project_code"],columns=["tenderee","win_tenderer"])

                        if len(list_data)==1:
                            dtmp.setValue("merge_uuid",list_data[0]["uuid"],True)
                            print(item["docid"],list_data[0]["uuid"])

        return dtmp.getProperties().get("merge_uuid","")
        # dtmp.update_row(self.ots_client)


    def test_merge(self):
        import pandas as pd
        import queue
        def producer(columns=[document_tmp_status,document_tmp_page_time,document_tmp_docchannel,document_tmp_product,document_tmp_fingerprint,document_tmp_tenderee,document_tmp_agency,document_tmp_project_code,document_tmp_project_name,document_tmp_doctitle_refine,document_tmp_doctitle,document_tmp_sub_docs_json]):
            list_test_item = []
            should_q = BoolQuery(should_queries=[
                TermQuery("docchannel",101),
                TermQuery("docchannel",119),
                TermQuery("docchannel",120)
            ])
            bool_query = BoolQuery(must_queries=[
                TermQuery("page_time","2022-04-22"),
                should_q,
                TermQuery("save",1)

            ])
            rows,next_token,total_count,is_all_succeed = self.ots_client.search("document_tmp","document_tmp_index",
                                                                                SearchQuery(bool_query,sort=Sort(sorters=[FieldSort("docid")]),limit=100,get_total_count=True),
                                                                                ColumnsToGet(columns,return_type=ColumnReturnType.SPECIFIED))
            log("flow_dumplicate producer total_count:%d"%total_count)
            list_dict = getRow_ots(rows)
            for _dict in list_dict:
                list_test_item.append(_dict)
            _count = len(list_dict)
            while next_token:
                rows,next_token,total_count,is_all_succeed = self.ots_client.search("document_tmp","document_tmp_index",
                                                                                    SearchQuery(bool_query,next_token=next_token,limit=100,get_total_count=True),
                                                                                    ColumnsToGet(columns,return_type=ColumnReturnType.SPECIFIED))
                list_dict = getRow_ots(rows)
                for _dict in list_dict:
                    list_test_item.append(_dict)
                _count += len(list_dict)
                print("%d/%d"%(_count,total_count))
            return list_test_item

        from BaseDataMaintenance.model.ots.project import Project
        def comsumer_handle(item,result_queue,ots_client):
            item["merge_uuid"] = self.merge_document(item)
            if item["merge_uuid"]!="":
                _dict = {"uuid":item["merge_uuid"]}
                _p = Project(_dict)
                _p.fix_columns(self.ots_client,["zhao_biao_page_time"],True)
                if _p.getProperties().get("zhao_biao_page_time","")!="":
                    item["是否有招标"] = "是"


        list_test_item = producer()
        task_queue = queue.Queue()
        for item in list_test_item:
            task_queue.put(item)
        mt = MultiThreadHandler(task_queue,comsumer_handle,None,30,1,ots_client=self.ots_client)
        mt.run()
        keys = [document_tmp_docid,document_tmp_docchannel,document_tmp_project_code,document_tmp_project_name,document_tmp_tenderee,document_tmp_doctitle_refine,"win_tenderer","bidding_budget","win_bid_price","merge_uuid","是否有招标"]
        df_data = {}
        for k in keys:
            df_data[k] = []
        for item in list_test_item:
            for k in keys:
                df_data[k].append(item.get(k,""))

        df = pd.DataFrame(df_data)
        df.to_excel("test_merge.xlsx",columns=keys)


    def flow_merge(self,process_count=10000,status_from=[71,80],status_to=[81,90]):
        def producer(columns=[document_tmp_save,document_tmp_page_time,document_tmp_docchannel,document_tmp_product,document_tmp_fingerprint,document_tmp_tenderee,document_tmp_agency,document_tmp_project_code,document_tmp_project_name,document_tmp_doctitle_refine,document_tmp_doctitle,document_tmp_sub_docs_json]):
            bool_query = BoolQuery(must_queries=[RangeQuery(document_tmp_status,*status_from,True,True)])
            rows,next_token,total_count,is_all_succeed = self.ots_client.search("document_tmp","document_tmp_index",
                                                                                SearchQuery(bool_query,sort=Sort(sorters=[FieldSort("docid")]),limit=100,get_total_count=True),
                                                                                ColumnsToGet(columns,return_type=ColumnReturnType.SPECIFIED))
            log("flow_merge producer total_count:%d"%total_count)
            list_dict = getRow_ots(rows)
            for _dict in list_dict:
                self.queue_merge.put(_dict)
            _count = len(list_dict)
            while next_token and _count<process_count:
                rows,next_token,total_count,is_all_succeed = self.ots_client.search("document_tmp","document_tmp_index",
                                                                                    SearchQuery(bool_query,next_token=next_token,limit=100,get_total_count=True),
                                                                                    ColumnsToGet(columns,return_type=ColumnReturnType.SPECIFIED))
                list_dict = getRow_ots(rows)
                for _dict in list_dict:
                    self.queue_merge.put(_dict)
                _count += len(list_dict)
        def comsumer():
            mt = MultiThreadHandler(self.queue_merge,comsumer_handle,None,10,1,ots_client=self.ots_client)
            mt.run()


        def comsumer_handle(item,result_queue,ots_client):
            self.merge_document(item,status_to)

        # producer()
        # comsumer()
        pass

    def flow_syncho(self,status_from=[71,80],status_to=[81,90]):
        pass

    def flow_remove(self,process_count=flow_process_count,status_from=flow_remove_status_from):

        def producer():
            current_date = getCurrent_date("%Y-%m-%d")
            tmp_date = timeAdd(current_date,-4)
            bool_query = BoolQuery(must_queries=[RangeQuery(document_tmp_status,*status_from,True,True),
                                                 RangeQuery(document_tmp_crtime,range_to="%s 00:00:00"%(tmp_date))])
            rows,next_token,total_count,is_all_succeed = self.ots_client.search("document_tmp","document_tmp_index",
                                                                                SearchQuery(bool_query,sort=Sort(sorters=[FieldSort("docid")]),limit=100,get_total_count=True),
                                                                                ColumnsToGet(return_type=ColumnReturnType.NONE))
            log("flow_remove producer total_count:%d"%total_count)
            list_dict = getRow_ots(rows)
            for _dict in list_dict:
                self.queue_remove.put(_dict)
            _count = len(list_dict)
            while next_token:
                rows,next_token,total_count,is_all_succeed = self.ots_client.search("document_tmp","document_tmp_index",
                                                                                    SearchQuery(bool_query,next_token=next_token,limit=100,get_total_count=True),
                                                                                    ColumnsToGet(return_type=ColumnReturnType.NONE))
                list_dict = getRow_ots(rows)
                for _dict in list_dict:
                    self.queue_remove.put(_dict)
                _count += len(list_dict)
        def comsumer():
            mt = MultiThreadHandler(self.queue_remove,comsumer_handle,None,10,1,ots_client=self.ots_client)
            mt.run()


        def comsumer_handle(item,result_queue,ots_client):
            dtmp = Document_tmp(item)
            dtmp.delete_row(self.ots_client)
            dhtml = Document_html(item)
            dhtml.delete_row(self.ots_client)
        producer()
        comsumer()

    def flow_remove_project_tmp(self,process_count=flow_process_count):

        def producer():
            current_date = getCurrent_date("%Y-%m-%d")
            tmp_date = timeAdd(current_date,-6*31)
            bool_query = BoolQuery(must_queries=[
                                                 RangeQuery(project_page_time,range_to="%s"%(tmp_date))])
            rows,next_token,total_count,is_all_succeed = self.ots_client.search("project2_tmp","project2_tmp_index",
                                                                                SearchQuery(bool_query,sort=Sort(sorters=[FieldSort("page_time")]),limit=100,get_total_count=True),
                                                                                ColumnsToGet(return_type=ColumnReturnType.NONE))
            log("flow_remove project2_tmp producer total_count:%d"%total_count)
            list_dict = getRow_ots(rows)
            for _dict in list_dict:
                self.queue_remove_project.put(_dict)
            _count = len(list_dict)
            while next_token:
                rows,next_token,total_count,is_all_succeed = self.ots_client.search("document_tmp","document_tmp_index",
                                                                                    SearchQuery(bool_query,next_token=next_token,limit=100,get_total_count=True),
                                                                                    ColumnsToGet(return_type=ColumnReturnType.NONE))
                list_dict = getRow_ots(rows)
                for _dict in list_dict:
                    self.queue_remove_project.put(_dict)
                _count += len(list_dict)
        def comsumer():
            mt = MultiThreadHandler(self.queue_remove_project,comsumer_handle,None,10,1,ots_client=self.ots_client)
            mt.run()


        def comsumer_handle(item,result_queue,ots_client):
            ptmp = Project_tmp(item)
            ptmp.delete_row(self.ots_client)
        producer()
        comsumer()

    def start_flow_dumplicate(self):
        schedule = BlockingScheduler()
        schedule.add_job(self.flow_remove,"cron",hour="20")
        schedule.add_job(self.flow_remove_project_tmp,"cron",hour="20")
        schedule.add_job(self.flow_dumplicate,"cron",second="*/10")
        schedule.start()

    def start_flow_merge(self):
        schedule = BlockingScheduler()
        schedule.add_job(self.flow_merge,"cron",second="*/10")
        schedule.start()


def download_attachment():
    ots_client = getConnect_ots()
    queue_attachment = Queue()

    auth = getAuth()
    oss2.defaults.connection_pool_size = 100
    oss2.defaults.multiget_num_threads = 20
    attachment_bucket_name = "attachment-hub"
    if is_internal:
        bucket_url = "http://oss-cn-hangzhou-internal.aliyuncs.com"
    else:
        bucket_url = "http://oss-cn-hangzhou.aliyuncs.com"
    bucket = oss2.Bucket(auth,bucket_url,attachment_bucket_name)
    current_path = os.path.dirname(__file__)
    def producer():
        columns = [document_tmp_attachment_path]
        bool_query = BoolQuery(must_queries=[RangeQuery(document_tmp_crtime,"2022-03-29 15:00:00","2022-03-29 17:00:00",True,True)])
        rows,next_token,total_count,is_all_succeed = ots_client.search("document_tmp","document_tmp_index",
                                                                            SearchQuery(bool_query,sort=Sort(sorters=[FieldSort("status",SortOrder.DESC)]),limit=100,get_total_count=True),
                                                                            ColumnsToGet(columns,return_type=ColumnReturnType.SPECIFIED))
        log("flow_attachment producer total_count:%d"%total_count)
        list_dict = getRow_ots(rows)
        for _dict in list_dict:
            queue_attachment.put(_dict)
        _count = len(list_dict)
        while next_token:
            rows,next_token,total_count,is_all_succeed = ots_client.search("document_tmp","document_tmp_index",
                                                                                SearchQuery(bool_query,next_token=next_token,limit=100,get_total_count=True),
                                                                                ColumnsToGet(columns,return_type=ColumnReturnType.SPECIFIED))
            list_dict = getRow_ots(rows)
            for _dict in list_dict:
                queue_attachment.put(_dict)
            _count += len(list_dict)
    def comsumer():
        mt = MultiThreadHandler(queue_attachment,comsumer_handle,None,10,1)
        mt.run()


    def getAttachments(list_filemd5,columns_to_get=[attachment_filemd5,attachment_path,attachment_size,attachment_attachmenthtml,attachment_filetype,attachment_docids,attachment_status,attachment_swfUrls]):
        list_attachment = []
        rows_to_get = []
        for _md5 in list_filemd5[:50]:
            if _md5 is None:
                continue
            primary_key = [(attachment_filemd5,_md5)]
            rows_to_get.append(primary_key)
        req = BatchGetRowRequest()
        req.add(TableInBatchGetRowItem(attachment_table_name,rows_to_get,columns_to_get,None,1))
        try:
            result = ots_client.batch_get_row(req)
            attach_result = result.get_result_by_table(attachment_table_name)
            for item in attach_result:
                if item.is_ok:
                    _dict = getRow_ots_primary(item.row)
                    if _dict is not None:
                        list_attachment.append(attachment(_dict))

        except Exception as e:
            log(str(list_filemd5))
            log("attachProcess comsumer error %s"%str(e))
        return list_attachment

    def comsumer_handle(item,result_queue):
        page_attachments = json.loads(item.get(document_tmp_attachment_path,"[]"))
        if len(page_attachments)==0:
            pass
        else:
            list_fileMd5 = []
            for _atta in page_attachments:
                list_fileMd5.append(_atta.get(document_tmp_attachment_path_filemd5))
            list_attach = getAttachments(list_fileMd5)
            for attach in list_attach:
                filemd5 = attach.getProperties().get(attachment_filemd5)
                _status = attach.getProperties().get(attachment_status)
                _filetype = attach.getProperties().get(attachment_filetype)
                _size = attach.getProperties().get(attachment_size)
                _path = attach.getProperties().get(attachment_path)
                _uuid = uuid4()
                objectPath = attach.getProperties().get(attachment_path)
                localpath = os.path.join(current_path,"download","%s.%s"%(filemd5,_filetype))

                try:
                    if _size>ATTACHMENT_LARGESIZE:
                        pass
                    else:
                        downloadFile(bucket,objectPath,localpath)
                except Exception as e:
                    traceback.print_exc()

    producer()
    comsumer()

def test_attachment_interface():
    current_path = os.path.dirname(__file__)
    task_queue = Queue()
    def producer():
        _count = 0
        list_filename = os.listdir(os.path.join(current_path,"download"))
        for _filename in list_filename:
            _count += 1
            _type = _filename.split(".")[1]
            task_queue.put({"path":os.path.join(current_path,"download",_filename),"file_type":_type})
            if _count>=500:
                break

    def comsumer():
        mt = MultiThreadHandler(task_queue,comsumer_handle,None,10)
        mt.run()

    def comsumer_handle(item,result_queue):
        _path = item.get("path")
        _type = item.get("file_type")
        _data_base64 = base64.b64encode(open(_path,"rb").read())
        #调用接口处理结果
        start_time = time.time()
        _success,_html,swf_images = getAttachDealInterface(_data_base64,_type)
        log("%s result:%s takes:%d"%(_path,str(_success),time.time()-start_time))

    producer()
    comsumer()

class Dataflow_attachment(Dataflow):

    def __init__(self):
        Dataflow.__init__(self)

    def flow_attachment_process(self):
        self.process_comsumer()

    def monitor_attachment_process(self):
        alive_count = 0
        for _t in self.process_list_thread:
            if _t.is_alive():
                alive_count += 1
        log("attachment_process alive:%d total:%d"%(alive_count,len(self.process_list_thread)))

    def process_comsumer(self):
        self.process_list_thread = []
        thread_count = 60

        for i in range(thread_count):
            self.process_list_thread.append(Thread(target=self.process_comsumer_handle))

        for t in self.process_list_thread:
            t.start()

        for t in self.process_list_thread:
            t.join()


    def process_comsumer_handle(self):
        while 1:
            _flag = False
            try:
                item = self.queue_attachment_ocr.get(True,timeout=0.2)
                log("attachment get doc:%s"%(str(item.get("item",{}).get("docid"))))

                self.attachment_recognize(item,None)
                log("attachment get doc:%s succeed"%(str(item.get("item",{}).get("docid"))))
            except Exception as e:
                _flag = True
                pass
            try:
                item = self.queue_attachment_not_ocr.get(True,timeout=0.2)
                log("attachment get doc:%s"%(str(item.get("item",{}).get("docid"))))

                self.attachment_recognize(item,None)
                log("attachment get doc:%s succeed"%(str(item.get("item",{}).get("docid"))))
            except Exception as e:
                _flag = True and _flag
                pass
            if _flag:
                time.sleep(2)

    def attachment_recognize(self,_dict,result_queue):

        item = _dict.get("item")
        list_attach = _dict.get("list_attach")

        dhtml = Document_html({"partitionkey":item.get("partitionkey"),
                               "docid":item.get("docid")})
        dhtml.fix_columns(self.ots_client,["dochtmlcon"],True)


        _dochtmlcon = dhtml.getProperties().get("dochtmlcon","")
        _succeed,list_html,swf_urls = self.rec_attachments_by_interface(list_attach,_dochtmlcon,save=True)
        log(str(swf_urls))
        if not _succeed:
            item[document_tmp_status] = random.randint(*flow_attachment_status_failed_to)
        else:
            dhtml.updateSWFImages(swf_urls)
            dhtml.updateAttachment(list_html)
            dhtml.update_row(self.ots_client)
            item[document_tmp_status] = random.randint(*flow_attachment_status_succeed_to)
            item[document_tmp_attachment_extract_status] = 1
        log("document:%d get attachments with result:%s"%(item.get("docid"),str(_succeed)))
        dtmp = Document_tmp(item)
        dtmp.update_row(self.ots_client)



    def flow_attachment(self):
        self.flow_attachment_producer()
        self.flow_attachment_producer_comsumer()

    def getAttachments(self,list_filemd5,columns_to_get=[attachment_filemd5,attachment_path,attachment_size,attachment_attachmenthtml,attachment_filetype,attachment_docids,attachment_status,attachment_swfUrls]):
        list_attachment = []
        rows_to_get = []
        for _md5 in list_filemd5[:50]:
            if _md5 is None:
                continue
            primary_key = [(attachment_filemd5,_md5)]
            rows_to_get.append(primary_key)
        req = BatchGetRowRequest()
        req.add(TableInBatchGetRowItem(attachment_table_name,rows_to_get,columns_to_get,None,1))
        try:
            result = self.ots_client.batch_get_row(req)
            attach_result = result.get_result_by_table(attachment_table_name)
            for item in attach_result:
                if item.is_ok:
                    _dict = getRow_ots_primary(item.row)
                    if _dict is not None:
                        list_attachment.append(attachment(_dict))

        except Exception as e:
            log(str(list_filemd5))
            log("attachProcess comsumer error %s"%str(e))
        return list_attachment

    def flow_attachment_producer(self,columns=[document_tmp_attachment_path,document_tmp_crtime]):

        qsize_ocr = self.queue_attachment_ocr.qsize()
        qsize_not_ocr = self.queue_attachment_not_ocr.qsize()


        log("queue_attachment_ocr:%d,queue_attachment_not_ocr:%d"%(qsize_ocr,qsize_not_ocr))
        #选择加入数据场景
        if min(qsize_ocr,qsize_not_ocr)>200 or max(qsize_ocr,qsize_not_ocr)>1000:
            return

        #去重
        set_docid = set()
        set_docid = set_docid | set(self.list_attachment_ocr) | set(self.list_attachment_not_ocr)

        if qsize_ocr>0:
            self.list_attachment_ocr = self.list_attachment_ocr[-qsize_ocr:]
        else:
            self.list_attachment_ocr = []
        if qsize_not_ocr>0:
            self.list_attachment_not_ocr = self.list_attachment_not_ocr[-qsize_not_ocr:]
        else:
            self.list_attachment_not_ocr = []


        try:
            bool_query = BoolQuery(must_queries=[
                RangeQuery(document_tmp_status,*flow_attachment_status_from,True,True),
                # TermQuery(document_tmp_docid,234925191),
            ])

            rows,next_token,total_count,is_all_succeed = self.ots_client.search("document_tmp","document_tmp_index",
                                                                                SearchQuery(bool_query,sort=Sort(sorters=[FieldSort("status",SortOrder.DESC)]),limit=100,get_total_count=True),
                                                                                ColumnsToGet(columns,return_type=ColumnReturnType.SPECIFIED))
            log("flow_attachment producer total_count:%d"%total_count)
            list_dict = getRow_ots(rows)
            _count = 0
            for _dict in list_dict:
                docid = _dict.get(document_tmp_docid)
                if docid in set_docid:
                    continue
                self.queue_attachment.put(_dict,True)

                _count += 1
            while next_token and _count<flow_process_count:
                rows,next_token,total_count,is_all_succeed = self.ots_client.search("document_tmp","document_tmp_index",
                                                                                    SearchQuery(bool_query,next_token=next_token,limit=100,get_total_count=True),
                                                                                    ColumnsToGet(columns,return_type=ColumnReturnType.SPECIFIED))
                list_dict = getRow_ots(rows)
                for _dict in list_dict:
                    docid = _dict.get(document_tmp_docid)
                    if docid in set_docid:
                        continue
                    self.queue_attachment.put(_dict,True)
                    _count += 1
                log("add attachment count:%d"%(_count))
        except Exception as e:
            log("flow attachment producer error:%s"%(str(e)))
            traceback.print_exc()

    def flow_attachment_producer_comsumer(self):
        log("start flow_attachment comsumer")
        mt = MultiThreadHandler(self.queue_attachment,self.comsumer_handle,None,10,1)
        mt.run()

    def set_queue(self,_dict):
        list_attach = _dict.get("list_attach")
        to_ocr = False
        for attach in list_attach:
            if attach.getProperties().get(attachment_filetype) in ["bmp","jpeg","jpg","png","swf","pdf","tif"]:
                to_ocr = True
                break
        if to_ocr:
            self.queue_attachment_ocr.put(_dict,True)
            # self.list_attachment_ocr.append(_dict.get("item").get(document_tmp_docid))
        else:
            self.queue_attachment_not_ocr.put(_dict,True)
            # self.list_attachment_not_ocr.append(_dict.get("item").get(document_tmp_docid))

    def comsumer_handle(self,item,result_queue):
        try:
            page_attachments = json.loads(item.get(document_tmp_attachment_path,"[]"))
            if len(page_attachments)==0:
                item[document_tmp_status] = random.randint(*flow_attachment_status_succeed_to)
                dtmp = Document_tmp(item)
                dtmp.update_row(self.ots_client)
            else:
                list_fileMd5 = []
                for _atta in page_attachments:
                    list_fileMd5.append(_atta.get(document_tmp_attachment_path_filemd5))

                list_attach = self.getAttachments(list_fileMd5)

                #未上传成功的2小时内不处理
                if len(page_attachments)!=len(list_attach) and time.mktime(time.localtime())-time.mktime(time.strptime(item.get(document_tmp_crtime),"%Y-%m-%d %H:%M:%S"))<7200:
                    item[document_tmp_status] = 1
                    dtmp = Document_tmp(item)
                    dtmp.update_row(self.ots_client)
                    return
                self.set_queue({"item":item,"list_attach":list_attach})
        except Exception as e:
            traceback.print_exc()

    def start_flow_attachment(self):
        schedule = BlockingScheduler()
        schedule.add_job(self.flow_attachment_process,"cron",second="*/20")
        schedule.add_job(self.flow_attachment,"cron",second="*/10")
        schedule.start()

class Dataflow_extract(Dataflow):

    def __init__(self):
        Dataflow.__init__(self)

    def flow_extract_producer(self,columns=[document_tmp_page_time,document_tmp_doctitle,document_tmp_docchannel,document_tmp_status,document_tmp_original_docchannel,document_tmp_web_source_no]):

        q_size = self.queue_extract.qsize()
        if q_size>100:
            return


        set_docid = set(self.list_extract)
        if q_size>0:
            self.list_extract = self.list_extract[-q_size:]
        else:
            self.list_extract = []


        try:
            bool_query = BoolQuery(must_queries=[RangeQuery(document_tmp_status,*flow_extract_status_from,True,True)])
            rows,next_token,total_count,is_all_succeed = self.ots_client.search("document_tmp","document_tmp_index",
                                                                                SearchQuery(bool_query,sort=Sort(sorters=[FieldSort("status",SortOrder.ASC)]),limit=100,get_total_count=True),
                                                                                ColumnsToGet(columns,return_type=ColumnReturnType.SPECIFIED))
            log("flow_extract producer total_count:%d"%total_count)
            list_dict = getRow_ots(rows)
            for _dict in list_dict:
                docid = _dict.get(document_tmp_docid)
                if docid in set_docid:
                    self.list_extract.insert(0,docid)
                    continue
                else:
                    self.queue_extract.put(_dict)
                    self.list_extract.append(docid)
            _count = len(list_dict)
            while next_token and _count<flow_process_count:
                rows,next_token,total_count,is_all_succeed = self.ots_client.search("document_tmp","document_tmp_index",
                                                                                    SearchQuery(bool_query,next_token=next_token,limit=100,get_total_count=True),
                                                                                    ColumnsToGet(columns,return_type=ColumnReturnType.SPECIFIED))
                list_dict = getRow_ots(rows)
                for _dict in list_dict:
                    docid = _dict.get(document_tmp_docid)
                    if docid in set_docid:
                        self.list_extract.insert(0,docid)
                        continue
                    else:
                        self.queue_extract.put(_dict)
                        self.list_extract.append(docid)
                _count += len(list_dict)
        except Exception as e:
            log("flow extract producer error:%s"%(str(e)))
            traceback.print_exc()

    def flow_extract(self,):
        self.comsumer()

    def comsumer(self):
        mt = MultiThreadHandler(self.queue_extract,self.comsumer_handle,None,35,1,True)
        mt.run()

    def comsumer_handle(self,item,result_queue):
        dhtml = Document_html({"partitionkey":item.get("partitionkey"),
                               "docid":item.get("docid")})
        dhtml.fix_columns(self.ots_client,["dochtmlcon"],True)

        item[document_tmp_dochtmlcon] = dhtml.getProperties().get(document_tmp_dochtmlcon,"")
        _extract = Document_extract({})
        _extract.setValue(document_extract2_partitionkey,item.get(document_partitionkey))
        _extract.setValue(document_extract2_docid,item.get(document_docid))
        all_done = 1
        if all_done:
            data = item
            resp = requests.post(self.other_url,json=data,headers=self.header)
            if (resp.status_code >=200 and resp.status_code<=210):
                _extract.setValue(document_extract2_other_json,resp.content.decode("utf8"),True)
            else:
                all_done = -1

        data = {}
        for k,v in item.items():
            data[k] = v
        data["timeout"] = 240
        data["doc_id"] = data.get(document_tmp_docid)
        data["content"] = data.get(document_tmp_dochtmlcon,"")
        if document_tmp_dochtmlcon in data:
            data.pop(document_tmp_dochtmlcon)
        data["title"] = data.get(document_tmp_doctitle,"")
        data["web_source_no"] = item.get(document_tmp_web_source_no,"")
        data["original_docchannel"] = item.get(document_tmp_original_docchannel,"")
        if all_done:
            resp = requests.post(self.extract_url,json=data,headers=self.header)
            if (resp.status_code >=200 and resp.status_code<=210):
                _extract.setValue(document_extract2_extract_json,resp.content.decode("utf8"),True)
            else:
                all_done = -2
        if all_done:
            resp = requests.post(self.industy_url,json=data,headers=self.header)
            if (resp.status_code >=200 and resp.status_code<=210):
                _extract.setValue(document_extract2_industry_json,resp.content.decode("utf8"),True)
            else:
                all_done = -3
        _dict = {document_partitionkey:item.get(document_tmp_partitionkey),
                 document_docid:item.get(document_tmp_docid),
                 }
        dtmp = Document_tmp(_dict)
        if all_done!=1:
            sentMsgToDD("要素提取失败：docid:%d with result:%d"%(item.get(document_tmp_docid),all_done))
            dtmp.setValue(document_tmp_status,random.randint(*flow_extract_status_failed_to),True)
            dtmp.update_row(self.ots_client)

        else:

            dtmp.setValue(document_tmp_status,random.randint(*flow_extract_status_succeed_to),True)
            dtmp.update_row(self.ots_client)

            # 插入接口表,上线放开
            _extract.setValue(document_extract2_status,random.randint(1,50),True)
            _extract.update_row(self.ots_client)

        log("process docid:%d %s"%(data["doc_id"],str(all_done)))


    def start_flow_extract(self):
        schedule = BlockingScheduler()
        schedule.add_job(self.flow_extract_producer,"cron",second="*/10")
        schedule.add_job(self.flow_extract,"cron",second="*/10")
        schedule.start()

class Dataflow_dumplicate(Dataflow):

    class DeleteListener():
        def __init__(self,conn,_func,*args,**kwargs):
            self.conn = conn
            self._func = _func

        def on_error(self, headers,*args,**kwargs):
            log('received an error %s' % str(headers.body))

        def on_message(self, headers,*args,**kwargs):
            try:
                message_id = headers.headers["message-id"]
                body = headers.body
                log("get message %s"%(message_id))
                self._func(_dict={"frame":headers,"conn":self.conn},result_queue=None)
            except Exception as e:
                traceback.print_exc()
                pass

        def __del__(self):
            self.conn.disconnect()

    def __init__(self,start_delete_listener=True):
        Dataflow.__init__(self,)
        self.c_f_get_extractCount = f_get_extractCount()
        self.c_f_get_package = f_get_package()
        logging.basicConfig(level = logging.info,format = '%(asctime)s - %(name)s - %(levelname)s - %(message)s')

        self.fix_doc_docid = None
        self.bdm = BaseDataMonitor()

        if start_delete_listener:
            self.delete_comsumer_counts = 2

            self.doc_delete_queue = "/queue/doc_delete_queue"
            self.doc_delete_result = "/queue/doc_delete_result"

            self.pool_mq_ali = ConnectorPool(1,10,getConnect_activateMQ_ali)

            for _ in range(self.delete_comsumer_counts):
                conn = getConnect_activateMQ_ali()
                listener = self.DeleteListener(conn,self.delete_doc_handle)
                createComsumer(listener,self.doc_delete_queue)


    def get_dict_time(self,_extract,keys=["time_bidclose","time_bidopen","time_bidstart","time_commencement","time_completion","time_earnestMoneyEnd","time_earnestMoneyStart","time_getFileEnd","time_getFileStart","time_publicityEnd","time_publicityStart","time_registrationEnd","time_registrationStart","time_release"]):
        dict_time = {}
        for k in keys:
            dict_time[k] = _extract.get(k)
        return dict_time


    def post_extract(self,_dict):
        win_tenderer,bidding_budget,win_bid_price,_ = self.f_decode_sub_docs_json(_dict.get(document_tmp_project_code),_dict.get(document_tmp_project_name),_dict.get(document_tmp_tenderee),_dict.get(document_tmp_agency),_dict.get(document_tmp_sub_docs_json))
        _dict["win_tenderer"] = win_tenderer
        _dict["bidding_budget"] = bidding_budget
        _dict["win_bid_price"] = win_bid_price
        extract_json = _dict.get(document_tmp_extract_json,"{}")
        _extract = json.loads(extract_json)
        _dict["product"] = ",".join(_extract.get("product",[]))
        _dict["fingerprint"] = _extract.get("fingerprint","")
        _dict["project_codes"] = _extract.get("code",[])
        if len(_dict["project_codes"])>0:
            _dict["project_code"] = _dict["project_codes"][0]
        else:
            _dict["project_code"] = ""
        _dict["doctitle_refine"] = _extract.get("doctitle_refine","")
        _dict["nlp_enterprise"] = str({"indoctextcon":_extract.get("nlp_enterprise",[]),
                                       "notindoctextcon":_extract.get("nlp_enterprise_attachment",[])})
        _dict["extract_count"] = self.c_f_get_extractCount.evaluate(extract_json)
        _dict["package"] = self.c_f_get_package.evaluate(extract_json)
        _dict["project_name"] = _extract.get("name","")
        _dict["dict_time"] = self.get_dict_time(_extract)

    def dumplicate_fianl_check(self,base_list):
        the_group = base_list
        the_group.sort(key=lambda x:x["confidence"],reverse=True)

        _index = 0
        base_fingerprint = "None"

        if len(base_list)>0:
            base_fingerprint = base_list[0]["fingerprint"]
        for _i in range(1,len(base_list)):
            _dict1 = base_list[_i]
            fingerprint_less = _dict1["fingerprint"]
            _pass = True
            if fingerprint_less==base_fingerprint:
                _index = _i
                continue
            for _j in range(min(_i,10)):
                _dict2 = base_list[_j]
                _prob = self.dumplicate_check(_dict1,_dict2,_dict2.get("min_counts",10),b_log=False)
                # print("_prob:",_prob)
                if _prob<=0.1:
                    _pass = False
                    break
            log("checking index:%d"%(_i))
            _index = _i
            if not _pass:
                _index -= 1
                break

        if _index>=1:
            # #对重复入库的进行去重
            # _l = the_group[:_index+1]
            # set_fingerprint = set()
            # final_l = []
            # for _dict in _l:
            #     fingerprint_less = _dict["fingerprint"]
            #     if fingerprint_less in set_fingerprint:
            #         continue
            #     else:
            #         final_l.append(_dict)
            #         set_fingerprint.add(fingerprint_less)
            return the_group[:_index+1]
        return []

    def dumplicate_check(self,_dict1,_dict2,min_counts,b_log=False):
        document_less = _dict1
        docid_less = _dict1["docid"]
        docchannel_less = document_less["docchannel"]
        page_time_less = document_less["page_time"]
        doctitle_refine_less = document_less["doctitle_refine"]
        project_codes_less = document_less["project_codes"]
        nlp_enterprise_less = document_less["nlp_enterprise"]
        tenderee_less = document_less["tenderee"]
        agency_less = document_less["agency"]
        win_tenderer_less = document_less["win_tenderer"]
        bidding_budget_less = document_less["bidding_budget"]
        win_bid_price_less = document_less["win_bid_price"]
        product_less = document_less["product"]
        package_less = document_less["package"]
        json_time_less = document_less["dict_time"]
        project_name_less = document_less["project_name"]
        fingerprint_less = document_less["fingerprint"]
        extract_count_less = document_less["extract_count"]
        web_source_no_less = document_less.get("web_source_no")

        document_greater = _dict2
        docid_greater = _dict2["docid"]
        page_time_greater = document_greater["page_time"]
        docchannel_greater = document_greater["docchannel"]
        doctitle_refine_greater = document_greater["doctitle_refine"]
        project_codes_greater = document_greater["project_codes"]
        nlp_enterprise_greater = document_greater["nlp_enterprise"]
        tenderee_greater = document_greater["tenderee"]
        agency_greater = document_greater["agency"]
        win_tenderer_greater = document_greater["win_tenderer"]
        bidding_budget_greater = document_greater["bidding_budget"]
        win_bid_price_greater = document_greater["win_bid_price"]
        product_greater = document_greater["product"]
        package_greater = document_greater["package"]
        json_time_greater = document_greater["dict_time"]
        project_name_greater = document_greater["project_name"]
        fingerprint_greater = document_greater["fingerprint"]
        extract_count_greater = document_greater["extract_count"]
        web_source_no_greater = document_greater.get("web_source_no")

        hard_level=1
        if web_source_no_less==web_source_no_greater=="17397-3":
            hard_level=2

        return check_dumplicate_rule(docid_less,docid_greater,fingerprint_less,fingerprint_greater,project_codes_less,project_codes_greater,tenderee_less,tenderee_greater,agency_less,agency_greater,win_tenderer_less,win_tenderer_greater,bidding_budget_less,bidding_budget_greater,win_bid_price_less,win_bid_price_greater,project_name_less,project_name_greater,doctitle_refine_less,doctitle_refine_greater,extract_count_less,extract_count_greater,docchannel_less,docchannel_greater,page_time_less,page_time_greater,product_less,product_greater,nlp_enterprise_less,nlp_enterprise_greater,package_less,package_greater,json_time_less,json_time_greater,min_counts,b_log=b_log,hard_level=hard_level)


    def dumplicate_check_bak(self,_dict1,_dict2,min_counts,b_log=False):
        document_less = _dict1
        docid_less = _dict1["docid"]
        docchannel_less = document_less["docchannel"]
        page_time_less = document_less["page_time"]
        doctitle_refine_less = document_less["doctitle_refine"]
        project_codes_less = document_less["project_codes"]
        nlp_enterprise_less = document_less["nlp_enterprise"]
        tenderee_less = document_less["tenderee"]
        agency_less = document_less["agency"]
        win_tenderer_less = document_less["win_tenderer"]
        bidding_budget_less = document_less["bidding_budget"]
        win_bid_price_less = document_less["win_bid_price"]
        product_less = document_less["product"]
        package_less = document_less["package"]
        json_time_less = document_less["dict_time"]
        project_name_less = document_less["project_name"]
        fingerprint_less = document_less["fingerprint"]
        extract_count_less = document_less["extract_count"]

        document_greater = _dict2
        docid_greater = _dict2["docid"]
        page_time_greater = document_greater["page_time"]
        doctitle_refine_greater = document_greater["doctitle_refine"]
        project_codes_greater = document_greater["project_codes"]
        nlp_enterprise_greater = document_greater["nlp_enterprise"]
        tenderee_greater = document_greater["tenderee"]
        agency_greater = document_greater["agency"]
        win_tenderer_greater = document_greater["win_tenderer"]
        bidding_budget_greater = document_greater["bidding_budget"]
        win_bid_price_greater = document_greater["win_bid_price"]
        product_greater = document_greater["product"]
        package_greater = document_greater["package"]
        json_time_greater = document_greater["dict_time"]
        project_name_greater = document_greater["project_name"]
        fingerprint_greater = document_greater["fingerprint"]
        extract_count_greater = document_greater["extract_count"]


        if fingerprint_less==fingerprint_greater:
            return 1

        same_count = 0
        all_count = 8
        if len(set(project_codes_less) & set(project_codes_greater))>0:
            same_count += 1
        if getLength(tenderee_less)>0 and tenderee_less==tenderee_greater:
            same_count += 1
        if getLength(agency_less)>0 and agency_less==agency_greater:
            same_count += 1
        if getLength(win_tenderer_less)>0 and win_tenderer_less==win_tenderer_greater:
            same_count += 1
        if getLength(bidding_budget_less)>0 and bidding_budget_less==bidding_budget_greater:
            same_count += 1
        if getLength(win_bid_price_less)>0 and win_bid_price_less==win_bid_price_greater:
            same_count += 1
        if getLength(project_name_less)>0 and project_name_less==project_name_greater:
            same_count += 1
        if getLength(doctitle_refine_less)>0 and doctitle_refine_less==doctitle_refine_greater:
            same_count += 1
        base_prob = 0
        if min_counts<3:
            base_prob = 0.9
        elif min_counts<5:
            base_prob = 0.8
        elif min_counts<8:
            base_prob = 0.7
        else:
            base_prob = 0.6
        _prob = base_prob*same_count/all_count
        if _prob<0.1 and min(extract_count_less,extract_count_greater)<=3:
            _prob = 0.15
        if _prob<0.1:
            return _prob

        check_result = {"pass":1}
        if docchannel_less in (51,102,103,104,115,116,117):
            if doctitle_refine_less!=doctitle_refine_greater:
                if page_time_less!=page_time_greater:
                    check_result["docchannel"] = 0
                    check_result["pass"] = 0
                else:
                    check_result["docchannel"] = 2
        if not check_doctitle(doctitle_refine_less,doctitle_refine_greater,project_codes_less,project_codes_greater):
            check_result["doctitle"] = 0
            check_result["pass"] = 0
            if b_log:
                logging.info("%d-%d,check_doctitle_failed:%s==%s"%(docid_less,docid_greater,str(doctitle_refine_less),str(doctitle_refine_greater)))
        else:
            check_result["doctitle"] = 2

        #added check
        if not check_codes(project_codes_less,project_codes_greater):
            check_result["code"] = 0
            check_result["pass"] = 0
            if b_log:
                logging.info("%d-%d,check_code_failed:%s==%s"%(docid_less,docid_greater,str(project_codes_less),str(project_codes_greater)))
        else:
            if getLength(project_codes_less)>0 and getLength(project_codes_greater)>0 and len(set(project_codes_less) & set(project_codes_greater))>0:
                check_result["code"] = 2
            else:
                check_result["code"] = 1


        if not check_product(product_less,product_greater):
            check_result["product"] = 0
            check_result["pass"] = 0
            if b_log:
                logging.info("%d-%d,check_product_failed:%s==%s"%(docid_less,docid_greater,str(product_less),str(product_greater)))
        else:
            if getLength(product_less)>0 and getLength(product_greater)>0:
                check_result["product"] = 2
            else:
                check_result["product"] = 1

        if not check_demand():
            check_result["pass"] = 0

        if not check_entity(nlp_enterprise_less,nlp_enterprise_greater,
                            tenderee_less,tenderee_greater,
                            agency_less,agency_greater,
                            win_tenderer_less,win_tenderer_greater):
            check_result["entity"] = 0
            check_result["pass"] = 0
            if b_log:
                logging.info("%d-%d,check_entity_failed:%s==%s==%s==%s==%s==%s==%s==%s"%(docid_less,docid_greater,str(nlp_enterprise_less),str(nlp_enterprise_greater),str(tenderee_less),str(tenderee_greater),str(agency_less),str(agency_greater),str(win_tenderer_less),str(win_tenderer_greater)))
        else:
            if docchannel_less in (51,52,103,105,114,118) and getLength(tenderee_less)>0 and getLength(tenderee_greater)>0:
                check_result["entity"] = 2
            elif docchannel_less in (101,119,120) and getLength(win_tenderer_less)>0 and getLength(win_tenderer_greater)>0:
                check_result["entity"] = 2
            else:
                check_result["entity"] = 1

        if not check_money(bidding_budget_less,bidding_budget_greater,
                           win_bid_price_less,win_bid_price_greater):
            if b_log:
                logging.info("%d-%d,check_money_failed:%s==%s==%s==%s"%(docid_less,docid_greater,str(bidding_budget_less),str(bidding_budget_greater),str(win_bid_price_less),str(win_bid_price_greater)))
            check_result["money"] = 0
            check_result["pass"] = 0
        else:
            if docchannel_less in (51,52,103,105,114,118) and getLength(bidding_budget_less)>0 and getLength(bidding_budget_greater)>0:
                check_result["money"] = 2
            elif docchannel_less in (101,119,120) and getLength(win_bid_price_less)>0 and getLength(win_bid_price_greater)>0:
                check_result["money"] = 2
            else:
                check_result["money"] = 1

        #added check
        if not check_package(package_less,package_greater):
            if b_log:
                logging.info("%d-%d,check_package_failed:%s==%s"%(docid_less,docid_greater,str(package_less),str(package_greater)))
            check_result["package"] = 0
            check_result["pass"] = 0
        else:
            if getLength(package_less)>0 and getLength(package_greater)>0:
                check_result["package"] = 2
            else:
                check_result["package"] = 1

        #added check
        if not check_time(json_time_less,json_time_greater):
            if b_log:
                logging.info("%d-%d,check_time_failed:%s==%s"%(docid_less,docid_greater,str(json_time_less),str(json_time_greater)))
                if isinstance(json_time_less,dict):
                    time_less = json_time_less
                else:
                    time_less = json.loads(json_time_less)
                if isinstance(json_time_greater,dict):
                    time_greater = json_time_greater
                else:
                    time_greater = json.loads(json_time_greater)
                for k,v in time_less.items():
                    if getLength(v)>0:
                        v1 = time_greater.get(k,"")
                        if getLength(v1)>0:
                            if v!=v1:
                                log("%d-%d,key:%s"%(docid_less,docid_greater,str(k)))

            check_result["time"] = 0
            check_result["pass"] = 0
        else:
            if getLength(json_time_less)>10 and getLength(json_time_greater)>10:
                check_result["time"] = 2
            else:
                check_result["time"] = 1

        if check_result.get("pass",0)==0:
            if b_log:
                logging.info(str(check_result))

            if check_result.get("money",1)==0:
                return 0

            if check_result.get("entity",1)==2 and check_result.get("code",1)==2 and check_result.get("doctitle",2)==2 and check_result.get("product",2)==2 and check_result.get("money",0)==2:
                return _prob
            else:
                return 0
            if check_result.get("time",1)==0:
                return 0
        return _prob

    def search_data_by_query(self,item,_query,confidence,retry_times=3,merge=False,table_name="document_tmp",table_index="document_tmp_index",sort_column="docid",singleNum_keys=["tenderee","win_tenderer"],contain_keys=[],multiNum_keys=[],notlike_keys=["project_code"],columns=[document_tmp_docchannel,document_tmp_web_source_no,document_tmp_doctitle_refine,document_tmp_project_code,document_tmp_project_name,document_tmp_tenderee,document_tmp_agency,document_tmp_sub_docs_json,document_tmp_extract_count]):

        for _ in range(retry_times):
            try:
                _time = time.time()
                check_time = 0
                if isinstance(_query,list):
                    bool_query = BoolQuery(should_queries=_query)
                else:
                    bool_query = _query
                rows,next_token,total_count,is_all_succeed = self.ots_client.search(table_name,table_index,
                                                                                    SearchQuery(bool_query,sort=Sort(sorters=[FieldSort(sort_column)]),limit=30,get_total_count=True),
                                                                                    ColumnsToGet(columns,return_type=ColumnReturnType.SPECIFIED))
                list_dict = getRow_ots(rows)
                list_data = []
                for _dict in list_dict:
                    self.post_extract(_dict)
                    _docid = _dict.get(document_tmp_docid)
                    if merge:
                        list_data.append(_dict)
                    else:
                        if _docid!=item.get(document_tmp_docid):
                            _time1 = time.time()
                            confidence = self.dumplicate_check(item,_dict,total_count,b_log=False)
                            check_time+= time.time()-_time1

                            _dict["confidence"] = confidence
                            _dict["min_counts"] = total_count

                            print("check====",item.get("docid"),_dict.get("docid"),confidence)

                            if not confidence<0.1:
                                list_data.append(_dict)
                all_time = time.time()-_time
                # log("check:%d rows takes%.4f,check%.4f"%(len(list_dict),all_time-check_time,check_time))
                return list_data
            except Exception as e:
                traceback.print_exc()
        return []

    def add_data_by_query(self,item,base_list,set_docid,_query,confidence,table_name,table_index,singleNum_keys=["tenderee","win_tenderer"],contain_keys=[],multiNum_keys=[],notlike_keys=["project_code"],columns=[document_tmp_save,document_tmp_status,document_tmp_product,document_tmp_docchannel,document_tmp_web_source_no,document_tmp_doctitle_refine,document_tmp_project_code,document_tmp_project_name,document_tmp_tenderee,document_tmp_agency,document_tmp_sub_docs_json,document_tmp_extract_count]):
        list_dict = self.search_data_by_query(item,_query,confidence,table_name=table_name,table_index=table_index,singleNum_keys=singleNum_keys,contain_keys=contain_keys,multiNum_keys=multiNum_keys,notlike_keys=notlike_keys,columns=columns)
        for _dict in list_dict:
            _docid = _dict.get(document_tmp_docid)
            confidence = _dict["confidence"]
            print("confidence",_docid,confidence)
            if confidence>0.1:
                if _docid not in set_docid:
                    base_list.append(_dict)
                    set_docid.add(_docid)
            set_docid.add(_docid)

    def appendRule(self,list_rules,_dict,base_dict,must_not_dict,confidence,item,b_log=False):
        for k,v in _dict.items():
            if getLength(v)==0:
                return
        _dict.update(base_dict)
        if b_log:
            log(str(_dict))
        _query = self.generate_dumplicate_query(_dict,must_not_dict)
        _rule = {"confidence":confidence,
                 "item":item,
                 "query":_query,
                 "singleNum_keys":[],
                 "contain_keys":[],
                 "multiNum_keys":[]}
        list_rules.append(_rule)

    def translate_dumplicate_rules(self,status_from,item,get_all=False,to_log=False):
        docchannel,project_code,project_name,tenderee,agency,doctitle_refine,win_tenderer,bidding_budget,win_bid_price,page_time,fingerprint,product = self.get_dump_columns(item)
        current_date = getCurrent_date("%Y-%m-%d")
        if page_time=='':
            page_time = current_date

        if page_time>=timeAdd(current_date,-2):
            table_name = "document_tmp"
            table_index = "document_tmp_index"
            base_dict = {
                "docchannel":item.get("docchannel",52),
                "status":[status_from[0]],
                "page_time":[timeAdd(page_time,-2),timeAdd(page_time,2)]
            }
            must_not_dict = {"save":0,"docid":item.get("docid")}
            doctitle_refine_name = "doctitle_refine"
        else:
            table_name = "document"
            table_index = "document_index"
            if get_all:
                _status = [201,450]
            else:
                _status = [201,300]
            base_dict = {
                "docchannel":item["docchannel"],
                "status":_status,
                "page_time":[timeAdd(page_time,-2),timeAdd(page_time,2)]
            }
            must_not_dict = {"docid":item.get("docid")}
            doctitle_refine_name = "doctitle"



        list_rules = []
        singleNum_keys = ["tenderee","win_tenderer"]

        confidence = 100
        self.appendRule(list_rules,{document_tmp_fingerprint:fingerprint},base_dict,must_not_dict,confidence,item,b_log=to_log)
        confidence = 90
        _dict = {document_tmp_agency:agency,
                 "win_tenderer":win_tenderer,
                 "win_bid_price":win_bid_price}
        self.appendRule(list_rules,_dict,base_dict,must_not_dict,confidence,item,b_log=to_log)
        _dict = {document_tmp_agency:agency,
                 "win_tenderer":win_tenderer,
                 "bidding_budget":bidding_budget}
        self.appendRule(list_rules,_dict,base_dict,must_not_dict,confidence,item,b_log=to_log)
        _dict = {document_tmp_agency:agency,
                 "win_bid_price":win_bid_price,
                 "bidding_budget":bidding_budget}
        self.appendRule(list_rules,_dict,base_dict,must_not_dict,confidence,item,b_log=to_log)
        _dict = {win_tenderer:win_tenderer,
                 "win_bid_price":win_bid_price,
                 "bidding_budget":bidding_budget}
        self.appendRule(list_rules,_dict,base_dict,must_not_dict,confidence,item,b_log=to_log)
        _dict = {"tenderee":tenderee,
                 "win_tenderer":win_tenderer,
                 "win_bid_price":win_bid_price}
        self.appendRule(list_rules,_dict,base_dict,must_not_dict,confidence,item,b_log=to_log)
        _dict = {"tenderee":tenderee,
                 "win_tenderer":win_tenderer,
                 "bidding_budget":bidding_budget}
        self.appendRule(list_rules,_dict,base_dict,must_not_dict,confidence,item,b_log=to_log)

        _dict = {"tenderee":tenderee,
                 "win_bid_price":win_bid_price,
                 "bidding_budget":bidding_budget}
        self.appendRule(list_rules,_dict,base_dict,must_not_dict,confidence,item,b_log=to_log)
        _dict = {"tenderee":tenderee,
                 "agency":agency,
                 "win_tenderer":win_tenderer}
        self.appendRule(list_rules,_dict,base_dict,must_not_dict,confidence,item,b_log=to_log)
        _dict = {"tenderee":tenderee,
                 "agency":agency,
                 "win_bid_price":win_bid_price}
        self.appendRule(list_rules,_dict,base_dict,must_not_dict,confidence,item,b_log=to_log)

        _dict = {"tenderee":tenderee,
                 "agency":agency,
                 "bidding_budget":bidding_budget}
        self.appendRule(list_rules,_dict,base_dict,must_not_dict,confidence,item,b_log=to_log)

        confidence=85
        _dict = {"tenderee":tenderee,
                 "agency":agency
                 }
        self.appendRule(list_rules,_dict,base_dict,must_not_dict,confidence,item,b_log=to_log)
        _dict = {"tenderee":tenderee,
                 "project_codes":project_code
                 }
        self.appendRule(list_rules,_dict,base_dict,must_not_dict,confidence,item,b_log=to_log)
        _dict = {"tenderee":tenderee,
                 "project_name":project_name
                 }
        self.appendRule(list_rules,_dict,base_dict,must_not_dict,confidence,item,b_log=to_log)

        if getLength(product)>0:
            l_p = product.split(",")
            _dict = {"tenderee":tenderee,
                     "product":l_p[0]
                     }
            self.appendRule(list_rules,_dict,base_dict,must_not_dict,confidence,item,b_log=to_log)

        _dict = {"tenderee":tenderee,
                 "win_tenderer":win_tenderer
                 }
        self.appendRule(list_rules,_dict,base_dict,must_not_dict,confidence,item,b_log=to_log)

        _dict = {"tenderee":tenderee,
                 "win_bid_price":win_bid_price
                 }
        self.appendRule(list_rules,_dict,base_dict,must_not_dict,confidence,item,b_log=to_log)

        _dict = {"tenderee":tenderee,
                 "bidding_budget":bidding_budget
                 }
        self.appendRule(list_rules,_dict,base_dict,must_not_dict,confidence,item,b_log=to_log)

        _dict = {"tenderee":tenderee,
                 doctitle_refine_name:doctitle_refine
                 }
        self.appendRule(list_rules,_dict,base_dict,must_not_dict,confidence,item,b_log=to_log)

        _dict = {"agency":agency,
                 "project_codes":project_code
                 }
        self.appendRule(list_rules,_dict,base_dict,must_not_dict,confidence,item,b_log=to_log)

        _dict = {"agency":agency,
                 "project_name":project_name
                 }
        self.appendRule(list_rules,_dict,base_dict,must_not_dict,confidence,item,b_log=to_log)

        _dict = {"project_codes":project_code,
                 "project_name":project_name
                 }
        self.appendRule(list_rules,_dict,base_dict,must_not_dict,confidence,item,b_log=to_log)

        _dict = {"project_codes":project_code,
                 "win_tenderer":win_tenderer
                 }
        self.appendRule(list_rules,_dict,base_dict,must_not_dict,confidence,item,b_log=to_log)

        _dict = {"project_codes":project_code,
                 "win_bid_price":win_bid_price
                 }
        self.appendRule(list_rules,_dict,base_dict,must_not_dict,confidence,item,b_log=to_log)

        _dict = {"project_codes":project_code,
                 "bidding_budget":bidding_budget
                 }
        self.appendRule(list_rules,_dict,base_dict,must_not_dict,confidence,item,b_log=to_log)

        _dict = {"project_codes":project_code,
                 doctitle_refine_name:doctitle_refine
                 }
        self.appendRule(list_rules,_dict,base_dict,must_not_dict,confidence,item,b_log=to_log)

        _dict = {"project_name":project_name,
                 "win_tenderer":win_tenderer
                 }
        self.appendRule(list_rules,_dict,base_dict,must_not_dict,confidence,item,b_log=to_log)

        _dict = {"project_name":project_name,
                 "win_bid_price":win_bid_price
                 }
        self.appendRule(list_rules,_dict,base_dict,must_not_dict,confidence,item,b_log=to_log)

        _dict = {"project_name":project_name,
                 "bidding_budget":bidding_budget
                 }
        self.appendRule(list_rules,_dict,base_dict,must_not_dict,confidence,item,b_log=to_log)

        _dict = {"project_name":project_name,
                 doctitle_refine_name:doctitle_refine
                 }
        self.appendRule(list_rules,_dict,base_dict,must_not_dict,confidence,item,b_log=to_log)

        _dict = {"win_tenderer":win_tenderer,
                 "win_bid_price":win_bid_price
                 }
        self.appendRule(list_rules,_dict,base_dict,must_not_dict,confidence,item,b_log=to_log)

        _dict = {"win_tenderer":win_tenderer,
                 "bidding_budget":bidding_budget
                 }
        self.appendRule(list_rules,_dict,base_dict,must_not_dict,confidence,item,b_log=to_log)

        _dict = {"win_tenderer":win_tenderer,
                 doctitle_refine_name:doctitle_refine
                 }
        self.appendRule(list_rules,_dict,base_dict,must_not_dict,confidence,item,b_log=to_log)

        _dict = {"win_bid_price":win_bid_price,
                 "bidding_budget":bidding_budget
                 }
        self.appendRule(list_rules,_dict,base_dict,must_not_dict,confidence,item,b_log=to_log)

        _dict = {"win_bid_price":win_bid_price,
                 doctitle_refine_name:doctitle_refine
                 }
        self.appendRule(list_rules,_dict,base_dict,must_not_dict,confidence,item,b_log=to_log)

        _dict = {"bidding_budget":bidding_budget,
                 doctitle_refine_name:doctitle_refine
                 }
        self.appendRule(list_rules,_dict,base_dict,must_not_dict,confidence,item,b_log=to_log)

        confidence=80
        _dict = {doctitle_refine_name:doctitle_refine}
        self.appendRule(list_rules,_dict,base_dict,must_not_dict,confidence,item,b_log=to_log)
        _dict = {"project_codes":project_code}
        self.appendRule(list_rules,_dict,base_dict,must_not_dict,confidence,item,b_log=to_log)

        confidence=70
        _dict = {"project_name":project_name}
        self.appendRule(list_rules,_dict,base_dict,must_not_dict,confidence,item,b_log=to_log)

        return list_rules,table_name,table_index


    def flow_dumplicate(self,process_count=flow_process_count,status_from=flow_dumplicate_status_from):
        def producer(columns=[document_tmp_status,document_tmp_save,document_tmp_page_time,document_tmp_docchannel,document_tmp_tenderee,document_tmp_agency,document_tmp_doctitle,document_tmp_sub_docs_json,document_tmp_extract_json,document_attachment_extract_status,document_update_document]):
            q_size = self.queue_dumplicate.qsize()
            log("dumplicate queue size %d"%(q_size))
            if q_size>flow_process_count//3:
                return
            bool_query = BoolQuery(must_queries=[
                RangeQuery(document_tmp_status,*status_from,True,True),
                # TermQuery("docid",271983871)
            ])
            rows,next_token,total_count,is_all_succeed = self.ots_client.search("document_tmp","document_tmp_index",
                                                                                SearchQuery(bool_query,sort=Sort(sorters=[FieldSort(document_update_document,SortOrder.DESC),FieldSort("docid",SortOrder.DESC)]),limit=100,get_total_count=True),
                                                                                ColumnsToGet(columns,return_type=ColumnReturnType.SPECIFIED))
            log("flow_dumplicate producer total_count:%d"%total_count)
            list_dict = getRow_ots(rows)
            for _dict in list_dict:
                docid = _dict.get(document_tmp_docid)
                if docid in self.dumplicate_set:
                    continue
                self.dumplicate_set.add(docid)
                self.queue_dumplicate.put(_dict)
            _count = len(list_dict)
            while next_token and _count<flow_process_count:
                rows,next_token,total_count,is_all_succeed = self.ots_client.search("document_tmp","document_tmp_index",
                                                                                    SearchQuery(bool_query,next_token=next_token,limit=100,get_total_count=True),
                                                                                    ColumnsToGet(columns,return_type=ColumnReturnType.SPECIFIED))
                list_dict = getRow_ots(rows)
                for _dict in list_dict:
                    docid = _dict.get(document_tmp_docid)
                    if docid in self.dumplicate_set:
                        continue
                    self.dumplicate_set.add(docid)
                    self.queue_dumplicate.put(_dict)
                _count += len(list_dict)

            _l = list(self.dumplicate_set)
            _l.sort(key=lambda x:x,reverse=True)
            self.dumplicate_set = set(_l[:flow_process_count]) | set(_l[-flow_process_count:])
        def comsumer():
            mt = MultiThreadHandler(self.queue_dumplicate,self.dumplicate_comsumer_handle,None,60,1,ots_client=self.ots_client)
            mt.run()

        producer()
        # comsumer()

    def flow_dumpcate_comsumer(self):
        from multiprocessing import Process
        process_count = 2
        thread_count = 30
        list_process = []
        def start_thread():
            mt = MultiThreadHandler(self.queue_dumplicate,self.dumplicate_comsumer_handle,None,thread_count,1,need_stop=False,ots_client=self.ots_client)
            mt.run()

        for _ in range(process_count):
            p = Process(target=start_thread)
            list_process.append(p)
        for p in list_process:
            p.start()
        for p in list_process:
            p.join()

        # mt = MultiThreadHandler(self.queue_dumplicate,self.dumplicate_comsumer_handle,None,40,1,ots_client=self.ots_client)
        # mt.run()


    def search_docs(self,list_docids,columns_to_get = [document_doctitle,document_tmp_save,document_bidway,document_status,document_page_time,document_info_source,document_fingerprint,document_docchannel,document_life_docchannel,document_area,document_province,document_city,document_district,document_tmp_sub_docs_json,document_industry,document_info_type,document_qcodes,document_project_name,document_project_code,document_tenderee,document_tenderee_addr,document_tenderee_phone,document_tenderee_contact,document_agency,document_agency_phone,document_agency_contact,project_procurement_system,document_project_codes,document_product,document_moneysource,document_time_bidclose,document_time_bidopen,document_time_bidstart,document_time_commencement,document_time_completion,document_time_earnest_money_start,document_time_earnest_money_end,document_time_get_file_end,document_time_get_file_start,document_time_publicity_end,document_time_publicity_start,document_time_registration_end,document_time_registration_start,document_time_release,document_tmp_extract_count,document_nlp_enterprise,document_nlp_enterprise_attachment]):
        '''
        根据docid查询公告内容，先查询document_tmp，再查询document
        :param list_docids:
        :return:
        '''
        list_docs = []
        set_fingerprint = set()
        for _docid in list_docids:
            docid = int(_docid)
            _dict = {document_partitionkey:getPartitionKey(docid),
                     document_docid:docid}
            _doc = Document_tmp(_dict)
            _exists = _doc.fix_columns(self.ots_client,columns_to_get,True)
            if not _exists:
                _doc = Document(_dict)
                _exists = _doc.fix_columns(self.ots_client,columns_to_get,True)
            if _exists:
                _fingerprint = _doc.getProperties().get(document_fingerprint)
                if _fingerprint in set_fingerprint:
                    continue
                set_fingerprint.add(_fingerprint)
                list_docs.append(_doc)
        for _doc in list_docs:
            try:
                _sub_docs_json = _doc.getProperties().get(document_tmp_sub_docs_json)
                if _sub_docs_json is not None:
                    _doc.setValue("sub_docs",json.loads(_sub_docs_json),False)

            except Exception as e:
                traceback.print_exc()
        list_docs.sort(key=lambda x:x.getProperties().get(document_page_time,""))
        return list_docs

    def is_same_package(self,_dict1,_dict2):
        sub_project_name1 = _dict1.get(project_sub_project_name,"")
        if sub_project_name1=="Project":
            sub_project_name1 = ""
        win_tenderer1 = _dict1.get(project_win_tenderer,"")
        win_bid_price1 = _dict1.get(project_win_bid_price,0)
        bidding_budget1 = _dict1.get(project_bidding_budget,0)

        sub_project_name2 = _dict2.get(project_sub_project_name,"")
        if sub_project_name2=="Project":
            sub_project_name2 = ""
        win_tenderer2 = _dict2.get(project_win_tenderer,"")
        win_bid_price2 = _dict2.get(project_win_bid_price,0)
        bidding_budget2 = _dict2.get(project_bidding_budget,0)

        _set = set([a for a in [sub_project_name1,sub_project_name2] if a!=""])
        if len(_set)>1:
            return False
        _set = set([a for a in [win_tenderer1,win_tenderer2] if a!=""])
        if len(_set)>1:
            return False
        _set = set([a for a in [win_bid_price1,win_bid_price2] if a!=0])
        if len(_set)>1:
            return False
        _set = set([a for a in [bidding_budget1,bidding_budget2] if a!=0])
        if len(_set)>1:
            return False
        return True



    def getUpdate_dict(self,_dict):
        update_dict = {}
        for k,v in _dict.items():
            if v is None:
                continue
            if isinstance(v,str):
                if v=="":
                    continue
            if isinstance(v,(float,int)):
                if v==0:
                    continue
            update_dict[k] = v
        return update_dict




    def update_projects_by_document(self,docid,save,projects):
        '''
        更新projects中对应的document的属性
        :param docid:
        :param projects: 项目集合
        :param action:add/delete add时附加唯一属性,delete时删除唯一属性
        :return:
        '''
        list_docs = self.search_docs([docid])
        docs = [_doc.getProperties() for _doc in list_docs]

        project_dict = generate_common_properties(docs)

        list_package_properties = generate_packages_properties(docs)

        _dict = {}
        #更新公共属性

        _replace_replace = False
        v = project_dict.get(document_district,"")
        if not (v is None or v=="" or v=="[]" or v=="未知"):
            _replace_replace = True
        for k,v in project_dict.items():
            if not _replace_replace:
                if k in [document_district,document_city,document_province,document_area]:
                    continue
            if v is None or v=="" or v=="[]" or v=="未知":
                continue
            if k in (project_project_dynamics,project_product,project_project_codes,project_docids):
                continue
            _dict[k] = v
        for _proj in projects:
            _proj.update(_dict)
        for _proj in projects:
            if _proj.get(project_page_time,"")<project_dict.get(project_page_time,""):
                _proj[project_page_time] = project_dict.get(project_page_time,"")

        #拼接属性
        append_dict = {}
        set_docid = set()
        set_product = set()
        set_code = set()
        set_nlp_enterprise = set()
        set_nlp_enterprise_attachment = set()
        for _proj in projects:
            _docids = _proj.get(project_docids,"")
            _codes = _proj.get(project_project_codes,"")
            _product = _proj.get(project_product,"")

            set_docid = set(_docids.split(","))
            if save==1:
                set_docid.add(str(docid))
            else:
                if str(docid) in set_docid:
                    set_docid.remove(str(docid))
            set_code = set_code | set(_codes.split(","))
            set_product = set_product | set(_product.split(","))
            try:
                set_nlp_enterprise |= set(json.loads(_proj.get(project_nlp_enterprise,"[]")))
                set_nlp_enterprise_attachment |= set(json.loads(_proj.get(project_nlp_enterprise_attachment,"[]")))
            except Exception as e:
                pass

            set_code = set_code | set(project_dict.get(project_project_codes,"").split(","))
            set_product = set_product | set(project_dict.get(project_product,"").split(","))

            try:
                set_nlp_enterprise |= set(json.loads(project_dict.get(project_nlp_enterprise,"[]")))
                set_nlp_enterprise_attachment |= set(json.loads(project_dict.get(project_nlp_enterprise_attachment,"[]")))
            except Exception as e:
                pass


            append_dict[project_docids] = ",".join([a for a in list(set_docid) if a!=""])
            append_dict[project_docid_number] = len(set_docid)
            append_dict[project_project_codes] = ",".join([a for a in list(set_code) if a!=""])
            append_dict[project_product] = ",".join([a for a in list(set_product) if a!=""])

            append_dict[project_nlp_enterprise] = json.dumps(list(set_nlp_enterprise)[:100],ensure_ascii=False)
            append_dict[project_nlp_enterprise_attachment] = json.dumps(list(set_nlp_enterprise_attachment)[:100],ensure_ascii=False)


            dict_dynamic = {}
            set_docid = set()
            _dynamic = json.loads(_proj.get(project_project_dynamics,"[]"))
            for _dy in _dynamic:
                _docid = _dy.get("docid")
                dict_dynamic[_docid] = _dy
            _dynamic = json.loads(project_dict.get(project_project_dynamics,"[]"))
            for _dy in _dynamic:
                _docid = _dy.get("docid")
                dict_dynamic[_docid] = _dy
            list_dynamics = []
            for k,v in dict_dynamic.items():
                list_dynamics.append(v)
            list_dynamics.sort(key=lambda x:x.get(document_page_time,""))

            append_dict[project_project_dynamics] = json.dumps(list_dynamics[:100],ensure_ascii=False)
            _proj.update(append_dict)


        dict_package = {}
        for _pp in projects:
            _counts = 0
            sub_project_name = _pp.get(project_sub_project_name,"")
            if sub_project_name=="Project":
                sub_project_name = ""
            win_tenderer = _pp.get(project_win_tenderer,"")
            win_bid_price = _pp.get(project_win_bid_price,0)
            bidding_budget = _pp.get(project_bidding_budget,0)
            if win_tenderer!="" and bidding_budget!=0:
                _key = "%s-%s-%s"%(sub_project_name,str(win_tenderer),str(bidding_budget))
                dict_package[_key] = _pp
                _counts += 1
            if win_tenderer!="" and  win_bid_price!=0:
                _key = "%s-%s-%s"%(sub_project_name,win_tenderer,str(win_bid_price))
                dict_package[_key] = _pp
                _counts +=1
            if _counts==0:
                if win_tenderer!="":
                    _key = "%s-%s"%(sub_project_name,win_tenderer)
                    dict_package[_key] = _pp
                    _counts += 1
                if bidding_budget!=0:
                    _key = "%s-%s"%(sub_project_name,str(bidding_budget))
                    dict_package[_key] = _pp
                    _counts += 1


        #更新私有属性
        for _pp in list_package_properties:

            flag_update = False
            sub_project_name = _pp.get(project_sub_project_name,"")
            if sub_project_name=="Project":
                sub_project_name = ""
            win_tenderer = _pp.get(project_win_tenderer,"")
            win_bid_price = _pp.get(project_win_bid_price,0)
            bidding_budget = _pp.get(project_bidding_budget,0)
            if win_tenderer!="" and bidding_budget!=0:
                _key = "%s-%s-%s"%(sub_project_name,str(win_tenderer),str(bidding_budget))
                if _key in dict_package:
                    if self.is_same_package(_pp,dict_package[_key]):
                        ud = self.getUpdate_dict(_pp)
                        self.set_project_uuid(ud,dict_package[_key].get("uuid"))
                        dict_package[_key].update(ud)
                        flag_update = True
                        continue
            if win_tenderer!="" and  win_bid_price!=0:
                _key = "%s-%s-%s"%(sub_project_name,win_tenderer,str(win_bid_price))
                if _key in dict_package:
                    if self.is_same_package(_pp,dict_package[_key]):
                        ud = self.getUpdate_dict(_pp)
                        self.set_project_uuid(ud,dict_package[_key].get("uuid"))
                        dict_package[_key].update(ud)
                        flag_update = True
                        continue
            if win_tenderer!="":
                _key = "%s-%s"%(sub_project_name,win_tenderer)
                if _key in dict_package:
                    if self.is_same_package(_pp,dict_package[_key]):
                        ud = self.getUpdate_dict(_pp)
                        self.set_project_uuid(ud,dict_package[_key].get("uuid"))
                        dict_package[_key].update(ud)
                        flag_update = True
                        continue
            if bidding_budget!=0:
                _key = "%s-%s"%(sub_project_name,str(bidding_budget))
                if _key in dict_package:
                    if self.is_same_package(_pp,dict_package[_key]):
                        ud = self.getUpdate_dict(_pp)
                        self.set_project_uuid(ud,dict_package[_key].get("uuid"))
                        dict_package[_key].update(ud)
                        flag_update = True
                        continue
            if not flag_update:
                _pp.update(project_dict)
                projects.append(_pp)


                _counts = 0
                if win_tenderer!="" and bidding_budget!=0:
                    _key = "%s-%s-%s"%(sub_project_name,str(win_tenderer),str(bidding_budget))
                    dict_package[_key] = _pp
                _counts += 1
                if win_tenderer!="" and  win_bid_price!=0:
                    _key = "%s-%s-%s"%(sub_project_name,win_tenderer,str(win_bid_price))
                    dict_package[_key] = _pp
                    _counts +=1
                if _counts==0:
                    if win_tenderer!="":
                        _key = "%s-%s"%(sub_project_name,win_tenderer)
                        dict_package[_key] = _pp
                        _counts += 1
                    if bidding_budget!=0:
                        _key = "%s-%s"%(sub_project_name,str(bidding_budget))
                        dict_package[_key] = _pp
                        _counts += 1




    def delete_projects_by_document(self,docid):
        '''
        更新projects中对应的document的属性
        :param docid:
        :param projects: 项目集合
        :param action:add/delete add时附加唯一属性,delete时删除唯一属性
        :return:
        '''

        set_docid = set()
        list_delete_projects = []
        list_projects = self.search_projects_with_document([docid])
        for _proj in list_projects:
            _p = {}
            _docids = _proj.get(project_docids,"")
            print(_proj.get(project_uuid))
            _p["delete_uuid"] = _proj.get(project_uuid)
            _p["to_delete"] = True
            list_delete_projects.append(_p)
            if _docids!="":
                set_docid = set_docid | set(_docids.split(","))
        if str(docid) in set_docid:
            set_docid.remove(str(docid))
        list_docid = list(set_docid)

        list_projects = []
        if len(list_docid)>0:
            list_docs = self.search_docs(list_docid)
            list_projects = self.generate_projects_from_document(list_docs)
            list_projects = dumplicate_projects(list_projects)
        list_projects.extend(list_delete_projects)
        project_json = to_project_json(list_projects)
        print("delete_json",project_json)
        return project_json


    def delete_doc_handle(self,_dict,result_queue):
        headers = _dict.get("frame")
        conn = _dict.get("conn")
        log("==========delete")
        if headers is not None:
            message_id = headers.headers["message-id"]
            body = headers.body
            item = json.loads(body)
            docid = item.get("docid")
            if docid is None:
                return
            delete_result = self.delete_projects_by_document(docid)

            _uuid = uuid4().hex
            _d = {PROJECT_PROCESS_UUID:_uuid,
                  PROJECT_PROCESS_CRTIME:1,
                  PROJECT_PROCESS_PROJECTS:delete_result}
            _pp = Project_process(_d)
            if _pp.update_row(self.ots_client):
                ackMsg(conn,message_id)
            #取消插入结果队列,改成插入project_process表
            # if send_msg_toacmq(self.pool_mq_ali,delete_result,self.doc_delete_result):
            #     ackMsg(conn,message_id)

    def generate_common_properties(self,list_docs):
        '''
        #通用属性生成
        :param list_docis:
        :return:
        '''
        #计数法选择
        choose_dict = {}
        project_dict = {}
        for _key in [document_bidway,document_industry,document_info_type,document_info_source,document_qcodes,document_project_name,document_project_code,document_tenderee,document_tenderee_addr,document_tenderee_phone,document_tenderee_contact,document_agency,document_agency_phone,document_agency_contact,project_procurement_system,document_moneysource,document_time_bidclose,document_time_bidopen,document_time_bidstart,document_time_commencement,document_time_completion,document_time_earnest_money_start,document_time_earnest_money_end,document_time_get_file_end,document_time_get_file_start,document_time_publicity_end,document_time_publicity_start,document_time_registration_end,document_time_registration_start,document_time_release,document_tmp_extract_count]:
            for _doc in list_docs:
                _value = _doc.getProperties().get(_key,"")
                if _value!="":
                    if _key not in choose_dict:
                        choose_dict[_key] = {}
                    if _value not in choose_dict[_key]:
                        choose_dict[_key][_value] = 0
                    choose_dict[_key][_value] += 1


        _find = False
        for _key in [document_district,document_city,document_province,document_area]:
            area_dict = {}
            for _doc in list_docs:
                loc = _doc.getProperties().get(_key,"未知")
                if loc not in ('全国','未知',"0"):
                    if loc not in area_dict:
                        area_dict[loc] = 0
                    area_dict[loc] += 1
            list_loc = []
            for k,v in area_dict.items():
                list_loc.append([k,v])
            list_loc.sort(key=lambda x:x[1],reverse=True)
            if len(list_loc)>0:
                project_dict[document_district] = _doc.getProperties().get(document_district)
                project_dict[document_city] = _doc.getProperties().get(document_city)
                project_dict[document_province] = _doc.getProperties().get(document_province)
                project_dict[document_area] = _doc.getProperties().get(document_area)
                _find = True
                break
        if not _find:
            if len(list_docs)>0:
                project_dict[document_district] = list_docs[0].getProperties().get(document_district)
                project_dict[document_city] = list_docs[0].getProperties().get(document_city)
                project_dict[document_province] = list_docs[0].getProperties().get(document_province)
                project_dict[document_area] = list_docs[0].getProperties().get(document_area)



        for _key,_value in choose_dict.items():
            _l = []
            for k,v in _value.items():
                _l.append([k,v])
            _l.sort(key=lambda x:x[1],reverse=True)
            if len(_l)>0:
                _v = _l[0][0]
                if _v in ('全国','未知'):
                    if len(_l)>1:
                        _v = _l[1][0]
                project_dict[_key] = _v


        list_dynamics = []
        docid_number = 0
        visuable_docids = []
        zhao_biao_page_time = ""
        zhong_biao_page_time = ""
        list_codes = []

        list_product = []
        p_page_time = ""
        remove_docids = set()
        for _doc in list_docs:
            table_name = _doc.getProperties().get("table_name")
            status = _doc.getProperties().get(document_status,0)
            _save = _doc.getProperties().get(document_tmp_save,1)
            doctitle = _doc.getProperties().get(document_doctitle,"")
            docchannel = _doc.getProperties().get(document_docchannel)
            page_time = _doc.getProperties().get(document_page_time,"")
            _docid = _doc.getProperties().get(document_docid)
            _bidway = _doc.getProperties().get(document_bidway,"")
            _docchannel = _doc.getProperties().get(document_life_docchannel,0)
            project_codes = _doc.getProperties().get(document_project_codes)
            product = _doc.getProperties().get(document_product)
            sub_docs = _doc.getProperties().get("sub_docs",[])

            is_multipack = True if len(sub_docs)>1 else False
            extract_count = _doc.getProperties().get(document_tmp_extract_count,0)

            if product is not None:
                list_product.extend(product.split(","))

            if project_codes is not None:
                _c = project_codes.split(",")
                list_codes.extend(_c)

            if p_page_time=="":
                p_page_time = page_time

            if zhao_biao_page_time=="" and _docchannel in (51,52,102,103,114):
                zhao_biao_page_time = page_time
            if zhong_biao_page_time=="" and _docchannel in (101,118,119,120):
                zhong_biao_page_time = page_time
            is_visuable = 0
            if table_name=="document":
                if status>=201 and status<=300:
                    docid_number +=1
                    visuable_docids.append(str(_docid))
                    is_visuable = 1
                else:
                    remove_docids.add(str(_docid))
            else:
                if _save==1:
                    docid_number +=1
                    visuable_docids.append(str(_docid))
                    is_visuable = 1
                else:
                    remove_docids.add(str(_docid))
            list_dynamics.append({document_docid:_docid,
                                     document_doctitle:doctitle,
                                     document_docchannel:_docchannel,
                                     document_bidway:_bidway,
                                     document_page_time:page_time,
                                     document_status:201 if is_visuable==1 else 401,
                                  "is_multipack":is_multipack,
                                  document_tmp_extract_count:extract_count
                                  }
                                    )

        project_dict[project_project_dynamics] = json.dumps(list_dynamics,ensure_ascii=False)
        project_dict[project_docid_number] = docid_number
        project_dict[project_docids] = ",".join(list(set(visuable_docids)-remove_docids))
        if zhao_biao_page_time !="":
            project_dict[project_zhao_biao_page_time] = zhao_biao_page_time
        if zhong_biao_page_time !="":
            project_dict[project_zhong_biao_page_time] = zhong_biao_page_time
        project_dict[project_project_codes] = ",".join(list(set(list_codes)))
        project_dict[project_page_time] = p_page_time
        project_dict[project_product] = ",".join(list(set(list_product)))

        return project_dict


    def generate_packages_properties(self,list_docs):
        '''
        生成分包属性
        :param list_docs:
        :return:
        '''

        list_properties = []
        set_key = set()
        for _doc in list_docs:
            _dict = {}
            sub_docs = _doc.getProperties().get("sub_docs")
            if sub_docs is not None:
                for _d in sub_docs:
                    sub_project_code = _d.get(project_sub_project_code,"")
                    sub_project_name = _d.get(project_sub_project_name,"")
                    win_tenderer = _d.get(project_win_tenderer,"")
                    win_bid_price = _d.get(project_win_bid_price,"")
                    _key = "%s-%s-%s-%s"%(sub_project_code,sub_project_name,win_tenderer,win_bid_price)
                    if _key in set_key:
                        continue
                    set_key.add(_key)
                    list_properties.append(_d)
        return list_properties


    def generate_projects_from_document(self,list_docs):
        '''
        #通过公告生成projects
        :param list_docids:
        :return:
        '''
        #判断标段数

        list_projects = generate_projects([doc.getProperties() for doc in list_docs])

        return list_projects

    def search_projects_with_document(self,list_docids):
        '''
        通过docid集合查询对应的projects
        :param list_docids:
        :return:
        '''
        print("==",list_docids)
        list_should_q = []
        for _docid in list_docids:
            list_should_q.append(TermQuery("docids",_docid))
        bool_query = BoolQuery(should_queries=list_should_q)
        _query = {"query":bool_query,"limit":20}
        list_project_dict = getDocument(_query,self.ots_client,[
            project_uuid,project_docids,project_zhao_biao_page_time,
            project_zhong_biao_page_time,
            project_page_time,
            project_area,
            project_province,
            project_city,
            project_district,
            project_info_type,
            project_industry,
            project_qcodes,
            project_project_name,
            project_project_code,
            project_project_codes,
            project_project_addr,
            project_tenderee,
            project_tenderee_addr,
            project_tenderee_phone,
            project_tenderee_contact,
            project_agency,
            project_agency_phone,
            project_agency_contact,
            project_sub_project_name,
            project_sub_project_code,
            project_bidding_budget,
            project_win_tenderer,
            project_win_bid_price,
            project_win_tenderer_manager,
            project_win_tenderer_phone,
            project_second_tenderer,
            project_second_bid_price,
            project_second_tenderer_manager,
            project_second_tenderer_phone,
            project_third_tenderer,
            project_third_bid_price,
            project_third_tenderer_manager,
            project_third_tenderer_phone,
            project_procurement_system,
            project_bidway,
            project_dup_data,
            project_docid_number,
            project_project_dynamics,
            project_product,
            project_moneysource,
            project_service_time,
            project_time_bidclose,
            project_time_bidopen,
            project_time_bidstart,
            project_time_commencement,
            project_time_completion,
            project_time_earnest_money_start,
            project_time_earnest_money_end,
            project_time_get_file_end,
            project_time_get_file_start,
            project_time_publicity_end,
            project_time_publicity_start,
            project_time_registration_end,
            project_time_registration_start,
            project_time_release,
            project_dup_docid,
            project_info_source,
            project_nlp_enterprise,
            project_nlp_enterprise_attachment,
        ],sort="page_time",table_name="project2",table_index="project2_index")

        return list_project_dict

    def set_project_uuid(self,_dict,_uuid):
        if _uuid is not None and _uuid!="":
            if "uuid" in _dict:
                _dict["uuid"] = "%s,%s"%(_dict["uuid"],_uuid)
            else:
                _dict["uuid"] = _uuid



    def getMerge_rules(self,page_time,project_codes,project_name,tenderee,agency,product,sub_project_name,bidding_budget,win_tenderer,win_bid_price,province,city,district):

        whole_time_start = time.time()
        _time = time.time()
        list_query = []

        list_code = [a for a in project_codes.split(",") if a!='']
        should_q_code = BoolQuery(should_queries=[MatchQuery(project_project_codes,a) for a in list_code[:20]])

        # print("should_q_code",[a for a in list_code[:20]])
        should_q_cod = BoolQuery(should_queries=[MatchQuery(project_project_code,a) for a in list_code[:20]])
        list_product = [a for a in product.split(",") if a!='']
        should_q_product = BoolQuery(should_queries=[MatchQuery(project_product,a) for a in list_product[:20]])

        should_q_area = None
        if province!="" or city!="" or district!="":
            should_q = []
            if province not in ("","全国","未知") and province is not None:
                should_q.append(TermQuery(project_province,province))
            if city not in ("","全国","未知") and city is not None:
                should_q.append(TermQuery(project_city,city))
            if district not in ("","全国","未知") and district is not None:
                should_q.append(TermQuery(project_district,district))
            if len(should_q)>0:
                should_q_area = BoolQuery(should_queries=should_q)


        prepare_time = time.time()-_time

        _time = time.time()
        # log("list_code %s"%(str(list_code)))
        # log("list_product %s"%(str(list_product)))
        # log("tenderee %s"%(tenderee))
        # log("bidding_budget %s"%(bidding_budget))
        # log("win_tenderer %s"%(win_tenderer))
        # log("win_bid_price %s"%(win_bid_price))
        # log("project_name %s"%(project_name))
        log_time = time.time()-_time


        _time = time.time()
        if tenderee!="" and len(list_code)>0:
            _query = [TermQuery(project_tenderee,tenderee),
                                             should_q_code,
                                             ]
            list_query.append([_query,2])

            _query = [TermQuery(project_tenderee,tenderee),
                      should_q_cod
                      ]
            list_query.append([_query,2])

        if tenderee!="" and len(list_product)>0:
            _query = [TermQuery(project_tenderee,tenderee),
                      should_q_product]
            list_query.append([_query,2])

        if tenderee!="" and project_name!="":
            _query = [TermQuery(project_tenderee,tenderee),
                                             TermQuery(project_project_name,project_name)]
            list_query.append([_query,2])

        if tenderee!="" and agency!="":
            _query = [TermQuery(project_tenderee,tenderee),
                      TermQuery(project_agency,agency)]
            list_query.append([_query,1])

        if tenderee!="" and float(bidding_budget)>0:
            _query = [TermQuery(project_tenderee,tenderee),
                                             TermQuery(project_bidding_budget,bidding_budget)]
            list_query.append([_query,2])

        if float(bidding_budget)>0 and float(win_bid_price)>0:
            _query = [TermQuery(project_bidding_budget,bidding_budget),
                      TermQuery(project_win_bid_price,win_bid_price)]
            list_query.append([_query,2])


        if tenderee!="" and win_tenderer!="":
            _query = [TermQuery(project_tenderee,tenderee),
                      TermQuery(project_win_tenderer,win_tenderer)]
            list_query.append([_query,2])

        if agency!="" and win_tenderer!="":
            _query = [TermQuery(project_agency,agency),
                      TermQuery(project_win_tenderer,win_tenderer)]
            list_query.append([_query,2])

        if agency!="" and len(list_product)>0:
            _query = [TermQuery(project_agency,agency),
                      should_q_product]
            list_query.append([_query,2])

        if win_tenderer!="" and len(list_code)>0:
            _query = [TermQuery(project_win_tenderer,win_tenderer),
                                             should_q_code]
            list_query.append([_query,2])

            _query = [TermQuery(project_win_tenderer,win_tenderer),
                      should_q_cod]
            list_query.append([_query,2])

        if win_tenderer!="" and float(win_bid_price)>0:
            _query = [TermQuery(project_win_tenderer,win_tenderer),
                                             TermQuery(project_win_bid_price,win_bid_price)]
            list_query.append([_query,2])

        if win_tenderer!="" and float(bidding_budget)>0:
            _query = [TermQuery(project_win_tenderer,win_tenderer),
                      TermQuery(project_bidding_budget,bidding_budget)]
            list_query.append([_query,2])

        if len(list_code)>0 and len(list_product)>0:
            _query = [should_q_code,
                      should_q_product]
            list_query.append([_query,2])

        if len(list_code)>0:
            _query = [
                      should_q_code]
            list_query.append([_query,1])

            _query = [
                should_q_cod]
            list_query.append([_query,1])

        if project_name!="" and project_name is not None:
            _query = [
                      TermQuery(project_project_name,project_name)]
            list_query.append([_query,1])
            _query_title = [MatchPhraseQuery(project_doctitles,project_name)]
            list_query.append([_query_title,1])
        if len(list_product)>0 and should_q_area is not None:
            _query = [should_q_area,
                      should_q_product]
            list_query.append([_query,1])

        generate_time = time.time()-_time
        whole_time = time.time()-whole_time_start
        log("projects merge rules whole_time:%.3f prepare_time:%.3f log_time:%.3f generate_time:%.3f"%(whole_time,prepare_time,log_time,generate_time))
        return list_query




    def merge_projects(self,list_projects,b_log=False,check_columns=[project_uuid,project_zhao_biao_page_time,project_zhong_biao_page_time,project_page_time,project_project_name,project_project_code,project_project_codes,project_tenderee,project_agency,project_sub_project_name,project_sub_project_code,project_bidding_budget,project_win_tenderer,project_win_bid_price,project_project_dynamics,project_product,project_time_bidclose,project_time_bidopen,project_time_bidstart,project_time_commencement,project_time_completion,project_time_earnest_money_start,project_time_earnest_money_end,project_time_get_file_end,project_time_get_file_start,project_time_publicity_end,project_time_publicity_start,project_time_registration_end,project_time_registration_start,project_time_release,project_nlp_enterprise,project_nlp_enterprise_attachment,project_docids,project_area,project_province,project_city,project_district,project_info_type,project_industry,project_qcodes,project_project_addr,project_tenderee_addr,project_agency_phone,project_agency_contact,project_tenderee_phone,project_tenderee_contact,project_win_tenderer_manager,project_win_tenderer_phone,project_second_tenderer,project_second_bid_price,project_second_tenderer_manager,project_second_tenderer_phone,project_third_tenderer,project_third_bid_price,project_third_tenderer_manager,project_third_tenderer_phone,project_procurement_system,project_bidway,project_dup_data,project_docid_number,project_moneysource,project_service_time,project_dup_docid,project_info_source],fix_columns=[project_area,project_province,project_city,project_district,project_info_type,project_industry,project_qcodes,project_project_addr,project_tenderee_addr,project_agency_phone,project_agency_contact,project_tenderee_phone,project_tenderee_contact,project_win_tenderer_manager,project_win_tenderer_phone,project_second_tenderer,project_second_bid_price,project_second_tenderer_manager,project_second_tenderer_phone,project_third_tenderer,project_third_bid_price,project_third_tenderer_manager,project_third_tenderer_phone,project_procurement_system,project_bidway,project_dup_data,project_docid_number,project_moneysource,project_service_time,project_dup_docid,project_info_source]):
        '''
        对项目进行合并
        :return:
        '''

        try:
            whole_time_start = time.time()
            set_uuid = set()
            for _proj in list_projects:
                _uuid = _proj.get("uuid")
                if _uuid is not None:
                    set_uuid = set_uuid | set(_uuid.split(","))
            must_not_q = []
            for _uuid in list(set_uuid):
                must_not_q.append(TermQuery("uuid",_uuid))


            projects_merge_count = 0
            projects_check_rule_time = 0
            projects_update_time = 0
            projects_query_time = 0
            projects_prepare_time = 0
            current_date = getCurrent_date("%Y-%m-%d")
            min_date = timeAdd(current_date,-35,format="%Y-%m-%d")
            search_table = "project2"
            search_table_index = "project2_index_formerge"
            project_cls = Project
            docids = ""
            for _proj in list_projects[:30]:

                docids = _proj.get(project_docids,"")
                page_time = _proj.get(project_page_time,"")
                project_codes = _proj.get(project_project_codes,"")
                project_name = _proj.get(project_project_name,"")
                tenderee = _proj.get(project_tenderee,"")
                agency = _proj.get(project_agency,"")
                product = _proj.get(project_product,"")
                sub_project_name = _proj.get(project_sub_project_name,"")
                bidding_budget = _proj.get(project_bidding_budget,-1)
                win_tenderer = _proj.get(project_win_tenderer,"")
                win_bid_price = _proj.get(project_win_bid_price,-1)

                province = _proj.get(project_province,"")
                city = _proj.get(project_city,"")
                district = _proj.get(project_district,"")

                page_time_less = timeAdd(page_time,-150)
                page_time_greater = timeAdd(page_time,120)
                sub_project_q = TermQuery(project_sub_project_name,sub_project_name) if sub_project_name.replace("Project","")!="" else None
                _time = time.time()
                list_must_query = self.getMerge_rules(page_time,project_codes,project_name,tenderee,agency,product,sub_project_name,bidding_budget,win_tenderer,win_bid_price,province,city,district)


                list_merge_data = []

                search_table = "project2"
                search_table_index = "project2_index_formerge"
                project_cls = Project

                print("page_time,min_date",page_time,min_date)
                if page_time>=min_date:
                    search_table = "project2_tmp"
                    search_table_index = "project2_tmp_index"
                    project_cls = Project_tmp


                _step = 4
                _begin = 0
                must_queries = []


                if page_time_less is not None and page_time_greater is not None:
                    must_queries = [RangeQuery(project_page_time,page_time_less,page_time_greater,True,True),
                                ]

                print("page_time_less,page_time_greater",page_time,page_time_less,page_time_greater)
                #sub_project_name非必要条件
                # if sub_project_q is not None:
                #     must_queries.append(sub_project_q)

                projects_prepare_time += time.time()-_time
                _time = time.time()
                while _begin<len(list_must_query):
                    list_should_q = []
                    _limit = 20
                    for must_q,_count in list_must_query[_begin:_begin+_step]:
                        must_q1 = list(must_q)
                        must_q1.extend(must_queries)
                        list_should_q.append(BoolQuery(must_queries=must_q1))

                        # _limit += _count*5
                    _query = BoolQuery(
                                       should_queries=list_should_q,
                                       must_not_queries=must_not_q[:100]
                    )
                    # rows,next_token,total_count,is_all_succeed = self.ots_client_merge.search("project2","project2_index_formerge",
                    #                                                                     SearchQuery(_query,limit=_limit),
                    #                                                                     columns_to_get=ColumnsToGet(column_names=[project_uuid,project_docids,project_zhao_biao_page_time,project_zhong_biao_page_time,project_page_time,project_area,project_province,project_city,project_district,project_info_type,project_industry,project_qcodes,project_project_name,project_project_code,project_project_codes,project_project_addr,project_tenderee,project_tenderee_addr,project_tenderee_phone,project_tenderee_contact,project_agency,project_agency_phone,project_agency_contact,project_sub_project_name,project_sub_project_code,project_bidding_budget,project_win_tenderer,project_win_bid_price,project_win_tenderer_manager,project_win_tenderer_phone,project_second_tenderer,project_second_bid_price,project_second_tenderer_manager,project_second_tenderer_phone,project_third_tenderer,project_third_bid_price,project_third_tenderer_manager,project_third_tenderer_phone,project_procurement_system,project_bidway,project_dup_data,project_docid_number,project_project_dynamics,project_product,project_moneysource,project_service_time,project_time_bidclose,project_time_bidopen,project_time_bidstart,project_time_commencement,project_time_completion,project_time_earnest_money_start,project_time_earnest_money_end,project_time_get_file_end,project_time_get_file_start,project_time_publicity_end,project_time_publicity_start,project_time_registration_end,project_time_registration_start,project_time_release,project_dup_docid,project_info_source,project_nlp_enterprise,project_nlp_enterprise_attachment],return_type=ColumnReturnType.SPECIFIED))

                    rows,next_token,total_count,is_all_succeed = self.ots_client_merge.search(search_table,search_table_index,
                                                                                              SearchQuery(_query,limit=_limit),
                                                                                              columns_to_get=ColumnsToGet(column_names=check_columns,return_type=ColumnReturnType.SPECIFIED))
                    list_data = getRow_ots(rows)

                    list_merge_data.extend(list_data)

                    # print(list_data)
                    for _data in list_data:
                        must_not_q.append(TermQuery(project_uuid,_data.get(project_uuid)))

                    _begin += _step
                projects_query_time += time.time()-_time
                #优先匹配招标金额相近的
                projects_merge_count = len(list_merge_data)
                list_merge_data.sort(key=lambda x:x.get(project_page_time,""))
                list_merge_data.sort(key=lambda x:x.get(project_bidding_budget,-1))
                # log(page_time_less+"=="+page_time_greater)
                # log("list_merge_data:%s"%(str(list_merge_data)))
                list_check_data = []
                for _data in list_merge_data:
                    _time = time.time()
                    _check,_prob = check_merge_rule(_proj,_data,b_log=b_log,return_prob=True)
                    if b_log:
                        log(str(_check))
                    projects_check_rule_time += time.time()-_time
                    if _check:
                        list_check_data.append([_data,_prob])

                list_check_data.sort(key=lambda x:x[1],reverse=True)
                for _data,_ in list_check_data:
                    _time = time.time()
                    _check,_prob = check_merge_rule(_proj,_data,b_log=b_log,return_prob=True)
                    projects_check_rule_time += time.time()-_time
                    _time = time.time()
                    if _check:
                        # o_proj = project_cls(_data)
                        # o_proj.fix_columns(self.ots_client,fix_columns,True)
                        # for k in fix_columns:
                        #     _data[k] = o_proj.getProperties().get(k)

                        update_projects_by_project(_data,[_proj])
                        projects_update_time += time.time()-_time

            whole_time = time.time()-whole_time_start
            log("%s %s merge_project whole_time:%.3f projects_prepare_time:%.3f projects_query_time:%.3f projects_merge_count:%d rules%d projects_check_rule_time %.3f projects_update_time %.3f"%(search_table,docids,whole_time,projects_prepare_time,projects_query_time,projects_merge_count,len(list_must_query),projects_check_rule_time,projects_update_time))

            return list_projects
        except Exception as e:
            traceback.print_exc()
            assert 1==2








    def merge_document_real(self,item,dup_docid,table_name,save,status_to=None,b_log=False):
        '''
        实时项目合并
        :param item:
        :param dup_docid:重复的公告集合
        :param status_to:
        :return:
        '''
        try:
            list_docids = []
            _docid = item.get(document_tmp_docid)
            list_docids.append(_docid)
            if isinstance(dup_docid,list):
                list_docids.extend(dup_docid)
            list_docids = [a for a in list_docids if a is not None]

            _time = time.time()
            list_projects = self.search_projects_with_document(list_docids)
            # log("search projects takes:%.3f"%(time.time()-_time))
            if len(list_projects)==0:
                # _time = time.time()
                list_docs = self.search_docs(list_docids)
                # log("search document takes:%.3f"%(time.time()-_time))
                # _time = time.time()
                list_projects = self.generate_projects_from_document(list_docs)
                # log("generate projects takes:%.3f"%(time.time()-_time))
            else:
                _time = time.time()
                self.update_projects_by_document(_docid,save,list_projects)
                # log("update projects takes:%.3f"%(time.time()-_time))
            _time = time.time()
            list_projects = dumplicate_projects(list_projects)
            # log("dumplicate projects takes:%.3f"%(time.time()-_time))
            _time = time.time()
            list_projects = self.merge_projects(list_projects,b_log)
            # log("merge projects takes:%.3f"%(time.time()-_time))

            _time = time.time()
            dumplicate_document_in_merge(list_projects)
            log("dumplicate document %d takes:%.3f"%(len(list_projects),time.time()-_time))

            _time = time.time()
            project_json = to_project_json(list_projects)
            # log("json projects takes:%.3f"%(time.time()-_time))
            if b_log:
                log("project_json:%s"%project_json)
            return project_json
        except Exception as e:
            raise RuntimeError("error on dumplicate")

    def is_exist_fingerprint(self,final_list,_docid,_fingerprint,table_name):
        set_fingerprint = set()
        for _i in range(1,len(final_list)):
            _dict = final_list[_i]
            b_docid = _dict[document_tmp_docid]
            _save = _dict.get(document_tmp_save,0)
            _status = _dict.get(document_tmp_status,0)
            if table_name=="document":
                if _status>=201 and _status<=300:
                    _save = 1
            fingerprint_less = _dict.get(document_tmp_fingerprint,"")
            if b_docid==_docid:
                pass
            else:
                if _save==1:
                    set_fingerprint.add(fingerprint_less)
        print("_fingerprint",_fingerprint)
        print(set_fingerprint)
        if _fingerprint in set_fingerprint:
            return True
        return False



    def dumplicate_comsumer_handle(self,item,result_queue,ots_client,get_all=False,upgrade=True):
        try:
            start_time = time.time()
            self.post_extract(item)

            base_list = []
            set_docid = set()

            list_rules,table_name,table_index = self.translate_dumplicate_rules(flow_dumplicate_status_from,item,get_all=get_all,to_log=False)
            # print("len_rules",len(list_rules),table_name,table_index)
            list_rules.sort(key=lambda x:x["confidence"],reverse=True)
            _i = 0
            step = 5

            item["confidence"] = 999
            if item.get(document_tmp_docid) not in set_docid:
                base_list.append(item)
                set_docid.add(item.get(document_tmp_docid))

            while _i<len(list_rules):
                must_not_q = []
                if len(base_list)>0:
                    must_not_q = [TermQuery("docid",a) for a in list(set_docid)[-100:]]
                _query = BoolQuery(should_queries=[_rule["query"] for _rule in list_rules[_i:_i+step]],
                                   must_not_queries=must_not_q)
                _rule = list_rules[_i]
                confidence = _rule["confidence"]
                singleNum_keys = _rule["singleNum_keys"]
                contain_keys = _rule["contain_keys"]
                multiNum_keys = _rule["multiNum_keys"]
                self.add_data_by_query(item,base_list,set_docid,_query,confidence,table_name=table_name,table_index=table_index,singleNum_keys=singleNum_keys,contain_keys=contain_keys,multiNum_keys=multiNum_keys,columns=[document_tmp_status,document_tmp_save,document_tmp_page_time,document_tmp_docchannel,document_tmp_tenderee,document_tmp_agency,document_tmp_doctitle_refine,document_tmp_sub_docs_json,document_tmp_extract_json,document_tmp_web_source_no,document_tmp_fingerprint,document_attachment_extract_status])
                _i += step



            _time = time.time()
            # log("%d start final check with length:%d"%(item["docid"],len(base_list)))
            final_list = self.dumplicate_fianl_check(base_list)

            exist_finterprint = self.is_exist_fingerprint(final_list,item.get(document_tmp_docid),item.get(document_tmp_fingerprint),table_name)
            # log("%d final_check takes:%.2f"%(item["docid"],time.time()-_time))
            best_docid = self.get_best_docid(final_list)

            final_list_docid = [a["docid"] for a in final_list]
            # log("%d:final_list_docid:%s"%(item["docid"],str(final_list_docid)))
            _d = {"partitionkey":item["partitionkey"],
                  "docid":item["docid"],
                  "status":random.randint(*flow_dumplicate_status_to),
                  document_tmp_opertime:getCurrent_date(format="%Y-%m-%d %H:%M:%S")
                  }
            dtmp = Document_tmp(_d)


            dup_docid = set()
            for _dict in final_list:
                dup_docid.add(_dict.get(document_tmp_docid))
            if item.get(document_tmp_docid) in dup_docid:
                dup_docid.remove(item.get(document_tmp_docid))


            remove_list = []


            if len(final_list)==0 or best_docid==item.get(document_tmp_docid):
                dtmp.setValue(document_tmp_save,1,True)
                # dtmp.setValue(document_tmp_merge_uuid,self.merge_document(item,flow_dumplicate_status_to),True)
                dmp_docid = ",".join([str(a) for a in list(dup_docid)])
                for _dict in final_list:
                    if _dict.get(document_tmp_docid) in dup_docid:
                        remove_list.append(_dict)
            else:
                dtmp.setValue(document_tmp_save,0,True)
                if best_docid in dup_docid:
                    dup_docid.remove(best_docid)
                    for _dict in final_list:
                        if _dict.get(document_tmp_docid) in dup_docid:
                            remove_list.append(_dict)
                    dmp_docid = ",".join([str(a) for a in list(dup_docid)])
                    dmp_docid = "%d,%s"%(best_docid,dmp_docid)
                else:
                    dmp_docid = ",".join([str(a) for a in list(dup_docid)])
                    for _dict in final_list:
                        if _dict.get(document_tmp_docid) in dup_docid:
                            remove_list.append(_dict)

            list_docids = list(dup_docid)
            list_docids.append(best_docid)
            b_log = False if upgrade else True

            if item.get(document_update_document)=="true":
                dtmp.setValue(document_tmp_save,1,True)

            if exist_finterprint and dtmp.getProperties().get(document_tmp_save)==0:
                log("exist_finterprint %s"%(str(item.get(document_tmp_docid))))
                dtmp.setValue(document_tmp_projects,"[]",True)
            else:
                dtmp.setValue(document_tmp_projects,self.merge_document_real(item,list_docids,table_name,dtmp.getProperties().get(document_tmp_save),flow_dumplicate_status_to,b_log),True)
            log("upgrate %s save:%s:docid:%d,final_list:%d,rules:%d,best_docid:%s,dmp_docid:%s"%(str(upgrade),dtmp.getProperties().get(document_tmp_save),item.get(document_tmp_docid),len(final_list),len(list_rules),str(best_docid),dmp_docid))
            if upgrade:
                if table_name=="document_tmp":
                    self.changeSaveStatus(remove_list)

                # print(dtmp.getProperties())
                dtmp.setValue(document_tmp_dup_docid,dmp_docid,True)
                dtmp.setValue(document_tmp_best_docid,best_docid,True)
                _flag = dtmp.update_row(self.ots_client)
                if not _flag:
                    for i in range(10):
                        list_proj_json = dtmp.getProperties().get(document_tmp_projects)
                        if list_proj_json is not None:
                            list_proj = json.loads(list_proj_json)
                            dtmp.setValue(document_tmp_projects,json.dumps(list_proj[:len(list_proj)//2]),True)
                            if dtmp.update_row(self.ots_client):
                                break


            # log("dump takes %.2f"%(time.time()-start_time))
        except Exception as e:
            traceback.print_exc()
            log("error on dumplicate of %s"%(str(item.get(document_tmp_docid))))


    def fix_doc_which_not_in_project(self):
        '''
        将成品公告中不存在于project2的数据取出，并放入document_tmp中重新进行去重和合并
        :return:
        '''
        def fix_doc_handle(item,result_queue):
            _docid = item.get(document_tmp_docid)
            b_q = BoolQuery(must_queries=[TermQuery(project_docids,str(_docid))])

            rows,next_token,total_count,is_all_succeed = self.ots_client.search("project2","project2_index",
                                                                                SearchQuery(b_q,get_total_count=True),
                                                                                ColumnsToGet(return_type=ColumnReturnType.NONE))
            if total_count==0:
                log("fix_doc:%s not in project2"%(str(_docid)))
                d_tmp = Document_tmp(item)
                d_tmp.setValue(document_tmp_status,flow_dumplicate_status_from[0],True)
                d_tmp.update_row(self.ots_client)



        if self.fix_doc_docid is None:
            current_date = getCurrent_date(format="%Y-%m-%d %H:%M:%S")
            before_date = timeAdd(current_date,0,format="%Y-%m-%d %H:%M:%S",minutes=-5)
            bool_query = BoolQuery(must_queries=[
                TermQuery(document_tmp_save,1),
                RangeQuery(document_tmp_status,flow_dumplicate_status_to[0]),
                RangeQuery(document_tmp_opertime,before_date)
            ])
        else:
            bool_query = BoolQuery(must_queries=[
                TermQuery(document_tmp_save,1),
                RangeQuery(document_tmp_status,flow_dumplicate_status_to[0]),
                RangeQuery(document_tmp_docid,self.fix_doc_docid)
            ])

        list_data = []
        rows,next_token,total_count,is_all_succeed = self.ots_client.search("document_tmp","document_tmp_index",
                                                                            SearchQuery(bool_query,sort=Sort(sorters=[FieldSort("docid",SortOrder.ASC)]),get_total_count=True,limit=100),
                                                                            ColumnsToGet(return_type=ColumnReturnType.NONE))
        list_d = getRow_ots(rows)
        list_data.extend(list_d)
        while next_token:
            rows,next_token,total_count,is_all_succeed = self.ots_client.search("document_tmp","document_tmp_index",
                                                                                SearchQuery(bool_query,next_token=next_token,get_total_count=True,limit=100),
                                                                                ColumnsToGet(return_type=ColumnReturnType.NONE))
            list_d = getRow_ots(rows)
            list_data.extend(list_d)
            print("%d/%d"%(len(list_data),total_count))
        if len(list_data)>0:
            self.fix_doc_docid = list_data[-1].get(document_tmp_docid)
            log("current fix_doc_docid:%s"%(str(self.fix_doc_docid)))
            task_queue = Queue()
            for _data in list_data:
                task_queue.put(_data)

            mt = MultiThreadHandler(task_queue,fix_doc_handle,None,30)
            mt.run()



    def start_flow_dumplicate(self):
        schedule = BlockingScheduler()
        schedule.add_job(self.flow_dumplicate,"cron",second="*/40")
        schedule.add_job(self.flow_dumpcate_comsumer,"cron",second="*/10")
        schedule.add_job(self.bdm.monitor_dumplicate,"cron",minute="*/10")
        # schedule.add_job(self.fix_doc_which_not_in_project,"cron",minute="55")
        schedule.start()

    def changeSaveStatus(self,list_dict):
        for _dict in list_dict:
            if _dict.get(document_tmp_save,1)==1:
                _d = {"partitionkey":_dict["partitionkey"],
                      "docid":_dict["docid"],
                      document_tmp_save:0
                      }
                _d_tmp = Document_tmp(_d)
                _d_tmp.update_row(self.ots_client)



    def test_dumplicate(self,docid):
        columns=[document_tmp_status,document_tmp_page_time,document_tmp_docchannel,document_tmp_tenderee,document_tmp_agency,document_tmp_doctitle,document_tmp_sub_docs_json,document_tmp_extract_json,document_tmp_web_source_no,document_tmp_fingerprint,document_attachment_extract_status]
        bool_query = BoolQuery(must_queries=[
            TermQuery("docid",docid)
        ])
        rows,next_token,total_count,is_all_succeed = self.ots_client.search("document","document_index",
                                                                            SearchQuery(bool_query,sort=Sort(sorters=[FieldSort("docid")]),limit=100,get_total_count=True),
                                                                            ColumnsToGet(columns,return_type=ColumnReturnType.SPECIFIED))
        log("flow_dumplicate producer total_count:%d"%total_count)
        if total_count==0:
            rows,next_token,total_count,is_all_succeed = self.ots_client.search("document_tmp","document_tmp_index",
                                                                                SearchQuery(bool_query,sort=Sort(sorters=[FieldSort("docid")]),limit=100,get_total_count=True),
                                                                                ColumnsToGet(columns,return_type=ColumnReturnType.SPECIFIED))
        list_dict = getRow_ots(rows)

        for item in list_dict:
            self.dumplicate_comsumer_handle(item,None,self.ots_client,get_all=True,upgrade=False)
            return

    def test_merge(self,list_docid_less,list_docid_greater):
        list_docs_less = self.search_docs(list_docid_less)
        list_projects_less = self.generate_projects_from_document(list_docs_less)

        list_docs_greater = self.search_docs(list_docid_greater)
        list_projects_greater = self.generate_projects_from_document(list_docs_greater)

        list_projects_less.extend(list_projects_greater)
        list_projects = dumplicate_projects(list_projects_less,b_log=True)
        project_json = to_project_json(list_projects)
        log("project_json:%s"%project_json)
        return project_json

    def getRemainDoc(self,docid):
        columns=[document_tmp_status,document_tmp_page_time,document_tmp_docchannel,document_tmp_tenderee,document_tmp_agency,document_tmp_doctitle,document_tmp_sub_docs_json,document_tmp_extract_json]
        bool_query = BoolQuery(must_queries=[
            TermQuery("docid",docid)
        ])
        rows,next_token,total_count,is_all_succeed = self.ots_client.search("document","document_index",
                                                                            SearchQuery(bool_query,sort=Sort(sorters=[FieldSort("docid")]),limit=100,get_total_count=True),
                                                                            ColumnsToGet(columns,return_type=ColumnReturnType.SPECIFIED))
        list_dict = getRow_ots(rows)

        if len(list_dict)>0:
            item = list_dict[0]
            start_time = time.time()
            self.post_extract(item)

            base_list = []
            set_docid = set()

            list_rules,table_name,table_index = self.translate_dumplicate_rules(flow_dumplicate_status_from,item,to_log=True)

            list_rules.sort(key=lambda x:x["confidence"],reverse=True)
            _i = 0
            step = 5

            item["confidence"] = 999
            if item.get(document_tmp_docid) not in set_docid:
                base_list.append(item)
                set_docid.add(item.get(document_tmp_docid))

            while _i<len(list_rules):
                must_not_q = []
                if len(base_list)>0:
                    must_not_q = [TermQuery("docid",a) for a in list(set_docid)[-100:]]
                _query = BoolQuery(should_queries=[_rule["query"] for _rule in list_rules[_i:_i+step]],
                                   must_not_queries=must_not_q)
                _rule = list_rules[_i]
                confidence = _rule["confidence"]
                singleNum_keys = _rule["singleNum_keys"]
                contain_keys = _rule["contain_keys"]
                multiNum_keys = _rule["multiNum_keys"]
                self.add_data_by_query(item,base_list,set_docid,_query,confidence,table_name=table_name,table_index=table_index,singleNum_keys=singleNum_keys,contain_keys=contain_keys,multiNum_keys=multiNum_keys,columns=[document_tmp_status,document_tmp_save,document_tmp_page_time,document_tmp_docchannel,document_tmp_tenderee,document_tmp_agency,document_tmp_doctitle_refine,document_tmp_sub_docs_json,document_tmp_extract_json])
                _i += step



            _time = time.time()
            log("%d start final check with length:%d"%(item["docid"],len(base_list)))
            final_list = self.dumplicate_fianl_check(base_list)
            log("%d final_check takes:%.2f"%(item["docid"],time.time()-_time))
            best_docid = self.get_best_docid(final_list)
            return best_docid
        return None



if __name__ == '__main__':
    # df = Dataflow()
    # df.flow_init()
    # df.flow_test()
    # df.test_merge()
    # df.start_flow_attachment()
    # df.start_flow_extract()
    # df.start_flow_dumplicate()
    # # df.start_flow_merge()
    # df.start_flow_remove()

    # download_attachment()
    # test_attachment_interface()
    df_dump = Dataflow_dumplicate(start_delete_listener=False)
    # df_dump.start_flow_dumplicate()
    a = time.time()
    df_dump.test_dumplicate(335604520)
    # df_dump.test_merge([292315564],[287890754])
    # df_dump.flow_remove_project_tmp()
    print("takes",time.time()-a)
    # df_dump.fix_doc_which_not_in_project()
    # df_dump.delete_projects_by_document(16288036)
    # log("=======")
    # for i in range(3):
    #     time.sleep(20)
    #
    # a = {"docid":74295123}
    # send_msg_toacmq(df_dump.pool_mq_ali,json.dumps(a),df_dump.doc_delete_queue)