dev-skills/scripts/diagrams.py

#!/usr/bin/env python3
"""Проверка mermaid-диаграмм в markdown этого репозитория.

Диаграммы заведены там, где структура — граф или автомат: порядок проходов
ревью, жизненный цикл записи по индексам, исходы задачи в спринте, храповик
промоута, счётчик калибровки, граф вызовов между плагинами.

Проверка нужна по одной причине: **синтаксическая ошибка в блоке не видна при
чтении**. Текст диаграммы выглядит правдоподобно, `git diff` показывает разумную
строку, ревью её пропускает — а отрендерить не удаётся, и читатель видит вместо
схемы полотно исходника либо сообщение об ошибке. Первый раз это ловилось тем,
что автор не забыл прогнать рендер руками; на «не забыл» проверки не строятся.

Чего проверка **не** ловит: расхождение диаграммы с прозой вокруг неё. Это
второй дом для одного факта, и удержать его в синхроне может только правило
старшинства, записанное рядом с каждой диаграммой (в конвейере ревью старший —
граф, он и есть алгоритм; в остальных местах старшая — проза, диаграмма там
сводка). Механической сверки для этого нет: между текстом и графом нет
дословного соответствия, которое можно было бы сличить, — в отличие от копий
правил, где оно есть и где его сверяет copies.py.

Рендерит `mermaid-cli`: бинарь `mmdc`, если он на PATH, иначе
`npx --yes @mermaid-js/mermaid-cli`. Ни того ни другого нет — это код 3, а не
молчаливый успех: проверка, отчитавшаяся «диаграммы в порядке», ничего не
отрендерив, хуже отсутствующей.

Chromium запускается с `--no-sandbox`: на современных дистрибутивах
непривилегированные user namespaces выключены, и без флага рендер падает на
«No usable sandbox». Содержимое здесь своё и локальное, так что песочница ничего
не защищает — она только мешает запуску.

Дорого здесь не чтение markdown, а рендер: каждый блок — отдельный запуск
mermaid-cli со своим chromium, секунда с лишним. Отсюда два рычага, и оба нужны
гейту коммита:

- **блоки собираются все сразу, а рендерятся параллельно.** Сбор — обход файлов,
  он же и определяет порядок вывода; рендер ждёт подпроцесс и потому пускается
  пулом потоков. Порядок находок от этого не плывёт: он берётся из порядка
  сбора, а не из порядка ответов;
- **проверять можно не весь репозиторий, а названные файлы.** `diagrams.py
  путь.md …` смотрит только их — так гейт платит за диаграммы ровно того файла,
  который правят. Без аргументов обходится весь репозиторий, как и раньше.

Коды выхода — тот же словарь, что у tasks.py, docs.py и copies.py:
  0  все диаграммы рендерятся
  1  диаграмма не рендерится
  2  ошибка употребления: аргументы
  3  окружение: не тот каталог, нет mermaid-cli
  4  внутренний сбой
"""

from __future__ import annotations

import argparse
import json
import os
import re
import shutil
import subprocess
import sys
import tempfile
from concurrent.futures import ThreadPoolExecutor
from pathlib import Path

OK, DRIFT, USAGE, ENV, INTERNAL = 0, 1, 2, 3, 4

# Потолок параллели. Каждый рендер — свой chromium, а он стоит сотни мегабайт:
# на машине с 24 ядрами упереться в память дешевле, чем в процессор. Восемь
# снимают почти весь выигрыш и не рискуют ничем.
MAX_WORKERS = 8

FENCE_OPEN = re.compile(r"^\s*```mermaid\s*$")
FENCE_CLOSE = re.compile(r"^\s*```\s*$")

# Каталоги, где markdown не наш: клоны, окружения, временное.
SKIP = {".git", ".venv", "node_modules", "tmp", ".ruff_cache"}

# Стек вызовов puppeteer/mermaid полезного не несёт — режем по первой его строке.
NOISE = re.compile(r"^(Parser\.|\s+at )")
# Строки хода работы: не ошибка, а прогресс рендера.
PROGRESS = re.compile(r"^(Generating|Found \d+ mermaid)")


class Block:
    """Один блок ```mermaid: где начался и что внутри."""

    def __init__(self, path: Path, line: int, text: str, root: Path) -> None:
        self.path = path
        self.line = line
        self.text = text
        self.where = f"{path.relative_to(root).as_posix()}:{line}"


def markdown(root: Path, named: list[Path]) -> list[Path]:
    """Какие файлы смотреть: названные или весь репозиторий.

    Названные фильтруются теми же правилами, что и обход: только `*.md`, только
    внутри корня, без пропускаемых каталогов. Гейт передаёт сюда staged-файлы
    списком, в котором есть и скрипты, и удалённое, — отбор его дело, а не
    вызывающего.
    """
    if not named:
        return sorted(root.rglob("*.md"))
    out = []
    for path in named:
        full = (path if path.is_absolute() else root / path).resolve()
        if full.suffix == ".md" and full.is_file() and full.is_relative_to(root):
            out.append(full)
    return sorted(set(out))


def collect(root: Path, named: list[Path]) -> list[Block]:
    """Все mermaid-блоки, в порядке обхода. Порядок вывода берётся отсюда."""
    found: list[Block] = []
    for path in markdown(root, named):
        if any(part in SKIP for part in path.relative_to(root).parts):
            continue
        lines = path.read_text(encoding="utf-8").splitlines()
        start: int | None = None
        body: list[str] = []
        for i, line in enumerate(lines, start=1):
            if start is None:
                if FENCE_OPEN.match(line):
                    start, body = i, []
                continue
            if FENCE_CLOSE.match(line):
                found.append(Block(path, start, "\n".join(body) + "\n", root))
                start = None
                continue
            body.append(line)
        if start is not None:
            found.append(Block(path, start, "\n".join(body) + "\n", root))
    return found


def renderer() -> list[str] | None:
    """Команда рендера или None, если mermaid-cli недоступен."""
    mmdc = shutil.which("mmdc")
    if mmdc:
        return [mmdc]
    if shutil.which("npx"):
        return ["npx", "--yes", "@mermaid-js/mermaid-cli"]
    return None


def render(cmd: list[str], block: Block, workdir: Path, config: Path,
           slot: int) -> str | None:
    """Отрендерить блок. None — получилось, иначе сообщение об ошибке.

    `slot` разводит временные файлы: рендеры идут параллельно, и одно имя на
    всех означало бы, что блоки затирают исходники друг друга — с находками,
    которые не воспроизводятся поодиночке.
    """
    src = workdir / f"d{slot}.mmd"
    src.write_text(block.text, encoding="utf-8")
    out = workdir / f"d{slot}.svg"
    done = subprocess.run(
        [*cmd, "-p", str(config), "-i", str(src), "-o", str(out)],
        capture_output=True,
        text=True,
        check=False,
    )
    if done.returncode == 0 and out.exists():
        out.unlink()
        return None
    noise = f"{done.stdout}\n{done.stderr}"
    useful = []
    for line in noise.splitlines():
        if NOISE.match(line):
            break
        if line.strip() and not PROGRESS.match(line.strip()):
            useful.append(line.strip())
    return "\n    ".join(useful[:8]) or f"код {done.returncode} без сообщения"


def main() -> int:
    ap = argparse.ArgumentParser(description="Проверка mermaid-диаграмм.")
    ap.add_argument("paths", nargs="*", type=Path,
                    help="какие файлы смотреть; без них — весь репозиторий")
    ap.add_argument("--dir", default=".", help="корень репозитория")
    args = ap.parse_args()

    root = Path(args.dir).resolve()
    if not (root / ".claude-plugin").is_dir():
        print(f"окружение: {root} не похож на корень репозитория"
              f" (нет .claude-plugin)", file=sys.stderr)
        return ENV

    blocks = collect(root, args.paths)
    if not blocks:
        print("диаграмм нет" if not args.paths
              else "диаграмм нет в названных файлах")
        return OK

    cmd = renderer()
    if cmd is None:
        print("окружение: нет ни mmdc, ни npx — рендер невозможен."
              " Поставь mermaid-cli (npm i -g @mermaid-js/mermaid-cli)"
              " или запусти проверку там, где есть npx.", file=sys.stderr)
        return ENV

    with tempfile.TemporaryDirectory(prefix="diagrams-") as tmp:
        workdir = Path(tmp)
        config = workdir / "puppeteer.json"
        config.write_text(json.dumps({"args": ["--no-sandbox"]}), encoding="utf-8")
        workers = max(1, min(MAX_WORKERS, len(blocks), os.cpu_count() or 1))
        with ThreadPoolExecutor(max_workers=workers) as pool:
            # Потоки, а не процессы: работа целиком в ожидании подпроцесса,
            # своего интерпретатора ей не надо. `map` сохраняет порядок блоков,
            # поэтому вывод не зависит от того, кто ответил первым.
            errors = list(pool.map(
                lambda pair: render(cmd, pair[1], workdir, config, pair[0]),
                enumerate(blocks)))
    broken = [(b, e) for b, e in zip(blocks, errors, strict=True) if e is not None]

    files = len({b.path for b in blocks})
    print(f"диаграмм {len(blocks)} в {files} файлах")
    if broken:
        print()
        for block, error in broken:
            print(f"НЕ РЕНДЕРИТСЯ  {block.where}\n    {error}")
        print(f"\nИтог: не рендерятся {len(broken)}."
              f" Блок правится в самом markdown — картинок в репозитории нет"
              f" и не заводится, диаграмма живёт исходником рядом с текстом.")
        return DRIFT

    print("все рендерятся")
    return OK


if __name__ == "__main__":
    try:
        sys.exit(main())
    except KeyboardInterrupt:
        sys.exit(INTERNAL)
    except Exception as e:  # noqa: BLE001 — последний рубеж, код 4 по словарю
        print(f"внутренний сбой ({type(e).__name__}): {e}", file=sys.stderr)
        sys.exit(INTERNAL)