Python爬虫解析MeterSphere平台执行报告并自动入库实战

目标与准备条件

本文旨在解决手动导出并整理MeterSphere大量场景执行报告的效率问题。通过编写Python自动化脚本，直接对接平台内部API获取JSON响应，自动完成数据清洗、状态判别与落库。运行前请确保已安装核心依赖包：requests（网络请求）、xlrd（表格读取）与xlutils（含copy方法实现追加写入）。同时准备好待处理的源表格文件（如autoinsurancereport.xls），并确认目标接口地址及鉴权Header权限无误。

第1步：遍历场景调用报告API

目的：批量拉取各场景下的原始执行日志，为后续结构解析提供数据源。
入口：已知的业务场景列表变量与平台API网关基址。
动作：使用循环遍历场景索引，动态拼接报告查询URL，携带认证Header发起GET请求。统一声明响应编码格式以避免中文乱码。
重要参数：响应编码设置为UTF-8，Header需包含有效的Authorization或Cookie字段。
可观察的完成状态：控制台打印出完整的场景节点URL，且内存中保留未经处理的JSON文本流。

# 遍历每个场景下API
for TestReporTGGet in range(0, TestReporTgCount001):
    TestReporTGName = TestReporTG[TestReporTGGet]
    
    # 特殊符号过滤与斜杠清理，防止路径拼接异常
    TestReporTGNameFiltration = re.sub(r'[!"#$%&\'*+,./:;<=>?@，。?★、…【】《》？“”‘！[\\]^_`{|}~\s]+', "",
                                       TestReporTGName)
    TestReporTGNameFiltration01 = re.sub(r'\\', "", TestReporTGNameFiltration)
    
    # 获取响应体url
    ResponseUrl = f"http://ms.***.*****:****/api/scenario/report/selectReportContent/{TestReporTGNameFiltration01}"
    print("场景下节点地址："+ResponseUrl)
    
    ResponseUrlResp = requests.get(url=ResponseUrl, headers=TestReportHeaders)
    ResponseUrlResp.encoding = "utf-8"
    ResponseUrlRespGet = ResponseUrlResp.text
    
    # 获取每个流程节点名字，进行过滤
    TestResponseResp = re.findall('","name":"(.+?)",', ResponseUrlRespGet)
    TestResponseRespGet=str(TestResponseResp)
    TestResponseRespGetFiltration=re.sub(r'[a-zA-Z0-9!"#$%&\'*+,./:;<=>?@，。?★、…【】《》？“”‘！[\\]^_`{|}~\s]+', "", TestResponseRespGet)
    print("根节点处理报告名称:" + TestResponseRespGetFiltration)
    
    #获取根节点成功或失败
    RootNodeSucceed=re.findall('"pass":(.+?)}]}}}',ResponseUrlRespGet)
    RootNodeSucceedGet=str(RootNodeSucceed)
    print("根节点失败或成功状态标识："+RootNodeSucceedGet)
    
    # 获取根节点返参数据
    ReturnParameter=re.findall('"body":"{(.+?)]}}}',ResponseUrlRespGet)
    ReturnParameterGet=str(ReturnParameter)
    ReturnParameterGettest=re.sub(r'\\', "", ReturnParameterGet)
    print("过滤后的报文字段"+ReturnParameterGettest)
    
    Namenge_list = list(ReturnParameterGet)
    is_contain_ch29 = '"error":0' in ResponseUrlRespGet
    
    #获取投保单号
    InsureGain=re.findall('proposalNo：(.+?)nqueryCode',ResponseUrlRespGet)
    InsureGainGet=str(InsureGain)
    InsureGainGetDispose=re.sub(r'[!"#$%&\'*+,./:;<=>?@，。?★、…【】《》？“”‘！[\\]^_`{|}~\s]+', "",InsureGainGet)
    InsureGainGetDispose01=re.sub(r'\\', "",InsureGainGetDispose)
    print("投保单号:",InsureGainGetDispose01)
    
    #获取报价单号
    PriceSheet=re.findall('quotationNo：(.+?)nquotationNo1：',ResponseUrlRespGet)
    PriceSheetGet=str(PriceSheet)
    PriceSheetGetDispose=re.sub(r'[!"#$%&\'*+,./:;<=>?@，。?★、…【】《》？“”‘！[\\]^_`{|}~\s]+', "",PriceSheetGet)
    PriceSheetGetDispose01=re.sub(r'\\', "",PriceSheetGetDispose)
    print("报价单号：",PriceSheetGetDispose01)
    
    #获取orderMessageList
    orderMessageList=re.findall('"assertions":(.+?)"pass":true',ResponseUrlRespGet)
    orderMessageListGet=str(orderMessageList)
    print("获取orderMessageList：",orderMessageListGet)
    print("===============================================================================")
    print("是否为：",is_contain_ch29)
    print("根节点的接口:"+ReportParticularsUrl)
    print("===============================================================================")

第2步：正则清洗与核心字段提取

目的：从非结构化JSON文本中精准剥离业务关键指标，剔除干扰字符。
入口：第1步缓存的ResponseUrlRespGet字符串对象。
动作：使用re.findall结合模式匹配定位节点名、根节点通过标识、返回报文主体，以及特定业务字段（投保单号、报价单号、断言消息）。对提取结果再次应用正则清理多余标点与空白符。
重要参数：正则表达式需保持原始字符串前缀r以防转义冲突；关键判定条件设为'"error":0' in ResponseUrlRespGet用于快速区分执行成败。
可观察的完成状态：控制台输出清晰过滤后的节点名称、状态标识、单据编号及消息列表，无乱码或越界截断。

第3步：状态校验与Excel持久化存储

目的：将解析完毕的结构化数据按标准表头格式写入本地电子表格，形成可回溯的运行档案。
入口：已打开的工作簿对象、当前系统时间戳、清洗后的各字段变量。
动作：锁定源文件首个Sheet，计算末行索引作为新数据起始位置。创建表头列（场景、报告名称、接口、返回状态码、投保单号、报价单号、状态、报文、路径、文档记录时间）。依据is_contain_ch29布尔值分流写入逻辑，对超长报文施加长度阈值拦截以防单元格报错。最终触发写回操作。
重要参数：写入坐标遵循(rows, columns, value)规范；阈值设定为32000字符以适配Office兼容标准；文件保存需覆盖原路径以保持追踪连续性。
可观察的完成状态：终端输出执行耗时提示，指定目录下的autoinsurancereport.xls体积增大且末尾新增一行完整记录。

# 读取原表格
workbook = xlrd.open_workbook("autoinsurancereport.xls")
all_sheet = workbook.sheet_names()
first_sheet = workbook.sheet_by_name(all_sheet[0])
rows = first_sheet.nrows

# 拷贝新的excel,并从rows行开始追加写入
new_workbook = copy(workbook)
new_sheet = new_workbook.get_sheet(0)

# 创建标题
new_sheet.write(0, 0, "场景")
new_sheet.write(0, 1, "报告名称")
new_sheet.write(0, 2, "接口")
new_sheet.write(0, 3, "返回状态码")
new_sheet.write(0, 4, "投保单号")
new_sheet.write(0, 5, "报价单号")
new_sheet.write(0, 6, "状态")
new_sheet.write(0, 7, "orderMessageList")
new_sheet.write(0, 8, "路径")
new_sheet.write(0, 9, "文档记录时间")

# （行，列，值）
new_sheet.write(rows, 0, GetSceneNameGain)
new_sheet.write(rows, 1, TestResponseRespGetFiltration)
new_sheet.write(rows, 2, ReportParticularsUrl)
new_sheet.write(rows, 3, GetStateGain)
new_sheet.write(rows, 4, InsureGainGetDispose01)
new_sheet.write(rows, 5, PriceSheetGetDispose01)

if is_contain_ch29 == True:
    new_sheet.write(rows, 6, "成功")
    new_sheet.write(rows, 8, "\u3000\u3000")
else:
    print("ReturnParameterGettest长度为:",len(ReturnParameterGettest))
    if len(ReturnParameterGettest)<=32000:
        new_sheet.write(rows, 8, ReturnParameterGettest)
    else:
        new_sheet.write(rows, 8, "执行失败，报文较长，文档无法进行存储")
    new_sheet.write(rows, 6, "失败或未执行")
    
new_sheet.write(rows, 7, orderMessageListGet)
new_sheet.write(rows, 9, str(current_time))
new_workbook.save("autoinsurancereport.xls")

完成判断与排错指南

脚本正常运行结束且目录内更新对应xls文件即视为完成。在实际跑批过程中可能遇到以下情况：
• 正则匹配为空：当接口结构调整导致字段名变更时，需同步更新re.findall内的键名模式，并补充空值 fallback 逻辑。
• 长文本写入阻断：Excel单单元格理论上限约32767字符，超过阈值会直接抛出异常。除预设阈值外，建议大型测试套件改用CSV导出或接入轻量级数据库存储。
• 模块导入缺失：上述copy()函数依赖于from xlutils.copy import copy声明，若运行提示NameError请补全依赖链。正则表达式务必保留r前缀，否则反斜杠转义易引发语法崩溃。

简要总结

利用Python对接MeterSphere底层API实现测试报告的自动化抓取与归档，能有效替代高频次的人工点击导出。核心在于熟练运用Requests发起会话、精准把控JSON结构特征以及合理裁剪正则边界。结合定时调度策略后，即可构建稳定的持续交付质量看板。以上就是Python爬虫解析MeterSphere平台执行报告并自动入库的详细内容，更多关于API接口自动化测试与数据抓取的资料请关注本站其它相关文章！