from __future__ import annotations import json import shutil import subprocess import tempfile from pathlib import Path from typing import Any import yaml from api.config import settings class MarkitectUnavailable(RuntimeError): """Raised when neither MarkiTect's Python API nor CLI can render templates.""" class MarkitectRenderError(RuntimeError): """Raised when MarkiTect rejects a template or data payload.""" def render_markdown_template(template_path: Path, data: dict[str, Any]) -> str: """Render a Markdown template with MarkiTect. Prefer the Python API when available. Fall back to the CLI because local workstations may have MarkiTect installed as a standalone tool. """ template_text = template_path.read_text(encoding="utf-8") try: from markitect_tool.template import render_template except ImportError: return _render_with_cli(template_path, data) try: return render_template(template_text, data, strict=True).markdown except Exception as exc: # MarkiTect raises its own TemplateError hierarchy. raise MarkitectRenderError(str(exc)) from exc def _render_with_cli(template_path: Path, data: dict[str, Any]) -> str: cli = _find_markitect_cli() if cli is None: raise MarkitectUnavailable( "MarkiTect is required to render RecentlyOnScope digests. " "Install markitect_tool, put `markitect` on PATH, or set STATE_HUB_MARKITECT_CLI_PATH." ) with tempfile.TemporaryDirectory(prefix="state-hub-markitect-") as tmp: data_path = Path(tmp) / "data.yaml" data_path.write_text(yaml.safe_dump(data, sort_keys=False), encoding="utf-8") result = _run_first_supported( [ [cli, "template", "render", str(template_path), "--data", str(data_path)], [cli, "template-render", str(template_path), str(data_path), "--strict"], ] ) if result.returncode != 0: detail = (result.stderr or result.stdout or "MarkiTect render failed").strip() raise MarkitectRenderError(detail) return result.stdout def inspect_markdown_template(template_path: Path) -> dict[str, Any]: """Return MarkiTect template analysis data where available.""" template_text = template_path.read_text(encoding="utf-8") try: from markitect_tool.template import analyze_template except ImportError: return _inspect_with_cli(template_path) analysis = analyze_template(template_text).to_dict() if not analysis.get("valid", False): raise MarkitectRenderError("; ".join(analysis.get("syntax_errors") or [])) return analysis def _inspect_with_cli(template_path: Path) -> dict[str, Any]: cli = _find_markitect_cli() if cli is None: raise MarkitectUnavailable( "MarkiTect is required to inspect RecentlyOnScope templates. " "Install markitect_tool, put `markitect` on PATH, or set STATE_HUB_MARKITECT_CLI_PATH." ) result = subprocess.run( [cli, "template", "inspect", str(template_path), "--format", "json"], check=False, capture_output=True, text=True, ) if _is_unsupported_cli_command(result): with tempfile.TemporaryDirectory(prefix="state-hub-markitect-") as tmp: data_path = Path(tmp) / "empty.yaml" data_path.write_text("{}\n", encoding="utf-8") legacy = subprocess.run( [cli, "template-render", str(template_path), str(data_path), "--lenient", "--validate"], check=False, capture_output=True, text=True, ) if legacy.returncode != 0: detail = (legacy.stderr or legacy.stdout or "MarkiTect template inspection failed").strip() raise MarkitectRenderError(detail) return {"valid": True, "variables": [], "legacy_cli": True} if result.returncode != 0: detail = (result.stderr or result.stdout or "MarkiTect template inspection failed").strip() raise MarkitectRenderError(detail) try: return json.loads(result.stdout) except json.JSONDecodeError as exc: raise MarkitectRenderError("MarkiTect returned invalid JSON during template inspection") from exc def _find_markitect_cli() -> str | None: candidates = [ settings.state_hub_markitect_cli_path, shutil.which("markitect"), "/home/worsch/bin/markitect", ] for candidate in candidates: if not candidate: continue path = Path(candidate).expanduser() if path.is_file(): return str(path) if shutil.which(str(candidate)): return str(candidate) return None def _run_first_supported(commands: list[list[str]]) -> subprocess.CompletedProcess[str]: last_result: subprocess.CompletedProcess[str] | None = None for command in commands: result = subprocess.run(command, check=False, capture_output=True, text=True) if not _is_unsupported_cli_command(result): return result last_result = result return last_result if last_result is not None else subprocess.CompletedProcess([], 1, "", "") def _is_unsupported_cli_command(result: subprocess.CompletedProcess[str]) -> bool: output = f"{result.stdout}\n{result.stderr}" return result.returncode != 0 and ( "No such command" in output or "Missing command" in output or "Got unexpected extra argument" in output )