项目背景:
由于日常监控的数据由Excel完成统计,需要大量重复的复制粘贴数据,效率地下,通过友盟的OpenAPI文档,开发一套Python脚本,自动更新友盟数据至表格中。
相关技术:
- JSON数据操作
- Pandas+openpyxl操作Excel文档
- datetime动态获取日期
- 面向对象编程
- logging模块记录程序运行日志
实现过程:
- 开发友盟数据API
- 通过引入友盟官方提供的Python SDK,获取新增、活跃、总用户量等常用指标,以字典的形式返回
- 所有的appkey、apiinfo放在配置文件中以变量存储,通过调用配置文件中的变量来实现传参,防止appkey等信息泄露
- 通过封装的方式,不同的APP使用不同的类进行封装
- 动态日期使用datetime模块获取,并以传参形式传入各个类中
- 使用logging模块记录程序运行日志,记录操作。
- 通过引入友盟官方提供的Python SDK,获取新增、活跃、总用户量等常用指标,以字典的形式返回
- 开发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)
网友评论