355 lines
10 KiB
Python
355 lines
10 KiB
Python
#!/usr/bin/env python3
|
|
"""Internal helpers for semantic Markdown heading repair plans.
|
|
|
|
The semantic decision belongs to the agent using this skill. This script keeps
|
|
the file rewrite narrow: it only changes ATX heading marker depth at explicit
|
|
line numbers from a reviewed plan.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import pathlib
|
|
import re
|
|
from dataclasses import dataclass
|
|
from typing import Any
|
|
|
|
|
|
HEADING_RE = re.compile(r"^( {0,3})(#{1,6})([ \t]+)(.*)$")
|
|
FENCE_RE = re.compile(r"^( {0,3})(`{3,}|~{3,})")
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class Heading:
|
|
line: int
|
|
level: int
|
|
text: str
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class Edit:
|
|
line: int
|
|
from_level: int | None
|
|
to_level: int
|
|
reason: str
|
|
|
|
|
|
def _split_body_newline(line: str) -> tuple[str, str]:
|
|
if line.endswith("\r\n"):
|
|
return line[:-2], "\r\n"
|
|
if line.endswith("\n"):
|
|
return line[:-1], "\n"
|
|
if line.endswith("\r"):
|
|
return line[:-1], "\r"
|
|
return line, ""
|
|
|
|
|
|
def _clean_heading_text(raw: str) -> str:
|
|
text = raw.strip()
|
|
closing = re.match(r"^(.*?)(?:[ \t]+#{1,}[ \t]*)$", text)
|
|
if closing:
|
|
text = closing.group(1).rstrip()
|
|
return text
|
|
|
|
|
|
def iter_headings(lines: list[str]) -> list[Heading]:
|
|
headings: list[Heading] = []
|
|
in_fence = False
|
|
fence_char = ""
|
|
fence_len = 0
|
|
|
|
for index, line in enumerate(lines, start=1):
|
|
body, _newline = _split_body_newline(line)
|
|
fence_match = FENCE_RE.match(body)
|
|
if fence_match:
|
|
marker = fence_match.group(2)
|
|
marker_char = marker[0]
|
|
if not in_fence:
|
|
in_fence = True
|
|
fence_char = marker_char
|
|
fence_len = len(marker)
|
|
elif marker_char == fence_char and len(marker) >= fence_len:
|
|
in_fence = False
|
|
fence_char = ""
|
|
fence_len = 0
|
|
continue
|
|
|
|
if in_fence:
|
|
continue
|
|
|
|
heading_match = HEADING_RE.match(body)
|
|
if not heading_match:
|
|
continue
|
|
|
|
hashes = heading_match.group(2)
|
|
text = _clean_heading_text(heading_match.group(4))
|
|
headings.append(Heading(line=index, level=len(hashes), text=text))
|
|
|
|
return headings
|
|
|
|
|
|
def inspect_payload(path: pathlib.Path) -> dict[str, Any]:
|
|
text = path.read_text(encoding="utf-8")
|
|
lines = text.splitlines(keepends=True)
|
|
return {
|
|
"source": str(path),
|
|
"heading_count": len(iter_headings(lines)),
|
|
"headings": [
|
|
{"line": item.line, "level": item.level, "text": item.text}
|
|
for item in iter_headings(lines)
|
|
],
|
|
}
|
|
|
|
|
|
def render_inspection(payload: dict[str, Any], output_format: str) -> str:
|
|
if output_format == "json":
|
|
return json.dumps(payload, ensure_ascii=False, indent=2) + "\n"
|
|
if output_format == "markdown":
|
|
return format_markdown_inspection(payload)
|
|
raise ValueError("output_format must be 'json' or 'markdown'")
|
|
|
|
|
|
def write_inspection(
|
|
source: pathlib.Path,
|
|
output: pathlib.Path,
|
|
output_format: str = "markdown",
|
|
) -> None:
|
|
output.parent.mkdir(parents=True, exist_ok=True)
|
|
output.write_text(
|
|
render_inspection(inspect_payload(source), output_format),
|
|
encoding="utf-8",
|
|
)
|
|
|
|
|
|
def format_markdown_inspection(payload: dict[str, Any]) -> str:
|
|
output = [
|
|
"# Markdown Heading Inspection",
|
|
"",
|
|
f"- Source: `{payload['source']}`",
|
|
f"- Heading count: {payload['heading_count']}",
|
|
"",
|
|
"| Line | Level | Text |",
|
|
"| --- | --- | --- |",
|
|
]
|
|
for heading in payload["headings"]:
|
|
output.append(
|
|
f"| {heading['line']} | {heading['level']} | {heading['text']} |"
|
|
)
|
|
output.append("")
|
|
return "\n".join(output)
|
|
|
|
|
|
def _unique_stem(path: pathlib.Path, seen: dict[str, int]) -> str:
|
|
base = path.stem
|
|
count = seen.get(base, 0) + 1
|
|
seen[base] = count
|
|
if count == 1:
|
|
return base
|
|
return f"{base}-{count}"
|
|
|
|
|
|
def _batch_item_paths(
|
|
source: pathlib.Path,
|
|
mode: str,
|
|
output_dir: pathlib.Path,
|
|
stem: str,
|
|
) -> dict[str, str]:
|
|
return {
|
|
"source": str(source),
|
|
"mode": mode,
|
|
"heading_map": str(output_dir / f"{stem}.heading-map.json"),
|
|
"plan": str(output_dir / f"{stem}.heading-plan.json"),
|
|
"fixed": str(output_dir / f"{stem}.fixed.md"),
|
|
"report": str(output_dir / f"{stem}.heading-report.md"),
|
|
}
|
|
|
|
|
|
def prepare_batch_scaffold(
|
|
files: list[pathlib.Path],
|
|
mode: str,
|
|
output_dir: pathlib.Path,
|
|
) -> list[dict[str, str]]:
|
|
if mode not in {"discussion", "artifact"}:
|
|
raise ValueError("mode must be 'discussion' or 'artifact'")
|
|
if not files:
|
|
raise ValueError("files must contain at least one Markdown path")
|
|
|
|
output_dir.mkdir(parents=True, exist_ok=True)
|
|
seen: dict[str, int] = {}
|
|
items: list[dict[str, str]] = []
|
|
|
|
for source in files:
|
|
source = pathlib.Path(source)
|
|
if not source.exists():
|
|
raise ValueError(f"source file does not exist: {source}")
|
|
stem = _unique_stem(source, seen)
|
|
item = _batch_item_paths(source, mode, output_dir, stem)
|
|
write_inspection(source, pathlib.Path(item["heading_map"]), "json")
|
|
pathlib.Path(item["plan"]).write_text(
|
|
json.dumps(
|
|
{
|
|
"mode": mode,
|
|
"source": item["source"],
|
|
"output": item["fixed"],
|
|
"report": item["report"],
|
|
"edits": [],
|
|
},
|
|
ensure_ascii=False,
|
|
indent=2,
|
|
)
|
|
+ "\n",
|
|
encoding="utf-8",
|
|
)
|
|
items.append(item)
|
|
|
|
write_batch_report(items, output_dir / "fix-title-batch-report.md")
|
|
return items
|
|
|
|
|
|
def write_batch_report(items: list[dict[str, str]], output: pathlib.Path) -> None:
|
|
lines = [
|
|
"# Fix Title Batch Report",
|
|
"",
|
|
"| Source | Mode | Heading Map | Plan | Fixed | Report |",
|
|
"| --- | --- | --- | --- | --- | --- |",
|
|
]
|
|
for item in items:
|
|
lines.append(
|
|
"| {source} | {mode} | {heading_map} | {plan} | {fixed} | {report} |".format(
|
|
**item
|
|
)
|
|
)
|
|
lines.append("")
|
|
output.write_text("\n".join(lines), encoding="utf-8")
|
|
|
|
|
|
def parse_plan(plan_path: pathlib.Path) -> tuple[str, list[Edit]]:
|
|
try:
|
|
payload = json.loads(plan_path.read_text(encoding="utf-8"))
|
|
except json.JSONDecodeError as exc:
|
|
raise ValueError(f"invalid JSON plan {plan_path}: {exc}") from exc
|
|
|
|
if not isinstance(payload, dict):
|
|
raise ValueError("plan must be a JSON object")
|
|
|
|
mode = str(payload.get("mode", "unspecified"))
|
|
raw_edits = payload.get("edits")
|
|
if not isinstance(raw_edits, list):
|
|
raise ValueError("plan must contain an edits list")
|
|
|
|
edits: list[Edit] = []
|
|
seen_lines: set[int] = set()
|
|
for index, raw_edit in enumerate(raw_edits, start=1):
|
|
if not isinstance(raw_edit, dict):
|
|
raise ValueError(f"edit {index} must be an object")
|
|
line = raw_edit.get("line")
|
|
from_level = raw_edit.get("from_level")
|
|
to_level = raw_edit.get("to_level")
|
|
reason = raw_edit.get("reason", "")
|
|
|
|
if not isinstance(line, int) or line <= 0:
|
|
raise ValueError(f"edit {index} has invalid line")
|
|
if line in seen_lines:
|
|
raise ValueError(f"duplicate edit for line {line}")
|
|
seen_lines.add(line)
|
|
|
|
if from_level is not None and (
|
|
not isinstance(from_level, int) or not 1 <= from_level <= 6
|
|
):
|
|
raise ValueError(f"edit {index} has invalid from_level")
|
|
if not isinstance(to_level, int) or not 1 <= to_level <= 6:
|
|
raise ValueError(f"edit {index} has invalid to_level")
|
|
if not isinstance(reason, str):
|
|
raise ValueError(f"edit {index} has invalid reason")
|
|
|
|
edits.append(
|
|
Edit(
|
|
line=line,
|
|
from_level=from_level,
|
|
to_level=to_level,
|
|
reason=reason.strip(),
|
|
)
|
|
)
|
|
|
|
return mode, edits
|
|
|
|
|
|
def apply_plan(
|
|
source: pathlib.Path,
|
|
plan: pathlib.Path,
|
|
output: pathlib.Path,
|
|
report: pathlib.Path | None,
|
|
) -> int:
|
|
lines = source.read_text(encoding="utf-8").splitlines(keepends=True)
|
|
mode, edits = parse_plan(plan)
|
|
headings_by_line = {heading.line: heading for heading in iter_headings(lines)}
|
|
changed: list[dict[str, Any]] = []
|
|
|
|
for edit in edits:
|
|
heading = headings_by_line.get(edit.line)
|
|
if heading is None:
|
|
raise ValueError(f"line {edit.line} is not an editable ATX heading")
|
|
if edit.from_level is not None and heading.level != edit.from_level:
|
|
raise ValueError(
|
|
f"line {edit.line} expected level {edit.from_level}, "
|
|
f"found level {heading.level}"
|
|
)
|
|
|
|
body, newline = _split_body_newline(lines[edit.line - 1])
|
|
heading_match = HEADING_RE.match(body)
|
|
if heading_match is None:
|
|
raise ValueError(f"line {edit.line} is not an editable ATX heading")
|
|
|
|
indent, _hashes, spacing, rest = heading_match.groups()
|
|
lines[edit.line - 1] = f"{indent}{'#' * edit.to_level}{spacing}{rest}{newline}"
|
|
changed.append(
|
|
{
|
|
"line": edit.line,
|
|
"from_level": heading.level,
|
|
"to_level": edit.to_level,
|
|
"text": heading.text,
|
|
"reason": edit.reason,
|
|
}
|
|
)
|
|
|
|
output.parent.mkdir(parents=True, exist_ok=True)
|
|
output.write_text("".join(lines), encoding="utf-8")
|
|
|
|
if report is not None:
|
|
report.parent.mkdir(parents=True, exist_ok=True)
|
|
report.write_text(
|
|
format_report(source, output, plan, mode, changed),
|
|
encoding="utf-8",
|
|
)
|
|
|
|
return len(changed)
|
|
|
|
|
|
def format_report(
|
|
source: pathlib.Path,
|
|
output: pathlib.Path,
|
|
plan: pathlib.Path,
|
|
mode: str,
|
|
changed: list[dict[str, Any]],
|
|
) -> str:
|
|
lines = [
|
|
"# Semantic Heading Repair Report",
|
|
"",
|
|
f"- Source: `{source}`",
|
|
f"- Output: `{output}`",
|
|
f"- Plan: `{plan}`",
|
|
f"- Mode: `{mode}`",
|
|
f"- Changed headings: {len(changed)}",
|
|
"",
|
|
"| Line | From | To | Heading | Reason |",
|
|
"| --- | --- | --- | --- | --- |",
|
|
]
|
|
for item in changed:
|
|
lines.append(
|
|
"| {line} | {from_level} | {to_level} | {text} | {reason} |".format(
|
|
**item
|
|
)
|
|
)
|
|
lines.append("")
|
|
return "\n".join(lines)
|