Python之数据结构改造（二）

冷不防 2022-04-25 10:32 395阅读 0赞

**背景：**

> 因为日志易处理的数据和听云吐出的数据不一致，无法解析画图，所以需要进行数据结构改造。

**思路：**

> 根据日志易解析要求，需要的数据结构如下：

{'aggregateBy': 1, 'appIns_id': 66, 'yAxis': {'gridLineColor': '#c0c0c0', 'gridLineDashStyle': 'ShortDot', 'min': 0.0, 'title': {'text': None, 'style': {'color': '#000000'}}, 'labels': {'color': '#000000'}, 'gridLineWidth': 1, 'tickColor': '#000000', 'tickUnit': '%'}, 'series': {'name': 'Error rate', 'color': '#3598db', 'type': 'area', 'params': '7', 'marker': {'symbol': 'circle', 'radius': 4}, 'data': {'single': {'y': 0.0, 'x': 1559286660000, 'tooltip': '{"title":"05-31 15:11--05-31 15:12","data":[{"title":"Error rate","value":"0.0","unit":"%"},{"title":"Errors","value":"0"},{"title":"Traffic","value":"2"}]}'}}}, 'chart': {'spacingTop': 15, 'height': 300, 'spacingRight': 10, 'width': 375, 'plotBackgroundColor': '#ffffff', 'backgroundColor': '#ffffff', 'spacingBottom': 10, 'plotBorderColor': '#575757', 'type': 'area', 'plotBorderWidth': 1, 'spacingLeft': 0}, 'xAxis': {'gridLineColor': '#c0c0c0', 'gridLineDashStyle': 'ShortDot', 'title': {'text': None, 'style': {'color': '#000000'}}, 'labels': {'color': '#000000'}, 'dateTimeLabelFormats': {'millisecond': '%H:%M', 'week': '%H:%M', 'hour': '%H:%M', 'month': '%H:%M', 'second': '%H:%M', 'year': '%H:%M', 'day': '%H:%M', 'minute': '%H:%M'}, 'gridLineWidth': 1, 'tickColor': '#000000', 'minTickInterval': 60000, 'type': 'datetime'}, 'appId': 14, 'legend': {'align': 'center', 'verticalAlign': 'bottom'}, 'aggregateValue': 12.300000000000001}

> 鉴于上述数据结构都是json格式，可以利用Py字典键不可重复的特性对数据进行改造，程序见下面：

**要改造的数据：**

**处理程序：**

#!/usr/bin/env python
    # -*- coding:utf-8 -*-
    import json
    import sys
    import time
    import requests
    from kafka import KafkaProducer
    
    
    reload(sys)
    sys.setdefaultencoding('utf8')
    
    
    name = '账号'
    pawd = '密码'
    doLoginurl = 'http://xxx.xxx.xxx/lens/manager/doLogin'
    toserver = 'http://xxx.xxx.xxx/server/manager/platformLogin?target=/manager/index&product=server'
    instancesxurl = 'http://xxx.xxx.xxx/server/application/instances'
    errors_url = 'http://xxx.xxx.xxx/server/highcharts-chart-data/LATEST/accounts/applications/charts/application-errors.json'
    resp_time_url = 'http://xxx.xxx.xxx/server/highcharts-chart-data/LATEST/accounts/applications/charts/application-application.json'
    
    
    p=KafkaProducer(bootstrap_servers=['xxx:9092','xxx:9092','xxx:9092'])
    topic='tingyun4rzy'
    
    
    def do_login():
        """
        登录听云
        :return:
        """
        sess = requests.session()
        sess.post(doLoginurl, data={'name': name, 'password': pawd})
        # 获取csrftoken
        res = sess.get(toserver)
        csrftoken = res.text.split('name="csrftoken"')[1].split('"')[3]
        res.request.headers['csrftoken'] = csrftoken
        headers = res.request.headers
    
        return sess, headers
    
    
    def get_app_ids():
        # 发送请求获取cookie
        s = requests.Session()
        pre_login = 'http://xxx.xxx.xxx/lens/login'
        s.get(pre_login)
    
        login_url = 'http://xxxx.xxx.xxx/server/manager/doLogin?name=账号&password=密码'
        s.get(login_url)
    
        post_login = 'http://xxx.xxx.xxx/server/manager/platformLogin?target=/manager/index&product=server'
        ret = s.get(login_url)
        # print ret
        header = {"X-Auth-Key": "xxx-xxx-xxx"}
    
        # 第四步：带上X-Auth-Key请求某个具体的API接口
        url_app_list = 'http://xxx.xxx.xxx/server/data/application/allApps?X-Auth-Key=xxx-xxx-xxx'
        ret = s.get(url_app_list)
        # 将所有的app信息加入kafka
        # x=p.send(topic,json.dumps(ret.json(),))
    
        ret = json.loads(ret.text)
        ids = []
        for i in ret['appList']:
            ids.append(i['id'])
        return ids
    
    
    def get_app_instances(id):
        sess, headers = do_login()
        data = {'applicationId': id, 'timePeriod': '30', 'timeZone': 'GMT+0800'}
        instance_res = sess.post(instancesxurl, data=data, headers=headers).text
        #index_res = sess.post('http://xxx.xxx.xxx/server/overview/newAppList/' + str(id) + '/list',data={'timePeriod': '30'}, headers=headers).text
        #####print('*=*instance_res*=*======================================================', instance_res)
        #if index_res:
        #    data = json.loads(index_res)
        #    x = p.send(topic, json.dumps(data, ))
        if instance_res:
            data = json.loads(instance_res)
            appIns_ids = []
            for i in data:
                appIns_ids.append(i['id'])
            #print(appIns_ids)
            #if appIns_ids:
            #    print('==>>>',appIns_ids)
            return appIns_ids 
    
    def get_app_errors(app_id):
        """
        获取app错误率
        """
        sess, headers = do_login()
        data = {'applicationId': app_id, 'timePeriod': '30', 'timeZone': 'GMT+0800', 'timeType': 1}
        errors_res = sess.post(errors_url, data=data, headers=headers).text
        #print('*=*errors_res*=*=====================================================', errors_res)
        if errors_res:
            data = json.loads(errors_res)
            data['appId'] = app_id
            data['xAxis'] = data.get('xAxis')[0]
            data['yAxis'] = data.get('yAxis')[0]
            #print(data)
            if data.get('series'):
                data['series'] = data.get('series')[0]
                #print(data.get('series').get('data'))
                data_list = data.get('series').get('data')
                dt_dict = {}
                for idx,val in enumerate(data_list):
                    dt_dict.update({'single':val})
                    #dt_dict.update({val.get('x'):val})
                     
                    data['series']['data'] = dt_dict
                    #print(data)
                    #处理获取的数据
                    x = p.send(topic, json.dumps(data, ))
    
    
    def get_appIns_errors(app_id,appIns_id):
        """
        获取实例错误率
        """
        sess, headers = do_login()
        data = {'applicationId': app_id,'instanceId':appIns_id, 'timePeriod': '30', 'timeZone': 'GMT+0800', 'timeType': 1}
        errors_res = sess.post(errors_url, data=data, headers=headers).text
        if errors_res:
            data = json.loads(errors_res)
            data['appId'] = app_id
            data['appIns_id'] = appIns_id
            data['xAxis'] = data.get('xAxis')[0]
            data['yAxis'] = data.get('yAxis')[0]
            if data.get('series'):
                data['series'] = data.get('series')[0]
                data_list = data.get('series').get('data')
                dt_dict = {}
                for idx,val in enumerate(data_list):
                    dt_dict.update({'single':val})
                    data['series']['data'] = dt_dict
                    x = p.send(topic, json.dumps(data, ))
                    print('dooooooooooooooneeeeeeeeeeee',data)
    
    
    def get_app_resp_time(app_id,appIns_id):
        """
        获取应用服务器响应时间
        """
        sess, headers = do_login()
        data = {'applicationId': app_id,'instanceId':appIns_id, 'timePeriod': '30', 'timeZone': 'GMT+0800', 'timeType': 1}
        resp_res = sess.post(resp_time_url, data=data, headers=headers).text
        if resp_res:
            data = json.loads(resp_res)
            data['appId'] = app_id
            data['appIns_id'] = appIns_id
            data['xAxis'] = data.get('xAxis')[0]
            data['yAxis'] = data.get('yAxis')[0]
            if data.get('series'):
                #data['series'] = data.get('series')[0]
                for i in data['series']:
                    data['series'] = i
                    data['factor'] = data['series']['name']
                    data_list = data.get('series').get('data')
                    dt_dict = {}
                    for idx,val in enumerate(data_list):
                        dt_dict.update({'single':val})
                        data['series']['data'] = dt_dict
                        x = p.send(topic, json.dumps(data, ))
                        print(data['series']['name'],'=====>>>',data)
                        
        
    if __name__ == '__main__':
        app_ids = get_app_ids()
        while 1:
            for app_id in app_ids:
                appIns_ids = get_app_instances(app_id)
                #print(appIns_ids)
                if appIns_ids:
                    for appIns_id in appIns_ids:
                        get_appIns_errors(app_id,appIns_id)
                        #get_app_resp_time(app_id,appIns_id)
                #get_app_errors(app_id)
    
            time.sleep(120)