BaseDataMaintenance/BaseDataMaintenance/dataMonitor/data_monitor_fjs.py

import os,sys

sys.path.append(os.path.dirname(__file__)+"/../../")
from apscheduler.schedulers.blocking import BlockingScheduler

from BaseDataMaintenance.dataSource.source import getConnect_ots,getConnect_activateMQ

from BaseDataMaintenance.dataSource.interface import *
from BaseDataMaintenance.common.Utils import *

from tablestore import *
from BaseDataMaintenance.dataSource.setttings import *
from queue import Queue
from BaseDataMaintenance.common.multiThread import MultiThreadHandler


from BaseDataMaintenance.maintenance.dataflow_settings import *

import pandas as pd


flow_attachment_log_path = "/data/python/flow_attachment.log"

flow_extract_log_path = "/data/python/flow_extract.log"

flow_init_path = "/data/python/flow_init.log"


flow_init_log_dir = "/data/python"
flow_init_check_dir = "/data/python/flow_init_check"


flow_dumplicate_log_path = "/python_log/flow_dumplicate.log"


class BaseDataMonitor():

    def __init__(self):
        self.ots_client = getConnect_ots()
        self.recieviers = ["1175730271@qq.com","531870502@qq.com"]

        self.list_proposed_count = []
        self.current_path = os.path.dirname(__file__)

    def cmd_execute(self,_cmd):
        with os.popen(_cmd) as f:
            return f.read()

    def get_last_tenmin_time(self,nums=15):
        current_time = getCurrent_date(format="%Y-%m-%d %H:%M:%S")

        last_ten_minite_time = timeAdd(current_time,0,"%Y-%m-%d %H:%M:%S",-10)
        return last_ten_minite_time[:nums]

    def check_document_uuid(self,log_filename):

        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_tmp","document_tmp_index",
                                                                           SearchQuery(bool_query,get_total_count=True),
                                                                           columns_to_get=ColumnsToGet(return_type=ColumnReturnType.NONE))

            _item["exists"] = total_count
        check_filename = "%s_check.xlsx"%(log_filename)
        list_uuid = []
        task_queue = Queue()
        dict_tolong = {}
        if not os.path.exists(check_filename) and os.path.exists(log_filename):
            _regrex = "delete\s+(?P<tablename>bxkc[^\s]+)\s+.*ID='(?P<uuid>.+)'"
            _regrex_tolong = "msg too long:(?P<uuid>[^,]+),\d+"
            with open(log_filename,"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})
                    _match = re.search(_regrex_tolong,_line)
                    if _match is not None:
                        _uuid = _match.groupdict().get("uuid")
                        dict_tolong[_uuid] = 1


            if list_uuid==0:
                _msg = "数据遗漏检查出错"
                sentMsgToDD(_msg,ACCESS_TOKEN_DATAWORKS,atAll=True)
                # sendEmail(smtp_host,smtp_username,smtp_password,self.recieviers,_msg)

            ots_client = getConnect_ots()

            for _d in list_uuid:
                task_queue.put(_d)
            mt = MultiThreadHandler(task_queue,_handle,None,30)
            mt.run()
            df_data = {"uuid":[],
                       "tablename":[],
                       "exists":[],
                       "tolong":[]}

            for _data in list_uuid:
                for k,v in df_data.items():
                    if k!="tolong":
                        v.append(_data.get(k))
                df_data["tolong"].append(dict_tolong.get(_data["uuid"],0))
            df2 = pd.DataFrame(df_data)
            df2.to_excel(check_filename)

    def monitor_init(self):

        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_tmp","document_tmp_index",
                                                                           SearchQuery(bool_query,get_total_count=True),
                                                                           columns_to_get=ColumnsToGet(return_type=ColumnReturnType.NONE))

            _item["exists"] = total_count
        try:
            current_date = getCurrent_date("%Y-%m-%d")

            last_date = timeAdd(current_date,-1,"%Y-%m-%d")

            if not os.path.exists(flow_init_check_dir):
                os.mkdir(flow_init_check_dir)

            log_filename = os.path.join(flow_init_log_dir,"flow_init_%s.log"%(last_date))
            check_filename = os.path.join(flow_init_check_dir,"flow_init_%s.xlsx"%(last_date))

            list_uuid = []
            task_queue = Queue()
            dict_tolong = {}
            if not os.path.exists(check_filename) and os.path.exists(log_filename):
                _regrex = "delete\s+(?P<tablename>bxkc[^\s]+)\s+.*ID='(?P<uuid>.+)'"
                _regrex_tolong = "msg too long:(?P<uuid>[^,]+),\d+"
                with open(log_filename,"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})
                        _match = re.search(_regrex_tolong,_line)
                        if _match is not None:
                            _uuid = _match.groupdict().get("uuid")
                            dict_tolong[_uuid] = 1


                if list_uuid==0:
                    _msg = "数据遗漏检查出错"
                    sentMsgToDD(_msg,ACCESS_TOKEN_DATAWORKS,atAll=True)
                    # sendEmail(smtp_host,smtp_username,smtp_password,self.recieviers,_msg)

                ots_client = getConnect_ots()

                for _d in list_uuid:
                    task_queue.put(_d)
                mt = MultiThreadHandler(task_queue,_handle,None,30)
                mt.run()
                df_data = {"uuid":[],
                           "tablename":[],
                           "exists":[],
                           "tolong":[]}

                for _data in list_uuid:
                    for k,v in df_data.items():
                        if k!="tolong":
                            v.append(_data.get(k))
                    df_data["tolong"].append(dict_tolong.get(_data["uuid"],0))
                df2 = pd.DataFrame(df_data)
                df2.to_excel(check_filename)

            counts = 0
            df_data = pd.read_excel(check_filename)
            for _exists,_tolong in zip(df_data["exists"],df_data["tolong"]):
                if _exists==0 and _tolong==0:
                    counts += 1
            if counts>0:
                _msg = "数据遗漏检查报警,%s有%s条公告遗漏，详见%s"%(last_date,str(counts),check_filename)
                sentMsgToDD(_msg,ACCESS_TOKEN_DATAWORKS,atAll=True)
                # sendEmail(smtp_host,smtp_username,smtp_password,self.recieviers,_msg)



        except Exception as e:
            _msg = "数据遗漏检查报错"
            sentMsgToDD(_msg,ACCESS_TOKEN_DATAWORKS,atAll=True)
            # sendEmail(smtp_host,smtp_username,smtp_password,self.recieviers,_msg)
            traceback.print_exc()

    def monitor_init_fjs(self):

        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_tmp","document_tmp_index",
                                                                           SearchQuery(bool_query,get_total_count=True),
                                                                           columns_to_get=ColumnsToGet(return_type=ColumnReturnType.NONE))

            _item["exists"] = total_count

        try:
            current_date = getCurrent_date("%Y-%m-%d")

            last_date = timeAdd(current_date,-1,"%Y-%m-%d")

            if not os.path.exists(flow_init_check_dir):
                os.mkdir(flow_init_check_dir)

            log_filename = os.path.join(flow_init_log_dir,"flow_init.log")
            check_filename = os.path.join(flow_init_check_dir,"flow_init_2024-05-21.xlsx")

            list_uuid = []
            task_queue = Queue()
            dict_tolong = {}
            if not os.path.exists(check_filename) and os.path.exists(log_filename):
                print('--------- 1 ----------')
                _regrex = "delete\s+(?P<tablename>bxkc[^\s]+)\s+.*ID='(?P<uuid>.+)'"
                _regrex_tolong = "msg too long:(?P<uuid>[^,]+),\d+"
                with open(log_filename,"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})
                        _match = re.search(_regrex_tolong,_line)
                        if _match is not None:
                            _uuid = _match.groupdict().get("uuid")
                            dict_tolong[_uuid] = 1


                if list_uuid==0:
                    _msg = "数据遗漏检查出错"
                    sentMsgToDD(_msg,ACCESS_TOKEN_DATAWORKS,atAll=True)
                    # sendEmail(smtp_host,smtp_username,smtp_password,self.recieviers,_msg)

                ots_client = getConnect_ots()

                for _d in list_uuid:
                    task_queue.put(_d)
                mt = MultiThreadHandler(task_queue,_handle,None,30)
                mt.run()
                df_data = {"uuid":[],
                           "tablename":[],
                           "exists":[],
                           "tolong":[]}

                for _data in list_uuid:
                    for k,v in df_data.items():
                        if k!="tolong":
                            v.append(_data.get(k))
                    df_data["tolong"].append(dict_tolong.get(_data["uuid"],0))
                df2 = pd.DataFrame(df_data)
                df2.to_excel(check_filename)

            counts = 0
            df_data = pd.read_excel(check_filename)
            for _exists,_tolong in zip(df_data["exists"],df_data["tolong"]):
                if _exists==0 and _tolong==0:
                    counts += 1
            if counts>0:
                _msg = "数据遗漏检查报警,%s有%s条公告遗漏，详见%s"%(last_date,str(counts),check_filename)
                sentMsgToDD(_msg,ACCESS_TOKEN_DATAWORKS,atAll=True)
                # sendEmail(smtp_host,smtp_username,smtp_password,self.recieviers,_msg)



        except Exception as e:
            _msg = "数据遗漏检查报错"
            sentMsgToDD(_msg,ACCESS_TOKEN_DATAWORKS,atAll=True)
            # sendEmail(smtp_host,smtp_username,smtp_password,self.recieviers,_msg)
            traceback.print_exc()

    def monitor_attachment(self):
        from BaseDataMaintenance.java.MQInfo import getAllQueueSize,getQueueSize
        try:
            # query = BoolQuery(must_queries=[
            #     RangeQuery("status",0,11),
            # ])
            #
            # rows,next_token,total_count_todeal,is_all_succeed = self.ots_client.search("document_tmp","document_tmp_index",
            #                                                                            SearchQuery(query,None,True),
            #                                                                            columns_to_get=ColumnsToGet(return_type=ColumnReturnType.NONE))
            total_count_todeal = getQueueSize("dataflow_attachment")

            if total_count_todeal>100:
                # query = BoolQuery(must_queries=[
                #     RangeQuery("crtime",self.get_last_tenmin_time(16))
                # ])
                #
                # rows,next_token,total_count_todeal_t_a,is_all_succeed = self.ots_client.search("document_tmp","document_tmp_index",
                #                                                                              SearchQuery(query,None,True),
                #                                                                              columns_to_get=ColumnsToGet(return_type=ColumnReturnType.NONE))
                #
                # query = BoolQuery(must_queries=[
                #     RangeQuery("status",0,11),
                #     RangeQuery("crtime",self.get_last_tenmin_time(16))
                # ])
                #
                # rows,next_token,total_count_todeal_t,is_all_succeed = self.ots_client.search("document_tmp","document_tmp_index",
                #                                                                            SearchQuery(query,None,True),
                #                                                                            columns_to_get=ColumnsToGet(return_type=ColumnReturnType.NONE))





                #通过命令行获取日志情况
                # _cmd = 'cat %s | grep -c "%s.*process filemd5"'%(flow_attachment_log_path,self.get_last_tenmin_time())
                # log(_cmd)
                # process_count = self.cmd_execute(_cmd)
                #
                # _cmd = 'cat %s | grep -c "%s.*process filemd5.*True"'%(flow_attachment_log_path,self.get_last_tenmin_time())
                # log(_cmd)
                # process_succeed_count = self.cmd_execute(_cmd)
                #
                # _cmd = 'cat %s | grep -c "%s.*delete sql"'%(flow_init_path,self.get_last_tenmin_time())
                # log(_cmd)
                # init_count = self.cmd_execute(_cmd)

                # _msg = "附件提取队列报警：队列堆积%s条公告，最近十分钟处理公告附件数：%s，处理成功数：%s"%(str(total_count_todeal),str(process_count),str(process_succeed_count))

                #通过读取文件获取日志情况
                dict_type = {}
                _pattern = "%s.*process filemd5\:[^\s]* (?P<result>(True|False)) of type\:(?P<type>[^\s]*).*recognize takes (?P<costtime>\d+)s"%(re.escape(self.get_last_tenmin_time()))
                with open(flow_attachment_log_path,"r",encoding="utf8") as f:
                    while True:
                        line = f.readline()
                        if not line:
                            break
                        _match = re.search(_pattern,str(line))
                        if _match is not None:
                            _type = _match.groupdict().get("type")
                            _result = _match.groupdict().get("result")
                            _costtime = _match.groupdict().get("costtime")
                            if _type not in dict_type:
                                dict_type[_type] = {"success":0,"fail":0,"success_costtime":0,"fail_costtime":0}
                            if _result=="True":
                                dict_type[_type]["success"] += 1
                                dict_type[_type]["success_costtime"] += int(_costtime)
                            else:
                                dict_type[_type]["fail"] += 1
                                dict_type[_type]["fail_costtime"] += int(_costtime)

                process_count = 0
                process_succeed_count = 0
                _msg_type = ""
                for k,v in dict_type.items():
                    process_count += v.get("success",0)+v.get("fail",0)
                    process_succeed_count += v.get("success",0)
                    _msg_type += "\n类型%s\n\t成功%s,消耗%s秒,%.2f秒/个,\n\t失败%s,消耗%s秒,%.2f秒/个"%(k,str(v.get("success")),str(v.get("success_costtime")),v.get("success_costtime")/max(1,v.get("success")),str(v.get("fail")),str(v.get("fail_costtime")),v.get("fail_costtime")/max(1,v.get("fail")))

                _msg = "附件提取队列报警：队列堆积%s条公告，最近十分钟处理公告附件数：%s，处理成功数：%s"%(str(total_count_todeal),str(process_count),str(process_succeed_count))
                sentMsgToDD(_msg+_msg_type,ACCESS_TOKEN_DATAWORKS)
                # sendEmail(smtp_host,smtp_username,smtp_password,self.recieviers,_msg)
        except Exception as e:
            traceback.print_exc()

    def monitor_extract(self):
        from BaseDataMaintenance.java.MQInfo import getAllQueueSize,getQueueSize
        try:
            # query = BoolQuery(must_queries=[
            #     RangeQuery("status",11,61),
            # ])
            #
            # rows,next_token,total_count_todeal,is_all_succeed = self.ots_client.search("document_tmp","document_tmp_index",
            #                                                                     SearchQuery(query,None,True),
            #                                                                     columns_to_get=ColumnsToGet(return_type=ColumnReturnType.NONE))
            total_count_init = getQueueSize("dataflow_init")
            if total_count_init>=100:
                _msg = "同步队列报警，有%s条数据滞留"%(str(total_count_init))
                sentMsgToDD(_msg,ACCESS_TOKEN_DATAWORKS,atAll=True)
                # sendEmail(smtp_host,smtp_username,smtp_password,self.recieviers,_msg)


            total_count_todeal = getQueueSize("dataflow_extract")

            if total_count_todeal>100:
                _cmd = 'cat %s | grep "%s" | grep -c "process.*docid"'%(flow_extract_log_path,self.get_last_tenmin_time())
                log(_cmd)
                process_count = self.cmd_execute(_cmd)
                _cmd = 'cat %s | grep "%s" | grep -c "process.*docid.*1$"'%(flow_extract_log_path,self.get_last_tenmin_time())
                log(_cmd)
                success_count = self.cmd_execute(_cmd)

                _cmd = 'cat %s | grep "%s" | grep -c "fingerprint.*exists docid"'%(flow_extract_log_path,self.get_last_tenmin_time())
                log(_cmd)
                exists_count = self.cmd_execute(_cmd)

                _cmd = 'cat %s | grep -c "%s.*delete sql"'%(flow_init_path,self.get_last_tenmin_time())
                log(_cmd)
                init_count = self.cmd_execute(_cmd)


                # query = BoolQuery(must_queries=[
                #                                                                                RangeQuery("crtime",self.get_last_tenmin_time(16))
                #                                                                            ])
                #
                # rows,next_token,total_count_todeal_t_a,is_all_succeed = self.ots_client.search("document_tmp","document_tmp_index",
                #                                                                                SearchQuery(query,None,True),
                #                                                                                columns_to_get=ColumnsToGet(return_type=ColumnReturnType.NONE))
                #
                # query = BoolQuery(must_queries=[
                #     RangeQuery("status",11,61),
                #     RangeQuery("crtime",self.get_last_tenmin_time(16))
                # ])
                #
                # rows,next_token,total_count_todeal_t,is_all_succeed = self.ots_client.search("document_tmp","document_tmp_index",
                #                                                                              SearchQuery(query,None,True),
                #                                                                              columns_to_get=ColumnsToGet(return_type=ColumnReturnType.NONE))

                _msg = "要素提取队列报警：队列堆积%s条公告，最近十分钟入库数：%s,最近十分钟处理公告数：%s，其中成功处理数：%s，查库免提取数：%s"%(str(total_count_todeal),str(init_count),str(process_count),str(success_count),str(exists_count))
                atAll=False
                if success_count==0:
                    atAll=True
                    _msg += "\n疑似流程出现问题，请相关负责人尽快处理"
                sentMsgToDD(_msg,ACCESS_TOKEN_DATAWORKS,atAll=atAll)
                # sendEmail(smtp_host,smtp_username,smtp_password,self.recieviers,_msg)
        except Exception as e:
            traceback.print_exc()

    def monitor_proposedBuilding(self):
        current_date = getCurrent_date("%Y-%m-%d")

        current_hour = getCurrent_date("%H")


        query = BoolQuery(must_queries=[
            RangeQuery("update_time",current_date),
            WildcardQuery("docids","*")
        ])

        rows,next_token,total_count_doc,is_all_succeed = self.ots_client.search("designed_project","designed_project_index",
                                                                                   SearchQuery(query,None,True),
                                                                                   columns_to_get=ColumnsToGet(return_type=ColumnReturnType.NONE))


        query = BoolQuery(must_queries=[
            RangeQuery("update_time",current_date),
            WildcardQuery("spids","*")
        ])

        rows,next_token,total_count_sp,is_all_succeed = self.ots_client.search("designed_project","designed_project_index",
                                                                            SearchQuery(query,None,True),
                                                                            columns_to_get=ColumnsToGet(return_type=ColumnReturnType.NONE))

        total_count = total_count_doc+total_count_sp
        if not (current_hour in ("00","23","24")):
            _msg = "拟在建生成报警：当天生成的拟在建数量为:%d，其中公告生成：%d，审批项目生成：%d"%(total_count,total_count_doc,total_count_sp)
            atAll=False
            if total_count==0:
                atAll=True
                _msg += "\n疑似流程出现问题，请相关负责人尽快处理"
            sentMsgToDD(_msg,ACCESS_TOKEN_DATAWORKS,atAll=atAll)
            # sendEmail(smtp_host,smtp_username,smtp_password,self.recieviers,_msg)


    def monitor_sychr(self):
        current_date = getCurrent_date("%Y-%m-%d")

        last_date = timeAdd(current_date,-1,"%Y-%m-%d")

        query = BoolQuery(must_queries=[
            RangeQuery("status",*flow_sychro_status_from,True,True),
        ])

        rows,next_token,total_count,is_all_succeed = self.ots_client.search("document_tmp","document_tmp_index",
                                                                            SearchQuery(query,None,True),
                                                                            columns_to_get=ColumnsToGet(return_type=ColumnReturnType.NONE))

        if total_count>=200:
            _msg = "数据流报警：待同步到成品表公告数为:%d"%(total_count)
            sentMsgToDD(_msg,ACCESS_TOKEN_DATAWORKS)
            # sendEmail(smtp_host,smtp_username,smtp_password,self.recieviers,_msg)

    def monitor_preproject(self):
        current_date = getCurrent_date("%Y-%m-%d")

        last_date = timeAdd(current_date,-7,"%Y-%m-%d")

        query = BoolQuery(must_queries=[
            RangeQuery("crtime",last_date),
        ])

        rows,next_token,total_count,is_all_succeed = self.ots_client.search("preproject","preproject_index",
                                                                            SearchQuery(query,None,True),
                                                                            columns_to_get=ColumnsToGet(return_type=ColumnReturnType.NONE))
        log("preproject count:%d"%(total_count))
        if total_count<=10*10000:
            _msg = "周期项目生成报警：最近一周生成/更新的周期项目数量为:%d"%(total_count)
            sentMsgToDD(_msg,ACCESS_TOKEN_DATAWORKS)
            # sendEmail(smtp_host,smtp_username,smtp_password,self.recieviers,_msg)

    def monitor_dumplicate(self):

        current_date = getCurrent_date("%Y-%m-%d")

        last_date = timeAdd(current_date,-1,"%Y-%m-%d")

        query = BoolQuery(must_queries=[
            TermQuery("page_time",last_date),
            RangeQuery("status",71),

        ])

        rows,next_token,total_count_lastday,is_all_succeed = self.ots_client.search("document_tmp","document_tmp_index",
                                                                            SearchQuery(query,None,True),
                                                                            columns_to_get=ColumnsToGet(return_type=ColumnReturnType.NONE))

        query = BoolQuery(must_queries=[
            RangeQuery("status",66,71),

        ])

        rows,next_token,total_count_to_dump,is_all_succeed = self.ots_client.search("document_tmp","document_tmp_index",
                                                                                    SearchQuery(query,None,True),
                                                                                    columns_to_get=ColumnsToGet(return_type=ColumnReturnType.NONE))

        query = BoolQuery(must_queries=[
            TermQuery("page_time",last_date),
            RangeQuery("status",71),
            TermQuery("save",0)
        ])

        rows,next_token,total_count_lastday_dump,is_all_succeed = self.ots_client.search("document_tmp","document_tmp_index",
                                                                                    SearchQuery(query,None,True),
                                                                                    columns_to_get=ColumnsToGet(return_type=ColumnReturnType.NONE))
        # if total_count_lastday_dump/total_count_lastday<0.2:
        #     _msg = "公告去重报警，%s入库公告数：%d，其中去重数：%d,去重率：%.2f"%(last_date,total_count_lastday,total_count_lastday_dump,total_count_lastday_dump/total_count_lastday)
        #     sentMsgToDD(_msg,ACCESS_TOKEN_DATAWORKS)
        #     # sendEmail(smtp_host,smtp_username,smtp_password,self.recieviers,_msg)

        # if total_count_to_dump>2000:
        #     _msg = "公告去重报警，待去重数量:%s"%(str(total_count_to_dump))
        #     sentMsgToDD(_msg,ACCESS_TOKEN_DATAWORKS)
        #     # sendEmail(smtp_host,smtp_username,smtp_password,self.recieviers,_msg)

        #成品表监控
        query = BoolQuery(must_queries=[
            TermQuery("page_time",last_date),
        ])

        rows,next_token,total_count_lastday,is_all_succeed = self.ots_client.search("document","document_index",
                                                                                    SearchQuery(query,None,True),
                                                                                    columns_to_get=ColumnsToGet(return_type=ColumnReturnType.NONE))
        query = BoolQuery(must_queries=[
            TermQuery("page_time",last_date),
            RangeQuery("status",401,451),
        ])

        rows,next_token,total_count_lastday_dump,is_all_succeed = self.ots_client.search("document","document_index",
                                                                                         SearchQuery(query,None,True),
                                                                                         columns_to_get=ColumnsToGet(return_type=ColumnReturnType.NONE))
        if total_count_lastday_dump/total_count_lastday<0.2:
            _msg = "公告成品去重报警，%s入库公告数：%d，其中去重数：%d,去重率：%.2f"%(last_date,total_count_lastday,total_count_lastday_dump,total_count_lastday_dump/total_count_lastday)
            sentMsgToDD(_msg,ACCESS_TOKEN_DATAWORKS)
            # sendEmail(smtp_host,smtp_username,smtp_password,self.recieviers,_msg)

        query = BoolQuery(must_queries=[
            RangeQuery("status",*flow_dumplicate_status_from,True,True),
        ])

        rows,next_token,total_count,is_all_succeed = self.ots_client.search("document_tmp","document_tmp_index",
                                                                            SearchQuery(query,None,True),
                                                                            columns_to_get=ColumnsToGet(return_type=ColumnReturnType.NONE))

        if total_count>=1000:
            _cmd = 'cat %s | grep -c "%s.*upgrate True save"'%(flow_dumplicate_log_path,self.get_last_tenmin_time())
            process_count = self.cmd_execute(_cmd)
            atAll = False
            if process_count=="":
                process_count = 0
            if int(process_count)==0:
                atAll = True
            _msg = "数据流报警：待去重公告数为:%d,最近十分钟去重数为：%s"%(total_count,str(process_count))
            sentMsgToDD(_msg,ACCESS_TOKEN_DATAWORKS,atAll=atAll)
            # sendEmail(smtp_host,smtp_username,smtp_password,self.recieviers,_msg)


    def monitor_merge(self):
        current_date = getCurrent_date("%Y-%m-%d")
        last_date = timeAdd(current_date,-1,"%Y-%m-%d")
        check_count = 20000
        query = BoolQuery(must_queries=[
            RangeQuery("status",201,301)
        ])
        list_doc = []
        queue_docid = Queue()
        rows,next_token,total_count_lastday,is_all_succeed = self.ots_client.search("document","document_index",
                                                                                    SearchQuery(query,sort=Sort(sorters=[FieldSort("docid",SortOrder.DESC)]),get_total_count=True),
                                                                                    columns_to_get=ColumnsToGet(["docchannel"],return_type=ColumnReturnType.SPECIFIED))
        list_data = getRow_ots(rows)
        list_doc.extend(list_data)
        while next_token and len(list_doc)<check_count:
            rows,next_token,total_count_lastday,is_all_succeed = self.ots_client.search("document","document_index",
                                                                                        SearchQuery(query,next_token=next_token,limit=100,get_total_count=True),
                                                                                        columns_to_get=ColumnsToGet(["docchannel"],return_type=ColumnReturnType.SPECIFIED))
            list_data = getRow_ots(rows)
            list_doc.extend(list_data)
        for _doc in list_doc:
            queue_docid.put(_doc)
        def _handle(item,result_queue):
            docid = item.get("docid")
            _query = BoolQuery(must_queries=[TermQuery("docids",str(docid))])
            rows,next_token,total_count_lastday,is_all_succeed = self.ots_client.search("project2","project2_index",
                                                                                        SearchQuery(_query,None,limit=100,get_total_count=True),
                                                                                        columns_to_get=ColumnsToGet(["zhao_biao_page_time","zhong_biao_page_time","docid_number"],return_type=ColumnReturnType.SPECIFIED))
            list_data = getRow_ots(rows)
            item["projects"] = list_data

        mt = MultiThreadHandler(queue_docid,_handle,None,30)
        mt.run()

        not_find_count = 0
        zhongbiao_find_zhaobiao = 0
        zhongbiao_count = 0
        multi_count = 0
        list_notfind = []
        list_multi = []
        for _doc in list_doc:
            if len(_doc.get("projects",[]))==0:
                not_find_count += 1
                list_notfind.append(str(_doc.get("docid")))
            if len(_doc.get("projects",[]))>=2:
                multi_count += 1
                list_multi.append(str(_doc.get("docid")))
            if _doc.get("docchannel") in (101,118,119,120):
                zhongbiao_count += 1
                if len(_doc.get("projects",[]))==1:
                    _project = _doc.get("projects")[0]
                    if _project.get("zhao_biao_page_time","")!="":
                        zhongbiao_find_zhaobiao += 1

        if not_find_count>0 or multi_count>0 or zhongbiao_find_zhaobiao/zhongbiao_count<0.8:
            if not_find_count>0 or multi_count>0:
                current_time = getCurrent_date(format="%Y-%m-%d_%H%M%S")
                logname = os.path.join(self.current_path,"log","%s.log"%current_time)
                with open(logname,"w",encoding="utf8") as f:
                    f.write(",".join(list_notfind))
                    f.write("\n")
                    f.write(",".join(list_multi))
                _msg = "公告合并报警：近%d条成品公告，有%d条未生成项目，有%d条公告找到多个项目，详见%s;其中有%d条中标公告且找到招标公告数为%d，比率为%.3f"%(check_count,not_find_count,multi_count,logname,zhongbiao_count,zhongbiao_find_zhaobiao,zhongbiao_find_zhaobiao/zhongbiao_count)
                sentMsgToDD(_msg,ACCESS_TOKEN_DATAWORKS)
                # sendEmail(smtp_host,smtp_username,smtp_password,self.recieviers,_msg)
            else:
                _msg = "公告合并报警：近%d条成品公告其中有%d条中标公告且找到招标公告数为%d，比率为%.3f"%(check_count,zhongbiao_count,zhongbiao_find_zhaobiao,zhongbiao_find_zhaobiao/zhongbiao_count)
                sentMsgToDD(_msg,ACCESS_TOKEN_DATAWORKS)
                # sendEmail(smtp_host,smtp_username,smtp_password,self.recieviers,_msg)


        query = BoolQuery(must_queries=[
            TermQuery("page_time",last_date),
            RangeQuery("status",71),

        ])

        rows,next_token,total_count_lastday,is_all_succeed = self.ots_client.search("document_tmp","document_tmp_index",
                                                                                    SearchQuery(query,None,True),
                                                                                    columns_to_get=ColumnsToGet(return_type=ColumnReturnType.NONE))
        if total_count_lastday<10*10000:
            _msg = "公告成品入库报警，%s入库公告数：%d"%(last_date,total_count_lastday)
            sentMsgToDD(_msg,ACCESS_TOKEN_DATAWORKS)
            # sendEmail(smtp_host,smtp_username,smtp_password,self.recieviers,_msg)


    def start_monitor(self):
        #数据监控

        scheduler = BlockingScheduler()

        # scheduler.add_job(self.monitor_attachment,"cron",minute="*/10")
        scheduler.add_job(self.monitor_extract,"cron",minute="*/10")
        scheduler.add_job(self.monitor_proposedBuilding,"cron",hour="*/3")
        # scheduler.add_job(self.monitor_dumplicate,"cron",minute="*/10")
        scheduler.add_job(self.monitor_sychr,"cron",minute="*/10")
        scheduler.add_job(self.monitor_preproject,"cron",hour="8")
        scheduler.add_job(self.monitor_merge,"cron",hour="*/1")
        scheduler.add_job(self.monitor_init,"cron",hour="*/3")
        scheduler.start()

    def start_attach_monitor(self):
        #附件监控
        scheduler = BlockingScheduler()

        scheduler.add_job(self.monitor_attachment,"cron",minute="*/10")
        scheduler.start()


if __name__ == '__main__':

    dm = BaseDataMonitor()
    dm.monitor_init_fjs()

    # # dm.start_monitor()
    # log_filename = "C:\\Users\\Administrator\\Desktop\\flow_init_2023-02-03.log"
    # dm.check_document_uuid(log_filename)

    # sentMsgToDD("报警test_msg",ACCESS_TOKEN_DATAWORKS)
    # dm.monitor_proposedBuilding()
    # print(dm.get_last_tenmin_time(16))