Python + HAR PRO版本

借鉴51Testing软件测试网公众号发布的性能工具【告别手工分析!Python + HAR 一键生成页面性能测试报告】,做了一些内容优化,如下:
代码拆解分析

  1. 核心类结构:HARPerformanceAnalyzer
    代码定义了一个名为 HARPerformanceAnalyzer 的类,采用了面向对象的设计思路。
  • init (初始化):

  • 接收 HAR 文件路径。

  • 使用 json.load 读取文件内容。

  • 使用 haralyzer.HarParser 库解析原始 JSON 数据,将其转换为更易操作的对象(self.pages 和 self.entries)。

  1. 关键功能模块解析
    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、预加载资源。

  • 输出: 生成一份包含“问题描述”和“优化建议”的报告。

  1. Excel 报告生成 (export_to_excel)
    该方法使用 pandas.ExcelWriter 配合 openpyxl 引擎,生成一个包含 4 个工作表 (Sheet) 的 Excel 文件,适应不同人群阅读

  2. 代码流程图示
    为了让你更直观地理解数据是如何流动的,请看下面的流程图:

  • Input: 读取 vuejs_org.har 文件。

  • Parse: HarParser 将 JSON 结构化。

  • Analyze:

    遍历 Entries 提取网络层数据。

    提取 Pages 提取渲染层数据。

    Diagnose: 对比阈值(如 >500ms),生成建议。

    Output: 利用 Pandas 将数据写入 Excel 的不同 Sheet。

  1. 潜在的改进点
    虽然这段代码已经很完善,但如果用于生产环境,可以考虑以下微调:
  • 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 文件,这样就能实现从“访问”到“出报告”的全链路自动化了。

©著作权归作者所有,转载或内容合作请联系作者
【社区内容提示】社区部分内容疑似由AI辅助生成,浏览时请结合常识与多方信息审慎甄别。
平台声明:文章内容(如有图片或视频亦包括在内)由作者上传并发布,文章内容仅代表作者本人观点,简书系信息发布平台,仅提供信息存储服务。

相关阅读更多精彩内容

友情链接更多精彩内容