pymongo統計資料庫並寫入excel
阿新 • • 發佈：2020-11-24
excel.py如下
# encoding:utf-8 from bson.objectid import ObjectId import pymongo import time import xlwt from operator import itemgetter import collections from tools import * client = pymongo.MongoClient(host='10.x.x.75', port=27017) # 統計ABCDE_db資料庫裡機構的基本資訊 db = client.ABCDE_db collection = db.organizations result= collection.find() #org_nums = result.count() org_nums = collection.estimated_document_count() def gen_row_info(): for item in result[1:]: # 客戶名稱 org_name = item.get("name") # 建立者 creator = item.get("creator") # 郵箱 email = item.get("email")# 建立時間 create_time = timeFormat(item.get("createTime")) # 更新時間 update_time = timeFormat(item.get("updateTime")) # 客戶資料庫名 org_db_name = nameFormat(item.get("_id")) # 資料庫大小 org_data_size = data_size(org_db_name) print(org_db_name)# 使用者數量 user_num = count_simple(org_db_name, "users") # 現場數量 site_num = count_simple(org_db_name, "site") # 閘道器數量 device_num = count_simple(org_db_name, "device") # 線上裝置量 online_device_num = count_complex(org_db_name, "device", {"online": 1}) # 是否大客戶，閘道器數量大於20臺 is_big_customer = "是" if device_num > 20 else "否" # 是否有許可權分組 is_group = judge_simple(org_db_name, "groups") # 是否有自定義角色 is_userdefined_role = judge_complex(org_db_name, "roles", {'name': {'$nin': ["admin", "DeviceManager", "DeviceSense"]}}) # 是否有標籤 is_tag = judge_simple(org_db_name, "tags") # 是否有流量預警 is_data_alarm = judge_complex(org_db_name, "alarm.config", {'emailList': {'$ne': ""}}) # 是否有告警 is_alarm = judge_simple(org_db_name, "alarms") # 是否有控制器 is_machine = judge_simple(org_db_name, "machine") # 是否有組態畫面 is_scada = judge_simple(org_db_name, "scada") # 是否有遠端維護 is_channel = judge_simple(org_db_name, "channel") # 是否有遠端命令 is_task = judge_complex(org_db_name, "tasks", {"type": 2}) # 裝置型號 device_model_details = device_model_detail(org_db_name, "device") row_info = collections.OrderedDict() row_info['org_name'] = org_name row_info["email"] = email row_info["creator"] = creator row_info["creater_time"] = create_time row_info["update_time"] = update_time row_info["org_data_size"] = org_data_size row_info["user_num"] = user_num row_info["site_num"] = site_num row_info["device_num"] = device_num row_info["online_device_num"] = online_device_num row_info["is_big_customer"] = is_big_customer row_info["is_group"] = is_group row_info["is_userdefined_role"] = is_userdefined_role row_info["is_tag"] = is_tag row_info["is_data_alarm"] = is_data_alarm row_info["is_alarm"] = is_alarm row_info["is_machine"] = is_machine row_info["is_scada"] = is_scada row_info["is_channel"] = is_channel row_info["is_task"] = is_task row_info["device_model_details"] = device_model_details yield row_info f = xlwt.Workbook() sheet1 = f.add_sheet("sheet1", cell_overwrite_ok=True) style = xlwt.XFStyle() style.alignment = get_alignment() # 寫入標題, 合併單元格 title1 = ( "客戶名稱", "郵箱", "建立者", "建立時間", "更新時間", "資料庫大小", "使用者數量", "現場數量", "閘道器數量", "線上裝置總量", "是否大客戶" ) title2 = ( "許可權分組", "自定義角色", "標籤", "流量預警", "告警", "組態畫面", "控制器", "遠端維護", "遠端命令" ) column1 = 0 for t in title1: sheet1.write_merge(10, 12, column1, column1, t, style) column1 += 1 sheet1.write_merge(10, 11, 11, 19, "功能使用情況", style) column2 = 11 for t in title2: sheet1.write(12, column2, t, style) column2 += 1 sheet1.write_merge(10, 12, 20, 60, "裝置型號", style) # 裝置型號資料單元格合併 j = 13 while j < org_nums+10: sheet1.write_merge(j, j, 20, 60) j += 1 # 寫入資料 info = gen_row_info() row_list = list(info) rows_by_online = sorted(row_list, key=itemgetter('online_device_num'), reverse=True) #print(rows_by_online) # 從excle的第14行開始寫入資料 row = 13 for d in rows_by_online: col = 0 for one in d.values(): sheet1.write(row, col, one) col += 1 row += 1 f.save('test2.xls')
自定義函式tools.py
import xlwt import time from bson.objectid import ObjectId import pymongo client = pymongo.MongoClient(host='10.x.x.75', port=27017) def get_alignment(): alignment = xlwt.Alignment() # 0x01(左端對齊)、0x02(水平方向上居中對齊)、0x03(右端對齊) alignment.horz = 0x02 # 0x00(上端對齊)、 0x01(垂直方向上居中對齊)、0x02(底端對齊) alignment.vert = 0x01 # 設定自動換行 # alignment.wrap = 1 return alignment def timeFormat(x): x = time.localtime(x) return time.strftime("%F", x) def nameFormat(x): x = str(x) x = x.upper()+"_db" return x def count_simple(db, coll): org_db = client[db] coll = org_db[coll] num = coll.estimated_document_count() return num def count_complex(db, coll, exp): org_db = client[db] coll = org_db[coll] num = coll.count_documents(exp) return num def judge_simple(db, coll): org_db = client[db] coll = org_db[coll] num = coll.estimated_document_count() y = "是" if num > 0 else "否" return y def judge_complex(db, coll, exp): org_db = client[db] coll = org_db[coll] num = coll.count_documents(exp) y = "是" if num > 0 else "否" return y def device_model_detail(db, coll): org_db = client[db] coll = org_db[coll] result = coll.aggregate([{'$group': {'_id': '$model', 'modelCount': {'$sum': 1}}}]) model_list = [] for i in result: model_dict_list = [(k, v) for k, v in i.items()] model_type = model_dict_list[0][1] model_num = str(model_dict_list[1][1]) model_count = "{}({})".format(model_type, model_num) model_list.append(model_count) device_model_details = ",".join(model_list) return device_model_details def data_size(db): org_db = client[db] stats = org_db.command("dbstats") size = stats.get("storageSize")/1024/1024 return round(size)