美文网首页Python
项目:Python自动获取友盟数据API

项目:Python自动获取友盟数据API

作者: cef802fb1136 | 来源:发表于2020-03-09 17:53 被阅读0次

项目背景:

​ 由于日常监控的数据由Excel完成统计,需要大量重复的复制粘贴数据,效率地下,通过友盟的OpenAPI文档,开发一套Python脚本,自动更新友盟数据至表格中。

相关技术:

-   JSON数据操作
-   Pandas+openpyxl操作Excel文档
-   datetime动态获取日期
-   面向对象编程
-   logging模块记录程序运行日志

实现过程:

  • 开发友盟数据API
    • 通过引入友盟官方提供的Python SDK,获取新增、活跃、总用户量等常用指标,以字典的形式返回
      • 所有的appkey、apiinfo放在配置文件中以变量存储,通过调用配置文件中的变量来实现传参,防止appkey等信息泄露
      • 通过封装的方式,不同的APP使用不同的类进行封装
      • 动态日期使用datetime模块获取,并以传参形式传入各个类中
      • 使用logging模块记录程序运行日志,记录操作。
  • 开发Python自动化操作Excel脚本,用于更新数据
    • 使用初始化类属性设定好需要操作的文件名、路径、工作表、字段名称
      • 使用OS模块拷贝副本,防止破坏源文件
      • 使用Pandas模块获取单列的最大行数,用于下一行(在最大行数的基础上+1)自增数据
      • 使用Openpyxl模块中的load_workbook方法操作Excel文档特定的工作表
      • 接收友盟API中传入的字典数据并进行字典数据解析
      • 将解析的数据对应工作表中的每一列,并自增数据
      • 最后保存

目录结构

目录结构

完整代码如下:

### 获取友盟数据

import aop, aop.api
import logging
import apiinfo, appkey    # 外部文件引入的方式引入开发者信息
from datetime import date, timedelta

# 收集日志
LOG_FORMAT = "%(asctime)s %(name)s %(levelname)s %(pathname)s %(message)s "
DATE_FORMAT = '%Y-%m-%d %H:%M:%S'
logging.basicConfig(level=logging.DEBUG, format=LOG_FORMAT, datefmt=DATE_FORMAT,
                    filename="./logs/umeng_get_datalog.log")

# 全局变量
date_1_curr = (date.today() + timedelta(days=-1)).strftime('%Y-%m-%d')
date_2_curr = (date.today() + timedelta(days=-2)).strftime('%Y-%m-%d')
date_3_curr = (date.today() + timedelta(days=-3)).strftime('%Y-%m-%d')


class GetWZmetroData(object):

    def __init__(self):
        aop.set_default_server('gateway.open.umeng.com')
        aop.set_default_appinfo(apiinfo.wzmetroApiKey, apiinfo.wzmetroApiSecurity)

    def getYesterdayData_ios(self):
        # 获取昨天IOS统计数据
        req = aop.api.UmengUappGetYesterdayDataRequest()
        try:
            resp = req.get_response(None, appkey=appkey.wzmetro_ios)
            logging.info('获取IOS昨天统计数据成功!')
            return resp

        except Exception as e:
            # 其它未知异常
            logging.error(f'其它未知异常:{e}')
            print(e)

    def getYesterdayData_android(self):
        # 获取昨天安卓统计数据
        req = aop.api.UmengUappGetYesterdayDataRequest()
        try:
            resp = req.get_response(None, appkey=appkey.wzmetro_android)
            logging.info('获取安卓昨天统计数据成功!')
            return resp

        except Exception as e:
            # 其它未知异常
            logging.error(f'其它未知异常:{e}')
            print(e)
            
            
if __name__ == '__main__':
    ds = GetWZmetroData()
    ios_data = ds.getYesterdayData_ios()
    print(ios_data)
    android_data = ds.getYesterdayData_android()
    print(android_data)
### 自动化Excel操作
from openpyxl import load_workbook
from datetime import date, timedelta
import pandas as pd
import os
import logging
from getUmengData import GetHSmetroData, GetWZmetroData

LOG_FORMAT = "%(asctime)s %(name)s %(levelname)s %(pathname)s %(message)s "
DATE_FORMAT = '%Y-%m-%d %H:%M:%S'
logging.basicConfig(level=logging.DEBUG, format=LOG_FORMAT, datefmt=DATE_FORMAT, filename="./logs/excel_operate.log")

# 创建对象
wz_ios_data = GetWZmetroData().getYesterdayData_ios()
wz_android_data = GetWZmetroData().getYesterdayData_android()


class Writing_to_excel(object):
    def __init__(self):
        """
        基本参数配置:时间参数、文件路径、源文件名、临时文件名、工作表名、字段名
        """
        self.data_ytd = (date.today() + timedelta(days=-1)).strftime("%Y/%m/%d")
        self.filepath = '/Users/jason/Desktop/'
        self.filename = 'test_data.xlsx'
        self.filename_replace = 'tmp_test_data.xlsx'
        self.sheet_name = '汇总'
        self.column_name = '日期'

    def get_next_row(self):
        """
        拷贝出新文件,获取行号
        防止破坏源文件
        :return: next_row
        """
        os.system(f'cp {self.filepath}{self.filename} {self.filepath}{self.filename_replace}')
        pd_data = pd.read_excel(f'{self.filepath}{self.filename_replace}', sheet_name=self.sheet_name)
        col1 = pd_data[[self.column_name]].dropna()
        _maxrow = col1.shape[0]
        next_row = _maxrow + 2
        pd_data.to_excel(f'{self.filepath}{self.filename_replace}', sheet_name=self.sheet_name)
        os.system(f'rm {self.filepath}{self.filename_replace}')
        self.next_row = next_row
        return

    def write_data_to_excel(self):
        """
        写入数据
        :return:
        """
        wb = load_workbook(f'{self.filepath}{self.filename}')
        stotal = wb[self.sheet_name]
        stotal[f'A{self.next_row}'] = self.data_ytd
        stotal[f'G{self.next_row}'] = wz_ios_data['yesterdayData']['newUsers'] + wz_android_data['yesterdayData']['newUsers']
        stotal[f'H{self.next_row}'] = wz_ios_data['yesterdayData']['activityUsers'] + wz_android_data['yesterdayData']['activityUsers']
        wb.save(f'{self.filepath}{self.filename}')


if __name__ == '__main__':
    try:
        wte = Writing_to_excel()
        wte.get_next_row()
        wte.write_data_to_excel()
        logging.info(f"{wte.filename}   文件数据操作完成")
        print(f"{wte.filename}  文件数据操作完成")
    except Exception as e:
        logging.error(f"{wte.filename}操作失败!原因:{e}")
        print(e)

相关文章

网友评论

    本文标题:项目:Python自动获取友盟数据API

    本文链接:https://www.haomeiwen.com/subject/xjapdhtx.html