"""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_]*$") 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"} @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 ensure_user_schemes(base: Path | None = None) -> Path: """ On first run copy bundled template next to the app. Never overwrite an existing user file. """ dest = user_schemes_path(base) if dest.is_file(): return dest 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) return dest def _load_schemes_file(path: Path) -> dict[str, DiagramScheme]: if not path.is_file(): return {} try: raw = json.loads(path.read_text(encoding="utf-8")) except (OSError, json.JSONDecodeError) as exc: _log.warning("Не удалось прочитать схемы %s: %s", path, exc) return {} if not isinstance(raw, dict): return {} 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_from_path(path: Path) -> dict[str, DiagramScheme]: return _load_schemes_file(path) def save_schemes(schemes: dict[str, DiagramScheme], path: Path | None = None) -> Path: dest = path or user_schemes_path() payload = {sid: scheme.to_dict() for sid, scheme in sorted(schemes.items())} dest.parent.mkdir(parents=True, exist_ok=True) dest.write_text( json.dumps(payload, ensure_ascii=False, indent=2) + "\n", encoding="utf-8", ) return dest 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: 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: 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- / 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] 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"