elasticsearch.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
# @Version     : Python 3.11.4
# @Software    : Sublime Text 4
# @Author      : StudentCWZ
# @Email       : StudentCWZ@outlook.com
# @Date        : 2023/10/28 12:19
# @File        : elasticsearch.py
# @Description :
"""

import datetime

from application.extensions.init_elasticsearch import es


class ElasticsearchUtil:
    @classmethod
    def dsl(cls, _start: str, _end: str, size=10000) -> dict:
        """
        Setting dsl

        :param _start: start time
        :param _end: end time
        :param size: data number
        :return: dsl
        """
        _dsl = {
            "size": size,
            "query": {
                "bool": {
                    "must": [{
                        "match_phrase": {
                            "tag": {
                                "query": "global"
                            }
                        }
                    },
                        {
                        "range": {
                            "@timestamp": {
                                "gte": (datetime.datetime.strptime(_start, "%Y-%m-%d %H:%M:%S") -
                                        datetime.timedelta(hours=8)).strftime("%Y-%m-%dT%H:%M:%SZ"),
                                "lte": (datetime.datetime.strptime(_end, "%Y-%m-%d %H:%M:%S") -
                                        datetime.timedelta(hours=8)).strftime("%Y-%m-%dT%H:%M:%SZ")
                            }
                        }
                    }
                    ],
                    "filter": [{
                        "match_all": {}
                    }],
                    "should": [],
                    "must_not": []
                }
            }
        }
        return _dsl

    @classmethod
    def search(cls, _index: str, _dsl: dict, _scroll="5m") -> dict:
        """
        Search data

        :param _index: index name
        :param _dsl: dsl
        :param _scroll: scroll time
        :return: data after search
        """
        return es.search(index=_index, scroll=_scroll, body=_dsl)

    @classmethod
    def scroll_search(cls, _id,  _scroll="5m") -> dict:
        """
        Search data

        :param _id: scroll id
        :param _scroll: scroll time
        :return: data after search by scroll
        """
        return es.scroll(scroll_id=_id, scroll=_scroll, request_timeout=30)