data_quality_server/es.py

#!/usr/bin/env python
# -*- coding:utf-8 -*-
# author : liumiaomiao
#从es库中导出数据到测试环境mongo库
from util.es_tools import esutil
from pymongo import MongoClient
def ES_bidding(es_query):
    """
    操作样例:直接拉取数据
    """
    db_config = {
        # es
        'es_host': '127.0.0.1',
        'es_port': 9800,
        'es_http_auth': ('test_dataQuality','t9s3Ed0gBBowf5'),  # 重新申请
        'timeout': 10000,
        # 'index': "projectset",
        'index': "bidding",
        'size': 1000,
        # mongo存的数据库表
        'mg_host': '192.168.3.206',
        'mg_port': 27080,
        'database': 'data_quality',
        'collection': 'bidding_20231221'
    }
    query = es_query
    # 传入查询语句query 以及配置信息
    esutil.es_query_save(query, **db_config)

def run():
    # 根据ES语句查找bidding
    es_query = {"track_total_hits": True,
                "query": {"bool": {"must": [{"range": {"publishtime": {"from": "1703001600", "to": "1703048399"}}}]}}}
    # es_query = {"track_total_hits": True,
    #             "query": {"bool": {"must": [{"range": {"publishtime": {"from": "1691337600", "to": "1691424000"}}},
    #                                         {"terms": {"subtype": ["中标", "合同","成交"]}}]}}}
    ES_bidding(es_query)

run()