Private
Public Access
feat(scripts): add scripts/aggregate_directives.py (presets -> clean body aggregation)
Aggregates v1.md bodies from a preset markdown into a single output stream. NEVER reads meta.md (pollution fix). Stdlib-only; supports stdout and -o. 5 tests in tests/test_aggregate_directives.py cover: success path, no-meta-md pollution, missing-file error, -o flag, no provenance leakage.
This commit is contained in:
@@ -0,0 +1,97 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Aggregate directive bodies from a preset markdown file."""
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import re
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
if hasattr(sys.stdout, "reconfigure"):
|
||||
try:
|
||||
sys.stdout.reconfigure(encoding="utf-8")
|
||||
except Exception:
|
||||
pass
|
||||
if hasattr(sys.stderr, "reconfigure"):
|
||||
try:
|
||||
sys.stderr.reconfigure(encoding="utf-8")
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
REPO_ROOT = Path(__file__).resolve().parent.parent
|
||||
ENTRY_RE = re.compile(r"^\s*-\s+(?P<name>[A-Za-z0-9_]+)\s*:\s*(?P<path>\S+)\s*$")
|
||||
EXPECTED_PATTERN_TEMPLATE = r"^conductor/directives/{name}/v\d+\.md$"
|
||||
|
||||
|
||||
def parse_preset(preset_path: Path) -> list[tuple[str, Path]]:
|
||||
entries: list[tuple[str, Path]] = []
|
||||
for line in preset_path.read_text(encoding="utf-8").splitlines():
|
||||
m = ENTRY_RE.match(line)
|
||||
if not m:
|
||||
continue
|
||||
name = m.group("name")
|
||||
rel_path = m.group("path")
|
||||
abs_path = (REPO_ROOT / rel_path).resolve()
|
||||
entries.append((name, abs_path))
|
||||
return entries
|
||||
|
||||
|
||||
def aggregate(preset_path: Path, output_path: Path | None) -> int:
|
||||
entries = parse_preset(preset_path)
|
||||
if not entries:
|
||||
print("ERROR: no directive entries found in " + str(preset_path), file=sys.stderr)
|
||||
return 1
|
||||
missing: list[tuple[str, Path]] = []
|
||||
bodies: list[tuple[str, str]] = []
|
||||
for name, path in entries:
|
||||
if not path.exists():
|
||||
missing.append((name, path))
|
||||
continue
|
||||
rel = path.relative_to(REPO_ROOT).as_posix()
|
||||
pattern = EXPECTED_PATTERN_TEMPLATE.format(name=re.escape(name))
|
||||
if not re.match(pattern, rel):
|
||||
print("WARNING: " + name + ": path " + str(path) + " does not match pattern conductor/directives/" + name + "/vN.md; including anyway", file=sys.stderr)
|
||||
try:
|
||||
body = path.read_text(encoding="utf-8")
|
||||
except UnicodeDecodeError as e:
|
||||
print("ERROR: " + name + ": UTF-8 decode error in " + str(path) + ": " + str(e), file=sys.stderr)
|
||||
continue
|
||||
bodies.append((name, body))
|
||||
if missing:
|
||||
for name, path in missing:
|
||||
print("ERROR: missing file for directive '" + name + "': " + str(path), file=sys.stderr)
|
||||
return 1
|
||||
out_lines: list[str] = []
|
||||
for name, body in bodies:
|
||||
out_lines.append("=" * 40)
|
||||
out_lines.append(name)
|
||||
out_lines.append("=" * 40)
|
||||
out_lines.append("")
|
||||
out_lines.append(body.rstrip(chr(10)))
|
||||
out_lines.append("")
|
||||
out_lines.append("")
|
||||
rendered = chr(10).join(out_lines)
|
||||
if output_path:
|
||||
output_path.write_text(rendered, encoding="utf-8")
|
||||
print("Wrote " + str(len(bodies)) + " directives to " + str(output_path), file=sys.stderr)
|
||||
else:
|
||||
sys.stdout.write(rendered)
|
||||
if not rendered.endswith(chr(10)):
|
||||
sys.stdout.write(chr(10))
|
||||
return 0
|
||||
|
||||
|
||||
def main() -> int:
|
||||
parser = argparse.ArgumentParser(description="Aggregate directive bodies from a preset markdown file.")
|
||||
parser.add_argument("preset", type=Path, help="Path to the preset markdown file")
|
||||
parser.add_argument("-o", "--output", type=Path, default=None, help="Write output to file instead of stdout")
|
||||
args = parser.parse_args()
|
||||
if not args.preset.exists():
|
||||
print("ERROR: preset file not found: " + str(args.preset), file=sys.stderr)
|
||||
return 1
|
||||
output = args.output.resolve() if args.output else None
|
||||
return aggregate(args.preset.resolve(), output)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main())
|
||||
Reference in New Issue
Block a user