借鉴51Testing软件测试网公众号发布的性能工具【告别手工分析!Python + HAR 一键生成页面性能测试报告】,做了一些内容优化,如下:
代码拆解分析
- 核心类结构:HARPerformanceAnalyzer
代码定义了一个名为 HARPerformanceAnalyzer 的类,采用了面向对象的设计思路。
init (初始化):
接收 HAR 文件路径。
使用 json.load 读取文件内容。
使用 haralyzer.HarParser 库解析原始 JSON 数据,将其转换为更易操作的对象(self.pages 和 self.entries)。
- 关键功能模块解析
A. 数据清洗与分类 (_get_resource_type)
功能: 并不是所有的 URL 后缀都可靠,该方法优先检查 HTTP 响应头中的 Content-Type (MIME type)。
逻辑: 将资源标准化为 HTML, CSS, JS, 图片, 字体, JSON接口 等。这对于后续分析“哪类资源拖慢了页面”至关重要。
B. 请求级性能分析 (analyze_request_details)
这是代码中最核心的部分,它遍历了每一个网络请求(Entry):
时间分片提取: 提取了 DNS、TCP、SSL、TTFB (Wait)、Download (Receive) 等细分耗时。
慢资源标记: 设定了一个 500ms 的阈值,超过此时间的请求会被标记为“慢资源”。
异常检测: 检查 HTTP 状态码是否 >= 400,标记为错误请求。
输出: 返回一个包含所有请求详情的 pandas.DataFrame。
C. 页面级关键指标 (analyze_page_timings)
利用浏览器提供的 Navigation Timing API 数据(存在于 HAR 的 page.timings 中):
DOM Ready: DOM 内容加载完成的时间。
Load Event: 页面完全加载的时间。
FCP (First Contentful Paint): 首次内容绘制时间(用户看到第一个像素的时间)。
逻辑: 代码中包含了一个简单的容错处理(如果某些指标为 -1,则归零)。
D. 智能瓶颈识别 (identify_bottlenecks)
这是一个非常有价值的“专家系统”模块。它不只是列出数据,还会根据预设规则自动“诊断”问题:
规则 1: 如果慢资源 > 0 → 建议压缩图片、拆分代码。
规则 2: 如果总请求数 > 80 → 建议合并文件、使用雪碧图。
规则 3: 如果 FCP > 1800ms → 建议内联关键 CSS、预加载资源。
输出: 生成一份包含“问题描述”和“优化建议”的报告。
Excel 报告生成 (export_to_excel)
该方法使用 pandas.ExcelWriter 配合 openpyxl 引擎,生成一个包含 4 个工作表 (Sheet) 的 Excel 文件,适应不同人群阅读代码流程图示
为了让你更直观地理解数据是如何流动的,请看下面的流程图:
Input: 读取 vuejs_org.har 文件。
Parse: HarParser 将 JSON 结构化。
-
Analyze:
遍历 Entries 提取网络层数据。
提取 Pages 提取渲染层数据。
Diagnose: 对比阈值(如 >500ms),生成建议。
Output: 利用 Pandas 将数据写入 Excel 的不同 Sheet。
- 潜在的改进点
虽然这段代码已经很完善,但如果用于生产环境,可以考虑以下微调:
CDN 识别: 目前代码只提取了域名,可以增加逻辑判断该域名是否属于 CDN(如匹配 CNAME 或常见 CDN 域名后缀)。
压缩检测: 检查 Response Header 中的 Content-Encoding,如果文本资源没有 gzip 或 br,则视为一个优化点。
缓存检查: 检查 Cache-Control 或 Expires 头,分析静态资源是否设置了合理的缓存策略。
# -*- coding: utf-8 -*-
# @Time : 2026/1/6 14:03
# @Author : xuhao
# @File : har_to_excel.py
# @Software: PyCharm
import json
import os
import pandas as pd
from datetime import datetime
from typing import Dict, List
import traceback
# 尝试导入 haralyzer,如果没安装也不影响核心逻辑(为了兼容性)
try:
from haralyzer import HarParser
except ImportError:
pass
class HARFinalAnalyzer:
def __init__(self, har_file_path: str):
"""初始化分析器,加载HAR文件"""
self.har_file = har_file_path
if not os.path.exists(har_file_path):
raise FileNotFoundError(f"未找到文件: {har_file_path}")
# 1. 加载原始 JSON 数据
print(f"📂 正在加载文件: {har_file_path} ...")
try:
with open(self.har_file, "r", encoding="utf-8") as f:
self.har_data = json.load(f)
except Exception as e:
raise ValueError(f"无法读取文件,请确认并非二进制文件或编码错误: {e}")
# 2. 格式完整性校验
if "log" not in self.har_data:
keys = list(self.har_data.keys())
raise ValueError(f"HAR 格式无效:缺少 'log' 根节点。当前顶层键: {keys}")
if "entries" not in self.har_data["log"]:
raise ValueError("HAR 格式无效:'log' 节点下缺少 'entries' 数据。")
# 3. 初始化数据源 (直接使用字典操作,最稳健)
self.raw_pages = self.har_data["log"].get("pages", [])
self.entries = self.har_data["log"]["entries"]
# 常见 CDN 关键字库
self.cdn_keywords = [
'cdn', 'cloudfront', 'akamai', 'fastly', 'azureedge', 'googleusercontent',
'aliyuncs', 'blob.core', 'oss-', 'cos.', 'qiniu', 'upai', 'static.'
]
def _get_header_value(self, headers: List[Dict], header_name: str) -> str:
"""安全提取Header值"""
header_name = header_name.lower()
if not headers: return ""
for h in headers:
if h.get('name', '').lower() == header_name:
return h.get('value', '')
return ""
def _get_resource_type(self, url: str, content_type: str) -> str:
"""判断资源类型"""
if not content_type: content_type = ""
c_type = content_type.lower()
if "text/html" in c_type:
return "HTML"
elif "text/css" in c_type:
return "CSS"
elif "javascript" in c_type or "json" in c_type:
return "JS/JSON"
elif "image/" in c_type:
return "Image"
elif "font/" in c_type or "application/font" in c_type:
return "Font"
elif "video/" in c_type:
return "Video"
return "Other"
def _check_compression(self, resource_type: str, size_bytes: int, encoding: str) -> str:
"""检查压缩状态"""
# 仅检查 > 1KB 的文本资源
if resource_type in ["HTML", "CSS", "JS/JSON"] and size_bytes > 1024:
if not encoding or encoding.lower() not in ['gzip', 'br', 'deflate']:
return "❌ 未开启压缩"
return f"✅ 已开启 ({encoding})"
return "-"
def _check_caching(self, resource_type: str, cache_control: str) -> str:
"""检查缓存策略"""
static_types = ["CSS", "JS/JSON", "Image", "Font"]
if resource_type in static_types:
if not cache_control:
return "❌ 无缓存头"
if "no-store" in cache_control or "no-cache" in cache_control:
return "⚠️ 不缓存"
if "max-age" not in cache_control:
return "⚠️ 未设max-age"
return "✅ 有效缓存"
return "-"
def _is_cdn_domain(self, domain: str, headers: List[Dict]) -> str:
"""CDN 智能判断"""
for keyword in self.cdn_keywords:
if keyword in domain:
return "是 (域名匹配)"
server = self._get_header_value(headers, "server").lower()
x_cache = self._get_header_value(headers, "x-cache").lower()
via = self._get_header_value(headers, "via").lower()
if any(k in server for k in ['cloudflare', 'nginx', 'apache']) and "cdn" in domain:
return "是 (疑似)"
if x_cache or "cloudfront" in via or "akamai" in via:
return "是 (Header特征)"
return "否"
def analyze_request_details(self) -> pd.DataFrame:
"""深度分析请求详情"""
request_list = []
for idx, entry in enumerate(self.entries, 1):
req = entry["request"]
res = entry["response"]
timings = entry.get("timings", {})
url = req["url"]
try:
domain = url.split("//")[-1].split("/")[0]
except:
domain = "unknown"
mime_type = res.get("content", {}).get("mimeType", "")
res_type = self._get_resource_type(url, mime_type)
body_size = res.get("bodySize", 0)
content_size = res.get("content", {}).get("size", 0)
headers = res.get('headers', [])
encoding = self._get_header_value(headers, 'content-encoding')
cache_control = self._get_header_value(headers, 'cache-control')
compression_status = self._check_compression(res_type, content_size, encoding)
cache_status = self._check_caching(res_type, cache_control)
is_cdn = self._is_cdn_domain(domain, headers)
# 耗时处理 (防止 None)
def t(val):
return max(0, val) if val is not None else 0
item = {
"ID": idx,
"资源类型": res_type,
"URL (Last 50)": url[-50:],
"完整URL": url,
"状态码": res["status"],
"域名": domain,
"CDN检测": is_cdn,
"传输大小(KB)": round(body_size / 1024, 2),
"实际大小(KB)": round(content_size / 1024, 2),
"总耗时(ms)": round(entry.get("time", 0), 2),
"TTFB/等待(ms)": round(t(timings.get("wait", 0)), 2),
"下载耗时(ms)": round(t(timings.get("receive", 0)), 2),
"压缩状态": compression_status,
"缓存状态": cache_status,
"Cache-Control": cache_control[:50]
}
request_list.append(item)
return pd.DataFrame(request_list)
def analyze_page_timings(self) -> Dict:
"""分析页面加载核心指标"""
if not self.raw_pages:
return {"提示": "HAR文件中未包含页面导航(Pages)信息", "请求总数": len(self.entries)}
first_page = self.raw_pages[0]
t = first_page.get("pageTimings", {})
dom_ready = t.get("onContentLoad", -1)
on_load = t.get("onLoad", -1)
return {
"DOM Ready (ms)": round(dom_ready, 2) if dom_ready > 0 else "N/A",
"Page Load (ms)": round(on_load, 2) if on_load > 0 else "N/A",
"请求总数": len(self.entries)
}
def identify_bottlenecks(self, df: pd.DataFrame) -> pd.DataFrame:
"""生成智能诊断建议"""
bottlenecks = []
# 1. 压缩诊断
uncompressed = df[df['压缩状态'].str.contains("❌", na=False)]
if not uncompressed.empty:
bottlenecks.append({
"严重等级": "🔴 高",
"问题类型": "资源未压缩",
"涉及数量": len(uncompressed),
"问题描述": "检测到较大的文本文件(JS/CSS/HTML)未开启Gzip/Brotli压缩。",
"优化建议": "配置Nginx/Apache开启 gzip on; 或使用 brotli 模块。"
})
# 2. 缓存诊断
uncached = df[df['缓存状态'].str.contains("❌|⚠️", regex=True, na=False)]
if not uncached.empty:
bottlenecks.append({
"严重等级": "🟠 中",
"问题类型": "静态资源缓存缺失",
"涉及数量": len(uncached),
"问题描述": "静态资源缺少 Cache-Control 头或有效期过短。",
"优化建议": "设置长期缓存 (如 max-age=31536000) 并配合文件名Hash。"
})
# 3. 慢请求诊断
if '总耗时(ms)' in df.columns:
slow_req = df[df['总耗时(ms)'] > 1000]
if not slow_req.empty:
bottlenecks.append({
"严重等级": "🟠 中",
"问题类型": "慢速请求 (>1s)",
"涉及数量": len(slow_req),
"问题描述": "存在加载超过 1秒 的请求,阻碍渲染。",
"优化建议": "API需优化SQL;静态资源考虑CDN分发。"
})
# 4. 错误请求
errors = df[df['状态码'] >= 400]
if not errors.empty:
bottlenecks.append({
"严重等级": "🔴 高",
"问题类型": "HTTP 错误",
"涉及数量": len(errors),
"问题描述": "检测到 4xx 或 5xx 响应。",
"优化建议": "检查URL拼写或后端日志。"
})
if not bottlenecks:
bottlenecks.append({"严重等级": "🟢", "问题类型": "全项通过", "优化建议": "性能表现良好。"})
return pd.DataFrame(bottlenecks)
def generate_report(self):
"""执行全流程并生成报告"""
print("⚡️ [1/3] 正在解析数据...")
df_details = self.analyze_request_details()
print("⚡️ [2/3] 正在进行智能诊断...")
df_bottlenecks = self.identify_bottlenecks(df_details)
timings = self.analyze_page_timings()
# --- 核心修改:动态生成文件名 ---
# 1. 获取输入文件的基本名 (如 "vuejs_cn")
base_name = os.path.splitext(os.path.basename(self.har_file))[0]
# 2. 生成精确到秒的时间戳
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
# 3. 组合新文件名
output_file = f"{base_name}_Report_{timestamp}.xlsx"
print(f"⚡️ [3/3] 正在写入 Excel: {output_file} ...")
try:
with pd.ExcelWriter(output_file, engine='openpyxl') as writer:
# 写入各个 Sheet
df_bottlenecks.to_excel(writer, sheet_name='1.智能诊断报告', index=False)
pd.DataFrame([timings]).to_excel(writer, sheet_name='2.页面核心指标', index=False)
df_details.sort_values("总耗时(ms)", ascending=False).to_excel(writer, sheet_name='3.请求详情列表',
index=False)
uncompressed = df_details[df_details['压缩状态'].str.contains("❌", na=False)]
if not uncompressed.empty:
uncompressed.to_excel(writer, sheet_name='4.待优化-未压缩资源', index=False)
print(f"\n✅ 成功!报告已保存至:\n 👉 {os.path.abspath(output_file)}")
except PermissionError:
# 理论上有了时间戳不会触发这个,但为了保险还是加上
print(f"❌ 权限错误:文件 {output_file} 被占用。")
except Exception as e:
print(f"❌ 导出失败: {e}")
traceback.print_exc()
# --- 运行入口 ---
if __name__ == "__main__":
# 在这里修改你的文件名
HAR_PATH = "vuejs_cn.har"
if os.path.exists(HAR_PATH):
try:
analyzer = HARFinalAnalyzer(HAR_PATH)
analyzer.generate_report()
except Exception as e:
print(f"❌ 程序中断: {e}")
else:
print(f"⚠️ 文件不存在: {HAR_PATH}")
print("💡 提示: 请先从浏览器导出 HAR 文件,并将其重命名为代码中指定的名称。")
Tips:
使用 Python 的 Playwright 库来自动访问网页并生成这个 HAR 文件,这样就能实现从“访问”到“出报告”的全链路自动化了。