#!/usr/bin/env python3 """Perform dependency-free structural checks on an OKF Markdown bundle.""" from __future__ import annotations import argparse import json import re from pathlib import Path from pathlib import PurePosixPath from urllib.parse import unquote, urlsplit RESERVED = {"index.md", "log.md"} LINK_RE = re.compile(r"(? argparse.Namespace: parser = argparse.ArgumentParser(description="Lint an OKF v0.2 wiki bundle.") parser.add_argument("bundle", nargs="?", type=Path, default=Path("wiki")) return parser.parse_args() def frontmatter(text: str) -> tuple[str | None, str]: lines = text.splitlines() if not lines or lines[0].strip() != "---": return None, text for index in range(1, len(lines)): if lines[index].strip() == "---": return "\n".join(lines[1:index]), "\n".join(lines[index + 1 :]) return None, text def local_link_target(bundle: Path, source: Path, raw_target: str) -> Path | None: target = raw_target.strip().split(maxsplit=1)[0].strip("<>") parsed = urlsplit(target) if parsed.scheme or parsed.netloc or target.startswith("#"): return None path_text = unquote(parsed.path) if not path_text or not path_text.lower().endswith(".md"): return None if path_text.startswith("/"): return bundle / path_text.lstrip("/") return source.parent / path_text def lint_source_manifest( bundle: Path, errors: list[str], warnings: list[str] ) -> None: manifest_path = bundle / ".source-manifest.json" if not manifest_path.exists(): warnings.append("Root .source-manifest.json is missing") return try: manifest = json.loads(manifest_path.read_text(encoding="utf-8")) except (OSError, json.JSONDecodeError) as error: errors.append(f".source-manifest.json: invalid JSON: {error}") return if not isinstance(manifest, dict): errors.append(".source-manifest.json: root value must be an object") return if manifest.get("version") != 1: errors.append(".source-manifest.json: version must be 1") if manifest.get("hash_algorithm") != "sha256": errors.append(".source-manifest.json: hash_algorithm must be sha256") files = manifest.get("files") if not isinstance(files, dict): errors.append(".source-manifest.json: files must be an object") return version_path = Path(__file__).resolve().parent.parent / "VERSION" version = version_path.read_text(encoding="utf-8").strip() expected_processor = f"maintain-knowledge-wiki/{version}" for relative, entry in files.items(): path = PurePosixPath(relative) if not relative or path.is_absolute() or ".." in path.parts: errors.append(f".source-manifest.json: unsafe source path {relative!r}") continue if not isinstance(entry, dict): errors.append(f".source-manifest.json: entry must be an object: {relative}") continue status = entry.get("status") if status not in SOURCE_STATUSES: errors.append( f".source-manifest.json: invalid status for {relative}: {status!r}" ) digest = entry.get("sha256") if not isinstance(digest, str) or not SHA256_RE.fullmatch(digest): errors.append(f".source-manifest.json: invalid sha256 for {relative}") if not isinstance(entry.get("size"), int) or entry["size"] < 0: errors.append(f".source-manifest.json: invalid size for {relative}") if ( not isinstance(entry.get("mtime_ns"), str) or not entry["mtime_ns"].isdigit() ): errors.append(f".source-manifest.json: invalid mtime_ns for {relative}") processed_digest = entry.get("processed_sha256") if processed_digest is not None and ( not isinstance(processed_digest, str) or not SHA256_RE.fullmatch(processed_digest) ): errors.append( f".source-manifest.json: invalid processed_sha256 for {relative}" ) if status == "processed": if processed_digest != digest: errors.append( f".source-manifest.json: processed hash mismatch for {relative}" ) if not entry.get("processed_at") or not entry.get("processor_version"): errors.append( f".source-manifest.json: processed metadata missing for {relative}" ) elif entry.get("processor_version") != expected_processor: warnings.append( f"Source processed with another workflow version: {relative} " f"({entry.get('processor_version')})" ) elif status == "pending": warnings.append(f"Source pending ingestion: {relative}") elif status == "missing": warnings.append(f"Source missing: {relative}") elif status == "removed": if not entry.get("removed_at") or not entry.get("processor_version"): errors.append( f".source-manifest.json: removed metadata missing for {relative}" ) if not isinstance(entry.get("removed_reason"), str) or not entry[ "removed_reason" ].strip(): errors.append( f".source-manifest.json: removal reason missing for {relative}" ) replacements = entry.get("replaced_by", []) if not isinstance(replacements, list) or not all( isinstance(value, str) for value in replacements ): errors.append( f".source-manifest.json: replaced_by must be a string list for " f"{relative}" ) else: for replacement in replacements: replacement_path = PurePosixPath(replacement) if ( not replacement or replacement_path.is_absolute() or ".." in replacement_path.parts ): errors.append( f".source-manifest.json: unsafe replacement path " f"{replacement!r} for {relative}" ) elif replacement == relative: errors.append( f".source-manifest.json: source replaces itself: {relative}" ) elif replacement not in files: errors.append( f".source-manifest.json: unknown replacement " f"{replacement!r} for {relative}" ) elif not isinstance(files[replacement], dict): errors.append( f".source-manifest.json: invalid replacement entry " f"{replacement!r} for {relative}" ) elif files[replacement].get("status") not in { "processed", "removed", }: errors.append( f".source-manifest.json: replacement is not retained " f"or acknowledged " f"{replacement!r} for {relative}" ) if ( entry.get("processor_version") and entry.get("processor_version") != expected_processor ): warnings.append( f"Source removal processed with another workflow version: " f"{relative} ({entry.get('processor_version')})" ) def main() -> int: bundle = parse_args().bundle.expanduser().resolve() errors: list[str] = [] warnings: list[str] = [] if not bundle.is_dir(): print(f"ERROR: Bundle directory does not exist: {bundle}") return 2 markdown_files = sorted(bundle.rglob("*.md")) if not (bundle / "index.md").exists(): warnings.append("Root index.md is missing") if not (bundle / "log.md").exists(): warnings.append("Root log.md is missing") lint_source_manifest(bundle, errors, warnings) for path in markdown_files: relative = path.relative_to(bundle) text = path.read_text(encoding="utf-8") if path.name not in RESERVED: metadata, _ = frontmatter(text) if metadata is None: errors.append(f"{relative}: missing or unclosed YAML frontmatter") else: match = TYPE_RE.search(metadata) if not match or not match.group(1).strip().strip("'\""): errors.append(f"{relative}: missing or empty type") if path.name == "index.md" and path != bundle / "index.md": metadata, _ = frontmatter(text) if metadata is not None: errors.append(f"{relative}: only the root index may have frontmatter") if path.name == "log.md": for heading in re.findall(r"^## (.+?)\s*$", text, re.MULTILINE): if not re.fullmatch(r"\d{4}-\d{2}-\d{2}", heading): errors.append(f"{relative}: invalid log date heading '{heading}'") for raw_target in LINK_RE.findall(text): target = local_link_target(bundle, path, raw_target) if target is not None and not target.resolve().exists(): warnings.append(f"{relative}: broken link -> {raw_target}") for message in errors: print(f"ERROR: {message}") for message in warnings: print(f"WARN: {message}") print( f"Checked {len(markdown_files)} Markdown files: " f"{len(errors)} error(s), {len(warnings)} warning(s)." ) return 1 if errors else 0 if __name__ == "__main__": raise SystemExit(main())