97-documentation/ns_report_publisher/processing.py
from __future__ import annotations
import os
import re
from pathlib import Path
from .annotations import annotation_matches
from .model import (
ArtifactReference,
ArtifactTarget,
ProcessedDocument,
ProcessingAnnotation,
SourceDocument,
ValidationFinding,
)
from .knowledge import knowledge_presentation
from .resolver import (
artifact_reference_matches,
build_artifact_targets,
normalize_artifact_reference,
select_artifact_targets,
)
from .documentation_pipeline import mask_markdown_code
PROCESSED_ARTIFACT_REF_PREFIX = "\x00NS_ARTIFACT_REF_"
PROCESSED_ARTIFACT_REF_SUFFIX = "\x00"
PROCESSED_ANNOTATION_POINT_PREFIX = "\x00NS_ANNOTATION_POINT_"
PROCESSED_ANNOTATION_OPEN_PREFIX = "\x00NS_ANNOTATION_OPEN_"
PROCESSED_ANNOTATION_CLOSE_PREFIX = "\x00NS_ANNOTATION_CLOSE_"
PROCESSED_ANNOTATION_SUFFIX = "\x00"
MARKDOWN_LINK_RE = re.compile(r"(?<!!)\[([^\]]+)\]\(([^)\s]+)(?:\s+\"[^\"]*\")?\)")
HISTORICAL_WIKI_REF_RE = re.compile(r"\[\[([^\]|\n]+)(?:\|([^\]]+))?\]\]")
GLOSSARY_AZ_MARKER = "{{GLOSSARY-AZ}}"
def _repository_markdown_target(target: str) -> tuple[str, str] | None:
"""Return repository and normalized source path for a qualified report link.
The migration contains both source-shaped ``.md`` targets and already
materialized ``.html`` targets. Both denote the same source document for
semantic resolution; HTML is never passed through as an ordinary link.
"""
target = target.split("#", 1)[0].replace("\\", "/")
if "://" in target or target.startswith(("mailto:", "#")):
return None
repository, separator, path = target.partition(":")
if not separator or not repository or "/" in repository:
return None
path = path.lstrip("/")
lower = path.lower()
if lower.endswith(".html"):
if lower.endswith("/index.html"):
return repository, path[:-len("index.html")] + "README.md"
return repository, path[:-len(".html")] + ".md"
if lower.endswith((".md", ".markdown")):
return repository, path
return None
def _repository_markdown_path(target: str) -> str | None:
"""Compatibility helper returning only the normalized source path."""
resolved = _repository_markdown_target(target)
return resolved[1] if resolved is not None else None
def _reference_from_repository_link(
label: str,
target: str,
artifact_targets: dict[str, list[ArtifactTarget]],
) -> str | None:
resolved_target = _repository_markdown_target(target)
if resolved_target is None:
return None
repository, source_path = resolved_target
# Alias IDs (for example ``Promotion`` and ``ART-DEF-0049``) may point to
# the same document. They are one target, not an ambiguity. Keep the
# document identity as the deduplication key and honour the repository
# qualifier so equal paths in different repositories remain distinct.
matches: list[tuple[bool, str, str]] = []
for identifier, candidates in artifact_targets.items():
# Select after narrowing by the repository-qualified source path.
# A migrated twin at another path must not make the explicitly named
# document ambiguous.
path_candidates = [
candidate for candidate in candidates
if candidate.document.metadata.get("source_path", candidate.document.relative_path.as_posix()) == source_path
]
selected = select_artifact_targets(path_candidates)
if len(selected) != 1:
continue
document = selected[0].document
canonical = document.metadata.get("id", "").strip() or identifier
document_repository = document.metadata.get("repository_id", "").strip()
matches.append((document_repository == repository, document.relative_path.as_posix(), canonical))
# Prefer the explicitly named repository. A local/fallback document is
# retained only for standalone publications where the target repository is
# deliberately represented in the same fixture or source tree.
if any(is_target_repository for is_target_repository, _, _ in matches):
matches = [item for item in matches if item[0]]
by_document: dict[str, str] = {}
for _, document_path, canonical in matches:
by_document.setdefault(document_path, canonical)
unique = sorted(set(by_document.values()))
if len(unique) != 1:
return None
return f"[[{unique[0]}|{label.strip()}]]"
def _normalize_repository_markdown_links(
markdown: str,
artifact_targets: dict[str, list[ArtifactTarget]],
) -> str:
def replace(match: re.Match[str]) -> str:
resolved = _reference_from_repository_link(match.group(1), match.group(2), artifact_targets)
return resolved if resolved is not None else match.group(0)
return MARKDOWN_LINK_RE.sub(replace, markdown)
def _normalize_historical_wiki_references(markdown: str, document: SourceDocument) -> str:
"""Keep explicitly classified legacy identities readable, not unresolved."""
if document.metadata.get("reference_mode", "").strip() != "historical-evidence":
return markdown
def replace(match: re.Match[str]) -> str:
identifier = match.group(1).strip()
label = (match.group(2) or identifier).strip()
return f"[{label}](historical:{identifier})"
return HISTORICAL_WIKI_REF_RE.sub(replace, markdown)
def _knowledge_terms(
document: SourceDocument,
artifact_targets: dict[str, list[ArtifactTarget]],
) -> list[tuple[str, str]]:
"""Return unambiguous glossary/nugget titles, longest term first.
A knowledge document never annotates itself. The visible source stays
plain text; only the first matching occurrence in another document is
converted into the existing artifact-reference representation.
"""
if knowledge_presentation(document) is not None:
return []
terms: dict[str, str] = {}
ambiguous: set[str] = set()
for identifier, candidates in artifact_targets.items():
selected = select_artifact_targets(candidates)
if len(selected) != 1 or knowledge_presentation(selected[0].document) is None:
continue
canonical = selected[0].document.metadata.get("id", "").strip() or identifier
metadata = selected[0].document.metadata
aliases = [selected[0].document.title.strip(), metadata.get("primary_alias", "").strip()]
synonyms = metadata.get("synonyms", "")
if isinstance(synonyms, list):
aliases.extend(str(item).strip() for item in synonyms)
else:
aliases.extend(item.strip(" []\"'") for item in str(synonyms).split(","))
for title in aliases:
if not title:
continue
key = title.casefold()
existing = terms.get(key)
if existing and existing != canonical:
ambiguous.add(key)
else:
terms[key] = canonical
return sorted(
((title, identifier) for title, identifier in terms.items() if title not in ambiguous),
key=lambda item: (-len(item[0]), item[0]),
)
def _normalize_plain_knowledge_terms(
markdown: str,
document: SourceDocument,
artifact_targets: dict[str, list[ArtifactTarget]],
) -> str:
"""Convert only each document's first plain, unambiguous knowledge term.
Markdown headings, links, inline code and fenced code are deliberately
excluded. This preserves source wording and prevents false annotations
in examples or navigation labels.
"""
terms = _knowledge_terms(document, artifact_targets)
if not terms:
return markdown
seen: set[str] = set()
in_fence = False
output: list[str] = []
for line in markdown.splitlines(keepends=True):
if re.match(r"^\s*```", line):
in_fence = not in_fence
output.append(line)
continue
if in_fence or line.lstrip().startswith("#"):
output.append(line)
continue
protected: dict[str, str] = {}
def protect(match: re.Match[str]) -> str:
key = f"\x00NS_PLAIN_PROTECTED_{len(protected)}\x00"
protected[key] = match.group(0)
return key
# Existing explicit wiki references are already deliberate semantic
# markup. Protect the *whole* construct before looking for plain
# knowledge terms; otherwise the visible label is converted a second
# time, creating ``[[ID|[[ID|Label]]]]``. The resolver can only
# consume the inner construct and leaks the outer brackets to HTML.
value = re.sub(
r"!?\[[^\]]*\]\([^)]+\)|(?<!!)\[\[[^\[\]\n]+\]\]",
protect,
line,
)
for title, identifier in terms:
if identifier in seen:
continue
pattern = re.compile(rf"(?<![\w-])({re.escape(title)})(?![\w-])", re.IGNORECASE)
value, count = pattern.subn(lambda match: f"[[{identifier}|{match.group(1)}]]", value, count=1)
if count:
seen.add(identifier)
# A recognized knowledge term may have been written in inline code
# merely to mark a canonical technical spelling (for example
# `supersedes`). It must become one atomic interactive reference,
# not literal wiki syntax inside a <code> element.
value = re.sub(r"`(\[\[[^\[\]\n]+\]\])`", r"\1", value)
for key, original in protected.items():
value = value.replace(key, original)
output.append(value)
return "".join(output)
def _expand_glossary_az_index(
markdown: str,
artifact_targets: dict[str, list[ArtifactTarget]],
) -> str:
"""Materialize one complete, deterministic glossary index at a marker."""
if GLOSSARY_AZ_MARKER not in markdown:
return markdown
entries: dict[Path, tuple[str, str]] = {}
for candidates in artifact_targets.values():
selected = select_artifact_targets(candidates)
if len(selected) != 1:
continue
document = selected[0].document
identifier = document.metadata.get("id", "").strip()
artifact_type = document.metadata.get("artifact_type", "").strip().casefold()
if (
not identifier
or artifact_type not in {"glossary term", "glossary entry", "glossaryentry"}
or document.relative_path.parent.name != "entries"
):
continue
entries[document.relative_path] = (identifier, document.title.strip() or identifier)
grouped: dict[str, list[tuple[str, str]]] = {}
for identifier, title in sorted(entries.values(), key=lambda item: (item[1].casefold(), item[0])):
grouped.setdefault(title[:1].upper() if title else "#", []).append((identifier, title))
lines: list[str] = []
for letter in sorted(grouped, key=str.casefold):
lines.extend((f"### {letter}", ""))
lines.extend(f"- [[{identifier}|{title}]]" for identifier, title in grouped[letter])
lines.append("")
return markdown.replace(GLOSSARY_AZ_MARKER, "\n".join(lines).rstrip())
def _relative_href(current: SourceDocument, target: SourceDocument) -> str:
return Path(os.path.relpath(target.output_path, current.output_path.parent)).as_posix()
def _resolve_reference(
current: SourceDocument,
raw_reference: str,
artifact_targets: dict[str, list[ArtifactTarget]],
) -> tuple[ArtifactReference, ValidationFinding | None]:
identifier, visible_label = normalize_artifact_reference(raw_reference)
if not identifier:
reference = ArtifactReference(raw_reference, "", visible_label, None, "unresolved")
finding = ValidationFinding(
code="artifact-reference-invalid",
message=f"Ungültige Artefakt-Referenz: [[{raw_reference}]]",
reference=raw_reference,
)
return reference, finding
candidates = select_artifact_targets(artifact_targets.get(identifier, []))
if len(candidates) == 1:
target = candidates[0].document
knowledge = knowledge_presentation(target)
return (
ArtifactReference(
raw_reference=raw_reference,
identifier=identifier,
visible_label=visible_label,
href=_relative_href(current, target),
status="resolved",
knowledge_kind=knowledge.kind if knowledge else None,
short_text=knowledge.short_text if knowledge else target.metadata.get("summary", "").strip(),
detail_text=knowledge.detail_text if knowledge else "",
topic=knowledge.topic if knowledge else "",
term_type=knowledge.term_type if knowledge else "",
presentation_dimension=target.metadata.get("presentation_dimension", "").strip(),
artifact_type=target.metadata.get("artifact_type", "").strip(),
target_title=target.title,
),
None,
)
status = "ambiguous" if candidates else "unresolved"
code = "artifact-reference-ambiguous" if candidates else "artifact-reference-unresolved"
message = (
f"Mehrdeutige Artefakt-Referenz: [[{raw_reference}]]"
if candidates
else f"Nicht aufgelöste Artefakt-Referenz: [[{raw_reference}]]"
)
return (
ArtifactReference(raw_reference, identifier, visible_label, None, status),
ValidationFinding(code=code, message=message, reference=raw_reference),
)
def process_document(
document: SourceDocument,
artifact_targets: dict[str, list[ArtifactTarget]],
) -> ProcessedDocument:
references: list[ArtifactReference] = []
annotations: list[ProcessingAnnotation] = []
findings: list[ValidationFinding] = []
replacements: list[tuple[int, int, str]] = []
normalized_markdown = _normalize_historical_wiki_references(document.markdown, document)
normalized_markdown = _expand_glossary_az_index(normalized_markdown, artifact_targets)
normalized_markdown = _normalize_repository_markdown_links(normalized_markdown, artifact_targets)
normalized_markdown = _normalize_plain_knowledge_terms(
normalized_markdown, document, artifact_targets
)
masked = mask_markdown_code(normalized_markdown)
for match in annotation_matches(normalized_markdown, masked):
index = len(annotations)
annotations.append(ProcessingAnnotation(match.kind, match.content, match.block))
if match.block:
assert match.close_start is not None and match.close_end is not None
replacements.append(
(
match.open_start,
match.open_end,
f"{PROCESSED_ANNOTATION_OPEN_PREFIX}{index}{PROCESSED_ANNOTATION_SUFFIX}",
)
)
replacements.append(
(
match.close_start,
match.close_end,
f"{PROCESSED_ANNOTATION_CLOSE_PREFIX}{index}{PROCESSED_ANNOTATION_SUFFIX}",
)
)
else:
replacements.append(
(
match.start,
match.end,
f"{PROCESSED_ANNOTATION_POINT_PREFIX}{index}{PROCESSED_ANNOTATION_SUFFIX}",
)
)
for start, end, raw_reference in artifact_reference_matches(normalized_markdown):
reference, finding = _resolve_reference(document, raw_reference, artifact_targets)
index = len(references)
references.append(reference)
if finding is not None:
findings.append(finding)
replacements.append(
(start, end, f"{PROCESSED_ARTIFACT_REF_PREFIX}{index}{PROCESSED_ARTIFACT_REF_SUFFIX}")
)
processed_markdown = normalized_markdown
for start, end, replacement in sorted(replacements, reverse=True):
processed_markdown = processed_markdown[:start] + replacement + processed_markdown[end:]
return ProcessedDocument(
source=document,
markdown=processed_markdown,
artifact_references=tuple(references),
processing_annotations=tuple(annotations),
validation_findings=tuple(findings),
)
def merge_artifact_targets(
*catalogs: dict[str, list[ArtifactTarget]],
) -> dict[str, list[ArtifactTarget]]:
merged: dict[str, list[ArtifactTarget]] = {}
for catalog in catalogs:
for identifier, targets in catalog.items():
merged.setdefault(identifier, []).extend(targets)
return merged
def process_documents(
documents: list[SourceDocument],
external_artifact_targets: dict[str, list[ArtifactTarget]] | None = None,
) -> tuple[list[ProcessedDocument], dict[str, list[ArtifactTarget]]]:
local_artifact_targets = build_artifact_targets(documents)
artifact_targets = merge_artifact_targets(
external_artifact_targets or {},
local_artifact_targets,
)
return [process_document(document, artifact_targets) for document in documents], artifact_targets