使用python脚本统计es集群每天产生的数据大小

#!/usr/bin/evn python3
# -*-coding:utf-8-*-
# Author: zhaoyuanji
# Filename: es_size_everyday.py
# CreateTime: 2023/9/7 14:13

import os
from datetime import datetime
import xlsxwriter
from opensearchpy import OpenSearch
from collections import Counter


class get_store_day:

    def __init__(self,host,username,password):
        auth = (username,password)
        self.host = host
        self.client = OpenSearch(
            hosts=[{'host': host, 'port': 9200}],
            http_compress=True,  # enables gzip compression for request bodies
            http_auth=auth,
            use_ssl=True,
            verify_certs=True,
            ssl_assert_hostname=False,
            ssl_show_warn=False,
            ca_certs='root-ca.pem'
        )
        today = datetime.today().strftime('%Y%m%d')
        self.filename  = os.getenv('USERPROFILE') + "\\Desktop\\es_store_everyday_" + today + "_" + host + ".xlsx"

    def get_result(self):
        result = self.client.indices.stats(format="json", pretty=True)
        all_indices = []
        for indic in result['indices']:
            if not indic.startswith('.') and 'kibana' not in indic:
                k, v = indic.split('-')[-1], round(
                    result['indices'][indic]['total']['store']['size_in_bytes'] / 1024 / 1024, 2)   #获取索引大小，单位为MB
                all_indices.append({k: v})
        final_result = {}
        for sumar in all_indices:
            final_result = Counter(sumar) + Counter(final_result)
        # result = sorted(dict(final_result).items(), key=lambda item: item[0])

        return dict(final_result)

    def generate_xlxs(self):

        result = self.get_result()

        workbook = xlsxwriter.Workbook(self.filename)
        worksheet = workbook.add_worksheet()
        # 定义表头格式
        title_formatter = workbook.add_format({
            'bold': True,
            'border': 1,
            'align': 'center',
            'valign': 'vcenter',
            'fg_color': '#87CEFF'
        })

        # 定义内容格式
        data_formatter = workbook.add_format({
            'border': 1,
            'align': 'left',
            'valign': 'vcenter',
            'fg_color': '#EED8AE'
        })


        title = [u'日期', u'大小(MB)']
        worksheet.set_column('A:A', 10)
        worksheet.set_column('B:B', 10)
        worksheet.write_row("A1", title, title_formatter)

        summary = 0
        for i in range(1, len(result)):
            key = [key for key, value in result.items()][i-1]
            value = [value for key, value in result.items()][i-1]
            summary = summary + value
            worksheet.write_string(i, 0, key, data_formatter)
            worksheet.write_string(i, 1, str(round(value,2)), data_formatter)

        worksheet.write_string(len(result) , 0, "总大小", data_formatter)
        worksheet.write_string(len(result) , 1, str(round(summary, 2)), data_formatter)
        print("%s has been generated successfully with %d records." %( self.host , len(result)))
        # 将Pandas DataFrame写入Excel工作簿

        workbook.close()

        # 使用pandas进行排序操作，但是没有格式设置
        # df = pd.read_excel(self.filename)
        # df_sorted = df.sort_values(by='日期')
        # df_sorted.to_excel('new_filename.xlsx', index=False)
        # pd.read_excel('new_filename.xlsx', index_col=None).to_excel(self.filename, index=False)

if __name__ == '__main__':

    obj_uat = get_store_day('10.10.1.10','admin', 'admin')
    obj_pro = get_store_day('10.10.2.32','admin', 'admin')
    obj_uat.get_result()
    obj_uat.generate_xlxs()
    obj_pro.get_result()
    obj_pro.generate_xlxs()