quality_tools/lib/monitor_tools.py

#!/usr/bin/env python
# -*- coding:utf-8 -*-
# author : liumiaomiao
#从es库中导出数据到测试环境mongo库
from lib.es_tools import esutil
from datetime import datetime, timedelta
from lib.mongo_tools import MongoUtil,Data_save,MongoSentence
from lib.mysql_tools import MysqlUtil
from lib.clickhouse_tools import ClickhouseUtil, logger

# 定义一周的时间范围，转换为Unix时间戳格式
end_date = int(datetime.now().timestamp())
start_date = int((datetime.now() - timedelta(days=7)).timestamp())

class monitorTools:
    #标准库bidding-es 每周统计入库数量
    def es_bidding(self):
        """
        es链接
        """
        db_config = {
            # es
            'es_host': '127.0.0.1',
            'es_port': 19800,
            'es_http_auth': ('jianyuGr','we3g8glKfe#'),  # 重新申请
            'timeout': 10000,
            'index': "bidding"
        }
        query = {"query": {"bool": {"must": [{"range": {"comeintime": {"from": f"{start_date}", "to": f"{end_date}"}}}]}}}
        # 传入查询语句query 以及配置信息
        # es=esutil.get_es(db_config["es_host"], db_config["es_http_auth"], db_config["es_port"],db_config["index"])
        counts=esutil.get_es_count(query,**db_config)
        count = counts['count']
        print("标准库es-bidding每周入库数据量：",count)
        return count

    # 高质量库bidding-ai-es 每周统计入库数量
    def es_bidding_ai(self):
        """
        es链接
        """
        db_config = {
            # es
            'es_host': '127.0.0.1',
            'es_port': 19800,
            'es_http_auth': ('jianyuGr','we3g8glKfe#'),  # 重新申请
            'timeout': 10000,
            'index': "bidding_ai"
        }
        query = {"query": {"bool": {"must": [{"range": {"comeintime": {"from": f"{start_date}", "to": f"{end_date}"}}}]}}}
        # 传入查询语句query 以及配置信息
        # es=esutil.get_es(db_config["es_host"], db_config["es_http_auth"], db_config["es_port"],db_config["index"])
        counts=esutil.get_es_count(query,**db_config)
        count = counts['count']
        print("高质量库es-bidding每周入库数据量：",count)
        return count

    # 标准库bidding-es 碎片化数据每周统计入库数量
    def es_bidding_fragment(self):
        #正式环境
        db_config = {
            # es
            'es_host': '127.0.0.1',
            'es_port': 19800,
            'es_http_auth': ('jianyuGr', 'we3g8glKfe#'),  # 重新申请
            'timeout': 10000,
            'index': "bidding"
        }
        # #测试环境http://192.168.3.149:9201
        # db_config = {
        #     # es
        #     'es_host': '192.168.3.149',
        #     'es_port': 9201,
        #     # 'es_http_auth': ('jianyuGr', 'we3g8glKfe#'),  # 重新申请
        #     'timeout': 10000,
        #     'index': "bidding"
        # }
        # 定义要监控的字段值
        tags = [
            "情报_法务",
            "情报_财务审计",
            "情报_招标代理",
            "情报_管理咨询",
            "情报_保险",
            "情报_工程设计咨询",
            "情报_安防",
            "情报_印务商机",
            "情报_环境采购",
            "情报_家具招投标"
        ]

        # 初始化字典，将所有标签的计数设置为0
        data = {}
        for tag in tags:
            query = {
                "query": {"bool": {"must": [{"range": {"comeintime": {"from": f"{start_date}", "to": f"{end_date}"}}},
                                            {"term": {"tag_topinformation": tag}}]}}}
            count = esutil.get_es_count(query, **db_config)
            print(f"标准库es-bidding{tag}每周入库数据量：", count['count'])
            data[tag]=count['count']
        # 检查数据字典以确保所有标签都被更新
        print("数据字典内容:", data)  # 打印整个数据字典
        return data

    #拟在建es数据 每周统计入库数量
    def es_nzj(self):
        """
        es链接
        """
        db_config = {
            # es
            'es_host': '127.0.0.1',
            'es_port': 19800,
            'es_http_auth': ('jianyuGr', 'we3g8glKfe#'),  # 重新申请
            'timeout': 10000,
            'index': "proposed_v1"
        }
        query = {
            "query": {"match_all": {}}}
        # 传入查询语句query 以及配置信息
        # es=esutil.get_es(db_config["es_host"], db_config["es_http_auth"], db_config["es_port"],db_config["index"])
        counts = esutil.get_es_count(query, **db_config)
        count=counts['count']
        print("拟在建es入库数据总量：", count)
        return count

    #标准库bidding-mongo 每周统计入库数量
    def bidding(self):
        collection = MongoUtil.get_coon(host='127.0.0.1:27088', database='qfw',collection='bidding',authuser='viewdata',authpass='viewdata')
        query = { "comeintime": {"$gte": start_date, "$lt": end_date}}
        count=MongoSentence.count(collection,query)
        print("标准库bidding-mongo 每周统计入库数量",count)
        return count

    #高质量库bidding-mongo 每周统计入库数量
    def bidding_ai(self):
        collection = MongoUtil.get_coon(host='127.0.0.1:27088', database='qfw_ai',collection='bidding',authuser='viewdata',authpass='viewdata')
        query = { "comeintime": {"$gte": start_date, "$lt": end_date}}
        count=MongoSentence.count(collection,query)
        print("高质量库bidding-mongo 每周统计入库数量",count)
        return count

    #标准库bidding-mongo碎片化数据 每周统计入库数量
    def bidding_fragment(self):
        collection = MongoUtil.get_coon(host='127.0.0.1:27088', database='qfw',collection='bidding',authuser='viewdata',authpass='viewdata')
        # 定义要监控的字段值
        tags = [
            "情报_法务",
            "情报_财务审计",
            "情报_招标代理",
            "情报_管理咨询",
            "情报_保险",
            "情报_工程设计咨询",
            "情报_安防",
            "情报_印务商机",
            "情报_环境采购",
            "情报_家具招投标"
        ]
        data={}
        for tag in tags:
            query = {"comeintime": {"$gte": start_date, "$lt": end_date},"tag_topinformation":tag}
            count=MongoSentence.count(collection,query)
            print(f"标准库bidding-mongo{tag}每周统计入库数量",count)
            data[tag]=count
        return data

    #拟在建baseinfo-mysql 每周统计入库数量
    def nzj(self):
        # MySQL 数据库连接配置
        # mysql_db_config = {
        #     'host': '192.168.3.149',
        #     'port': 4000,
        #     'user': 'datagroup',
        #     'password': 'Dgrpdb#2024@36',
        #     'database': 'jianyu_subjectdb',
        #     'charset': 'utf8mb4'
        # }

        now = datetime.now()
        end_date = now.strftime("%Y-%m-%d %H:%M:%S")
        start_date = (datetime.now() - timedelta(days=7)).strftime("%Y-%m-%d %H:%M:%S")

        # SQL 查询
        mysql_query = "SELECT COUNT(*) FROM dwd_f_nzj_baseinfo WHERE createtime >= %s AND createtime <= %s"
        params = (start_date, end_date)
        conn=MysqlUtil.connect_to_mysql(host='192.168.3.149',port='4000',user='datagroup',password='Dgrpdb#2024@36',database='jianyu_subjectdb')
        count=MysqlUtil.execute_sql(conn,mysql_query,params)
        print("拟在建baseinfo-mysql每周统计入库数量", count)
        return count

     #人脉数据，每周统计入库数量
    def connections(self):
        try:
            query = f"SELECT COUNT(*) FROM information.transaction_info_all WHERE create_time >={start_date} AND create_time <={end_date}"
            # conn=ClickhouseUtil.connect_to_clickhouse(host='192.168.3.207',port='19000',user='jytop',password='pwdTopJy123',database='information')
            conn=ClickhouseUtil.connect_to_clickhouse(host='127.0.0.1',port='9000',user='biservice',password='Bi_top95215#',database='information')
            count=ClickhouseUtil.execute_sql(conn,query)
            result=count[0][0]
            print("人脉数据每周统计入库数量", result)
            return result
        except Exception as e:
            logger.error("An error occurred: %s", e)
            raise

    #统计结果入库
    def save_to_mongo(self,title,count):
        collection=Data_save.save_con(host='192.168.3.149',port=27180,database='data_quality',collection='statistics')
        now = datetime.now()
        timestamp = int(now.timestamp())
        document = {
            title: {
                "timestamp": timestamp,
                "count": count
            }
        }
        Data_save.insert_one(collection,document)


monitor=monitorTools()