scripts/read_stats.py - main - miker/custom-git-ai

scripts/read_stats.py
361 строка13 KB
Mike S
final fix
10 авг 2026, 01:14
10 авг 2026, 01:149defae3
#!/usr/bin/env python3
"""Сводная статистика по всем записям ``refs/ai/artifacts``.

Обходит хранилище артефактов (уровень 2) и считает: сколько сессий, каких
инструментов и моделей, сколько сообщений и вырезанных секретов, какой период
покрыт. Если в репозитории есть ноты ``refs/notes/ai``, дополняет картину
связкой уровней: доли кода ИИ и человека в процентах (по аттестациям нот),
сколько строк ИИ записано за каждой сессией, в скольких коммитах она
участвовала, и не осталось ли висячих ссылок.

Автономный, как и read_artifacts.py: только git и стандартная библиотека —
работает на клоне, где git-ai не установлен.

    python3 scripts/read_stats.py [--repo ПУТЬ] [--json] [--top N]
"""

from __future__ import annotations

import argparse
import json
import subprocess
import sys
import time
from typing import Dict, List, Optional

ARTIFACTS_REF = "refs/ai/artifacts"
SESSIONS_DIR = "sessions"
NOTES_REF = "ai"
REDACTED = "«вырезано git-ai»"


def git(repo: str, *args: str, input_bytes: Optional[bytes] = None) -> bytes:
    proc = subprocess.run(
        ["git", "-C", repo, *args], input=input_bytes, capture_output=True
    )
    if proc.returncode != 0:
        raise RuntimeError(
            f"git {' '.join(args)}: {proc.stderr.decode('utf-8', 'replace').strip()}"
        )
    return proc.stdout


def try_git(repo: str, *args: str) -> Optional[bytes]:
    try:
        return git(repo, *args)
    except RuntimeError:
        return None


# -- чтение ref (та же механика, что в read_artifacts.py) ---------------------


def sessions_tree(repo: str) -> Dict[str, str]:
    out = try_git(repo, "ls-tree", "-z", f"{ARTIFACTS_REF}:{SESSIONS_DIR}")
    if out is None:
        return {}
    tree: Dict[str, str] = {}
    for chunk in out.split(b"\x00"):
        if not chunk:
            continue
        head, _, name = chunk.partition(b"\t")
        parts = head.split(b" ")
        if len(parts) == 3 and parts[1] == b"blob" and name:
            tree[name.decode("utf-8", "replace")] = parts[2].decode()
    return tree


def read_blobs(repo: str, shas: List[str]) -> Dict[str, Optional[bytes]]:
    if not shas:
        return {}
    payload = "".join(f"{sha}\n" for sha in shas).encode()
    out = git(repo, "cat-file", "--batch", input_bytes=payload)
    result: Dict[str, Optional[bytes]] = {}
    pos = 0
    for sha in shas:
        eol = out.index(b"\n", pos)
        header = out[pos:eol].split()
        pos = eol + 1
        if len(header) < 3:
            result[sha] = None
            continue
        size = int(header[2])
        result[sha] = out[pos:pos + size]
        pos += size + 1
    return result


# -- разбор нот: строки ИИ и ссылки по сессиям --------------------------------


def count_ranges(spec: str) -> int:
    """«1-4,9,12-13» → 7, не разворачивая диапазоны."""
    total = 0
    for chunk in spec.split(","):
        chunk = chunk.strip()
        if not chunk:
            continue
        lo, _, hi = chunk.partition("-")
        try:
            total += (int(hi) - int(lo) + 1) if hi else 1
        except ValueError:
            continue
    return max(0, total)


def notes_stats(repo: str) -> dict:
    """Свод по нотам: строки ИИ/человека и разрез по сессиям.

    Возвращает ``{"per_session": {s_id: {"ai_lines", "commits", "linked"}},
    "linked": [...], "ai_lines": N, "human_lines": N}``. Сирота — сессия, на
    которую нота ссылается ключом ``artifacts``, а файла в ref нет (проверяет
    вызывающий, здесь собираются все ссылки).
    """
    empty = {"per_session": {}, "linked": [], "ai_lines": 0, "human_lines": 0}
    listing = try_git(repo, "notes", f"--ref={NOTES_REF}", "list")
    if not listing:
        return empty

    pairs = []
    for line in listing.decode().splitlines():
        parts = line.split()
        if len(parts) == 2:
            pairs.append((parts[0], parts[1]))
    blobs = read_blobs(repo, [blob for blob, _ in pairs])

    per_session: Dict[str, dict] = {}
    linked: List[str] = []
    ai_lines = human_lines = 0
    for blob_sha, commit_sha in pairs:
        raw = blobs.get(blob_sha)
        if raw is None:
            continue
        text = "\n" + raw.decode("utf-8", "replace")
        attest, divider, meta = text.rpartition("\n---\n")
        if not divider:
            continue

        seen_here = set()
        for line in attest.splitlines():
            # Аттестация всегда с отступом: «  <ключ> <диапазоны>».
            # Без отступа идут пути файлов — их пропускаем.
            if not line.startswith("  "):
                continue
            key, _, spec = line.strip().partition(" ")
            if not spec:
                continue
            count = count_ranges(spec)
            if key.startswith("s_"):
                s_id = key.partition("::")[0]
                bucket = per_session.setdefault(
                    s_id, {"ai_lines": 0, "commits": 0, "linked": False}
                )
                bucket["ai_lines"] += count
                seen_here.add(s_id)
                ai_lines += count
            elif key.startswith("h_"):
                human_lines += count
        for s_id in seen_here:
            per_session[s_id]["commits"] += 1

        try:
            artifacts = json.loads(meta.strip().splitlines()[-1]).get("artifacts")
        except (ValueError, IndexError):
            continue
        if isinstance(artifacts, dict):
            for s_id in artifacts:
                per_session.setdefault(
                    s_id, {"ai_lines": 0, "commits": 0, "linked": False}
                )["linked"] = True
                linked.append(s_id)
    return {
        "per_session": per_session,
        "linked": linked,
        "ai_lines": ai_lines,
        "human_lines": human_lines,
    }


# -- сбор статистики ----------------------------------------------------------


def collect(repo: str) -> dict:
    tree = sessions_tree(repo)
    blobs = read_blobs(repo, list(tree.values()))
    chain = int(git(repo, "rev-list", "--count", ARTIFACTS_REF).decode().strip())

    by_tool: Dict[str, dict] = {}
    by_model: Dict[str, dict] = {}
    sessions: Dict[str, dict] = {}
    total_bytes = total_messages = total_chars = redactions = unreadable = 0
    first_at: Optional[float] = None
    last_at: Optional[float] = None

    for name, blob_sha in sorted(tree.items()):
        s_id = name[: -len(".json")]
        raw = blobs.get(blob_sha)
        total_bytes += len(raw or b"")
        try:
            record = json.loads((raw or b"").decode("utf-8"))
            assert isinstance(record, dict)
        except (ValueError, UnicodeDecodeError, AssertionError):
            unreadable += 1
            continue

        tool = str(record.get("tool") or "?")
        model = str(record.get("model") or "")
        messages = record.get("messages") or []
        chars = sum(len(str(m.get("text", ""))) for m in messages)
        cut = sum(str(m.get("text", "")).count(REDACTED) for m in messages)

        total_messages += len(messages)
        total_chars += chars
        redactions += cut
        for stamp in (record.get("created_at"), record.get("updated_at")):
            try:
                value = float(stamp)
            except (TypeError, ValueError):
                continue
            first_at = value if first_at is None else min(first_at, value)
            last_at = value if last_at is None else max(last_at, value)

        for bucket, key in ((by_tool, tool), (by_model, model or "не указана")):
            slot = bucket.setdefault(key, {"sessions": 0, "messages": 0})
            slot["sessions"] += 1
            slot["messages"] += len(messages)

        sessions[s_id] = {
            "tool": tool,
            "model": model,
            "messages": len(messages),
            "chars": chars,
            "updated_at": record.get("updated_at"),
        }

    from_notes = notes_stats(repo)
    linked = from_notes["linked"]
    for s_id, extra in from_notes["per_session"].items():
        if s_id in sessions:
            sessions[s_id]["ai_lines"] = extra["ai_lines"]
            sessions[s_id]["commits"] = extra["commits"]
    orphans = sorted(
        {s_id for s_id in linked if f"{s_id}.json" not in tree}
    )

    attributed = from_notes["ai_lines"] + from_notes["human_lines"]
    ai_pct = 100.0 * from_notes["ai_lines"] / attributed if attributed else 0.0

    return {
        "notes_ai_lines": from_notes["ai_lines"],
        "notes_human_lines": from_notes["human_lines"],
        "notes_ai_pct": round(ai_pct, 1),
        "notes_human_pct": round(100.0 - ai_pct, 1) if attributed else 0.0,
        "ref": ARTIFACTS_REF,
        "chain_commits": chain,
        "total_bytes": total_bytes,
        "sessions_total": len(tree),
        "unreadable": unreadable,
        "messages_total": total_messages,
        "prompt_chars_total": total_chars,
        "redactions": redactions,
        "first_at": first_at,
        "last_at": last_at,
        "by_tool": by_tool,
        "by_model": by_model,
        "sessions": sessions,
        "linked_from_notes": sorted(set(linked) & set(sessions)),
        "orphans": orphans,
    }


# -- вывод --------------------------------------------------------------------


def when(stamp) -> str:
    try:
        return time.strftime("%Y-%m-%d %H:%M", time.localtime(float(stamp)))
    except (TypeError, ValueError, OverflowError, OSError):
        return "?"


def print_report(stats: dict, top: int) -> None:
    print(
        f"{stats['ref']}: сессий {stats['sessions_total']}, "
        f"коммитов цепочки {stats['chain_commits']}, "
        f"{stats['total_bytes'] / 1024:.1f} КиБ"
    )
    if stats["unreadable"]:
        print(f"нечитаемых записей: {stats['unreadable']}")
    if stats["first_at"]:
        print(f"период: {when(stats['first_at'])} — {when(stats['last_at'])}")
    print(
        f"сообщений: {stats['messages_total']} "
        f"({stats['prompt_chars_total']} симв.), "
        f"вырезано секретов: {stats['redactions']}"
    )

    attributed = stats["notes_ai_lines"] + stats["notes_human_lines"]
    if attributed:
        print(
            f"код по нотам: ИИ {stats['notes_ai_lines']} строк "
            f"({stats['notes_ai_pct']}%), "
            f"человек {stats['notes_human_lines']} ({stats['notes_human_pct']}%)"
        )

    for title, bucket in (("по инструментам", stats["by_tool"]),
                          ("по моделям", stats["by_model"])):
        if not bucket:
            continue
        print(f"\n{title}:")
        rows = sorted(bucket.items(), key=lambda kv: -kv[1]["sessions"])
        for name, slot in rows:
            print(f"  {name:<24} сессий {slot['sessions']:<4} сообщений {slot['messages']}")

    with_lines = [
        (s_id, info) for s_id, info in stats["sessions"].items()
        if info.get("ai_lines")
    ]
    if with_lines:
        print(f"\nсвязка с нотами: строк ИИ за сессиями (топ {top}):")
        with_lines.sort(key=lambda kv: -kv[1]["ai_lines"])
        for s_id, info in with_lines[:top]:
            agent = info["tool"] + (f" / {info['model']}" if info["model"] else "")
            print(
                f"  {s_id}  {info['ai_lines']:>6} строк  "
                f"{info.get('commits', 0):>3} коммитов  {agent}"
            )
    linked = stats["linked_from_notes"]
    if linked:
        print(f"\nсессий со ссылками из нот: {len(linked)} из {stats['sessions_total']}")
    if stats["orphans"]:
        print(
            "ноты ссылаются на сессии, которых нет в ref: "
            + ", ".join(stats["orphans"])
        )


def main() -> int:
    parser = argparse.ArgumentParser(description=__doc__.splitlines()[0])
    parser.add_argument("--repo", default=".", help="путь к репозиторию (по умолчанию текущий)")
    parser.add_argument("--json", action="store_true", help="полный дамп в JSON")
    parser.add_argument("--top", type=int, default=10, help="размер топа сессий по строкам ИИ")
    args = parser.parse_args()

    if try_git(args.repo, "rev-parse", "--git-dir") is None:
        print(f"не git-репозиторий: {args.repo}", file=sys.stderr)
        return 2
    if not try_git(args.repo, "rev-parse", "--verify", "--quiet", ARTIFACTS_REF):
        print(f"в репозитории нет {ARTIFACTS_REF} — статистику считать не по чему")
        return 1

    stats = collect(args.repo)
    if args.json:
        json.dump(stats, sys.stdout, ensure_ascii=False, indent=2)
        print()
    else:
        print_report(stats, args.top)
    return 0


if __name__ == "__main__":
    sys.exit(main())