extension for ref resolve, explode, implode, weave, tangle

This commit is contained in:
tegwick 2026-05-04 02:25:49 +02:00
parent 8203f50fd5
commit 65bfc1aebf
39 changed files with 3959 additions and 25 deletions

View file

@ -0,0 +1,25 @@
"""Reversible explode/implode operations for Markdown documents."""
from markitect_tool.explode.engine import (
EXPLODE_MANIFEST_NAME,
ExplodeEntry,
ExplodeError,
ExplodeManifest,
ExplodeResult,
ImplodeResult,
explode_markdown_file,
implode_markdown_directory,
load_explode_manifest,
)
__all__ = [
"EXPLODE_MANIFEST_NAME",
"ExplodeEntry",
"ExplodeError",
"ExplodeManifest",
"ExplodeResult",
"ImplodeResult",
"explode_markdown_file",
"implode_markdown_directory",
"load_explode_manifest",
]

View file

@ -0,0 +1,324 @@
"""Manifest-first reversible explode/implode for Markdown files."""
from __future__ import annotations
import hashlib
import re
from dataclasses import asdict, dataclass, field
from pathlib import Path
from typing import Any
import yaml
from markitect_tool.core import Heading, parse_markdown
EXPLODE_MANIFEST_NAME = "markitect-explode.yaml"
class ExplodeError(ValueError):
"""Raised when explode or implode cannot preserve a safe roundtrip."""
@dataclass(frozen=True)
class ExplodeEntry:
"""One file entry in an exploded Markdown directory."""
kind: str
file: str
order: int
unit_id: str
line_start: int
line_end: int
heading_level: int | None = None
heading_text: str | None = None
content_hash: str = ""
def to_dict(self) -> dict[str, Any]:
return {key: value for key, value in asdict(self).items() if value is not None}
@dataclass(frozen=True)
class ExplodeManifest:
"""Manifest used to implode an exploded Markdown directory."""
version: int
source_path: str
source_hash: str
variant: str
frontmatter_raw: str = ""
entries: list[ExplodeEntry] = field(default_factory=list)
def to_dict(self) -> dict[str, Any]:
return {
"version": self.version,
"source_path": self.source_path,
"source_hash": self.source_hash,
"variant": self.variant,
"frontmatter_raw": self.frontmatter_raw,
"entries": [entry.to_dict() for entry in self.entries],
}
@dataclass(frozen=True)
class ExplodeResult:
"""Result of exploding a Markdown file into a directory."""
manifest_path: str
output_dir: str
manifest: ExplodeManifest
written_files: list[str]
def to_dict(self) -> dict[str, Any]:
return {
"manifest_path": self.manifest_path,
"output_dir": self.output_dir,
"manifest": self.manifest.to_dict(),
"written_files": self.written_files,
}
@dataclass(frozen=True)
class ImplodeResult:
"""Result of rebuilding Markdown from an explode manifest."""
markdown: str
manifest_path: str
source_hash: str
current_hash: str
entries: list[str]
def to_dict(self) -> dict[str, Any]:
return asdict(self)
def explode_markdown_file(
path: str | Path,
output_dir: str | Path,
*,
variant: str = "flat",
overwrite: bool = False,
) -> ExplodeResult:
"""Explode a Markdown file into section files plus a roundtrip manifest."""
if variant not in {"flat", "hierarchical"}:
raise ExplodeError("Explode variant must be `flat` or `hierarchical`")
source_path = Path(path)
target_dir = Path(output_dir)
markdown = source_path.read_text(encoding="utf-8")
if target_dir.exists() and any(target_dir.iterdir()) and not overwrite:
raise ExplodeError(f"Output directory is not empty: {target_dir}")
target_dir.mkdir(parents=True, exist_ok=True)
frontmatter_raw, body_start_line = _split_frontmatter_raw(markdown)
entries_with_text = _explode_entries(markdown, body_start_line, variant)
written_files: list[str] = []
entries: list[ExplodeEntry] = []
for entry, text in entries_with_text:
entry_path = _safe_entry_path(target_dir, entry.file)
entry_path.parent.mkdir(parents=True, exist_ok=True)
entry_path.write_text(text, encoding="utf-8")
written_files.append(str(entry_path))
entries.append(entry)
manifest = ExplodeManifest(
version=1,
source_path=str(source_path),
source_hash=_hash_text(markdown),
variant=variant,
frontmatter_raw=frontmatter_raw,
entries=entries,
)
manifest_path = target_dir / EXPLODE_MANIFEST_NAME
manifest_path.write_text(yaml.safe_dump(manifest.to_dict(), sort_keys=False), encoding="utf-8")
return ExplodeResult(
manifest_path=str(manifest_path),
output_dir=str(target_dir),
manifest=manifest,
written_files=written_files + [str(manifest_path)],
)
def implode_markdown_directory(
directory: str | Path,
*,
manifest_path: str | Path | None = None,
) -> ImplodeResult:
"""Implode a Markdown directory created by :func:`explode_markdown_file`."""
root = Path(directory)
manifest_file = Path(manifest_path) if manifest_path else root / EXPLODE_MANIFEST_NAME
manifest = load_explode_manifest(manifest_file)
parts = [manifest.frontmatter_raw]
entry_files: list[str] = []
for entry in manifest.entries:
entry_path = _safe_entry_path(root, entry.file)
if not entry_path.exists() or not entry_path.is_file():
raise ExplodeError(f"Exploded entry file not found: {entry.file}")
parts.append(entry_path.read_text(encoding="utf-8"))
entry_files.append(str(entry_path))
markdown = "".join(parts)
return ImplodeResult(
markdown=markdown,
manifest_path=str(manifest_file),
source_hash=manifest.source_hash,
current_hash=_hash_text(markdown),
entries=entry_files,
)
def load_explode_manifest(path: str | Path) -> ExplodeManifest:
"""Load an explode manifest from YAML."""
manifest_path = Path(path)
data = yaml.safe_load(manifest_path.read_text(encoding="utf-8"))
if not isinstance(data, dict):
raise ExplodeError("Explode manifest must be a mapping")
entries = data.get("entries", [])
if not isinstance(entries, list):
raise ExplodeError("Explode manifest entries must be a list")
return ExplodeManifest(
version=int(data.get("version", 1)),
source_path=str(data.get("source_path", "")),
source_hash=str(data.get("source_hash", "")),
variant=str(data.get("variant", "flat")),
frontmatter_raw=str(data.get("frontmatter_raw", "")),
entries=[_entry_from_mapping(entry) for entry in entries],
)
def _explode_entries(
markdown: str,
body_start_line: int,
variant: str,
) -> list[tuple[ExplodeEntry, str]]:
lines = markdown.splitlines(keepends=True)
headings = parse_markdown(markdown).headings
entries: list[tuple[ExplodeEntry, str]] = []
used_ids: dict[str, int] = {}
order = 0
first_heading_line = headings[0].line if headings else len(lines) + 1
preamble_text = "".join(lines[body_start_line - 1:first_heading_line - 1])
if preamble_text or not headings:
entry = ExplodeEntry(
kind="preamble",
file="00-preamble.md",
order=order,
unit_id="preamble",
line_start=body_start_line,
line_end=max(first_heading_line - 1, body_start_line),
content_hash=_hash_text(preamble_text),
)
entries.append((entry, preamble_text))
order += 1
hierarchy: dict[int, str] = {}
for index, heading in enumerate(headings):
start = heading.line
end = headings[index + 1].line - 1 if index + 1 < len(headings) else len(lines)
text = "".join(lines[start - 1:end])
unit_id = _dedupe_id(_slug(_heading_title(heading)), used_ids)
file_path = _entry_file_for_heading(heading, index + 1, unit_id, variant, hierarchy)
entry = ExplodeEntry(
kind="section",
file=file_path,
order=order,
unit_id=unit_id,
line_start=start,
line_end=end,
heading_level=heading.level,
heading_text=heading.text,
content_hash=_hash_text(text),
)
entries.append((entry, text))
order += 1
return entries
def _entry_file_for_heading(
heading: Heading,
index: int,
unit_id: str,
variant: str,
hierarchy: dict[int, str],
) -> str:
filename = f"{index:02d}-{unit_id}.md"
if variant == "flat":
return f"sections/{filename}"
for level in list(hierarchy):
if level >= heading.level:
del hierarchy[level]
parents = [hierarchy[level] for level in sorted(hierarchy) if level < heading.level]
hierarchy[heading.level] = f"{index:02d}-{unit_id}"
return str(Path(*parents, filename)) if parents else filename
def _entry_from_mapping(data: Any) -> ExplodeEntry:
if not isinstance(data, dict):
raise ExplodeError("Explode manifest entry must be a mapping")
return ExplodeEntry(
kind=str(data["kind"]),
file=str(data["file"]),
order=int(data["order"]),
unit_id=str(data["unit_id"]),
line_start=int(data["line_start"]),
line_end=int(data["line_end"]),
heading_level=int(data["heading_level"]) if data.get("heading_level") is not None else None,
heading_text=str(data["heading_text"]) if data.get("heading_text") is not None else None,
content_hash=str(data.get("content_hash", "")),
)
def _safe_entry_path(root: Path, relative_path: str) -> Path:
path = Path(relative_path)
if path.is_absolute():
raise ExplodeError(f"Exploded entry path must be relative: {relative_path}")
resolved = (root / path).resolve()
try:
resolved.relative_to(root.resolve())
except ValueError as exc:
raise ExplodeError(f"Exploded entry path escapes directory: {relative_path}") from exc
return resolved
def _split_frontmatter_raw(markdown: str) -> tuple[str, int]:
if not markdown.startswith("---\n"):
return "", 1
end = markdown.find("\n---", 4)
if end == -1:
return "", 1
closing_end = markdown.find("\n", end + 4)
if closing_end == -1:
closing_end = len(markdown)
else:
closing_end += 1
frontmatter_raw = markdown[:closing_end]
return frontmatter_raw, frontmatter_raw.count("\n") + 1
def _heading_title(heading: Heading) -> str:
text = re.sub(r"\s+\{#[A-Za-z0-9_.:-]+\}\s*$", "", heading.text.strip())
return text or "section"
def _dedupe_id(unit_id: str, used_ids: dict[str, int]) -> str:
count = used_ids.get(unit_id, 0) + 1
used_ids[unit_id] = count
return unit_id if count == 1 else f"{unit_id}-{count}"
def _slug(value: str) -> str:
slug = re.sub(r"[^a-z0-9_.:-]+", "-", value.strip().lower())
slug = re.sub(r"-+", "-", slug).strip("-")
return slug or "section"
def _hash_text(text: str) -> str:
return "sha256:" + hashlib.sha256(text.encode("utf-8")).hexdigest()