共计 3798 个字符,预计需要花费 10 分钟才能阅读完成。
导读 | 最近领导安排让我每周定时把 grafana 导出的 csv 文件进行统计汇总工作,需要处理的 csv 文件还是蛮多的,况且还要每周重复汇总处理。干脆写个脚本,每周执行一遍脚本,既方便还不会出错。 |
一、需求分析
1. 原始文件分析
原始文件是多个 csv 表格,第一列为时间戳,每 10 分钟统计生成一行,其余列为 ip 地址在该时间段内的访问次数
2. 处理结果分析
根据要求,统计每个 ip 地址在当天访问次数求和,汇总生成新表格,结果如下,并将所有 csv 文件按照文件名,分别汇总到不同的 sheet 下
二、代码逻辑
1. 流程分析
- 首先遍历指定目录下的.csv 文件,提取文件名生成数组
- 然后使用 pandas 库读取 csv 文件,提取日期和 ip,然后统计每个 ip 当天访问次数,生成新的 DataFrame
- 最后使用 xlwings 库将 pandas 处理后的 DataFrame 数据写入 excel 文件,指定文件名作为 sheet 名
2. 遍历指定目录下.csv 文件
主要用到了 os 模块中的 walk() 函数,可以遍历文件夹下所有的文件名。
def find_csv(path):
"""
查找目录下 csv 文件
:param path: 查找 csv 的目录路径
:return: csv 文件名 list
"""
csv_file = []
for root, dirs, files in os.walk(path):
for file in files:
if os.path.splitext(file)[1] == '.csv':
csv_file.append(os.path.join(root, file))
return csv_file
3. pandas 处理 csv 文件
pandas 是 python 环境下最有名的数据统计包,对于数据挖掘和数据分析,以及数据清洗等工作,用 pandas 再合适不过了,官方地址:https://www.pypandas.cn/
def summary_data(file):
"""
grafana 导出的 csv 文件处理汇总
:param file: csv 文件路径
:return: 处理完成后的 pandas 对象
"""
# 读取整个 csv 文件
csv_data = pd.read_csv(file, ';')
# 提取日期
csv_data["Time"] = csv_data["Time"].map(lambda Time: Time[0:10])
date = csv_data["Time"].drop_duplicates()
# 提取 IP
ip_list = csv_data.columns.values[1:]
# 生成新列表
result_data = []
for day in list(date):
ip_data = []
for ip in ip_list:
# 统计指定 ip 地址在指定日期的数据之和
ip_sum = csv_data.loc[csv_data['Time'] == day, ip].sum()
ip_data.append(ip_sum)
# print("日期:%s ip:%s 总计:%s" % (day, ip, ip_sum))
result_data.append(ip_data)
# 生成新的 DataFrame
result_df = pd.DataFrame(result_data, index=list(date), columns=ip_list)
# 添加行列统计
result_df['day_sum'] = result_df.apply(lambda x: x.sum(), axis=1)
result_df.loc['ip_sum'] = result_df.apply(lambda x: x.sum())
print(file, "处理完毕!")
return result_df
4. excel 数据写入
pandas 的 to_excel 方法也可以写入到 excel 文件,但是如果需要写入到指定的 sheet,就无法满足需求了,此时就需要用的 xlwings 或者 openpyxl 库,此处使用 xlwings,参考文档:https://www.xlwings.org/pro
def save_excel(data_df, file_name, excel_name):
"""
生成并写入新 excel 文件
:param data_df: pandas 数据对象
:param file_name: 传入文件名,作为生成的 sheet 名称
:param excel_name: 生成 excel 文件名
:return: null
"""sheet_name = file_name[file_name.rfind('/', 1) + 1:file_name.rfind('.', 1)]
wb = xlwings.Book(excel_name)
sheet = wb.sheets.add(name=sheet_name)
sheet.range("A1").value = data_df
wb.save()
wb.close()
print(sheet_name, "Sheet 写入完毕!")
5. 完整代码
import os import pandas as pd import xlwingsdef find_csv(path): """ 查找目录下 csv 文件 :param path: 查找 csv 的目录路径 :return: csv 文件名 list """ csv_file = [] for root, dirs, files in os.walk(path): for file in files: if os.path.splitext(file)[1] == '.csv': csv_file.append(os.path.join(root, file)) return csv_file
def summary_data(file): """ grafana 导出的 csv 文件处理汇总 :param file: csv 文件路径 :return: 处理完成后的 pandas 对象 """ # 读取整个 csv 文件 csv_data = pd.read_csv(file, ';') # 提取日期 csv_data["Time"] = csv_data["Time"].map(lambda Time: Time[0:10]) date = csv_data["Time"].drop_duplicates() # 提取 IP ip_list = csv_data.columns.values[1:] # 生成新列表 result_data = [] for day in list(date): ip_data = [] for ip in ip_list: ip_sum = csv_data.loc[csv_data['Time'] == day, ip].sum() ip_data.append(ip_sum) # print("日期:%s ip:%s 总计:%s" % (day, ip, ip_sum)) result_data.append(ip_data) result_df = pd.DataFrame(result_data, index=list(date), columns=ip_list) # 添加行列统计 result_df['day_sum'] = result_df.apply(lambda x: x.sum(), axis=1) result_df.loc['ip_sum'] = result_df.apply(lambda x: x.sum()) print(file, "处理完毕!") return result_dfdef save_excel(data_df, file_name, excel_name): """ 生成并写入新 excel 文件 :param data_df: pandas 数据对象 :param file_name: 传入文件名,作为生成的 sheet 名称 :param excel_name: 生成 excel 文件名 :return: null """sheet_name = file_name[file_name.rfind('/', 1) + 1:file_name.rfind('.', 1)] wb = xlwings.Book(excel_name) sheet = wb.sheets.add(name=sheet_name) sheet.range("A1").value = data_df wb.save() wb.close() print(sheet_name, "Sheet 写入完毕!")
if __name__ == '__main__': # 原始 csv 文件存放路径 path = './csv' # 生成 excel 文件名 excel_name = 'cm.xlsx' csv_file = find_csv(path) # 创建 excel 文件 new_excel = pd.DataFrame() new_excel.to_excel(excel_name) # 处理并写入 excel 文件 for file in csv_file: data_df = summary_data(file) save_excel(data_df, file, excel_name) # 删除默认 Sheet1 wb = xlwings.Book(excel_name) wb.sheets['Sheet1'].delete() wb.save() wb.close() print("数据汇总完毕, 生成文件路径 %s/%s" % (os.getcwd(), excel_name))
正文完
星哥说事-微信公众号
![post-qrcode](https://www.xgss.net/wp-content/uploads/2023/07/二维码-微信公众号256x256.jpg)