Implement RecentlyOnScope domain digest

api/services/markitect_templates.py

@@ -0,0 +1,149 @@
+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
+    )