BaseDataMaintenance/BaseDataMaintenance/maintenance/dataflow_mq.py

from BaseDataMaintenance.maintenance.dataflow import *
from BaseDataMaintenance.common.activateMQUtils import *
from BaseDataMaintenance.dataSource.source import getConnect_activateMQ,getConnection_postgres,getConnection_mysql,getConnection_oracle,getConnect_ots_capacity,getConnect_redis_doc
from BaseDataMaintenance.dataSource.setttings import *

from BaseDataMaintenance.model.postgres.attachment import Attachment_postgres
import os
from BaseDataMaintenance.common.ossUtils import *
from BaseDataMaintenance.dataSource.pool import ConnectorPool
from BaseDataMaintenance.model.ots.document import Document

from BaseDataMaintenance.common.Utils import article_limit
from BaseDataMaintenance.common.documentFingerprint import getFingerprint
from BaseDataMaintenance.model.postgres.document_extract import *
from BaseDataMaintenance.model.oracle.T_SHEN_PI_XIANG_MU import *

import sys
sys.setrecursionlimit(1000000)

from multiprocessing import Process

class ActiveMQListener():

    def __init__(self,conn,_queue,*args,**kwargs):
        self.conn = conn
        self._queue = _queue

    def on_error(self, headers):
        log("===============")
        log('received an error %s' % str(headers.body))

    def on_message(self, headers):
        log("====message====")
        message_id = headers.headers["message-id"]
        body = headers.body
        self._queue.put({"frame":headers,"conn":self.conn},True)

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

class Dataflow_ActivteMQ_attachment(Dataflow_attachment):

    class AttachmentMQListener():

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

        def on_error(self, headers):
            log("===============")
            log('received an error %s' % str(headers.body))

        def on_message(self, headers):
            try:
                log("get message of idx:%s"%(str(self._idx)))
                message_id = headers.headers["message-id"]
                body = headers.body

                _dict = {"frame":headers,"conn":self.conn,"idx":self._idx}
                self._func(_dict=_dict)
            except Exception as e:
                traceback.print_exc()
                pass


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

    def __init__(self):
        Dataflow_attachment.__init__(self)


        self.mq_attachment = "/queue/dataflow_attachment"
        self.mq_attachment_failed = "/queue/dataflow_attachment_failed"
        self.mq_extract = "/queue/dataflow_extract"

        self.queue_attachment_ocr = Queue()
        self.queue_attachment_not_ocr = Queue()
        self.comsumer_count = 90
        self.comsumer_process_count = 5
        self.retry_comsumer_count = 10
        self.retry_times = 5
        self.list_attachment_comsumer = []

        # for _i in range(self.comsumer_count):
        #     listener_attachment = self.AttachmentMQListener(getConnect_activateMQ(),self.queue_attachment)
        #     createComsumer(listener_attachment,self.mq_attachment)
        #     self.list_attachment_comsumer.append(listener_attachment)

        self.attach_pool = ConnectorPool(10,30,getConnection_postgres)
        self.redis_pool = ConnectorPool(10,30,getConnect_redis_doc)
        self.conn_mq = getConnect_activateMQ()
        self.pool_mq = ConnectorPool(10,30,getConnect_activateMQ)

        self.session = None

        # for _ in range(self.comsumer_process_count):
        #     listener_p = Process(target=self.start_attachment_listener)
        #     listener_p.start()

        listener_p = Process(target=self.start_attachment_listener)
        listener_p.start()



    def start_attachment_listener(self):
        for _i in range(self.comsumer_count):
            listener_attachment = self.AttachmentMQListener(getConnect_activateMQ(),self.attachment_listener_handler,_i)
            createComsumer(listener_attachment,self.mq_attachment)
            self.list_attachment_comsumer.append(listener_attachment)

        while 1:
            for i in range(len(self.list_attachment_comsumer)):
                if self.list_attachment_comsumer[i].conn.is_connected():
                    continue
                else:
                    listener = self.AttachmentMQListener(getConnect_activateMQ(),self.attachment_listener_handler,i)
                    createComsumer(listener,self.mq_attachment)
                    self.list_attachment_comsumer[i] = listener
            time.sleep(5)

    def monitor_listener(self):
        for i in range(len(self.list_attachment_comsumer)):
            if self.list_attachment_comsumer[i].conn.is_connected():
                continue
            else:
                listener = ActiveMQListener(getConnect_activateMQ(),self.queue_attachment)
                createComsumer(listener,self.mq_attachment)
                self.list_attachment_comsumer[i] = listener


    def process_failed_attachment(self):

        from BaseDataMaintenance.java.MQInfo import getQueueSize
        attachment_size = getQueueSize("dataflow_attachment")
        failed_attachment_size = getQueueSize("dataflow_attachment_failed")
        if attachment_size<100 and failed_attachment_size>0:
            list_comsumer = []
            for _i in range(self.retry_comsumer_count):
                listener_attachment = self.AttachmentMQListener(getConnect_activateMQ(),self.attachment_listener_handler,_i)
                list_comsumer.append(listener_attachment)
                createComsumer(listener_attachment,self.mq_attachment_failed)
            while 1:
                failed_attachment_size = getQueueSize("dataflow_attachment_failed")
                if failed_attachment_size==0:
                    break
                time.sleep(10)
            for _c in list_comsumer:
                _c.conn.disconnect()

    def attachment_listener_handler(self,_dict):
        try:
            frame = _dict["frame"]
            conn = _dict["conn"]
            message_id = frame.headers["message-id"]
            item = json.loads(frame.body)
            _idx = _dict.get("idx",1)
            page_attachments = json.loads(item.get(document_tmp_attachment_path,"[]"))
            _dochtmlcon = item.get(document_tmp_dochtmlcon,"")

            if random.random()<0.2:
                log("jump by random")
                if send_msg_toacmq(self.pool_mq,frame.body,self.mq_attachment):
                    ackMsg(conn,message_id)
                    return

            if len(page_attachments)==0:
                newitem ={"item":item,"list_attach":[],"message_id":message_id,"conn":conn}
            else:
                list_fileMd5 = []
                for _atta in page_attachments:
                    list_fileMd5.append(_atta.get(document_tmp_attachment_path_filemd5))

                list_attach = self.getAttachments(list_fileMd5,_dochtmlcon)

                newitem = {"item":item,"list_attach":list_attach,"message_id":message_id,"conn":conn}


            log("attachment get doc:%s"%(str(newitem.get("item",{}).get("docid"))))
            self.attachment_recognize(newitem,None)

            log("attachment get doc:%s succeed"%(str(newitem.get("item",{}).get("docid"))))

        except Exception as e:
            traceback.print_exc()

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

                _filemd5 = _attach.getProperties().get(attachment_filemd5)
                if _attach.getProperties().get(attachment_status) in (ATTACHMENT_PROCESSED,ATTACHMENT_TOOLARGE):
                    log("%s has processed or toolarge"%(_filemd5))
                    _html = _attach.getProperties().get(attachment_attachmenthtml,"")
                    if _html is None:
                        _html = ""

                    list_html.append({attachment_filemd5:_filemd5,
                                      "html":_html})
                else:
                    #has process_time then jump
                    if len(str(_attach.getProperties().get(attachment_process_time,"")))>10 and _attach.getProperties().get(attachment_status)!=ATTACHMENT_INIT and not (_attach.getProperties().get(attachment_status)>=ATTACHMENT_MC_FAILED_FROM and _attach.getProperties().get(attachment_status)<=ATTACHMENT_MC_FAILED_TO):
                        log("%s has process_time jump"%(_filemd5))
                        _html = _attach.getProperties().get(attachment_attachmenthtml,"")
                        if _html is None:
                            _html = ""

                        list_html.append({attachment_filemd5:_filemd5,
                                          "html":_html})
                    else:
                        log("%s requesting interface"%(_filemd5))
                        _succeed = self.request_attachment_interface(_attach,_dochtmlcon)
                        if not _succeed:
                            _not_failed = False

                        _html = _attach.getProperties().get(attachment_attachmenthtml,"")
                        if _html is None:
                            _html = ""
                        list_html.append({attachment_filemd5:_filemd5,
                                          "html":_html})

                if _attach.getProperties().get(attachment_filetype)=="swf":
                    # swf_urls.extend(json.loads(_attach.getProperties().get(attachment_swfUrls,"[]")))
                    _swf_urls = _attach.getProperties().get(attachment_swfUrls, "[]")
                    if _swf_urls:
                        _swf_urls = _swf_urls.replace('\\', '')
                    else:
                        _swf_urls = '[]'
                    _swf_urls = json.loads(_swf_urls)
                    swf_urls.extend(_swf_urls)

            if not _not_failed:
                return False,list_html,swf_urls
            return True,list_html,swf_urls

        except requests.ConnectionError as e1:
            raise e1
        except Exception as e:
            return False,list_html,swf_urls

    def attachment_recognize(self,_dict,result_queue):
        '''
        识别附件内容
        :param _dict: 附件内容
        :param result_queue:
        :return:
        '''

        try:
            item = _dict.get("item")
            list_attach = _dict.get("list_attach")
            conn = _dict["conn"]
            message_id = _dict.get("message_id")

            _retry_times = item.get("retry_times",0)
            dhtml = Document_html({"partitionkey":item.get("partitionkey"),
                                   "docid":item.get("docid")})

            _dochtmlcon = item.get(document_tmp_dochtmlcon,"")
            dhtml.setValue(document_tmp_dochtmlcon,_dochtmlcon,True)
            dhtml.delete_bidi_a()
            dtmp = Document_tmp(item)


            start_time = time.time()
            #调用识别接口
            _succeed,list_html,swf_urls = self.rec_attachments_by_interface(list_attach,_dochtmlcon,save=True)



            _to_ack = False
            if not _succeed and _retry_times<self.retry_times:
                item[document_tmp_status] = random.randint(*flow_attachment_status_failed_to)
                item["retry_times"] = _retry_times+1
                #失败次数大于5次就放入失败队列，此队列的数据会在空闲时间重新处理一次
                if item["retry_times"]>=self.retry_times:
                    send_msg_toacmq(self.pool_mq,json.dumps(item,cls=MyEncoder,ensure_ascii=False),self.mq_attachment_failed)

                send_succeed = send_msg_toacmq(self.pool_mq,json.dumps(item,cls=MyEncoder,ensure_ascii=False),self.mq_attachment)

                #失败保存
                if _retry_times==0:
                    dtmp.setValue(document_tmp_dochtmlcon,"",False)
                    dtmp.setValue(document_tmp_status,0,True)
                    if not dtmp.exists_row(self.ots_client):
                        dtmp.update_row(self.ots_client)
                        dhtml.update_row(self.ots_client)
                if send_succeed:
                    _to_ack = True

            else:
                try:
                    log("docid:%d,retry:%d swf_urls:%s list_html:%s"%(dhtml.getProperties().get(document_docid),_retry_times,str(swf_urls),str(len(list_html))))
                    dhtml.updateSWFImages(swf_urls)
                    dhtml.updateAttachment(list_html)

                    dtmp.setValue(document_tmp_attachment_extract_status,1,True)
                    dtmp.setValue(document_tmp_dochtmlcon,dhtml.getProperties().get(document_tmp_dochtmlcon),True)
                    send_succeed = send_msg_toacmq(self.pool_mq,json.dumps(dtmp.getProperties(),cls=MyEncoder),self.mq_extract)
                    if send_succeed:
                        _to_ack = True
                except Exception as e:
                    traceback.print_exc()

            if _to_ack:
                ackMsg(conn,message_id)
            log("document:%d get attachments with result:%s %s retry_times:%d"%(item.get("docid"),str(_succeed),str(_to_ack),_retry_times))
        except Exception as e:

            traceback.print_exc()
            if time.time()-start_time<10:
                item["retry_times"] -= 1
            if send_msg_toacmq(self.pool_mq,json.dumps(item,cls=MyEncoder,ensure_ascii=False),self.mq_attachment):
                ackMsg(conn,message_id)



    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)
        _path = attach.getProperties().get(attachment_path)
        _uuid = uuid4()
        objectPath = attach.getProperties().get(attachment_path)
        docids = attach.getProperties().get(attachment_docids)

        if objectPath is None:
            relative_path = "%s/%s"%(_uuid.hex[:4],_uuid.hex)
        else:
            relative_path = objectPath[5:].replace("//","/")
        localpath = "/FileInfo/%s"%(relative_path)
        if not os.path.exists(localpath):
            if not os.path.exists(os.path.dirname(localpath)):
                os.makedirs(os.path.dirname(localpath))
            local_exists = False
        else:
            local_exists = True
            _size = os.path.getsize(localpath)

        not_failed_flag = True
        try:

            d_start_time = time.time()
            if not local_exists:
                log("md5:%s path:%s not exists,start downloading"%(filemd5,objectPath))
                try:
                    download_succeed = downloadFile(self.bucket,objectPath,localpath)
                except Exception as e:
                    download_succeed = False
            else:
                log("md5:%s path:%s exists"%(filemd5,objectPath[5:]))
            if not (local_exists or download_succeed):
                _ots_attach = attachment(attach.getProperties_ots())
                _ots_exists = _ots_attach.fix_columns(self.ots_client,[attachment_attachmenthtml,attachment_classification,attachment_attachmentcon,attachment_path,attachment_status,attachment_filetype],True)
                log("md5:%s path:%s file not in local or oss,search ots.attachment"%(filemd5,objectPath))
                if _ots_attach.getProperties().get(attachment_attachmenthtml,"")!="" and str(_ots_attach.getProperties().get(attachment_status))!=str(ATTACHMENT_INIT):
                    attach.setValue(attachment_attachmenthtml,_ots_attach.getProperties().get(attachment_attachmenthtml,""))
                    attach.setValue(attachment_attachmentcon,_ots_attach.getProperties().get(attachment_attachmentcon,""))
                    attach.setValue(attachment_status,_ots_attach.getProperties().get(attachment_status,""))
                    attach.setValue(attachment_filetype,_ots_attach.getProperties().get(attachment_filetype,""))
                    attach.setValue(attachment_classification,_ots_attach.getProperties().get(attachment_classification,""))
                    # if attach.exists(self.attach_pool):
                    #     attach.update_row(self.attach_pool)
                    # else:
                    #     attach.insert_row(self.attach_pool)
                    self.putAttach_json_toRedis(filemd5,attach.getProperties())


                    try:
                        if os.exists(localpath):
                            os.remove(localpath)
                    except Exception as e:
                        pass
                    return True
                if _ots_exists:
                    objectPath = attach.getProperties().get(attachment_path)
                    download_succeed = downloadFile(self.bucket,objectPath,localpath)
                    if download_succeed:
                        log("md5:%s path:%s download file from oss succeed"%(filemd5,objectPath))
                    else:
                        log("md5:%s path:%s download file from ots failed=="%(filemd5,objectPath))
                else:
                    log("md5:%s path:%s not found in ots"%(filemd5,objectPath))
            if local_exists or download_succeed:
                _size = os.path.getsize(localpath)
                attach.setValue(attachment_size,_size,True)

                if _size>ATTACHMENT_LARGESIZE:
                    attach.setValue(attachment_status, ATTACHMENT_TOOLARGE,True)
                    log("attachment :%s of path:%s to large"%(filemd5,_path))
                    _ots_attach = attachment(attach.getProperties_ots())
                    _ots_attach.update_row(self.ots_client)

                    # #更新postgres
                    # if attach.exists(self.attach_pool):
                    #     attach.update_row(self.attach_pool)
                    # else:
                    #     attach.insert_row(self.attach_pool)
                    self.putAttach_json_toRedis(filemd5,attach.getProperties())


                    if local_exists:
                        upload_status = uploadFileByPath(self.bucket,localpath,objectPath)
                    os.remove(localpath)

                    return True

                time_download = time.time()-d_start_time

                #调用接口处理结果
                start_time = time.time()
                _filetype = attach.getProperties().get(attachment_filetype)

                # _data_base64 = base64.b64encode(open(localpath,"rb").read())
                # _success,_html,swf_images = getAttachDealInterface(_data_base64,_filetype)
                _success,_html,swf_images,classification = getAttachDealInterface(None,_filetype,path=localpath,session=self.session)
                log("process filemd5:%s %s of type:%s with size:%.3fM download:%ds recognize takes %ds,ret_size:%d"%(filemd5,str(_success),_filetype,round(_size/1024/1024,4),time_download,time.time()-start_time,len(_html)))
                if _success:
                    if len(_html)<5:
                        _html = ""
                else:
                    if len(_html)>1:
                        _html = "interface return error"
                    else:
                        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时，删除原来的swf_urls中的"\"
                if attach.getProperties().get(attachment_filetype) == "swf":
                    swf_urls = attach.getProperties().get(attachment_swfUrls, "[]")
                    swf_urls = swf_urls.replace('\\', '') if swf_urls else '[]'
                    swf_urls = json.loads(swf_urls)
                    attach.setValue(attachment_swfUrls, json.dumps(swf_urls, ensure_ascii=False), True)

                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,"")
                        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)
                    else:
                        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.setValue(attachment_classification,classification,True)


                #更新ots
                _ots_attach = attachment(attach.getProperties_ots())
                _ots_attach.update_row(self.ots_client) #线上再开放更新

                # #更新postgres
                # if attach.exists(self.attach_pool):
                #     attach.update_row(self.attach_pool)
                # else:
                #     attach.insert_row(self.attach_pool)
                self.putAttach_json_toRedis(filemd5,attach.getProperties())


                if local_exists:
                    upload_status = uploadFileByPath(self.bucket,localpath,objectPath)
                try:
                    if upload_status and os.exists(localpath):
                        os.remove(localpath)
                except Exception as e:
                    pass


                return True
            else:
                return True

        except requests.ConnectionError as e1:
            raise e1

        except oss2.exceptions.NotFound as e:
            return True

        except Exception as e:
            traceback.print_exc()



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

    def getAttachPath(self,filemd5,_dochtmlcon):
        _soup = BeautifulSoup(_dochtmlcon,"lxml")

        list_mark = ["data","filelink"]
        for _mark in list_mark:
            _find = _soup.find("a",attrs={_mark:filemd5})
            filelink = ""
            if _find is None:
                _find = _soup.find("img",attrs={_mark:filemd5})
                if _find is not None:
                    filelink = _find.attrs.get("src","")
            else:
                filelink = _find.attrs.get("href","")
            if filelink.find("bidizhaobiao")>=0:
                _path = filelink.split("/file")
                if len(_path)>1:
                    return _path[1]


    def getAttach_json_fromRedis(self,filemd5):
        db = self.redis_pool.getConnector()
        try:

            _key = "attach-%s"%(filemd5)
            _attach_json = db.get(_key)
            return _attach_json
        except Exception as e:
            log("getAttach_json_fromRedis error %s"%(str(e)))
        finally:
            try:
                if db.connection.check_health():
                    self.redis_pool.putConnector(db)
            except Exception as e:
                pass
        return None

    def putAttach_json_toRedis(self,filemd5,extract_dict):

        db = self.redis_pool.getConnector()
        try:
            new_dict = {}
            for k,v in extract_dict.items():
                if not isinstance(v,set):
                    new_dict[k] = v
            _key = "attach-%s"%(filemd5)
            _extract_json = db.set(str(_key),json.dumps(new_dict))
            db.expire(_key,3600*3)
            return _extract_json
        except Exception as e:
            log("putExtract_json_toRedis error%s"%(str(e)))
            traceback.print_exc()
        finally:
            try:
                if db.connection.check_health():
                    self.redis_pool.putConnector(db)
            except Exception as e:
                pass



    def getAttachments(self,list_filemd5,_dochtmlcon):
        conn = self.attach_pool.getConnector()
        #搜索postgres
        try:
            to_find_md5 = []
            for _filemd5 in list_filemd5[:50]:
                if _filemd5 is not None:
                    to_find_md5.append(_filemd5)

            conditions = ["filemd5 in ('%s')"%("','".join(to_find_md5))]
            list_attachment = []

            set_md5 = set()
            # list_attachment =  Attachment_postgres.select_rows(conn,Attachment_postgres,"attachment",conditions)
            # for _attach in list_attachment:
            #     set_md5.add(_attach.getProperties().get(attachment_filemd5))

            for _filemd5 in to_find_md5:
                _json = self.getAttach_json_fromRedis(_filemd5)

                if _json is not None:
                    set_md5.add(_filemd5)
                    list_attachment.append(Attachment_postgres(json.loads(_json)))

            log("select localpath database %d/%d"%(len(set_md5),len(to_find_md5)))

            for _filemd5 in to_find_md5:

                if _filemd5 not in set_md5:

                    _path = self.getAttachPath(_filemd5,_dochtmlcon)


                    log("getAttachments search in ots:%s"%(_filemd5))
                    _attach = {attachment_filemd5:_filemd5}
                    _attach_ots = attachment(_attach)
                    if _attach_ots.fix_columns(self.ots_client,[attachment_status,attachment_path,attachment_attachmenthtml,attachment_attachmentcon,attachment_filetype,attachment_swfUrls,attachment_process_time],True):
                        if _attach_ots.getProperties().get(attachment_status) is not None:
                            log("getAttachments find in ots:%s"%(_filemd5))
                            list_attachment.append(Attachment_postgres(_attach_ots.getProperties()))
                    else:
                        log("getAttachments search in path:%s"%(_filemd5))
                        if _path:
                            log("getAttachments find in path:%s"%(_filemd5))
                            if _path[0]=="/":
                                _path = _path[1:]
                            _filetype = _path.split(".")[-1]
                            _attach = {attachment_filemd5:_filemd5,
                                       attachment_filetype:_filetype,
                                       attachment_status:20,
                                       attachment_path:"%s/%s"%(_filemd5[:4],_path),
                                       attachment_crtime:getCurrent_date(format="%Y-%m-%d %H:%M:%S")}
                            list_attachment.append(Attachment_postgres(_attach))




            return list_attachment
        except Exception as e:
            log("attachProcess comsumer error %s"%str(e))
            log(str(to_find_md5))
            traceback.print_exc()
            return []
        finally:
            self.attach_pool.putConnector(conn)

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

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

        log("queue_attachment:%d,queue_attachment_ocr:%d,queue_attachment_not_ocr:%d"%(q_size,qsize_ocr,qsize_not_ocr))


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

    def flow_attachment_process(self):
        self.process_comsumer()

        # p = Process(target = self.process_comsumer)
        # p.start()
        # p.join()


    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)
        else:
            self.queue_attachment_not_ocr.put(_dict,True)

    def comsumer_handle(self,_dict,result_queue):
        try:
            frame = _dict["frame"]
            conn = _dict["conn"]
            message_id = frame.headers["message-id"]
            item = json.loads(frame.body)
            page_attachments = json.loads(item.get(document_tmp_attachment_path,"[]"))
            _dochtmlcon = item.get(document_tmp_dochtmlcon,"")

            if len(page_attachments)==0:
                self.set_queue({"item":item,"list_attach":[],"message_id":message_id,"conn":conn})
            else:
                list_fileMd5 = []
                for _atta in page_attachments:
                    list_fileMd5.append(_atta.get(document_tmp_attachment_path_filemd5))

                list_attach = self.getAttachments(list_fileMd5,_dochtmlcon)

                self.set_queue({"item":item,"list_attach":list_attach,"message_id":message_id,"conn":conn})
        except Exception as e:
            traceback.print_exc()

    def remove_attachment_postgres(self):
        current_date = getCurrent_date(format="%Y-%m-%d")
        last_date = timeAdd(current_date,-2,format="%Y-%m-%d")
        sql = " delete from attachment where crtime<='%s 00:00:00' "%(last_date)
        conn = self.attach_pool.getConnector()
        try:
            cursor = conn.cursor()
            cursor.execute(sql)
            conn.commit()
            self.attach_pool.putConnector(conn)
        except Exception as e:
            conn.close()



    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.add_job(self.flow_attachment_producer,"cron",second="*/10")
        # schedule.add_job(self.flow_attachment_producer_comsumer,"cron",second="*/10")
        # schedule.add_job(self.monitor_listener,"cron",minute="*/1")


        # schedule.add_job(self.monitor_attachment_process,"cron",second="*/10")
        # schedule.add_job(self.remove_attachment_postgres,"cron",hour="6")
        schedule.add_job(self.process_failed_attachment,"cron",minute="*/10")
        schedule.start()

class Dataflow_ActivteMQ_extract(Dataflow_extract):

    class ExtractListener():

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

        def on_message(self, headers):
            try:
                log("get message of idx:%d"%(self._idx))
                message_id = headers.headers["message-id"]
                body = headers.body
                log("get message %s crtime:%s"%(message_id,json.loads(body).get("crtime","")))
                self._func(_dict={"frame":headers,"conn":self.conn},result_queue=None)
            except Exception as e:
                traceback.print_exc()
                pass

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

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

    def __init__(self,create_listener=True):
        Dataflow_extract.__init__(self)

        self.industy_url = "http://127.0.0.1:15000/industry_extract"

        self.extract_interfaces = [["http://127.0.0.1:15030/content_extract",20],
                                   ["http://192.168.0.115:15030/content_extract",10]
                                   ]



        self.mq_extract = "/queue/dataflow_extract"
        self.mq_extract_failed = "/queue/dataflow_extract_failed"

        self.whole_weight = 0
        for _url,weight in self.extract_interfaces:
            self.whole_weight+= weight
        current_weight = 0
        for _i in range(len(self.extract_interfaces)):
            current_weight += self.extract_interfaces[_i][1]
            self.extract_interfaces[_i][1] = current_weight/self.whole_weight

        self.comsumer_count = 5
        # self.pool_postgres = ConnectorPool(10,self.comsumer_count,getConnection_postgres)
        self.pool_redis_doc = ConnectorPool(1,self.comsumer_count,getConnect_redis_doc)
        self.conn_mq = getConnect_activateMQ()
        self.pool_mq = ConnectorPool(1,30,getConnect_activateMQ)
        self.block_url = RLock()
        self.url_count = 0
        self.session = None


        self.list_extract_comsumer = []

        # for _i in range(self.comsumer_count):
        #     listener_extract = self.ExtractListener(getConnect_activateMQ(),self.comsumer_handle,_i)
        #     createComsumer(listener_extract,self.mq_extract)
        #     self.list_extract_comsumer.append(listener_extract)

        if create_listener:
            for ii in range(10):
                listener_p = Process(target=self.start_extract_listener)
                listener_p.start()

    def start_extract_listener(self):

        self.list_extract_comsumer = []

        for _i in range(self.comsumer_count):
            listener_extract = self.ExtractListener(getConnect_activateMQ(),self.comsumer_handle,_i)
            createComsumer(listener_extract,self.mq_extract)
            self.list_extract_comsumer.append(listener_extract)

        while 1:
            for _i in range(len(self.list_extract_comsumer)):
                if self.list_extract_comsumer[_i].conn.is_connected():
                    continue
                else:
                    listener = self.ExtractListener(getConnect_activateMQ(),self.comsumer_handle,_i)
                    createComsumer(listener,self.mq_extract)
                    self.list_extract_comsumer[_i] = listener
            time.sleep(5)

    def monitor_listener(self):
        for i in range(len(self.list_extract_comsumer)):
            if self.list_extract_comsumer[i].conn.is_connected():
                continue
            else:
                listener = self.ExtractListener(getConnect_activateMQ(),self.comsumer_handle)
                createComsumer(listener,self.mq_extract)
                self.list_extract_comsumer[i] = listener

    def getExtract_url(self):
        # _url_num = 0
        # with self.block_url:
        #     self.url_count += 1
        #     self.url_count %= self.whole_weight
        #     _url_num = self.url_count

        _r = random.random()
        # _r = _url_num/self.whole_weight
        for _i in range(len(self.extract_interfaces)):
            if _r<=self.extract_interfaces[_i][1]:
                return self.extract_interfaces[_i][0]

    def request_extract_interface(self,json,headers):
        # _i = random.randint(0,len(self.extract_interfaces)-1)
        # _i = 0
        # _url = self.extract_interfaces[_i]
        _url = self.getExtract_url()
        log("extract_url:%s"%(str(_url)))
        with requests.Session() as session:
            resp = session.post(_url,json=json,headers=headers,timeout=10*60)
        return resp


    def request_industry_interface(self,json,headers):
        resp = requests.post(self.industy_url,json=json,headers=headers)
        return resp


    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()
        log("queue extract size:%d"%(q_size))

    def process_extract_failed(self):
        def _handle(_dict,result_queue):
            frame = _dict.get("frame")
            message_id = frame.headers["message-id"]
            subscription = frame.headers.setdefault('subscription', None)
            conn = _dict.get("conn")
            body = frame.body
            if body is not None:
                item = json.loads(body)
                item["extract_times"] = 10
                if send_msg_toacmq(self.pool_mq,json.dumps(item,ensure_ascii=False),self.mq_extract):
                    ackMsg(conn,message_id,subscription)



        from BaseDataMaintenance.java.MQInfo import getQueueSize
        try:
            extract_failed_size = getQueueSize("dataflow_extract_failed")
            extract_size = getQueueSize("dataflow_extract")
            log("extract_failed_size %s extract_size %s"%(str(extract_failed_size),str(extract_size)))
            if extract_failed_size>0 and extract_size<100:
                failed_listener = self.ExtractListener(getConnect_activateMQ(),_handle,1)
                createComsumer(failed_listener,self.mq_extract_failed)
                while 1:
                    extract_failed_size = getQueueSize("dataflow_extract_failed")
                    if extract_failed_size==0:
                        break
                    time.sleep(10)
                failed_listener.conn.disconnect()
        except Exception as e:
            traceback.print_exc()

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

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


    def getExtract_json_fromDB(self,_fingerprint):
        conn = self.pool_postgres.getConnector()
        try:
            list_extract = Document_extract_postgres.select_rows(conn,Document_extract_postgres,"document_extract",[" fingerprint='%s'"%_fingerprint])
            if len(list_extract)>0:
                _extract = list_extract[0]
                return _extract.getProperties().get(document_extract_extract_json)
        except Exception as e:
            traceback.print_exc()
        finally:
            self.pool_postgres.putConnector(conn)
        return None

    def putExtract_json_toDB(self,fingerprint,docid,extract_json):
        _de = Document_extract_postgres({document_extract_fingerprint:fingerprint,
                                         document_extract_docid:docid,
                                         document_extract_extract_json:extract_json})
        _de.insert_row(self.pool_postgres,1)

    def getExtract_json_fromRedis(self,_fingerprint):
        db = self.pool_redis_doc.getConnector()
        try:
            _extract_json = db.get(_fingerprint)
            return _extract_json
        except Exception as e:
            log("getExtract_json_fromRedis error %s"%(str(e)))
        finally:
            try:
                if db.connection.check_health():
                    self.pool_redis_doc.putConnector(db)
            except Exception as e:
                pass
        return None

    def putExtract_json_toRedis(self,fingerprint,extract_json):

        db = self.pool_redis_doc.getConnector()
        try:
            _extract_json = db.set(str(fingerprint),extract_json)
            db.expire(fingerprint,3600*2)
            return _extract_json
        except Exception as e:
            log("putExtract_json_toRedis error%s"%(str(e)))
            traceback.print_exc()
        finally:
            try:

                if db.connection.check_health():
                    self.pool_redis_doc.putConnector(db)
            except Exception as e:
                pass




    def comsumer_handle(self,_dict,result_queue):
        try:
            log("start handle")
            frame = _dict["frame"]
            conn = _dict["conn"]
            message_id = frame.headers["message-id"]
            subscription = frame.headers.setdefault('subscription', None)
            item = json.loads(frame.body)
            dtmp = Document_tmp(item)

            dhtml = Document_html({"partitionkey":item.get("partitionkey"),
                                   "docid":item.get("docid")})

            extract_times = item.get("extract_times",0)+1
            item["extract_times"] = extract_times

            _dochtmlcon = item.get(document_tmp_dochtmlcon,"")

            html_len = len(_dochtmlcon)
            if html_len>50000:
                log("docid %s dochtmlcon too long len %d "%(str(item.get("docid")),html_len))
                try:
                    _dochtmlcon = re.sub("<html>|</html>|<body>|</body>", "", _dochtmlcon)
                    _soup = BeautifulSoup(_dochtmlcon,"html5lib")
                    if len(_dochtmlcon)>200000:
                        _find = _soup.find("div",attrs={"class":"richTextFetch"})
                        if _find is not None:
                            _find.decompose()
                    else:
                        _soup = article_limit(_soup,50000)
                    _dochtmlcon = str(_soup)
                except Exception as e:
                    traceback.print_exc()
                    ackMsg(conn,message_id,subscription)
                    return
                log("docid %s len %d limit to %d"%(str(item.get("docid")),html_len,len(_dochtmlcon)))


            dhtml.setValue(document_tmp_dochtmlcon,_dochtmlcon,True)
            _extract = Document_extract({})
            _extract.setValue(document_extract2_partitionkey,item.get(document_partitionkey))
            _extract.setValue(document_extract2_docid,item.get(document_docid))
            all_done = 1

            data = {}
            for k,v in item.items():
                data[k] = v
            data["timeout"] = 440
            data["doc_id"] = data.get(document_tmp_docid,0)
            # if data["docid"]<298986054 and data["docid"]>0:
            #     log("jump docid %s"%(str(data["docid"])))
            #     ackMsg(conn,message_id,subscription)
            #     return
            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["web_source_name"] = item.get(document_tmp_web_source_name,"")
            data["original_docchannel"] = item.get(document_tmp_original_docchannel,"")

            _fingerprint = getFingerprint(str(data["title"])+str(data["content"]))

            if all_done>0:
                _time = time.time()
                # extract_json = self.getExtract_json_fromDB(_fingerprint)
                extract_json = self.getExtract_json_fromRedis(_fingerprint)
                log("get json from db takes %.4f"%(time.time()-_time))
                # extract_json = None
                _docid = int(data["doc_id"])
                if extract_json is not None:
                    log("fingerprint %s exists docid:%s"%(_fingerprint,str(_docid)))
                    _extract.setValue(document_extract2_extract_json,extract_json,True)
                else:
                    resp = self.request_extract_interface(json=data,headers=self.header)
                    if (resp.status_code >=200 and resp.status_code<=213):
                        extract_json = resp.content.decode("utf8")
                        _extract.setValue(document_extract2_extract_json,extract_json,True)
                        _time = time.time()
                        # self.putExtract_json_toDB(_fingerprint,_docid,extract_json)
                        self.putExtract_json_toRedis(_fingerprint,extract_json)
                        log("get json to db takes %.4f"%(time.time()-_time))
                    else:
                        log("%s--%s"%(str(resp.status_code),resp.content.decode("utf8")))
                        all_done = -2
            # if all_done>0:
            #     resp = self.request_industry_interface(json=data,headers=self.header)
            #     if (resp.status_code >=200 and resp.status_code<=213):
            #         _extract.setValue(document_extract2_industry_json,resp.content.decode("utf8"),True)
            #     else:
            #         log("%s--%s"%(str(resp.status_code),resp.content.decode("utf8")))
            #         all_done = -3
            # _to_ack = False
            # if all_done>0 and len(_extract.getProperties().get(document_extract2_extract_json,""))<=2:
            #     all_done = -4
            _extract.setValue(document_extract2_industry_json,"{}",True)
            try:
                if all_done!=1:
                    sentMsgToDD("要素提取失败：docid:%d with result:%d"%(item.get(document_tmp_docid),all_done))
                    if extract_times>=10:
                        #process as succeed
                        dtmp.setValue(document_tmp_dochtmlcon,"",False)
                        dtmp.setValue(document_tmp_status,random.randint(*flow_extract_status_succeed_to),True)
                        dtmp.update_row(self.ots_client)
                        dhtml.update_row(self.ots_client)

                        #replace as {}
                        _extract.setValue(document_extract2_extract_json,"{}",True)
                        _extract.setValue(document_extract2_industry_json,"{}",True)
                        _extract.setValue(document_extract2_status,random.randint(1,50),True)
                        _extract.update_row(self.ots_client)
                        _to_ack = True
                    elif extract_times>5:
                        #transform to the extract_failed queue
                        if send_msg_toacmq(self.pool_mq,json.dumps(item,ensure_ascii=False),self.mq_extract_failed):
                            #process as succeed
                            dtmp.setValue(document_tmp_dochtmlcon,"",False)
                            dtmp.setValue(document_tmp_status,random.randint(*flow_extract_status_succeed_to),True)
                            dtmp.update_row(self.ots_client)
                            dhtml.update_row(self.ots_client)

                            #replace as {}
                            _extract.setValue(document_extract2_extract_json,"{}",True)
                            _extract.setValue(document_extract2_industry_json,"{}",True)
                            _extract.setValue(document_extract2_status,random.randint(1,50),True)
                            _extract.update_row(self.ots_client)
                            _to_ack = True
                    else:

                        send_succeed = send_msg_toacmq(self.pool_mq,json.dumps(item,ensure_ascii=False),self.mq_extract)

                        #失败保存
                        dtmp.setValue(document_tmp_dochtmlcon,"",False)
                        dtmp.setValue(document_tmp_status,60,True)
                        if not dtmp.exists_row(self.ots_client):
                            dtmp.update_row(self.ots_client)
                            dhtml.update_row(self.ots_client)
                        if send_succeed:
                            _to_ack = True
                else:
                    #process succeed
                    dtmp.setValue(document_tmp_dochtmlcon,"",False)
                    dtmp.setValue(document_tmp_status,random.randint(*flow_extract_status_succeed_to),True)
                    if _docid==290816703:
                        dtmp.setValue("test_json",extract_json,True)
                    dtmp.update_row(self.ots_client)
                    dhtml.update_row(self.ots_client)
                    _extract.setValue(document_extract2_status,random.randint(1,50),True)

                    _extract.update_row(self.ots_client)
                    _to_ack = True
            except Exception:
                traceback.print_exc()

            if _to_ack:
                ackMsg(conn,message_id,subscription)
            log("process %s docid:%d %s"%(str(_to_ack),data["doc_id"],str(all_done)))
        except requests.ConnectionError as e1:
            item["extract_times"] -= 1
            if send_msg_toacmq(self.pool_mq,json.dumps(item,ensure_ascii=False),self.mq_extract):
                ackMsg(conn,message_id,subscription)
        except Exception as e:
            traceback.print_exc()
            sentMsgToDD("要素提取失败：docid:%d with result:%s"%(item.get(document_tmp_docid),str(e)))
            log("process %s docid: failed message_id:%s"%(data["doc_id"],message_id))
            if extract_times>=10:
                #process as succeed
                dtmp.setValue(document_tmp_dochtmlcon,"",False)
                dtmp.setValue(document_tmp_status,random.randint(*flow_extract_status_succeed_to),True)
                dtmp.update_row(self.ots_client)
                dhtml.update_row(self.ots_client)
                #replace as {}
                _extract.setValue(document_extract2_extract_json,"{}",True)
                _extract.setValue(document_extract2_industry_json,"{}",True)
                _extract.setValue(document_extract2_status,random.randint(1,50),True)
                _extract.update_row(self.ots_client)
                ackMsg(conn,message_id,subscription)
            elif extract_times>5:
                #transform to the extract_failed queue
                if send_msg_toacmq(self.pool_mq,json.dumps(item,ensure_ascii=False),self.mq_extract_failed):
                    #process as succeed
                    dtmp.setValue(document_tmp_dochtmlcon,"",False)
                    dtmp.setValue(document_tmp_status,random.randint(*flow_extract_status_succeed_to),True)
                    dtmp.update_row(self.ots_client)
                    dhtml.update_row(self.ots_client)
                    #replace as {}
                    _extract.setValue(document_extract2_extract_json,"{}",True)
                    _extract.setValue(document_extract2_industry_json,"{}",True)
                    _extract.setValue(document_extract2_status,random.randint(1,50),True)
                    _extract.update_row(self.ots_client)
                    ackMsg(conn,message_id,subscription)
            else:
                #transform to the extract queue
                #失败保存
                dtmp.setValue(document_tmp_dochtmlcon,"",False)
                dtmp.setValue(document_tmp_status,60,True)
                if not dtmp.exists_row(self.ots_client):
                    dtmp.update_row(self.ots_client)
                    dhtml.update_row(self.ots_client)
                if send_msg_toacmq(self.pool_mq,json.dumps(item,ensure_ascii=False),self.mq_extract):
                    ackMsg(conn,message_id,subscription)


    def delete_document_extract(self,save_count=70*10000):
        conn = self.pool_postgres.getConnector()
        try:
            cursor = conn.cursor()
            sql = " select max(docid),min(docid) from document_extract "
            cursor.execute(sql)
            rows = cursor.fetchall()
            if len(rows)>0:
                maxdocid,mindocid = rows[0]

                d_mindocid = int(maxdocid)-save_count
                if mindocid<d_mindocid:
                    sql = " delete from document_extract where docid<%d"%d_mindocid
                    cursor.execute(sql)
                    conn.commit()

        except Exception as e:
            traceback.print_exc()
        finally:
            self.pool_postgres.putConnector(conn)

    def start_flow_extract(self):
        schedule = BlockingScheduler()
        schedule.add_job(self.flow_extract_producer,"cron",second="*/20")
        schedule.add_job(self.process_extract_failed,"cron",minute="*/5")
        # schedule.add_job(self.delete_document_extract,"cron",hour="*/5")
        # schedule.add_job(self.monitor_listener,"cron",minute="*/5")
        schedule.start()

from multiprocessing import RLock
docid_lock = RLock()

conn_mysql = None


def generateRangeDocid(nums):
    global conn_mysql
    while 1:
        try:
            with docid_lock:
                if conn_mysql is None:
                    conn_mysql = getConnection_mysql()
                cursor = conn_mysql.cursor()
                sql = "select serial_value from b2c_serial_no where serial_name='DocumentIdSerial'"
                cursor.execute(sql)
                rows = cursor.fetchall()
                current_docid = rows[0][0]
                next_docid = current_docid+1
                update_docid = current_docid+nums
                sql = " update b2c_serial_no set serial_value=%d where serial_name='DocumentIdSerial'"%(update_docid)
                cursor.execute(sql)
                conn_mysql.commit()
                return next_docid
        except Exception as e:
            conn_mysql = getConnection_mysql()


# 自定义jsonEncoder
class MyEncoder(json.JSONEncoder):
    def default(self, obj):
        if isinstance(obj, np.ndarray):
            return obj.tolist()
        elif isinstance(obj, bytes):
            return str(obj, encoding='utf-8')
        elif isinstance(obj, (np.float_, np.float16, np.float32,
                              np.float64,Decimal)):
            return float(obj)
        elif isinstance(obj,str):
            return obj
        return json.JSONEncoder.default(self, obj)


class Dataflow_init(Dataflow):

    class InitListener():

        def __init__(self,conn,*args,**kwargs):
            self.conn = conn
            self.get_count = 1000
            self.count = self.get_count
            self.begin_docid = None
            self.mq_init = "/queue/dataflow_init"
            self.mq_attachment = "/queue/dataflow_attachment"
            self.mq_extract = "/queue/dataflow_extract"
            self.pool_mq1 = ConnectorPool(1,4,getConnect_activateMQ)

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

        def getRangeDocid(self):
            begin_docid = generateRangeDocid(self.get_count)
            self.begin_docid = begin_docid
            self.count = 0

        def getNextDocid(self):
            if self.count>=self.get_count:
                self.getRangeDocid()
            next_docid = self.begin_docid+self.count
            self.count += 1
            return next_docid

        def on_message(self, headers):
            try:
                next_docid = int(self.getNextDocid())
                partitionkey = int(next_docid%500+1)
                message_id = headers.headers["message-id"]
                body = json.loads(headers.body)
                body[document_tmp_partitionkey] = partitionkey
                body[document_tmp_docid] = next_docid
                if body.get(document_original_docchannel) is None:
                    body[document_original_docchannel] = body.get(document_docchannel)
                page_attachments = body.get(document_tmp_attachment_path,"[]")
                _uuid = body.get(document_tmp_uuid,"")
                if page_attachments!="[]":
                    status = random.randint(1,10)
                    body[document_tmp_status] = status
                    if send_msg_toacmq(self.pool_mq1,json.dumps(body,cls=MyEncoder),self.mq_attachment):
                        log("uuid:%s with docid:%s"%(str(_uuid),str(next_docid)))
                        ackMsg(self.conn,message_id)
                    else:
                        log("send_msg_error on init listener")
                else:
                    status = random.randint(11,50)
                    body[document_tmp_status] = status
                    if send_msg_toacmq(self.pool_mq1,json.dumps(body,cls=MyEncoder),self.mq_extract):
                        log("uuid:%s with docid:%s"%(str(_uuid),str(next_docid)))
                        ackMsg(self.conn,message_id)
                    else:
                        log("send_msg_error on init listener")
            except Exception as e:
                traceback.print_exc()
                if send_msg_toacmq(self.pool_mq1,json.dumps(body,cls=MyEncoder),self.mq_init):
                    log("init error")
                    ackMsg(self.conn,message_id)

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

    def __init__(self):
        Dataflow.__init__(self)
        self.max_shenpi_id = None
        self.base_shenpi_id = 400000000000
        self.mq_init = "/queue/dataflow_init"

        self.mq_attachment = "/queue/dataflow_attachment"
        self.mq_extract = "/queue/dataflow_extract"
        self.pool_oracle = ConnectorPool(10,15,getConnection_oracle)
        self.pool_mq = ConnectorPool(10,30,getConnect_activateMQ)

        self.ots_capacity = getConnect_ots_capacity()

        self.init_comsumer_counts = 2
        self.list_init_comsumer = []
        for i in range(self.init_comsumer_counts):
            listener = self.InitListener(getConnect_activateMQ())
            createComsumer(listener,self.mq_init)
            self.list_init_comsumer.append(listener)


    def monitor_listener(self):
        for i in range(len(self.list_init_comsumer)):
            if self.list_init_comsumer[i].conn.is_connected():
                continue
            else:
                listener = self.InitListener(getConnect_activateMQ())
                createComsumer(listener,self.mq_init)
                self.list_init_comsumer[i] = listener



    def temp2mq(self,object):
        conn_oracle = self.pool_oracle.getConnector()

        try:
            list_obj = object.select_rows(conn_oracle,type(object),object.table_name,[],limit=1000)
            for _obj in list_obj:
                ots_dict = _obj.getProperties_ots()

                if len(ots_dict.get("dochtmlcon",""))>500000:
                    _obj.delete_row(conn_oracle)
                    log("msg too long:%s,%d"%(ots_dict.get("uuid"),len(ots_dict.get("dochtmlcon",""))))
                    continue

                if send_msg_toacmq(self.pool_mq,json.dumps(ots_dict,cls=MyEncoder),self.mq_init):
                    #删除数据，上线放开
                    _obj.delete_row(conn_oracle)
                else:
                    log("send_msg_error111:%s,%d"%(ots_dict.get("uuid"),len(ots_dict.get("dochtmlcon",""))))
            self.pool_oracle.putConnector(conn_oracle)
        except Exception as e:
            traceback.print_exc()
            self.pool_oracle.decrease()

    def shengpi2mq(self):

        conn_oracle = self.pool_oracle.getConnector()

        try:
            if self.max_shenpi_id is None:
                # get the max_shenpi_id
                _query = BoolQuery(must_queries=[ExistsQuery("id")])
                rows,next_token,total_count,is_all_succeed = self.ots_client.search("t_shen_pi_xiang_mu","t_shen_pi_xiang_mu_index",
                                                                                    SearchQuery(_query,sort=Sort(sorters=[FieldSort("id",SortOrder.DESC)]),limit=1))
                list_data = getRow_ots(rows)
                if len(list_data)>0:
                    max_shenpi_id = list_data[0].get("id")
                    if max_shenpi_id>self.base_shenpi_id:
                        max_shenpi_id -= self.base_shenpi_id
                    self.max_shenpi_id = max_shenpi_id
            if self.max_shenpi_id is not None:
                # select data in order
                list_data = T_SHEN_PI_XIANG_MU.select_rows(conn_oracle,self.max_shenpi_id,)

                # send data to mq one by one with max_shenpi_id updated
                for _data in list_data:
                    _id = _data.getProperties().get(T_SHEN_PI_XIANG_MU_ID)

                    ots_dict = _data.getProperties_ots()
                    if ots_dict["docid"]<self.base_shenpi_id:
                        ots_dict["docid"] += self.base_shenpi_id

                    if ots_dict.get(T_SHEN_PI_XIANG_MU_PAGE_ATTACHMENTS,"") !='[]':
                        if send_msg_toacmq(self.pool_mq,json.dumps(ots_dict,cls=MyEncoder),self.mq_attachment):
                            self.max_shenpi_id = _id
                        else:
                            log("sent shenpi message to mq failed %s"%(_id))
                            break
                    else:
                        if send_msg_toacmq(self.pool_mq,json.dumps(ots_dict,cls=MyEncoder),self.mq_extract):
                            self.max_shenpi_id = _id
                        else:
                            log("sent shenpi message to mq failed %s"%(_id))
                            break

        except Exception as e:
            traceback.print_exc()
            self.pool_oracle.decrease()



    def ots2mq(self):
        try:
            bool_query = BoolQuery(must_queries=[RangeQuery("status",1,51)])

            rows,next_token,total_count,is_all_succeed = self.ots_client.search("document","document_index",
                                                                                SearchQuery(bool_query,sort=Sort(sorters=[FieldSort(document_docid)]),get_total_count=True,limit=100),
                                                                                ColumnsToGet(return_type=ColumnReturnType.ALL))
            list_data = getRow_ots(rows)
            for _data in list_data:
                _d = {document_tmp_partitionkey:_data.get(document_tmp_partitionkey),
                      document_tmp_docid:_data.get(document_tmp_docid),
                      document_tmp_status:0}
                _document = Document(_d)
                page_attachments = _data.get(document_tmp_attachment_path,"[]")

                _document_html = Document(_data)
                _document_html.fix_columns(self.ots_capacity,[document_tmp_dochtmlcon],True)

                if page_attachments!="[]":
                    status = random.randint(1,10)
                    _data[document_tmp_status] = status
                    send_succeed = send_msg_toacmq(self.pool_mq,json.dumps(_document_html.getProperties(),cls=MyEncoder),self.mq_attachment)
                else:
                    status = random.randint(11,50)
                    _data[document_tmp_status] = status
                    send_succeed = send_msg_toacmq(self.pool_mq,json.dumps(_document_html.getProperties(),cls=MyEncoder),self.mq_extract)
                if send_succeed:
                    _document.update_row(self.ots_client)
                else:
                    log("send_msg_error2222")
            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,get_total_count=True,limit=100),
                                                                                    ColumnsToGet(return_type=ColumnReturnType.ALL))
                list_data = getRow_ots(rows)
                for _data in list_data:
                    _d = {document_tmp_partitionkey:_data.get(document_tmp_partitionkey),
                          document_tmp_docid:_data.get(document_tmp_docid),
                          document_tmp_status:0}
                    _document = Document(_d)
                    page_attachments = _data.get(document_tmp_attachment_path,"[]")

                    _document_html = Document(_data)
                    _document_html.fix_columns(self.ots_capacity,[document_tmp_dochtmlcon],True)

                    if page_attachments!="[]":
                        status = random.randint(1,10)
                        _data[document_tmp_status] = status
                        send_succeed = send_msg_toacmq(self.pool_mq,json.dumps(_document_html.getProperties(),cls=MyEncoder),self.mq_attachment)
                    else:
                        status = random.randint(11,50)
                        _data[document_tmp_status] = status
                        send_succeed = send_msg_toacmq(self.pool_mq,json.dumps(_document_html.getProperties(),cls=MyEncoder),self.mq_extract)
                    if send_succeed:
                        _document.update_row(self.ots_client)
                    else:
                        log("send_msg_error2222")
        except Exception as e:
            traceback.print_exc()

    def otstmp2mq(self):
        try:
            bool_query = BoolQuery(must_queries=[TermQuery("status",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(document_docid)]),get_total_count=True,limit=100),
                                                                                ColumnsToGet(return_type=ColumnReturnType.ALL))
            list_data = getRow_ots(rows)
            for _data in list_data:
                _d = {document_tmp_partitionkey:_data.get(document_tmp_partitionkey),
                      document_tmp_docid:_data.get(document_tmp_docid),
                      document_tmp_status:0}
                _document = Document_tmp(_d)
                page_attachments = _data.get(document_tmp_attachment_path,"[]")

                _document_html = Document_html(_data)
                _document_html.fix_columns(self.ots_client,[document_tmp_dochtmlcon],True)

                if page_attachments!="[]":
                    status = random.randint(1,10)
                    _data[document_tmp_status] = status
                    send_succeed = send_msg_toacmq(self.pool_mq,json.dumps(_document_html.getProperties(),cls=MyEncoder),self.mq_attachment)
                else:
                    status = random.randint(11,50)
                    _data[document_tmp_status] = status
                    send_succeed = send_msg_toacmq(self.pool_mq,json.dumps(_document_html.getProperties(),cls=MyEncoder),self.mq_extract)
                if send_succeed:
                    _document.setValue(document_tmp_status,1,True)
                    _document.update_row(self.ots_client)
                else:
                    log("send_msg_error2222")
            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.ALL))
                list_data = getRow_ots(rows)
                for _data in list_data:
                    _d = {document_tmp_partitionkey:_data.get(document_tmp_partitionkey),
                          document_tmp_docid:_data.get(document_tmp_docid),
                          document_tmp_status:0}
                    _document = Document_tmp(_d)
                    page_attachments = _data.get(document_tmp_attachment_path,"[]")

                    _document_html = Document_html(_data)
                    _document_html.fix_columns(self.ots_client,[document_tmp_dochtmlcon],True)

                    if page_attachments!="[]":
                        status = random.randint(1,10)
                        _data[document_tmp_status] = status
                        send_succeed = send_msg_toacmq(self.pool_mq,json.dumps(_document_html.getProperties(),cls=MyEncoder),self.mq_attachment)
                    else:
                        status = random.randint(11,50)
                        _data[document_tmp_status] = status
                        send_succeed = send_msg_toacmq(self.pool_mq,json.dumps(_document_html.getProperties(),cls=MyEncoder),self.mq_extract)
                    if send_succeed:
                        _document.setValue(document_tmp_status,1,True)
                        _document.update_row(self.ots_client)
                    else:
                        log("send_msg_error2222")
        except Exception as e:
            traceback.print_exc()


    def test_dump_docid(self):
        class TestDumpListener(ActiveMQListener):
            def on_message(self, headers):
                message_id = headers.headers["message-id"]
                body = headers.body
                self._queue.put(headers,True)
                ackMsg(self.conn,message_id)

        _queue = Queue()
        listener1 = TestDumpListener(getConnect_activateMQ(),_queue)
        listener2 = TestDumpListener(getConnect_activateMQ(),_queue)
        createComsumer(listener1,"/queue/dataflow_attachment")
        createComsumer(listener2,"/queue/dataflow_extract")

        time.sleep(10)
        list_item = []
        list_docid = []
        while 1:
            try:
                _item = _queue.get(timeout=2)
                list_item.append(_item)
            except Exception as e:
                break
        for item in list_item:
            _item = json.loads(item.body)
            list_docid.append(_item.get("docid"))
        log(list_docid[:10])
        log("len docid:%d set len:%d"%(len(list_docid),len(set(list_docid))))


    def start_dataflow_init(self):
        # self.test_dump_docid()
        from BaseDataMaintenance.model.oracle.CaiGouYiXiangTemp import CaiGouYiXiangTemp
        from BaseDataMaintenance.model.oracle.PaiMaiChuRangTemp import PaiMaiChuRangTemp
        from BaseDataMaintenance.model.oracle.ZhaoBiaoGongGaoTemp import ZhaoBiaoGongGaoTemp
        from BaseDataMaintenance.model.oracle.ZhaoBiaoYuGaoTemp import ZhaoBiaoYuGaoTemp
        from BaseDataMaintenance.model.oracle.ZhongBiaoXinXiTemp import ZhongBiaoXinXiTemp
        from BaseDataMaintenance.model.oracle.ZiShenJieGuoTemp import ZiShenJieGuoTemp
        from BaseDataMaintenance.model.oracle.ChanQuanJiaoYiTemp import ChanQuanJiaoYiTemp
        from BaseDataMaintenance.model.oracle.GongGaoBianGengTemp import GongGaoBianGeng
        from BaseDataMaintenance.model.oracle.KongZhiJiaTemp import KongZhiJiaTemp
        from BaseDataMaintenance.model.oracle.TuDiKuangChanTemp import TuDiKuangChanTemp
        from BaseDataMaintenance.model.oracle.ZhaoBiaoDaYiTemp import ZhaoBiaoDaYiTemp
        from BaseDataMaintenance.model.oracle.ZhaoBiaoWenJianTemp import ZhaoBiaoWenJianTemp
        schedule = BlockingScheduler()
        schedule.add_job(self.temp2mq,"cron",args=(CaiGouYiXiangTemp({}),),second="*/10")
        schedule.add_job(self.temp2mq,"cron",args=(PaiMaiChuRangTemp({}),),second="*/10")
        schedule.add_job(self.temp2mq,"cron",args=(ZhaoBiaoGongGaoTemp({}),),second="*/10")
        schedule.add_job(self.temp2mq,"cron",args=(ZhaoBiaoYuGaoTemp({}),),second="*/10")
        schedule.add_job(self.temp2mq,"cron",args=(ZhongBiaoXinXiTemp({}),),second="*/10")
        schedule.add_job(self.temp2mq,"cron",args=(ZiShenJieGuoTemp({}),),second="*/10")
        schedule.add_job(self.temp2mq,"cron",args=(ChanQuanJiaoYiTemp({}),),second="*/10")
        schedule.add_job(self.temp2mq,"cron",args=(GongGaoBianGeng({}),),second="*/10")
        schedule.add_job(self.temp2mq,"cron",args=(KongZhiJiaTemp({}),),second="*/10")
        schedule.add_job(self.temp2mq,"cron",args=(TuDiKuangChanTemp({}),),second="*/10")
        schedule.add_job(self.temp2mq,"cron",args=(ZhaoBiaoDaYiTemp({}),),second="*/10")
        schedule.add_job(self.temp2mq,"cron",args=(ZhaoBiaoWenJianTemp({}),),second="*/10")
        schedule.add_job(self.ots2mq,"cron",second="*/10")
        schedule.add_job(self.otstmp2mq,"cron",second="*/10")
        schedule.add_job(self.monitor_listener,"cron",minute="*/1")
        schedule.start()




def transform_attachment():
    from BaseDataMaintenance.model.ots.attachment import attachment
    from BaseDataMaintenance.model.postgres.attachment import Attachment_postgres
    from BaseDataMaintenance.dataSource.source import getConnection_postgres,getConnect_ots

    from threading import Thread
    from queue import Queue

    task_queue = Queue()

    def comsumer(task_queue,pool_postgres):
        while 1:
            _dict = task_queue.get(True)
            try:
                attach = Attachment_postgres(_dict)
                if not attach.exists(pool_postgres):
                    attach.insert_row(pool_postgres)
            except Exception as e:
                traceback.print_exc()

    def start_comsumer(task_queue):
        pool_postgres = ConnectorPool(10,30,getConnection_postgres)

        comsumer_count = 30
        list_thread = []
        for i in range(comsumer_count):
            _t = Thread(target=comsumer,args=(task_queue,pool_postgres))
            list_thread.append(_t)
        for _t in list_thread:
            _t.start()


    ots_client = getConnect_ots()
    _thread = Thread(target=start_comsumer,args=(task_queue,))
    _thread.start()
    bool_query = BoolQuery(must_queries=[
        RangeQuery(attachment_crtime,"2022-05-06"),
        BoolQuery(should_queries=[TermQuery(attachment_status,10),
                                  TermQuery(attachment_status,ATTACHMENT_TOOLARGE)])
    ])

    rows,next_token,total_count,is_all_succeed = ots_client.search("attachment","attachment_index",
                                                                   SearchQuery(bool_query,sort=Sort(sorters=[FieldSort(attachment_crtime,sort_order=SortOrder.DESC)]),get_total_count=True,limit=100),
                                                                   columns_to_get=ColumnsToGet(return_type=ColumnReturnType.ALL))

    list_dict = getRow_ots(rows)

    for _dict in list_dict:
        task_queue.put(_dict,True)
    _count = len(list_dict)
    while next_token:
        rows,next_token,total_count,is_all_succeed = ots_client.search("attachment","attachment_index",
                                                                       SearchQuery(bool_query,next_token=next_token,get_total_count=True,limit=100),
                                                                       columns_to_get=ColumnsToGet(return_type=ColumnReturnType.ALL))

        list_dict = getRow_ots(rows)
        for _dict in list_dict:
            task_queue.put(_dict,True)
        _count += len(list_dict)
        print("%d/%d,queue:%d"%(_count,total_count,task_queue.qsize()))


def del_test_doc():

    ots_client = getConnect_ots()
    bool_query = BoolQuery(must_queries=[RangeQuery("docid",range_to=0)])

    list_data = []
    rows,next_token,total_count,is_all_succeed = ots_client.search("document_tmp","document_tmp_index",
                                                                   SearchQuery(bool_query,sort=Sort(sorters=[FieldSort("docid")]),get_total_count=True,limit=100),
                                                                   columns_to_get=ColumnsToGet(return_type=ColumnReturnType.NONE))
    print(total_count)
    list_row = getRow_ots(rows)
    list_data.extend(list_row)
    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,get_total_count=True,limit=100),
                                                                       columns_to_get=ColumnsToGet(return_type=ColumnReturnType.NONE))

        list_row = getRow_ots(rows)
        list_data.extend(list_row)
    for _row in list_data:
        _doc = Document_tmp(_row)
        _doc.delete_row(ots_client)
        _html = Document_html(_row)
        if _html.exists_row(ots_client):
            _html.delete_row(ots_client)

def fixDoc_to_queue_extract():
    pool_mq = ConnectorPool(10,20,getConnect_activateMQ)
    try:
        ots_client = getConnect_ots()
        ots_capacity = getConnect_ots_capacity()
        bool_query = BoolQuery(must_queries=[
            RangeQuery("crtime","2022-05-31"),
            TermQuery("docchannel",114)
        ])

        list_data = []
        rows,next_token,total_count,is_all_succeed = ots_client.search("document","document_index",
                                                                       SearchQuery(bool_query,sort=Sort(sorters=[FieldSort("docid")]),get_total_count=True,limit=100),
                                                                       columns_to_get=ColumnsToGet(return_type=ColumnReturnType.ALL))
        print(total_count)
        list_row = getRow_ots(rows)
        list_data.extend(list_row)
        _count = len(list_row)
        while next_token:
            rows,next_token,total_count,is_all_succeed = ots_client.search("document","document_index",
                                                                           SearchQuery(bool_query,next_token=next_token,get_total_count=True,limit=100),
                                                                           columns_to_get=ColumnsToGet(return_type=ColumnReturnType.ALL))

            list_row = getRow_ots(rows)
            list_data.extend(list_row)
            _count  = len(list_data)
            print("%d/%d"%(_count,total_count))
        task_queue = Queue()
        for _row in list_data:
            if "all_columns" in _row:
                _row.pop("all_columns")
            _html = Document(_row)
            task_queue.put(_html)
        def _handle(item,result_queue):
            _html = item
            _html.fix_columns(ots_capacity,["dochtmlcon"],True)
            print(_html.getProperties().get(document_tmp_docid))
            send_msg_toacmq(pool_mq,json.dumps(_html.getProperties()),"/queue/dataflow_extract")
        mt = MultiThreadHandler(task_queue,_handle,None,30)
        mt.run()
    except Exception as e:
        traceback.print_exc()
    finally:
        pool_mq.destory()

def check_data_synchronization():
    # filepath = "C:\\Users\\Administrator\\Desktop\\to_check.log"
    # list_uuid = []
    # _regrex = "delete\s+(?P<tablename>[^\s]+)\s+.*ID='(?P<uuid>.+)'"
    # with open(filepath,"r",encoding="utf8") as f:
    #     while 1:
    #         _line = f.readline()
    #         if not _line:
    #             break
    #         _match = re.search(_regrex,_line)
    #         if _match is not None:
    #             _uuid = _match.groupdict().get("uuid")
    #             tablename = _match.groupdict.get("tablename")
    #             if _uuid is not None:
    #                 list_uuid.append({"uuid":_uuid,"tablename":tablename})
    # print("total_count:",len(list_uuid))

    import pandas as pd
    from BaseDataMaintenance.common.Utils import load


    task_queue = Queue()
    list_data = []
    df_data = load("uuid.pk")
    # df_data = pd.read_excel("check.xlsx")
    for uuid,tablename in zip(df_data["uuid"],df_data["tablename"]):
        _dict = {"uuid":uuid,
                 "tablename":tablename}
        list_data.append(_dict)
        task_queue.put(_dict)
    print("qsize:",task_queue.qsize())
    ots_client = getConnect_ots()
    def _handle(_item,result_queue):
        bool_query = BoolQuery(must_queries=[TermQuery("uuid",_item.get("uuid"))])

        rows,next_token,total_count,is_all_succeed = ots_client.search("document","document_index",
                                                                       SearchQuery(bool_query,get_total_count=True),
                                                                       columns_to_get=ColumnsToGet(return_type=ColumnReturnType.NONE))

        _item["exists"] = total_count
    mt = MultiThreadHandler(task_queue,_handle,None,30)
    mt.run()
    df_data = {"uuid":[],
               "tablename":[],
               "exists":[]}
    for _data in list_data:
        if _data["exists"]==0:
            for k,v in df_data.items():
                v.append(_data.get(k))
    import pandas as pd
    df2 = pd.DataFrame(df_data)
    df2.to_excel("check1.xlsx")

current_path = os.path.abspath(os.path.dirname(__file__))

def fixDoc_to_queue_init(filename=""):
    import pandas as pd
    from BaseDataMaintenance.model.oracle.GongGaoTemp import dict_oracle2ots
    if filename=="":
        filename = os.path.join(current_path,"check.xlsx")
    df = pd.read_excel(filename)
    if "docchannel" in dict_oracle2ots:
        dict_oracle2ots.pop("docchannel")
    row_name = ",".join(list(dict_oracle2ots.keys()))
    conn = getConnection_oracle()
    cursor = conn.cursor()
    _count = 0
    for uuid,tablename,_exists,_toolong in zip(df["uuid"],df["tablename"],df["exists"],df["tolong"]):
        if _exists==0 and _toolong==0:
            _count += 1
            _source = str(tablename).replace("_TEMP","")
            sql = " insert into %s(%s) select %s from %s where id='%s' "%(tablename,row_name,row_name,_source,uuid)
            cursor.execute(sql)
            log("%d:%s"%(_count,sql))
    conn.commit()
    conn.close()

if __name__ == '__main__':
    # di = Dataflow_init()
    # di.start_dataflow_init()
    # transform_attachment()
    # del_test_doc()
    # de = Dataflow_ActivteMQ_extract()
    # de.start_flow_extract()
    # fixDoc_to_queue_extract()
    # check_data_synchronization()
    # fixDoc_to_queue_init(filename="C:\\Users\\Administrator\\Desktop\\flow_init_2023-02-07.xlsx")
    current_date = getCurrent_date(format="%Y-%m-%d")
    last_date = timeAdd(current_date,-30,format="%Y-%m-%d")
    sql = " delete from attachment where crtime<='%s 00:00:00' "%(last_date)
    print(sql)