#!/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)