"""Диаграммы модели данных по частям: 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"], ), "07_consolidation": ( "Консолидация: версии, статусы, периметр, документы, блокировка", ["consolidation", "consolidation_version", "consolidation_status_history", "consolidation_lock", "consolidation_member", "consolidation_document"], ), "08_corrections_kfo": ( "Корректировки, гудвилл, результат КФО, подтверждение КФО клиента", ["correction", "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()