"""Размер ответов на проводе (сжатый, CDP) из сетевого лога сценариев - для report-data (network.jsonl в git не попадает). cd perf && .venv/bin/python analysis/netlog_sizes.py results/20261026-s1 --out report-data/20261026-s1/analysis По каждому шаблону пути данных (лист, отчёт проекта, своды) и шагу сценария: число ответов, медиана `body_size` (сжатый размер по CDP `request.sizes()`, gzip уровня 1 бэка) и медиана `duration_ms` в браузере. Распакованные размеры - в summary.md (bench, `bytes`) и `json_bytes_sheet` скролла. Вывод: /netlog-sizes.md. """ from __future__ import annotations import argparse import json import re import statistics from collections import defaultdict from pathlib import Path PATTERNS = [ (re.compile(r"/api/v1/form/(\d+)/sheet/(\w+)(\?.*)?$"), lambda m: f"GET /form/{m.group(1)}/sheet/{m.group(2)}{m.group(3) or ''}"), (re.compile(r"/api/v1/projects/(\d+)/report/(\d+)/(\w+)$"), lambda m: f"GET /projects/{{id}}/report/{{year}}/{m.group(3)}"), (re.compile(r"/api/v1/svod/(\d)/(\w+)/(\d+)"), lambda m: f"GET /svod/{m.group(1)}/{m.group(2)}/{{year}}"), (re.compile(r"/api/v1/projects/plan-summary"), lambda m: "GET /projects/plan-summary"), ] def main(argv=None): ap = argparse.ArgumentParser() ap.add_argument("run_dir") ap.add_argument("--out", required=True) a = ap.parse_args(argv) groups = defaultdict(lambda: {"size": [], "dur": []}) for line in (Path(a.run_dir) / "network.jsonl").read_text().splitlines(): r = json.loads(line) if r.get("method") != "GET" or r.get("status") != 200 or not r.get("body_size"): continue path = re.sub(r"^https?://[^/]+", "", r["url"]) for rx, name in PATTERNS: m = rx.search(path) if m: g = groups[(name(m), r.get("scenario"), r.get("role"))] g["size"].append(r["body_size"]) g["dur"].append(r.get("duration_ms") or 0) break md = [f"# Размер ответов на проводе (сжатый) по сетевому логу {Path(a.run_dir).name} (netlog_sizes.py)", "", "| запрос | сценарий | роль | ответов | медиана body_size, КБ | медиана duration_ms в браузере |", "|---|---|---|---|---|---|"] for (name, sc, role), g in sorted(groups.items(), key=lambda kv: -statistics.median(kv[1]["size"])): md.append(f"| {name} | {sc} | {role} | {len(g['size'])} | {statistics.median(g['size']) / 1024:.0f} | {statistics.median(g['dur']):.0f} |") out = Path(a.out) out.mkdir(parents=True, exist_ok=True) (out / "netlog-sizes.md").write_text("\n".join(md) + "\n") print("\n".join(md[:14])) if __name__ == "__main__": main()