"""Топ функций по self time из CPU-профилей (.cpuprofile, CDP Profiler) с привязкой к исходникам через source maps.

    cd perf && .venv/bin/python analysis/cpuprofile_top.py results/20261026-hot \
        --maps .build/prod-sourcemap --out report-data/20261026-s1/analysis

Профили пишет perf/e2e/scenarios/hotspots.spec.ts (PERF_HOT_MODE=cpu) на prod-сборке с source maps
(`perf/stand.sh web-sourcemap`, :8082; код совпадает с prod-сборкой :8080 с точностью до хешей имён чанков).

Для каждого профиля:
- self time (мс) по функциям: имя, исходный файл:строка (через source map), доля от активного времени;
- self time по пакетам / файлам исходников (`node_modules/<pkg>` или `web/src/...`);
- «приложение»: время каждого сэмпла приписывается ближайшему кадру стека из `web/src` (кто в коде приложения
  вызвал дорогую работу библиотек) - топ таких мест file:line;
- inclusive time функций `web/src` (сэмпл считается один раз на функцию).
Специальные узлы `(idle)`, `(program)`, `(garbage collector)` выводятся отдельно; `(idle)` в активное время не входит.
Вывод: <out>/cpuprofile-top.json и cpuprofile-top.md.
"""
from __future__ import annotations

import argparse
import bisect
import json
import re
from collections import Counter, defaultdict
from pathlib import Path

B64 = {c: i for i, c in enumerate("ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789+/")}
SPECIAL = {"(idle)", "(program)", "(garbage collector)", "(root)"}


def vlq_decode(seg: str) -> list[int]:
    out, val, shift = [], 0, 0
    for ch in seg:
        d = B64[ch]
        val += (d & 31) << shift
        if d & 32:
            shift += 5
            continue
        out.append(-(val >> 1) if val & 1 else val >> 1)
        val, shift = 0, 0
    return out


class SourceMap:
    def __init__(self, path: Path):
        m = json.loads(path.read_text())
        self.sources = [norm_source(s) for s in m["sources"]]
        self.names = m.get("names", [])
        self.lines: list[tuple[list[int], list[tuple]]] = []
        src = sline = scol = name = 0
        for line in m["mappings"].split(";"):
            col = 0
            cols, segs = [], []
            if line:
                for raw in line.split(","):
                    if not raw:
                        continue
                    v = vlq_decode(raw)
                    col += v[0]
                    if len(v) >= 4:
                        src += v[1]; sline += v[2]; scol += v[3]
                        nm = None
                        if len(v) >= 5:
                            name += v[4]; nm = name
                        cols.append(col); segs.append((src, sline, scol, nm))
            self.lines.append((cols, segs))

    def lookup(self, line0: int, col0: int):
        """0-based строка/колонка сгенерированного кода -> (source, line 1-based, col, name) или None."""
        if line0 < 0 or line0 >= len(self.lines):
            return None
        cols, segs = self.lines[line0]
        i = bisect.bisect_right(cols, col0) - 1
        if i < 0:
            return None
        s, l, c, n = segs[i]
        return self.sources[s], l + 1, c, (self.names[n] if n is not None and n < len(self.names) else None)


def norm_source(s: str) -> str:
    s = s.replace("\\", "/")
    m = re.search(r"(web/(?:node_modules|src)/.*)$", s)
    if m:
        return m.group(1)
    return s.lstrip("./")


def package_of(src: str) -> str:
    m = re.match(r"web/node_modules/((?:@[^/]+/)?[^/]+)/", src)
    if m:
        return "node_modules/" + m.group(1)
    if src.startswith("web/src/"):
        return src
    return src


class Resolver:
    def __init__(self, maps_root: Path):
        self.root = maps_root
        self.cache: dict[str, SourceMap | None] = {}

    def frame(self, cf: dict):
        url = cf.get("url") or ""
        fn = cf.get("functionName") or "(anonymous)"
        if not url:
            return {"fn": fn, "src": "", "line": None, "gen": ""}
        mpath = re.sub(r"^https?://[^/]+/", "", url).split("?")[0]
        sm = self.cache.get(mpath, False)
        if sm is False:
            p = self.root / (mpath + ".map")
            sm = SourceMap(p) if p.exists() else None
            self.cache[mpath] = sm
        gen = f"{mpath.rsplit('/', 1)[-1]}:{cf.get('lineNumber', -1) + 1}:{cf.get('columnNumber', -1) + 1}"
        if sm is None:
            return {"fn": fn, "src": mpath, "line": None, "gen": gen}
        r = sm.lookup(cf.get("lineNumber", -1), cf.get("columnNumber", -1))
        if not r:
            return {"fn": fn, "src": mpath, "line": None, "gen": gen}
        src, line, _col, _name = r
        return {"fn": fn, "src": src, "line": line, "gen": gen}


def analyze(path: Path, res: Resolver, top: int = 25) -> dict:
    prof = json.loads(path.read_text())
    nodes = {n["id"]: n for n in prof["nodes"]}
    parent: dict[int, int] = {}
    for n in prof["nodes"]:
        for c in n.get("children", []):
            parent[c] = n["id"]
    samples = prof.get("samples", [])
    deltas = prof.get("timeDeltas", [])
    # длительность сэмпла i = timeDeltas[i+1] (как DevTools); последний - медиана
    dur = [deltas[i + 1] if i + 1 < len(deltas) else 0 for i in range(len(samples))]
    fr_cache: dict[int, dict] = {}

    def fr(nid):
        if nid not in fr_cache:
            fr_cache[nid] = res.frame(nodes[nid]["callFrame"])
        return fr_cache[nid]

    special = Counter()
    self_fn = Counter()
    self_pkg = Counter()
    app_attr = Counter()
    app_incl = Counter()
    active = 0.0
    for nid, d in zip(samples, dur):
        ms = d / 1000
        f = fr(nid)
        if f["fn"] in SPECIAL and not f["src"]:
            special[f["fn"]] += ms
            if f["fn"] != "(idle)":
                active += ms
            continue
        active += ms
        key = f"{f['fn']} — {f['src']}:{f['line']}" if f["line"] else f"{f['fn']} — {f['src'] or '?'} ({f['gen']})"
        self_fn[key] += ms
        self_pkg[package_of(f["src"]) if f["src"] else "(native/unknown)"] += ms
        # ближайший кадр приложения (web/src) и inclusive по функциям приложения
        cur, attributed, seen = nid, False, set()
        while cur is not None:
            g = fr(cur)
            if g["src"].startswith("web/src/"):
                k = f"{g['fn']} — {g['src']}:{g['line']}"
                if not attributed:
                    app_attr[k] += ms
                    attributed = True
                if k not in seen:
                    app_incl[k] += ms
                    seen.add(k)
            cur = parent.get(cur)
        if not attributed:
            app_attr["(нет кадра web/src в стеке: библиотека/браузер)"] += ms
    total = sum(dur) / 1000

    def tbl(c: Counter):
        return [[k, round(v, 1), round(100 * v / active, 1) if active else 0] for k, v in c.most_common(top)]

    return {
        "file": path.name,
        "profile_ms": round(total, 1),
        "active_ms": round(active, 1),
        "special_ms": {k: round(v, 1) for k, v in special.items()},
        "self_by_function": tbl(self_fn),
        "self_by_package": tbl(self_pkg),
        "app_attribution": tbl(app_attr),
        "app_inclusive": tbl(app_incl),
    }


def main(argv=None):
    ap = argparse.ArgumentParser()
    ap.add_argument("run_dir")
    ap.add_argument("--maps", required=True, help="каталог сборки с .map (perf/.build/prod-sourcemap)")
    ap.add_argument("--out", required=True)
    ap.add_argument("--top", type=int, default=25)
    a = ap.parse_args(argv)
    run_dir, out = Path(a.run_dir), Path(a.out)
    out.mkdir(parents=True, exist_ok=True)
    res = Resolver(Path(a.maps))
    recs = {}
    p = run_dir / "hotspot_cpu.jsonl"
    if p.exists():
        for line in p.read_text().splitlines():
            r = json.loads(line)
            for art in r.get("artifacts", []):
                recs[Path(art).name] = r
    results = []
    for f in sorted((run_dir / "artifacts").glob("*.cpuprofile")):
        r = analyze(f, res, a.top)
        rec = recs.get(f.name)
        if rec:
            r.update(target=rec["target"], role=rec["role"], steps=rec["metrics"].get("steps"),
                     main_thread_ms=round(rec["metrics"].get("main_thread_ms", 0), 1))
        results.append(r)
        print(f"{f.name}: active {r['active_ms']} ms", flush=True)
    (out / "cpuprofile-top.json").write_text(json.dumps(results, ensure_ascii=False, indent=1))
    md = ["# CPU-профили действий: топ по self time (cpuprofile_top.py)", "",
          f"Профили: `{run_dir}/artifacts/*.cpuprofile` (hotspots.spec.ts, PERF_HOT_MODE=cpu, prod-сборка с source maps, сэмплы 200 мкс).",
          "Доля - от активного времени профиля (без `(idle)`). Профайлер добавляет накладные расходы: абсолютные мс выше, чем в scroll.spec; сравнивать доли.", ""]
    for r in results:
        md += [f"## {r['file']} — {r.get('target', '')} ({r.get('role', '')}), шагов {r.get('steps')}, main thread {r.get('main_thread_ms')} мс", "",
               f"Активное время {r['active_ms']} мс; служебные: {r['special_ms']}", ""]
        for title, key in (("Self time по функциям", "self_by_function"), ("Self time по пакетам/файлам", "self_by_package"),
                           ("Ближайший кадр web/src (кто вызвал работу)", "app_attribution"), ("Inclusive time функций web/src", "app_inclusive")):
            md += [f"### {title}", "", "| место | мс | % |", "|---|---|---|"]
            md += [f"| {k} | {v} | {pc} |" for k, v, pc in r[key][:15]]
            md.append("")
    (out / "cpuprofile-top.md").write_text("\n".join(md) + "\n")


if __name__ == "__main__":
    main()