"""Разбор trace-файлов скролла (Chrome trace, devtools.timeline): куда уходит время main thread.

    cd perf && .venv/bin/python analysis/timeline_breakdown.py results/20261026-s1 \
        --out report-data/20261026-s1/analysis [--glob '*.trace.json']

Для каждого `<run_dir>/artifacts/*.trace.json` (trace снимается на повторе 0, `traced=1`):
- выбирается поток CrRendererMain с наибольшей суммой `RunTask` (страница замера);
- по вложенности X-событий считается self time каждого события (dur минус дети);
- self time раскладывается по категориям как в DevTools Performance:
  scripting / rendering (style, layout, hit test, layerize) / painting / gc / other (RunTask без детей);
  неизвестное событие наследует категорию ближайшего известного предка;
- `FunctionCall` верхнего уровня (точка входа JS: функция и url:line:col бандла) - топ по суммарному dur;
- `EventDispatch` по типу события.
Число шагов колеса берётся из записи `traced=1` того же target/role в scroll.jsonl / scroll_coauthor.jsonl
(trace снят ровно на этом замере), отсюда *_per_step.

trace без CPU-профайлера (категории в lib/cdp.ts startTrace): функции внутри JS здесь не видны,
только точки входа. Топ функций по self time - analysis/cpuprofile_top.py.

Вывод: <out>/timeline-breakdown.json и timeline-breakdown.md (имя без «trace»: шаблон .gitignore
`perf/**/*trace*.json` исключает трассы).
"""
from __future__ import annotations

import argparse
import fnmatch
import json
from collections import Counter, defaultdict
from pathlib import Path

try:
    import orjson

    _loads = orjson.loads
except ImportError:  # pragma: no cover
    _loads = json.loads

SCRIPTING = {
    "FunctionCall", "EvaluateScript", "v8.compile", "v8.compileModule", "v8.evaluateModule", "TimerFire",
    "EventDispatch", "FireAnimationFrame", "RunMicrotasks", "XHRReadyStateChange", "XHRLoad",
    "FireIdleCallback", "V8.Execute", "v8.run", "V8.HandleInterrupts", "V8.StackGuard",
    "V8.BytecodeBudgetInterrupt", "v8.callFunction", "ParseHTML", "PerformanceObserver",
    "ResizeObserver::BroadcastObservations", "V8.ParseFunction", "V8.CompileLazy", "V8.CompileCode",
    "MutationObserver", "PostMessage", "HandlePostMessage", "v8.produceCache", "CallbackFunction",
}
RENDERING = {
    "UpdateLayoutTree", "Layout", "PrePaint", "HitTest", "Layerize", "UpdateLayer", "RecalculateStyles",
    "IntersectionObserverController::computeIntersections", "ScrollLayer", "UpdateLayerTree",
    "ComputeIntersections", "LayoutShift", "ParseAuthorStyleSheet", "InvalidateLayout",
}
PAINTING = {"Paint", "PaintImage", "Commit", "CompositeLayers", "Decode Image", "Draw LazyPixelRef",
            "Decode LazyPixelRef", "PaintSetup", "Rasterize", "RasterTask"}
GC = {"MinorGC", "MajorGC", "BlinkGC.AtomicPhase", "ThreadState::performIdleLazySweep", "BlinkGCMarking",
      "V8.GCIncrementalMarking", "V8.GCScavenger", "V8.GCCompactor", "V8.GCFinalizeMC", "V8.GC_MC_BACKGROUND_MARKING"}
CATS = ("scripting", "rendering", "painting", "gc", "other")


def category(name: str) -> str | None:
    if name in SCRIPTING:
        return "scripting"
    if name in RENDERING:
        return "rendering"
    if name in PAINTING:
        return "painting"
    if name in GC or name.startswith("V8.GC") or name.startswith("BlinkGC"):
        return "gc"
    return None


def iter_events(path: Path):
    """trace Playwright: одно событие на строку ('{...},'), обрамление '{"traceEvents":[' ... ']}'."""
    with open(path, "rb") as f:
        for raw in f:
            s = raw.strip()
            if s.endswith(b","):
                s = s[:-1]
            if not s.startswith(b"{") or s.startswith(b'{"traceEvents"'):
                continue
            try:
                yield _loads(s)
            except Exception:
                continue


def self_times(events: list[dict]) -> list[tuple[dict, float, str]]:
    """events - X-события одного потока. Возвращает (event, self_us, category) с наследованием категории."""
    evs = sorted(events, key=lambda e: (e["ts"], -e.get("dur", 0)))
    out: list[list] = []
    stack: list[list] = []  # [end_ts, idx_in_out, cat]
    for e in evs:
        ts, dur = e["ts"], e.get("dur", 0)
        while stack and stack[-1][0] <= ts:
            stack.pop()
        cat = category(e["name"]) or (stack[-1][2] if stack else None)
        rec = [e, float(dur), cat]
        if stack:
            parent = out[stack[-1][1]]
            parent[1] -= dur
        out.append(rec)
        stack.append([ts + dur, len(out) - 1, cat])
    return [(e, max(s, 0.0), c or "other") for e, s, c in out]


def analyze(path: Path) -> dict:
    thread_names: dict[tuple, str] = {}
    by_thread: dict[tuple, list] = defaultdict(list)
    for e in iter_events(path):
        ph = e.get("ph")
        key = (e.get("pid"), e.get("tid"))
        if ph == "M" and e.get("name") == "thread_name":
            thread_names[key] = e.get("args", {}).get("name", "")
        elif ph == "X" and "dur" in e:
            by_thread[key].append({"name": e.get("name"), "ts": e["ts"], "dur": e["dur"],
                                   "data": (e.get("args") or {}).get("data") if e.get("name") in ("FunctionCall", "EventDispatch", "TimerFire") else None})
    mains = [k for k, n in thread_names.items() if n == "CrRendererMain"]

    def runtask_sum(k):
        return sum(ev["dur"] for ev in by_thread.get(k, []) if ev["name"] == "RunTask")

    main = max(mains, key=runtask_sum) if mains else None
    evs = by_thread.get(main, [])
    st = self_times(evs)
    cat_ms = {c: 0.0 for c in CATS}
    name_self: Counter = Counter()
    for e, s, c in st:
        cat_ms[c] += s / 1000
        name_self[e["name"]] += s / 1000
    tasks = [ev for ev in evs if ev["name"] == "RunTask"]
    busy_ms = sum(ev["dur"] for ev in tasks) / 1000
    long_tasks = [ev["dur"] / 1000 for ev in tasks if ev["dur"] > 50_000]
    span_ms = (max(ev["ts"] + ev["dur"] for ev in evs) - min(ev["ts"] for ev in evs)) / 1000 if evs else 0
    # точки входа JS: FunctionCall верхнего уровня (не вложенный в другой FunctionCall)
    fc: Counter = Counter()
    fc_n: Counter = Counter()
    end_fc = -1
    for ev in sorted((x for x in evs if x["name"] == "FunctionCall"), key=lambda x: (x["ts"], -x["dur"])):
        if ev["ts"] < end_fc:
            continue
        end_fc = ev["ts"] + ev["dur"]
        d = ev.get("data") or {}
        url = (d.get("url") or "").rsplit("/", 1)[-1]
        k = f"{d.get('functionName') or '(anonymous)'} @ {url}:{d.get('lineNumber')}:{d.get('columnNumber')}"
        fc[k] += ev["dur"] / 1000
        fc_n[k] += 1
    ed: Counter = Counter()
    for ev in evs:
        if ev["name"] == "EventDispatch":
            ed[(ev.get("data") or {}).get("type", "?")] += ev["dur"] / 1000
    return {
        "file": path.name,
        "main_thread": list(main) if main else None,
        "span_ms": round(span_ms, 1),
        "runtask_ms": round(busy_ms, 1),
        "long_task_n": len(long_tasks),
        "long_task_ms": round(sum(long_tasks), 1),
        "category_ms": {k: round(v, 1) for k, v in cat_ms.items()},
        "top_events_self_ms": [[k, round(v, 1)] for k, v in name_self.most_common(15)],
        "top_entry_functions_ms": [[k, round(v, 1), fc_n[k]] for k, v in fc.most_common(8)],
        "event_dispatch_ms": [[k, round(v, 1)] for k, v in ed.most_common(8)],
    }


def traced_records(run_dir: Path) -> dict[str, dict]:
    """artifact path -> запись (traced=1), откуда steps и main_thread_ms."""
    out = {}
    for fn in ("scroll.jsonl", "scroll_coauthor.jsonl"):
        p = run_dir / fn
        if not p.exists():
            continue
        for line in p.read_text().splitlines():
            r = json.loads(line)
            for a in r.get("artifacts", []):
                out[Path(a).name] = r
    return out


def main(argv=None):
    ap = argparse.ArgumentParser()
    ap.add_argument("run_dir")
    ap.add_argument("--out", required=True)
    ap.add_argument("--glob", default="*.trace.json")
    a = ap.parse_args(argv)
    run_dir = Path(a.run_dir)
    out = Path(a.out)
    out.mkdir(parents=True, exist_ok=True)
    recs = traced_records(run_dir)
    res = []
    for p in sorted((run_dir / "artifacts").glob("*.trace.json")):
        if not fnmatch.fnmatch(p.name, a.glob):
            continue
        r = analyze(p)
        rec = recs.get(p.name)
        if rec:
            m = rec["metrics"]
            r.update(target=rec["target"], role=rec["role"], scenario=rec["scenario"], steps=m.get("steps"),
                     probe_main_thread_ms=round(m.get("main_thread_ms", 0), 1))
            steps = m.get("steps") or 0
            if steps:
                r["category_ms_per_step"] = {k: round(v / steps, 1) for k, v in r["category_ms"].items()}
                r["runtask_ms_per_step"] = round(r["runtask_ms"] / steps, 1)
        res.append(r)
        print(f"{p.name}: runtask {r['runtask_ms']} ms, {r['category_ms']}", flush=True)
    (out / "timeline-breakdown.json").write_text(json.dumps(res, ensure_ascii=False, indent=1))
    lines = ["# Разбивка main thread по trace первого повтора (timeline_breakdown.py)", "",
             f"run_dir: `{run_dir}`; категории как в DevTools (self time); ms на шаг колеса 200 px.", "",
             "| trace | target | role | steps | RunTask мс/шаг | scripting | rendering | painting | gc | other | топ событий по self time |",
             "|---|---|---|---|---|---|---|---|---|---|---|"]
    for r in res:
        ps = r.get("category_ms_per_step") or {}
        top = ", ".join(f"{k} {v:.0f}" for k, v in r["top_events_self_ms"][:4])
        lines.append(f"| {r['file']} | {r.get('target', '')} | {r.get('role', '')} | {r.get('steps', '')} | "
                     f"{r.get('runtask_ms_per_step', '')} | " + " | ".join(str(ps.get(c, '')) for c in CATS) + f" | {top} |")
    lines += ["", "## Точки входа JS (FunctionCall верхнего уровня, суммарный dur, мс; число вызовов)", ""]
    for r in res:
        lines.append(f"- {r['file']}: " + "; ".join(f"{k} — {v} мс ×{n}" for k, v, n in r["top_entry_functions_ms"][:4]))
    (out / "timeline-breakdown.md").write_text("\n".join(lines) + "\n")


if __name__ == "__main__":
    main()