#!/usr/bin/python # encoding=utf-8 # author: tangwy from __future__ import division import json import os, re import codecs import traceback from datetime import datetime, timedelta from collections import defaultdict from dashboard_data_conversion import ip_summary_data_format, account_summary_data_format, \ interface_summary_data_format, menu_summary_data_format, adjust_times, jobnum_region_dict,find_region_by_code from dataInterface.functions import CFunction from dataInterface.db.params import CPgSqlParam from ext_logging import logger TABLE_NAME = "ueba_analysis_schema.logs" DATA_TYPE = { "IP": 1, "ACCOUNT": 2, "INTERFACE": 3, "MENU": 4, } def pg_get_ip_group_data(startTime, endTime): """ IP维度查询 :param startTime: 开始时间, :param endTime: 结束时间, """ result = [] sql = """ select ip, jobnum, sum(count) from {TABLE_NAME} where logdate >= %s and logdate <= %s and data_type = %s group by ip, jobnum""".format(TABLE_NAME=TABLE_NAME) res = json.loads(CFunction.execute(CPgSqlParam(sql, params=(startTime, endTime, DATA_TYPE["IP"])))) if res: for item in res: result.append({ "ip": item[0], "jobnum": item[1], "count": item[2], }) return result def pg_get_account_group_data(startTime, endTime): """ 账号维度查询 :param startTime: 开始时间, :param endTime: 结束时间, """ result = [] sql = """ select account, jobnum, sum(count) from {TABLE_NAME} where logdate >= %s and logdate <= %s and data_type = %s group by account, jobnum""".format(TABLE_NAME=TABLE_NAME) res = json.loads(CFunction.execute(CPgSqlParam(sql, params=(startTime, endTime, DATA_TYPE["ACCOUNT"])))) if res: for item in res: # print("items:"+json.dumps(item)) # print("pg_get_account_group_data维度data："+json.dumps(item[0])) result.append({ "account": item[0], "jobnum": item[1], "count": item[2], }) return result def pg_get_interface_group_data(startTime, endTime): """ 接口维度查询 :param startTime: 开始时间, :param endTime: 结束时间, """ result = [] sql = """ select interface, ip, jobnum,account, sum(count) from {TABLE_NAME} where logdate >= %s and logdate <= %s and data_type = %s group by interface, ip, jobnum,account""".format(TABLE_NAME=TABLE_NAME) res = json.loads(CFunction.execute(CPgSqlParam(sql, params=(startTime, endTime, DATA_TYPE["INTERFACE"])))) if res: for item in res: result.append({ "interface": item[0], "ip": item[1], "jobnum": item[2], "account": item[3], "count": item[4], }) return result def pg_get_menu_group_data(startTime, endTime): """ 菜单维度查询 :param startTime: 开始时间, :param endTime: 结束时间, """ result = [] sql = """ select menu, ip,jobnum,account,sum(count) from {TABLE_NAME} where logdate >= %s and logdate <= %s and data_type = %s group by menu, ip, jobnum ,account""".format(TABLE_NAME=TABLE_NAME) res = json.loads(CFunction.execute(CPgSqlParam(sql, params=(startTime, endTime, DATA_TYPE["MENU"])))) if res: for item in res: result.append({ "menu": item[0], "ip": item[1], "jobnum": item[2], "account": item[3], "count": item[4], }) return result def pg_get_account_previous_company_count(startTime, endTime, data_type): """ 账号维度查询菜请求次数 :param startTime: 开始时间, :param endTime: 结束时间, :param data_type: 公司聚合类型 ACCOUNT or IP , """ result = defaultdict(int) if data_type in DATA_TYPE: data_type = DATA_TYPE[data_type] sql = """ select jobnum, account,sum(count) from {TABLE_NAME} where logdate >= %s and logdate <= %s and data_type = %s group by jobnum,account""".format(TABLE_NAME=TABLE_NAME) res = json.loads(CFunction.execute(CPgSqlParam(sql, params=(startTime, endTime, data_type)))) if res: for item in res: company = find_region_by_code(item[0], jobnum_region_dict) result[company] += item[2] return result def pg_get_ip_previous_company_count(startTime, endTime, data_type): """ 账号维度查询菜请求次数 :param startTime: 开始时间, :param endTime: 结束时间, :param data_type: 公司聚合类型 ACCOUNT or IP , """ result = defaultdict(int) if data_type in DATA_TYPE: data_type = DATA_TYPE[data_type] sql = """ select jobnum,ip, sum(count) from {TABLE_NAME} where logdate >= %s and logdate <= %s and data_type = %s group by jobnum,ip """.format(TABLE_NAME=TABLE_NAME) res = json.loads(CFunction.execute(CPgSqlParam(sql, params=(startTime, endTime, data_type)))) if res: for item in res: company = find_region_by_code(item[0], jobnum_region_dict) result[company] += item[2] return result def pg_get_previous_interface_count(startTime, endTime): """ 接口维度查询请求总次数 :param startTime: 开始时间, :param endTime: 结束时间, """ result = defaultdict(int) sql = """ select interface, ip, jobnum,account, sum(count) from {TABLE_NAME} where logdate >= %s and logdate <= %s and data_type = %s group by interface, ip, jobnum,account""".format(TABLE_NAME=TABLE_NAME) res = json.loads(CFunction.execute(CPgSqlParam(sql, params=(startTime, endTime, DATA_TYPE["INTERFACE"])))) if res: for item in res: result[item[0]] += item[4] return result def pg_get_previous_menu_count(startTime, endTime): """ 菜单维度查询请求总次数 :param startTime: 开始时间, :param endTime: 结束时间, """ result = defaultdict(int) sql = """ select menu, ip,jobnum,account, sum(count) from {TABLE_NAME} where logdate >= %s and logdate <= %s and data_type = %s group by menu, ip,jobnum,account""".format(TABLE_NAME=TABLE_NAME) res = json.loads(CFunction.execute(CPgSqlParam(sql, params=(startTime, endTime, DATA_TYPE["MENU"])))) if res: for item in res: result[item[0]] += item[4] return result def entry(data_type, start, end): # 前一段开始时间 date_format = "%Y-%m-%d %H:%M:%S" start = datetime.strptime(start, date_format) end = datetime.strptime(end, date_format) start = start.strftime('%Y-%m-%d') end = end.strftime('%Y-%m-%d') old_start,old_end = adjust_times(start, end) # print (old_start) # print (old_end) data = {} if data_type == "1": ip_summary_data = pg_get_ip_group_data(start, end) data = ip_summary_data_format(ip_summary_data) previous_company_dict = pg_get_ip_previous_company_count(old_start, old_end, "IP") for d in data["summary"]["ip"]: if previous_company_dict.get(d["company"], 0) == 0: d["trend"] = 0 # print("xxx") else: # print("dddd:"+str(d["req_frequency"])) # print(previous_company_dict.get(d["company"], 0)) d["trend"] = round( (d["req_frequency"] - previous_company_dict.get(d["company"], 0)) / previous_company_dict.get( d["company"], 0), 4) if data_type == "2": logger.info("准备获取账号维度数据") account_summary_data = pg_get_account_group_data(start, end) logger.info("准备做数据转换") data = account_summary_data_format(account_summary_data) logger.info("准备获取前一周期数据") previous_company_dict = pg_get_account_previous_company_count(old_start, old_end, "ACCOUNT") logger.info("完成前一周期数据") for d in data["summary"]["account"]: if previous_company_dict.get(d["company"], 0) == 0: d["trend"] = 0 else: # print("dddd:"+str(d["req_frequency"])) # print(previous_company_dict.get(d["company"], 0)) d["trend"] = round( (d["req_frequency"] - previous_company_dict.get(d["company"], 0)) / previous_company_dict.get( d["company"], 0), 4) if data_type == "3": logger.info("准备获取菜单维度数据") interface_summary_data = pg_get_interface_group_data(start, end) logger.info("准备做数据转换") data = interface_summary_data_format(interface_summary_data) logger.info("准备获取前一周期数据") previous_interface_dict = pg_get_previous_interface_count(old_start, old_end) logger.info("完成前一周期数据") for d in data["summary"]["interface"]: if previous_interface_dict.get(d["interface_addr"], 0) == 0: d["trend"] = 0 else: # print("dddd:"+str(d["req_frequency"])) # print(previous_interface_dict.get(d["interface_addr"], 0)) d["trend"] = round( (d["req_frequency"] - previous_interface_dict.get(d["interface_addr"], 0)) / previous_interface_dict.get( d["interface_addr"], 0), 4) logger.info("准备数返回") if data_type == "4": logger.info("准备获取菜单维度数据") menu_summary_data = pg_get_menu_group_data(start, end) logger.info("准备做数据转换") data = menu_summary_data_format(menu_summary_data) logger.info("准备获取前一周期数据") previous_menu_dict = pg_get_previous_menu_count(old_start, old_end) logger.info("完成前一周期数据") for d in data["summary"]["menu"]: if previous_menu_dict.get(d["menu_name"], 0) == 0: d["trend"] = 0 else: # print("dddd:"+str(d["req_frequency"])) # print(previous_menu_dict.get(d["menu_name"], 0)) d["trend"] = round( (d["req_frequency"] - previous_menu_dict.get(d["menu_name"], 0)) / previous_menu_dict.get( d["menu_name"], 0), 4) logger.info("准备数返回") return data # res = entry("1","2024-07-01 00:00:00","2024-07-20 23:59:59") # # print res # logger.info(json.dumps(res)) # res = entry("2","2024-07-01 00:00:00","2024-07-20 23:59:59") # logger.info(json.dumps(res)) # res = entry("3","2024-07-01 00:00:00","2024-07-20 23:59:59") # logger.info(json.dumps(res)) # res = entry("4","2024-07-01 00:00:00","2024-07-20 23:59:59") # logger.info(json.dumps(res))