arm-ka-docs/scripts/gen_model_diagrams.py

"""Диаграммы модели данных по частям: DBML → 9 частичных DBML → SVG (dbml-renderer) → PNG (resvg).

Запуск из корня репозитория: `make diagrams` (или `python3 scripts/gen_model_diagrams.py model/data-model-v5.17.dbml`).
Нужны Node.js и npx: пакеты рендера ставятся через `npx -y` с закреплёнными версиями (см. RENDERER, RASTERIZER).

Каждая часть — таблицы своей области плюс заглушки STUB (только `id`) для связей с таблицами других частей; в частичный
DBML копируются шапка «Ключи», решения DECISION-* и используемые enum'ы. Раскладка частей — как в
`context/04-data-model-guide.md`, «Девять частей модели»; новая таблица DBML, не попавшая ни в одну часть, — ошибка.
Промежуточные частичные DBML пишутся во временный каталог (с `--keep-parts DIR` — в DIR).
"""

import argparse
import pathlib
import re
import subprocess
import sys
import tempfile

RENDERER = "@softwaretechnik/dbml-renderer@1.0.31"
RASTERIZER = "@resvg/resvg-js-cli@2.6.2-beta.1"
PNG_WIDTH = "2200"

PARTS = {
    "01_groups_entities": (
        "ГК, ГСК/ГСЗ, организации, филиалы",
        ["regional_office", "company_group", "company_group_history", "legal_entity", "legal_entity_history", "legal_entity_alias", "related_group", "related_group_member"],
    ),
    "02_users_access": ("Пользователи и роли", ["app_user", "role", "permission", "role_permission", "user_role"]),
    "03_accounting_refs": (
        "План счетов, правила элиминации, обязательные документы",
        ["chart_of_account", "elimination_rule", "elimination_rule_account", "required_document_rule"],
    ),
    "04_files_jobs": (
        "Файлы, документы, S3, задания распознавания (NATS)",
        ["s3_object", "stored_file", "document", "recognition_job"],
    ),
    "05_osv_parsed": ("Разобранные ОСВ", ["osv_entry"]),
    "06_statements": (
        "Распознанные формы ББ/ОФР и логические проверки",
        [
            "financial_statement_meta",
            "financial_statement_column",
            "financial_statement_line",
            "financial_statement_value",
            "document_issue",
        ],
    ),
    "07_consolidation": (
        "Консолидация: версии, статусы, периметр, документы, блокировка",
        ["consolidation", "consolidation_version", "consolidation_status_history", "consolidation_lock", "consolidation_member", "consolidation_document"],
    ),
    "08_corrections_kfo": (
        "Корректировки, гудвилл, результат КФО, подтверждение КФО клиента",
        ["correction", "vgo_assignment", "goodwill", "kfo_result", "kfo_result_line", "kfo_result_line_company", "client_kfo_line"],
    ),
    "09_audit_ex": ("Комментарии, аудит, ЕХФД", ["comment", "audit_event", "ex_snapshot", "ex_snapshot_member", "ex_snapshot_item"]),
}

LEGEND = (
    "// Над каждой таблицей — блок «ИСТОЧНИК»: ТЗ АРМ КА BIQ-34 / ЕХ ТЗ BIQ-34649 / Методика; страница аналитики (md);\n"
    "//   экран макета dkb_24.09.26; версия модели, в которой таблица появилась (v4 = исходная DBML аналитики, v5+ = добавлено по ТЗ/макету).\n"
    "// Правило v5.3+: в модели только сущности из ТЗ или с экрана макета, плюс инфраструктура обмена (s3_object, recognition_job)."
)


def parse_items(src: str) -> dict[str, tuple[str, str]]:
    """Блоки `Enum`/`Table` с комментариями над ними: имя → (вид, текст блока)."""
    items = {}
    for m in re.finditer(r"^(Enum|Table)\s+(\w+)\s*\{", src, re.MULTILINE):
        i, depth = m.end(), 1
        while depth:
            c = src[i]
            depth += (c == "{") - (c == "}")
            i += 1
        before = src[: m.start()].rstrip("\n").split("\n")
        comments = []
        while before and before[-1].startswith("//"):
            comments.insert(0, before.pop())
        items[m.group(2)] = (m.group(1), ("\n".join(comments) + "\n" if comments else "") + src[m.start() : i])
    return items


def split_parts(src: str, src_name: str, version: str) -> dict[str, str]:
    """Текст частичного DBML для каждой части."""
    items = parse_items(src)
    enums = {k: v[1] for k, v in items.items() if v[0] == "Enum"}
    tables = {k: v[1] for k, v in items.items() if v[0] == "Table"}
    assigned = [t for _, ts in PARTS.values() for t in ts]
    if set(tables) != set(assigned):
        sys.exit(f"Раскладка частей не совпадает с таблицами DBML (поправить PARTS): {sorted(set(tables) ^ set(assigned))}")
    owner = {t: p for p, (_, ts) in PARTS.items() for t in ts}
    decisions = src[src.index("// DECISION-1") : src.index("// ============ ENUMS")].strip()
    conventions = src[src.index("// Ключи:") : src.index("// DECISION-1")].strip()
    index = "\n".join(f"//   part{p}: {t}" for p, (t, _) in PARTS.items())
    out = {}
    for pname, (title, ts) in PARTS.items():
        body = "\n\n".join(tables[t] for t in ts)
        used_enums = [e for e in enums if re.search(r"\b" + e + r"\b", body)]
        refs = set(re.findall(r"ref:\s*[<>-]\s*(\w+)\.", body))
        stubs = sorted(r for r in refs if r not in ts)
        stub = "\n\n".join(
            f"Table {r} {{\n  id uuid [pk]\n  Note: 'STUB — полное определение в part{owner[r]}'\n}}" for r in stubs
        )
        out[pname] = (
            f"// ДКБ · Модель данных {version} · part{pname}: {title}\n"
            f"// Часть общей модели {src_name} (единый источник истины; при правках менять обе).\n"
            f"// Части:\n{index}\n"
            "// Таблицы STUB ниже — заглушки (только id) для связей; их определения — в указанной части.\n"
            + LEGEND + "\n//\n" + conventions + "\n//\n" + decisions
            + "\n\n// ============ ENUMS ============\n" + "\n\n".join(enums[e] for e in used_enums)
            + ("\n\n// ============ STUBS (таблицы других частей) ============\n" + stub if stubs else "")
            + "\n\n// ============ TABLES ============\n" + body + "\n"
        )
        print(f"part{pname}: таблиц {len(ts)}, заглушек {len(stubs)}")
    return out


def render(part_dbml: pathlib.Path, svg: pathlib.Path, png: pathlib.Path) -> None:
    subprocess.run(["npx", "-y", RENDERER, "-i", str(part_dbml), "-o", str(svg)], check=True)
    subprocess.run(
        ["npx", "-y", RASTERIZER, "--fit-width", PNG_WIDTH, "--background", "white", str(svg), str(png)],
        check=True,
        stdout=subprocess.DEVNULL,
    )


def main() -> None:
    parser = argparse.ArgumentParser(description=__doc__.splitlines()[0])
    parser.add_argument("dbml", type=pathlib.Path, help="общая модель, например model/data-model-v5.17.dbml")
    parser.add_argument("--out", type=pathlib.Path, default=pathlib.Path("model/diagrams"), help="каталог SVG/PNG")
    parser.add_argument("--keep-parts", type=pathlib.Path, help="сохранить частичные DBML в этот каталог")
    args = parser.parse_args()

    m = re.search(r"data-model-(v[\d.]+)\.dbml$", args.dbml.name)
    if not m:
        sys.exit(f"Имя файла модели должно быть data-model-v<версия>.dbml: {args.dbml.name}")
    parts = split_parts(args.dbml.read_text(encoding="utf-8"), args.dbml.name, m.group(1))

    args.out.mkdir(parents=True, exist_ok=True)
    with tempfile.TemporaryDirectory() as tmp:
        parts_dir = args.keep_parts or pathlib.Path(tmp)
        parts_dir.mkdir(parents=True, exist_ok=True)
        for pname, text in parts.items():
            part_dbml = parts_dir / f"data-model-{m.group(1)}-part{pname}.dbml"
            part_dbml.write_text(text, encoding="utf-8")
            render(part_dbml, args.out / f"part{pname}.svg", args.out / f"part{pname}.png")
            print(f"  → {args.out}/part{pname}.svg, .png")


if __name__ == "__main__":
    main()