210 lines
8.0 KiB
Python
210 lines
8.0 KiB
Python
"""Independent semantic-preservation checks for rewritten Skill Markdown."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from collections import Counter
|
|
import re
|
|
|
|
from markdown_it import MarkdownIt
|
|
|
|
from .annotator import annotation_semantically_valid
|
|
from .document import DocumentError, FENCE_RE, INLINE_PROTECTED_RE, parse_document
|
|
from .models import GuardResult, Operation
|
|
|
|
|
|
CODE_FENCE_RE = re.compile(r"^[ \t]*(```+|~~~+).*?^[ \t]*\1[ \t]*$", re.M | re.S)
|
|
HEADING_LINE_RE = re.compile(r"^[ \t]*#{1,6}[ \t]+.*$", re.M)
|
|
MARKDOWN_MARKER_RE = re.compile(
|
|
r"^[ \t]*(?:[-+*]|\d+[.)])[ \t]+|[*_~>#`]+", re.M
|
|
)
|
|
PROTECTED_LITERAL_RE = re.compile(
|
|
INLINE_PROTECTED_RE.pattern
|
|
+ r"|\b(?:MUST(?:\s+NOT)?|不得|必须|禁止|仅可|只能|不能)\b",
|
|
re.I,
|
|
)
|
|
|
|
|
|
def _code_blocks(body: str) -> list[str]:
|
|
return CODE_FENCE_RE.findall(body)
|
|
|
|
|
|
def _full_code_blocks(body: str) -> list[str]:
|
|
blocks: list[str] = []
|
|
lines = body.splitlines(keepends=True)
|
|
index = 0
|
|
while index < len(lines):
|
|
fence = FENCE_RE.match(lines[index])
|
|
if not fence:
|
|
index += 1
|
|
continue
|
|
marker_char = fence.group(1)[0]
|
|
start = index
|
|
index += 1
|
|
while index < len(lines) and not re.match(
|
|
rf"^[ \t]*{re.escape(marker_char)}{{3,}}", lines[index]
|
|
):
|
|
index += 1
|
|
index = min(index + 1, len(lines))
|
|
blocks.append("".join(lines[start:index]))
|
|
return blocks
|
|
|
|
|
|
def _payload_counter(body: str) -> Counter[str]:
|
|
without_code = body
|
|
for code in _full_code_blocks(body):
|
|
without_code = without_code.replace(code, "", 1)
|
|
without_headings = HEADING_LINE_RE.sub("", without_code)
|
|
without_markers = MARKDOWN_MARKER_RE.sub("", without_headings)
|
|
return Counter(char.lower() for char in without_markers if char.isalnum())
|
|
|
|
|
|
def _adjust_expected_payload(payload: Counter[str], operations: list[Operation]) -> Counter[str]:
|
|
adjusted = payload.copy()
|
|
for operation in operations:
|
|
if operation.type == "DUPLICATE_EXACT" and operation.quote is not None:
|
|
adjusted.update(_payload_counter(operation.quote))
|
|
elif (
|
|
operation.type == "REPLACE_COREFERENCE_WITH_SOURCE_QUOTE"
|
|
and operation.quote is not None
|
|
and operation.replacement is not None
|
|
):
|
|
adjusted.subtract(
|
|
char.lower() for char in operation.quote if char.isalnum()
|
|
)
|
|
adjusted.update(
|
|
char.lower() for char in operation.replacement if char.isalnum()
|
|
)
|
|
elif operation.type == "SEMANTIC_REWRITE_BLOCK" and operation.quote is not None:
|
|
adjusted.subtract(_payload_counter(operation.quote))
|
|
adjusted.update(_payload_counter(operation.replacement or ""))
|
|
elif operation.type == "ADD_GROUNDED_SUMMARY":
|
|
adjusted.update(_payload_counter(operation.replacement or ""))
|
|
return +adjusted
|
|
|
|
|
|
def _protected_literals(body: str) -> Counter[str]:
|
|
return Counter(match.group(0) for match in PROTECTED_LITERAL_RE.finditer(body))
|
|
|
|
|
|
def _adjust_expected_protected(
|
|
literals: Counter[str], operations: list[Operation]
|
|
) -> Counter[str]:
|
|
adjusted = literals.copy()
|
|
for operation in operations:
|
|
if operation.type == "DUPLICATE_EXACT" and operation.quote is not None:
|
|
adjusted.update(
|
|
_protected_literals(operation.replacement or operation.quote)
|
|
)
|
|
elif (
|
|
operation.type == "REPLACE_COREFERENCE_WITH_SOURCE_QUOTE"
|
|
and operation.quote is not None
|
|
and operation.replacement is not None
|
|
):
|
|
adjusted.subtract(_protected_literals(operation.quote))
|
|
adjusted.update(_protected_literals(operation.replacement))
|
|
elif operation.type == "SEMANTIC_REWRITE_BLOCK" and operation.quote is not None:
|
|
adjusted.subtract(_protected_literals(operation.quote))
|
|
adjusted.update(_protected_literals(operation.replacement or ""))
|
|
elif operation.type == "ADD_GROUNDED_SUMMARY":
|
|
adjusted.update(_protected_literals(operation.replacement or ""))
|
|
elif (
|
|
operation.type == "FORMAT_HEADING_LABEL"
|
|
and operation.quote is not None
|
|
and operation.replacement is not None
|
|
):
|
|
# Heading-format passes may change casing or punctuation (for example,
|
|
# ``Must Follow`` to ``MUST FOLLOW:``). The operation records the
|
|
# exact source and replacement, so account for that deliberate,
|
|
# surface-only change rather than treating it as an untracked loss of
|
|
# a protected literal.
|
|
adjusted.subtract(_protected_literals(operation.quote))
|
|
adjusted.update(_protected_literals(operation.replacement))
|
|
return +adjusted
|
|
|
|
|
|
def run_semantic_guard(
|
|
original: str,
|
|
rewritten: str,
|
|
operations: list[Operation],
|
|
) -> GuardResult:
|
|
checks: dict[str, bool] = {}
|
|
failures: list[str] = []
|
|
try:
|
|
source = parse_document(original)
|
|
target = parse_document(rewritten)
|
|
checks["markdown_parseable"] = True
|
|
except DocumentError as exc:
|
|
return GuardResult(False, {"markdown_parseable": False}, [str(exc)])
|
|
try:
|
|
MarkdownIt("commonmark", {"html": True}).parse(target.body)
|
|
checks["commonmark_parseable"] = True
|
|
except Exception as exc: # pragma: no cover - markdown-it is intentionally permissive
|
|
checks["commonmark_parseable"] = False
|
|
failures.append(f"CommonMark parse failed: {exc}")
|
|
|
|
checks["frontmatter_exact"] = source.frontmatter == target.frontmatter
|
|
checks["code_blocks_exact"] = [
|
|
block.text for block in source.blocks if block.kind == "code"
|
|
] == [block.text for block in target.blocks if block.kind == "code"]
|
|
checks["protected_literals_preserved"] = _adjust_expected_protected(
|
|
_protected_literals(source.body), operations
|
|
) == _protected_literals(target.body)
|
|
expected_payload = _adjust_expected_payload(
|
|
_payload_counter(source.body), operations
|
|
)
|
|
checks["body_payload_preserved"] = expected_payload == _payload_counter(target.body)
|
|
duplicated_payloads = [
|
|
operation.replacement or operation.quote
|
|
for operation in operations
|
|
if operation.type == "DUPLICATE_EXACT" and operation.quote
|
|
]
|
|
checks["duplicated_spans_present"] = all(
|
|
payload is not None and payload in target.body
|
|
for payload in duplicated_payloads
|
|
)
|
|
semantic_operations = [
|
|
operation
|
|
for operation in operations
|
|
if operation.annotation_type is not None
|
|
and operation.block_id is not None
|
|
and operation.quote is not None
|
|
]
|
|
checks["operation_annotation_types_valid"] = all(
|
|
operation.block_id in source.block_index
|
|
and annotation_semantically_valid(
|
|
operation.annotation_type,
|
|
source.block_index[operation.block_id],
|
|
operation.quote,
|
|
)
|
|
for operation in semantic_operations
|
|
)
|
|
checks["emphasis_not_nested"] = all(
|
|
operation.type != "EMPHASIZE_IN_PLACE"
|
|
or operation.quote is None
|
|
or not re.search(r"\*\*|__", operation.quote)
|
|
for operation in operations
|
|
)
|
|
checks["semantic_rewrites_source_backed"] = all(
|
|
(
|
|
operation.type not in {"SEMANTIC_REWRITE_BLOCK", "ADD_GROUNDED_SUMMARY"}
|
|
or (
|
|
bool(operation.replacement)
|
|
and bool(operation.source_quotes)
|
|
and all(quote in source.body for quote in operation.source_quotes)
|
|
and (
|
|
operation.type != "SEMANTIC_REWRITE_BLOCK"
|
|
or (
|
|
operation.block_id in source.block_index
|
|
and operation.quote
|
|
== source.block_index[operation.block_id].text
|
|
)
|
|
)
|
|
)
|
|
)
|
|
for operation in operations
|
|
)
|
|
for name, passed in checks.items():
|
|
if not passed:
|
|
failures.append(name)
|
|
return GuardResult(not failures, checks, failures)
|