Igor20264
507 lines
16 KiB
Python
507 lines
16 KiB
Python
"""Configurable PlantUML diagram schemes (uml-c4, …) from md2gost.schemes.json."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import logging
|
|
import re
|
|
import shutil
|
|
from dataclasses import dataclass, field
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
from . import package_dir
|
|
from .diagram_includes import (
|
|
app_dir,
|
|
resolve_include_ref,
|
|
rewrite_http_includes_in_source,
|
|
)
|
|
|
|
_log = logging.getLogger(__name__)
|
|
|
|
SCHEMES_FILENAME = "md2gost.schemes.json"
|
|
SCHEME_ID_RE = re.compile(r"^[a-z][a-z0-9_]*$")
|
|
APP_SCHEME_AUTHOR = "md2gost"
|
|
BUILTIN_DIAGRAM_LANGS = frozenset({"uml", "plantuml"})
|
|
# Rendered via Kroki only (not PlantUML jar / schemes).
|
|
KROKI_DIAGRAM_LANGS = frozenset({"mermaid", "mmd"})
|
|
KROKI_TYPE_BY_LANG = {"mermaid": "mermaid", "mmd": "mermaid"}
|
|
# Local native renderers (not PlantUML / Kroki / schemes.json).
|
|
NATIVE_DIAGRAM_LANGS = {
|
|
"idef0": "idef0",
|
|
"uml-idef0": "idef0",
|
|
"dfd": "dfd",
|
|
"uml-dfd": "dfd",
|
|
"data-flow-diagram": "dfd",
|
|
"yourdon": "dfd",
|
|
"uml-yourdon": "dfd",
|
|
"dfd-gs": "dfd-gs",
|
|
"uml-dfd-gs": "dfd-gs",
|
|
"gane-sarson": "dfd-gs",
|
|
"uml-gane-sarson": "dfd-gs",
|
|
}
|
|
|
|
|
|
@dataclass
|
|
class SchemesMigrationResult:
|
|
added: list[str] = field(default_factory=list)
|
|
updated: list[str] = field(default_factory=list)
|
|
removed: list[str] = field(default_factory=list)
|
|
written: bool = False
|
|
|
|
@property
|
|
def changed(self) -> bool:
|
|
return bool(self.added or self.updated or self.removed)
|
|
|
|
|
|
@dataclass
|
|
class DiagramScheme:
|
|
id: str
|
|
title: str = ""
|
|
version: str = ""
|
|
author: str = ""
|
|
docs: str = ""
|
|
ai_prompt: str = ""
|
|
includes: list[str] = field(default_factory=list)
|
|
prefix: str = ""
|
|
postfix: str = ""
|
|
theme: str = ""
|
|
|
|
def to_dict(self) -> dict[str, Any]:
|
|
data: dict[str, Any] = {
|
|
"title": self.title,
|
|
"version": self.version,
|
|
"author": self.author,
|
|
"docs": self.docs,
|
|
"ai-prompt": self.ai_prompt,
|
|
"includes": list(self.includes),
|
|
"prefix": self.prefix,
|
|
"postfix": self.postfix,
|
|
}
|
|
if self.theme:
|
|
data["theme"] = self.theme
|
|
return data
|
|
|
|
@classmethod
|
|
def from_dict(cls, scheme_id: str, data: dict[str, Any]) -> DiagramScheme:
|
|
includes = data.get("includes") or []
|
|
if isinstance(includes, str):
|
|
includes = [includes]
|
|
return cls(
|
|
id=scheme_id,
|
|
title=str(data.get("title") or scheme_id),
|
|
version=str(data.get("version") or ""),
|
|
author=str(data.get("author") or ""),
|
|
docs=str(data.get("docs") or ""),
|
|
ai_prompt=str(data.get("ai-prompt") or data.get("ai_prompt") or ""),
|
|
includes=[str(x) for x in includes],
|
|
prefix=str(data.get("prefix") or ""),
|
|
postfix=str(data.get("postfix") or ""),
|
|
theme=str(data.get("theme") or ""),
|
|
)
|
|
|
|
|
|
def bundled_schemes_path() -> Path:
|
|
return Path(package_dir()) / "diagrams" / "schemes.json"
|
|
|
|
|
|
def user_schemes_path(base: Path | None = None) -> Path:
|
|
return (base or app_dir()) / SCHEMES_FILENAME
|
|
|
|
|
|
def is_app_owned_scheme(scheme: DiagramScheme) -> bool:
|
|
"""True if scheme is managed by the app template (author=md2gost)."""
|
|
return scheme.author.strip().lower() == APP_SCHEME_AUTHOR
|
|
|
|
|
|
def migrate_user_schemes(base: Path | None = None) -> SchemesMigrationResult:
|
|
"""
|
|
Add bundled schemes that are missing from the user file.
|
|
|
|
Existing entries are kept as written: edits of built-in ids, custom
|
|
schemes, and schemes with author=md2gost are not overwritten or removed.
|
|
A file that cannot be parsed is left untouched.
|
|
"""
|
|
result = SchemesMigrationResult()
|
|
path = user_schemes_path(base)
|
|
if not path.is_file():
|
|
return result
|
|
|
|
raw, err = _read_json_object(path)
|
|
if err or raw is None:
|
|
_log.warning("Миграция схем пропущена, файл не изменён (%s): %s", path, err)
|
|
return result
|
|
|
|
bundled = _load_schemes_file(bundled_schemes_path())
|
|
existing = {str(key).lower().strip() for key in raw}
|
|
for sid, scheme in bundled.items():
|
|
if sid in existing:
|
|
continue
|
|
raw[sid] = scheme.to_dict()
|
|
result.added.append(sid)
|
|
_log.info("Миграция схем: добавлена «%s»", sid)
|
|
|
|
if not result.added:
|
|
return result
|
|
|
|
path.write_text(
|
|
json.dumps(raw, ensure_ascii=False, indent=2) + "\n",
|
|
encoding="utf-8",
|
|
)
|
|
result.written = True
|
|
return result
|
|
|
|
|
|
def ensure_user_schemes(base: Path | None = None) -> Path:
|
|
"""
|
|
Ensure md2gost.schemes.json exists next to the app, then add any
|
|
bundled schemes that are not already in the user file.
|
|
"""
|
|
dest = user_schemes_path(base)
|
|
if not dest.is_file():
|
|
src = bundled_schemes_path()
|
|
dest.parent.mkdir(parents=True, exist_ok=True)
|
|
if src.is_file():
|
|
shutil.copyfile(src, dest)
|
|
_log.info("Создан файл схем: %s", dest)
|
|
else:
|
|
dest.write_text("{}\n", encoding="utf-8")
|
|
_log.warning("Шаблон схем не найден (%s), создан пустой %s", src, dest)
|
|
migrate_user_schemes(base)
|
|
return dest
|
|
|
|
|
|
def _read_json_object(path: Path) -> tuple[dict[str, Any] | None, str | None]:
|
|
"""Return (object, None). Missing file is ({}, None). Unreadable is (None, error)."""
|
|
if not path.is_file():
|
|
return {}, None
|
|
try:
|
|
# utf-8-sig: Notepad on Windows often writes a BOM, which json.loads rejects.
|
|
raw = json.loads(path.read_text(encoding="utf-8-sig"))
|
|
except (OSError, UnicodeError, json.JSONDecodeError) as exc:
|
|
return None, str(exc)
|
|
if not isinstance(raw, dict):
|
|
return None, "корень JSON должен быть объектом"
|
|
return raw, None
|
|
|
|
|
|
def _schemes_from_raw(raw: dict[str, Any]) -> dict[str, DiagramScheme]:
|
|
out: dict[str, DiagramScheme] = {}
|
|
for key, val in raw.items():
|
|
sid = str(key).lower().strip()
|
|
if not SCHEME_ID_RE.match(sid):
|
|
_log.warning("Пропуск схемы с недопустимым id: %s", key)
|
|
continue
|
|
if not isinstance(val, dict):
|
|
continue
|
|
out[sid] = DiagramScheme.from_dict(sid, val)
|
|
return out
|
|
|
|
|
|
def _load_schemes_file(path: Path) -> dict[str, DiagramScheme]:
|
|
raw, err = _read_json_object(path)
|
|
if err or raw is None:
|
|
if path.is_file():
|
|
_log.warning("Не удалось прочитать схемы %s: %s", path, err)
|
|
return {}
|
|
return _schemes_from_raw(raw)
|
|
|
|
|
|
def load_schemes_from_path(path: Path) -> dict[str, DiagramScheme]:
|
|
return _load_schemes_file(path)
|
|
|
|
|
|
def load_schemes_report(path: Path) -> tuple[dict[str, DiagramScheme], str | None]:
|
|
"""Load schemes. Error string is set when the file exists but cannot be parsed."""
|
|
raw, err = _read_json_object(path)
|
|
if err or raw is None:
|
|
if path.is_file():
|
|
_log.warning("Не удалось прочитать схемы %s: %s", path, err)
|
|
return {}, err
|
|
return _schemes_from_raw(raw), None
|
|
|
|
|
|
def save_schemes(
|
|
schemes: dict[str, DiagramScheme],
|
|
path: Path | None = None,
|
|
*,
|
|
removed_ids: set[str] | None = None,
|
|
) -> Path:
|
|
"""
|
|
Write schemes, keeping JSON entries the loader does not understand
|
|
(unknown ids, extra fields). ``removed_ids`` are normalized ids the user
|
|
deleted or renamed away.
|
|
"""
|
|
dest = path or user_schemes_path()
|
|
dest.parent.mkdir(parents=True, exist_ok=True)
|
|
raw, err = _read_json_object(dest)
|
|
if err or raw is None:
|
|
raw = {}
|
|
removed = {sid.lower() for sid in (removed_ids or set())}
|
|
norm_to_key: dict[str, str] = {}
|
|
for key in list(raw):
|
|
sid = str(key).lower().strip()
|
|
if sid in removed:
|
|
del raw[key]
|
|
continue
|
|
norm_to_key.setdefault(sid, str(key))
|
|
|
|
for sid in sorted(schemes):
|
|
scheme = schemes[sid]
|
|
data = scheme.to_dict()
|
|
key = norm_to_key.get(sid)
|
|
if key is None:
|
|
raw[sid] = data
|
|
continue
|
|
base = raw.get(key)
|
|
merged = dict(base) if isinstance(base, dict) else {}
|
|
merged.update(data)
|
|
if not scheme.theme:
|
|
merged.pop("theme", None)
|
|
raw[key] = merged
|
|
|
|
dest.write_text(
|
|
json.dumps(raw, ensure_ascii=False, indent=2) + "\n",
|
|
encoding="utf-8",
|
|
)
|
|
return dest
|
|
|
|
|
|
def claim_user_scheme_if_diverged(
|
|
scheme: DiagramScheme,
|
|
*,
|
|
bundled: dict[str, DiagramScheme] | None = None,
|
|
) -> DiagramScheme:
|
|
"""
|
|
If scheme id is in the app template and content differs while still
|
|
author=md2gost, clear author so the copy is marked as the user's.
|
|
"""
|
|
stock = (bundled if bundled is not None else _load_schemes_file(bundled_schemes_path())).get(
|
|
scheme.id
|
|
)
|
|
if stock is None:
|
|
return scheme
|
|
if not is_app_owned_scheme(scheme):
|
|
return scheme
|
|
if scheme.to_dict() == stock.to_dict():
|
|
return scheme
|
|
return DiagramScheme(
|
|
id=scheme.id,
|
|
title=scheme.title,
|
|
version=scheme.version,
|
|
author="",
|
|
docs=scheme.docs,
|
|
ai_prompt=scheme.ai_prompt,
|
|
includes=list(scheme.includes),
|
|
prefix=scheme.prefix,
|
|
postfix=scheme.postfix,
|
|
theme=scheme.theme,
|
|
)
|
|
|
|
|
|
def native_diagram_type(lang: str) -> str | None:
|
|
"""Kroki/PlantUML-independent diagram type, or None."""
|
|
lang = (lang or "").lower().strip()
|
|
return NATIVE_DIAGRAM_LANGS.get(lang)
|
|
|
|
|
|
def scheme_id_from_lang(lang: str) -> str | None:
|
|
"""Normalize fence lang to scheme id, or None for bare uml/plantuml."""
|
|
lang = (lang or "").lower().strip()
|
|
if not lang:
|
|
return None
|
|
if lang in BUILTIN_DIAGRAM_LANGS or lang in KROKI_DIAGRAM_LANGS or lang in NATIVE_DIAGRAM_LANGS:
|
|
return None
|
|
if lang.startswith("uml-"):
|
|
sid = lang[4:]
|
|
return sid if SCHEME_ID_RE.match(sid) else None
|
|
if SCHEME_ID_RE.match(lang):
|
|
return lang
|
|
return None
|
|
|
|
|
|
def load_merged_schemes(
|
|
*,
|
|
base: Path | None = None,
|
|
extra_path: Path | None = None,
|
|
md_dir: Path | None = None,
|
|
) -> dict[str, DiagramScheme]:
|
|
"""
|
|
Merge layers (later wins entirely per id):
|
|
1) bundled template
|
|
2) user file next to app (created on first run)
|
|
3) md2gost.schemes.json next to .md
|
|
4) explicit --schemes path
|
|
"""
|
|
ensure_user_schemes(base)
|
|
merged: dict[str, DiagramScheme] = {}
|
|
merged.update(_load_schemes_file(bundled_schemes_path()))
|
|
merged.update(_load_schemes_file(user_schemes_path(base)))
|
|
if md_dir:
|
|
near_md = Path(md_dir) / SCHEMES_FILENAME
|
|
if near_md.is_file():
|
|
user = user_schemes_path(base).resolve()
|
|
try:
|
|
if near_md.resolve() != user:
|
|
merged.update(_load_schemes_file(near_md))
|
|
except OSError:
|
|
merged.update(_load_schemes_file(near_md))
|
|
if extra_path:
|
|
merged.update(_load_schemes_file(Path(extra_path)))
|
|
return merged
|
|
|
|
|
|
_SCHEMES_CACHE: dict[str, DiagramScheme] | None = None
|
|
_SCHEMES_EXTRA: Path | None = None
|
|
_SCHEMES_MD_DIR: Path | None = None
|
|
|
|
|
|
def configure_schemes(
|
|
*,
|
|
schemes_path: str | None = None,
|
|
md_dir: str | None = None,
|
|
base: Path | None = None,
|
|
) -> dict[str, DiagramScheme]:
|
|
global _SCHEMES_CACHE, _SCHEMES_EXTRA, _SCHEMES_MD_DIR
|
|
_SCHEMES_EXTRA = Path(schemes_path) if schemes_path else None
|
|
_SCHEMES_MD_DIR = Path(md_dir) if md_dir else None
|
|
_SCHEMES_CACHE = load_merged_schemes(
|
|
base=base,
|
|
extra_path=_SCHEMES_EXTRA,
|
|
md_dir=_SCHEMES_MD_DIR,
|
|
)
|
|
return _SCHEMES_CACHE
|
|
|
|
|
|
def get_schemes() -> dict[str, DiagramScheme]:
|
|
global _SCHEMES_CACHE
|
|
if _SCHEMES_CACHE is None:
|
|
_SCHEMES_CACHE = load_merged_schemes()
|
|
return _SCHEMES_CACHE
|
|
|
|
|
|
def reload_schemes(**kwargs) -> dict[str, DiagramScheme]:
|
|
return configure_schemes(**kwargs)
|
|
|
|
|
|
def get_scheme(scheme_id: str) -> DiagramScheme | None:
|
|
return get_schemes().get(scheme_id)
|
|
|
|
|
|
def is_diagram_lang(lang: str) -> bool:
|
|
"""True if fence should render as a diagram (not a plain listing)."""
|
|
lang = (lang or "").lower().strip()
|
|
if lang in BUILTIN_DIAGRAM_LANGS or lang in KROKI_DIAGRAM_LANGS or lang in NATIVE_DIAGRAM_LANGS:
|
|
return True
|
|
if lang.startswith("uml-"):
|
|
# Force diagram path so missing scheme becomes an error, not a listing.
|
|
return bool(SCHEME_ID_RE.match(lang[4:]))
|
|
return lang in get_schemes()
|
|
|
|
|
|
def apply_scheme(
|
|
scheme: DiagramScheme,
|
|
body: str,
|
|
*,
|
|
base: Path | None = None,
|
|
) -> str:
|
|
"""Wrap body with includes / prefix / postfix / theme; resolve URL includes."""
|
|
text = body.strip()
|
|
has_start = text.lower().startswith("@start")
|
|
|
|
include_lines: list[str] = []
|
|
schemes_dir = user_schemes_path(base).parent
|
|
for ref in scheme.includes:
|
|
local = resolve_include_ref(ref, base=base, schemes_dir=schemes_dir)
|
|
include_lines.append(f"!include {local}")
|
|
|
|
theme_line = ""
|
|
if scheme.theme and "!theme" not in text.lower():
|
|
t = scheme.theme.strip()
|
|
if not t.lower().startswith("!theme"):
|
|
t = f"!theme {t}"
|
|
theme_line = t
|
|
|
|
prefix = scheme.prefix or ""
|
|
postfix = scheme.postfix or ""
|
|
|
|
# Avoid double @startuml / @enduml when body already has @start…
|
|
if has_start:
|
|
prefix_use = _strip_start_end_wrappers(prefix)
|
|
postfix_use = _strip_start_end_wrappers(postfix)
|
|
head_bits = [ln for ln in include_lines if ln]
|
|
if theme_line:
|
|
head_bits.append(theme_line)
|
|
if prefix_use.strip():
|
|
head_bits.append(prefix_use.strip())
|
|
if head_bits:
|
|
lines = text.splitlines()
|
|
text = lines[0] + "\n" + "\n".join(head_bits) + "\n" + "\n".join(lines[1:])
|
|
if postfix_use.strip():
|
|
text = text.rstrip() + "\n" + postfix_use.strip()
|
|
else:
|
|
parts: list[str] = []
|
|
p = prefix.rstrip("\n") if prefix else ""
|
|
pf = postfix.lstrip("\n") if postfix else ""
|
|
if not p.lstrip().lower().startswith("@start"):
|
|
parts.append("@startuml")
|
|
if p:
|
|
parts.append(p)
|
|
parts.extend(include_lines)
|
|
if theme_line:
|
|
parts.append(theme_line)
|
|
parts.append(text)
|
|
if pf:
|
|
parts.append(pf)
|
|
elif "@enduml" not in text.lower():
|
|
parts.append("@enduml")
|
|
text = "\n".join(parts)
|
|
|
|
text = rewrite_http_includes_in_source(text, base=base)
|
|
return text
|
|
|
|
|
|
def _strip_start_end_wrappers(chunk: str) -> str:
|
|
"""Remove leading @start… and trailing @end… lines from prefix/postfix."""
|
|
lines = chunk.splitlines()
|
|
while lines and lines[0].strip().lower().startswith("@start"):
|
|
lines = lines[1:]
|
|
while lines and lines[-1].strip().lower().startswith("@end"):
|
|
lines = lines[:-1]
|
|
return "\n".join(lines)
|
|
|
|
|
|
def prepare_with_schemes(lang: str, source: str, *, base: Path | None = None) -> tuple[str, str]:
|
|
"""
|
|
Prepare diagram source using schemes.
|
|
Returns (prepared_source, kroki_diagram_type).
|
|
Raises ValueError if uml-<id> / named scheme is missing.
|
|
"""
|
|
lang = (lang or "uml").lower().strip()
|
|
text = source.strip()
|
|
|
|
if lang in KROKI_DIAGRAM_LANGS:
|
|
return text, KROKI_TYPE_BY_LANG[lang]
|
|
|
|
native = native_diagram_type(lang)
|
|
if native:
|
|
return text, native
|
|
|
|
sid = scheme_id_from_lang(lang)
|
|
|
|
if sid is None:
|
|
# bare uml / plantuml
|
|
if not text.lower().startswith("@start"):
|
|
text = f"@startuml\n{text}\n@enduml"
|
|
text = rewrite_http_includes_in_source(text, base=base)
|
|
return text, "plantuml"
|
|
|
|
scheme = get_scheme(sid)
|
|
if scheme is None:
|
|
raise ValueError(
|
|
f"Схема диаграммы «{sid}» не найдена "
|
|
f"(оградка ```{lang}). Добавьте её в {SCHEMES_FILENAME}."
|
|
)
|
|
prepared = apply_scheme(scheme, text, base=base)
|
|
return prepared, "plantuml"
|