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

scripts/read_artifacts.py
218 строк8 KB
Mike S
Write to refs/ai/artifacts
09 авг 2026, 23:24
09 авг 2026, 23:24b0fd226
#!/usr/bin/env python3
"""Читатель ``refs/ai/artifacts`` без git-ai: только git и стандартная библиотека.

Обходит хранилище артефактов (уровень 2): перечисляет записи сессий с вершины
ref, читает их одним ``git cat-file --batch`` и, если есть ноты
``refs/notes/ai``, показывает, какие коммиты ссылаются на какую сессию.

Автономность — намеренная: формат обязан читаться на машине, где git-ai не
установлен (ревью чужого клона, CI), и этот скрипт — живая проверка обещания.

    python3 scripts/read_artifacts.py [--repo ПУТЬ] [--json] [--full]
"""

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"  # короткое имя для `git notes --ref=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 ---------------------------------------------------------------


def sessions_tree(repo: str) -> Dict[str, str]:
    """Имя файла сессии → sha блоба на вершине ref."""
    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]]:
    """Содержимое блобов пачкой: один процесс git на все записи."""
    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:  # "<sha> missing"
            result[sha] = None
            continue
        size = int(header[2])
        result[sha] = out[pos:pos + size]
        pos += size + 1  # завершающий \n после содержимого
    return result


def parse_record(raw: Optional[bytes]) -> Optional[dict]:
    if raw is None:
        return None
    try:
        data = json.loads(raw.decode("utf-8"))
    except (ValueError, UnicodeDecodeError):
        return None
    return data if isinstance(data, dict) else None


# -- ссылки из нот ------------------------------------------------------------


def note_links(repo: str) -> Dict[str, List[str]]:
    """s_id → коммиты, чьи ноты ссылаются на эту сессию (ключ ``artifacts``).

    Нота — это аттестации, строка ``---`` и одна строка JSON; ссылки лежат в
    JSON. Ноты читаются той же пачкой ``cat-file --batch``, что и записи.
    """
    listing = try_git(repo, "notes", f"--ref={NOTES_REF}", "list")
    if not listing:
        return {}

    pairs = []  # (sha блоба ноты, sha коммита)
    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])

    links: Dict[str, List[str]] = {}
    for blob_sha, commit_sha in pairs:
        raw = blobs.get(blob_sha)
        if raw is None:
            continue
        # Ведущий \n — чтобы разделитель нашёлся и у ноты без аттестаций,
        # которая начинается сразу со строки «---».
        text = "\n" + raw.decode("utf-8", "replace")
        _, divider, meta = text.rpartition("\n---\n")
        if not divider:
            continue
        try:
            artifacts = json.loads(meta.strip().splitlines()[-1]).get("artifacts")
        except (ValueError, IndexError):
            continue
        if isinstance(artifacts, dict):
            for s_id in artifacts:
                links.setdefault(s_id, []).append(commit_sha)
    return links


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


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


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("--full", action="store_true", 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
    tip = try_git(args.repo, "rev-parse", "--verify", "--quiet", ARTIFACTS_REF)
    if not tip:
        print(f"в репозитории нет {ARTIFACTS_REF} — артефакты не записывались "
              "или не доехали (git-ai artifacts fetch)")
        return 1

    tree = sessions_tree(args.repo)
    blobs = read_blobs(args.repo, list(tree.values()))
    links = note_links(args.repo)
    chain = git(args.repo, "rev-list", "--count", ARTIFACTS_REF).decode().strip()

    records = {
        name[: -len(".json")]: parse_record(blobs.get(sha))
        for name, sha in sorted(tree.items())
    }

    if args.json:
        json.dump(
            {
                "ref": ARTIFACTS_REF,
                "tip": tip.decode().strip(),
                "chain_commits": int(chain),
                "sessions": records,
                "referenced_by": links,
            },
            sys.stdout, ensure_ascii=False, indent=2,
        )
        print()
        return 0

    print(f"{ARTIFACTS_REF}: записей {len(records)}, коммитов цепочки {chain}")
    for s_id, record in records.items():
        print(f"\n{s_id}")
        if record is None:
            print("  запись не читается (битый JSON?)")
            continue
        tool = record.get("tool", "?")
        model = record.get("model", "")
        print(f"  агент      {tool}{' / ' + model if model else ''}")
        print(f"  обновлена  {when(record.get('updated_at'))}")
        commits = links.get(s_id)
        if commits:
            print(f"  в нотах    {', '.join(sha[:12] for sha in commits)}")
        messages = record.get("messages") or []
        if messages:
            shown = messages if args.full else messages[-3:]
            skipped = len(messages) - len(shown)
            print(f"  сообщений  {len(messages)}" + (f" (показаны последние {len(shown)})" if skipped else ""))
            for message in shown:
                text = str(message.get("text", "")).strip().replace("\n", " ")
                print(f"    [{when(message.get('at'))}] {text[:160]}")

    orphans = sorted(set(links) - set(records))
    if orphans:
        print(f"\nноты ссылаются на сессии, которых нет в ref: {', '.join(orphans)}")
        print("(сброс не случился или не доехал — git-ai artifacts flush / fetch)")
    return 0


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