- Add\Rework UI - Add Split Table and Listing - Add Support Customazeble schems
This commit is contained in:
Igor20264
@@ -0,0 +1,335 @@
|
||||
"""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-<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]
|
||||
|
||||
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"
|
||||
Reference in New Issue
Block a user