ckb/.agents/skills/ckb-export-okf/scripts/export_okf.py

#!/usr/bin/env python3
# Copyright 2026 Michał Kopeć
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
"""Export wiki/ as an Open Knowledge Format (OKF) v0.1 bundle at outputs/okf/.

Usage:
    python3 export_okf.py [--check] [--out DIR]

Run from the repository root (the directory containing wiki/ and outputs/).

--check   validate only: build the bundle in a temporary directory, run the
          conformance checks, print the report, and write nothing to outputs/.
--out     override the output directory (default: outputs/okf).

The transform is fully deterministic: two runs over an unchanged wiki/ produce
byte-identical output. Every directory walk and generated list is sorted.
"""
import argparse
import re
import shutil
import sys
import tempfile
from pathlib import Path

REPO_ROOT = Path.cwd()
WIKI = REPO_ROOT / "wiki"

OKF_VERSION = "0.1"
RESERVED = {"index.md", "log.md"}

# wiki/ frontmatter keys that ride along unchanged as OKF extension fields.
PASSTHROUGH_EXT = [
    "confidence",
    "quality",
    "retention",
    "supersedes",
    "superseded_by",
    "freshness_window_days",
]
# Keys with no valid home in an OKF bundle.
DROPPED = {"kb_schema_version"}

VERB_MAP = {
    "CREATE": "Creation",
    "UPDATE": "Update",
    "DELETE": "Deprecation",
    "RESTRUCTURE": "Update",
}


# --------------------------------------------------------------------------
# frontmatter
# --------------------------------------------------------------------------


def parse_frontmatter(text):
    """Return (dict, body). Flat `key: value` YAML only — that is all the
    schema uses. Unparseable or absent frontmatter yields ({}, text)."""
    if not text.startswith("---\n"):
        return {}, text
    end = text.find("\n---\n", 4)
    if end == -1:
        return {}, text
    raw = text[4:end]
    body = text[end + 5 :]
    fm = {}
    for line in raw.split("\n"):
        line = line.rstrip()
        if not line or line.lstrip().startswith("#"):
            continue
        if ":" not in line:
            continue
        key, _, value = line.partition(":")
        fm[key.strip()] = unquote(value.strip())
    return fm, body


def unquote(value):
    if len(value) >= 2 and value[0] == value[-1] and value[0] in ("'", '"'):
        return value[1:-1]
    return value


def yaml_scalar(value):
    """Emit a value that round-trips through the parser above."""
    s = str(value)
    if s == "":
        return '""'
    if s[0] in "\"'&*!|>%@`[]{},#" or s[-1] == ":" or ": " in s or s.strip() != s:
        return '"' + s.replace("\\", "\\\\").replace('"', '\\"') + '"'
    return s


def render_frontmatter(pairs):
    lines = ["---"]
    for key, value in pairs:
        lines.append(f"{key}: {yaml_scalar(value)}")
    lines.append("---")
    return "\n".join(lines) + "\n"


# --------------------------------------------------------------------------
# body text
# --------------------------------------------------------------------------


def first_h1(body):
    for line in body.split("\n"):
        if line.startswith("# "):
            return line[2:].strip()
    return None


def slug_title(filename):
    stem = Path(filename).stem
    return " ".join(w.capitalize() for w in re.split(r"[-_]+", stem) if w)


def strip_wikilinks(body, counters):
    """Drop the [[...]] half of every dual-link, keeping the markdown half."""

    def drop_pair(m):
        counters["wikilinks"] += 1
        return m.group("keep")

    # [[X]] / [text](path)   and   [text](path) / [[X]]
    body = re.sub(
        r"\[\[[^\]]*\]\]\s*/\s*(?P<keep>\[[^\]]*\]\([^)]*\))", drop_pair, body
    )
    body = re.sub(
        r"(?P<keep>\[[^\]]*\]\([^)]*\))\s*/\s*\[\[[^\]]*\]\]", drop_pair, body
    )

    # Any remaining bare wikilink degrades to its plain label.
    def bare(m):
        counters["wikilinks"] += 1
        label = m.group(1)
        return label.split("|", 1)[-1].strip()

    return re.sub(r"\[\[([^\]]*)\]\]", bare, body)


def rewrite_links(body, counters):
    """Strip the /wiki prefix from repo-root-absolute intra-wiki links and
    count (but never touch) cross-cascade linked//libs/ references."""

    def repl(m):
        target = m.group(2)
        if re.match(r"^\.{0,2}/?(linked|libs)/", target):
            counters["cascade_refs"] += 1
            return m.group(0)
        if target.startswith("/wiki/"):
            counters["rewritten"] += 1
            target = target[5:]
        elif target == "/wiki" or target == "/wiki/":
            counters["rewritten"] += 1
            target = "/"
        return f"[{m.group(1)}]({target})"

    return re.sub(r"\[([^\]]*)\]\(([^)]*)\)", repl, body)


def transform_body(body, counters):
    return rewrite_links(strip_wikilinks(body, counters), counters)


# --------------------------------------------------------------------------
# source tree
# --------------------------------------------------------------------------


class Page:
    def __init__(self, relpath, text):
        self.relpath = relpath  # PosixPath relative to wiki/
        self.fm, self.body = parse_frontmatter(text)
        self.title = first_h1(self.body) or slug_title(relpath.name)
        self.description = self.fm.get("tldr", "")


def read_tree(wiki):
    pages, logs, indexes, assets = {}, {}, {}, []
    for path in sorted(wiki.rglob("*")):
        if not path.is_file() or path.name.startswith("."):
            continue
        rel = path.relative_to(wiki)
        if path.suffix != ".md":
            assets.append(rel)
            continue
        text = path.read_text(encoding="utf-8")
        if path.name == "index.md":
            indexes[rel] = text
        elif path.name == "log.md":
            logs[rel] = text
        else:
            pages[rel] = Page(rel, text)
    return pages, logs, indexes, assets


# --------------------------------------------------------------------------
# emitters
# --------------------------------------------------------------------------


def emit_concept(page, counters, issues):
    ptype = page.fm.get("type", "").strip()
    if not ptype:
        ptype = "unknown"
        issues.append(f"{page.relpath}: no `type` in source frontmatter (exported as `unknown`)")

    pairs = [("type", ptype), ("title", page.title)]
    if page.description:
        pairs.append(("description", page.description))
    if page.fm.get("resource"):
        pairs.append(("resource", page.fm["resource"]))
    if page.fm.get("last_updated"):
        pairs.append(("timestamp", page.fm["last_updated"]))
    for key in PASSTHROUGH_EXT:
        if key in page.fm:
            pairs.append((key, page.fm[key]))
    for key in sorted(page.fm):
        if key in DROPPED or key in PASSTHROUGH_EXT:
            continue
        if key in ("type", "resource", "tldr", "last_updated"):
            continue
        pairs.append((key, page.fm[key]))

    body = transform_body(page.body, counters).lstrip("\n")
    return render_frontmatter(pairs) + "\n" + body.rstrip("\n") + "\n"


def index_entries(directory, pages, logs, indexes, assets):
    """Every direct child of `directory` (a PosixPath relative to wiki/, or
    Path('.') for the root), as (title, href, description) — sorted, so the
    generated list is stable across runs."""
    entries = []
    for rel, page in pages.items():
        if rel.parent == directory:
            entries.append((page.title, rel.name, page.description))
    for rel in logs:
        if rel.parent == directory:
            entries.append(("Change Log", rel.name, "Chronological record of changes in this directory."))
    for rel in assets:
        if rel.parent == directory:
            entries.append((rel.name, rel.name, ""))
    subdirs = set()
    for rel in list(pages) + list(logs) + list(indexes) + list(assets):
        parent = rel.parent
        while parent != Path("."):
            if parent.parent == directory:
                subdirs.add(parent)
            parent = parent.parent
    for sub in subdirs:
        sub_index = sub / "index.md"
        title = slug_title(sub.name)
        if sub_index in indexes:
            title = first_h1(parse_frontmatter(indexes[sub_index])[1]) or title
        entries.append((title, f"{sub.name}/index.md", ""))
    return sorted(entries, key=lambda e: e[1])


def emit_index(directory, pages, logs, indexes, assets, is_root):
    lines = []
    if is_root:
        lines.append(render_frontmatter([("okf_version", OKF_VERSION)]).rstrip("\n"))
        lines.append("")
    src = indexes.get(directory / "index.md" if directory != Path(".") else Path("index.md"), "")
    heading = first_h1(parse_frontmatter(src)[1]) if src else None
    lines.append(f"# {heading or slug_title(directory.name) or 'Index'}")
    lines.append("")
    for title, href, description in index_entries(directory, pages, logs, indexes, assets):
        if description:
            lines.append(f"* [{title}]({href}) - {description}")
        else:
            lines.append(f"* [{title}]({href})")
    return "\n".join(lines).rstrip("\n") + "\n"


LOG_HEADER_RE = re.compile(
    r"^##\s*\[?(?P<date>\d{4}-\d{2}-\d{2})(?:[ T](?P<time>\d{2}:\d{2}))?\]?\s*-\s*\[?(?P<action>[^\]\n]*?)\]?\s*$"
)
FIELD_RE = re.compile(r"^-\s*\*\*(?P<key>[^:*]+):?\*\*:?\s*(?P<value>.*)$")


def parse_log(text):
    """Return (heading, [entry dicts]) from a Rule B formatted log."""
    body = parse_frontmatter(text)[1]
    heading = first_h1(body)
    entries, current = [], None
    for line in body.split("\n"):
        m = LOG_HEADER_RE.match(line.strip())
        if m:
            if current:
                entries.append(current)
            current = {
                "date": m.group("date"),
                "time": m.group("time") or "",
                "action": (m.group("action") or "").strip(),
                "fields": {},
            }
            continue
        if current is None:
            continue
        f = FIELD_RE.match(line.strip())
        if f:
            current["fields"][f.group("key").strip().lower()] = f.group("value").strip()
    if current:
        entries.append(current)
    return heading, entries


def verb_for(action):
    for token in re.split(r"[/,\s]+", action.upper()):
        if token in VERB_MAP:
            return VERB_MAP[token]
    return "Update"


def files_to_links(value, counters):
    """`wiki/a.md`, `wiki/b.md` -> [/a.md](/a.md), [/b.md](/b.md)

    Only entries that name an actual file inside wiki/ become links. A
    Rule B entry may also name a directory or carry a parenthetical
    ("`wiki/decisions/` (directory)") — those stay plain text, since a link
    to them would not resolve inside the bundle."""
    out = []
    for chunk in re.split(r",\s*", value):
        chunk = chunk.strip()
        if not chunk:
            continue
        m = re.match(r"^`([^`]+)`(.*)$", chunk)
        name = (m.group(1) if m else chunk).strip()
        trailer = (m.group(2).strip() if m else "")
        if name.startswith("wiki/") and name.endswith(".md") and not re.search(r"\s", name):
            href = "/" + name[len("wiki/") :]
            rendered = f"[{href}]({href})"
        else:
            rendered = name
        out.append(f"{rendered} {trailer}".strip() if trailer else rendered)
    return ", ".join(out)


def emit_log(text, counters):
    heading, entries = parse_log(text)
    lines = [f"# {heading or 'Change Log'}", ""]
    by_date = {}
    for entry in entries:
        by_date.setdefault(entry["date"], []).append(entry)
    for date in sorted(by_date, reverse=True):
        lines.append(f"## {date}")
        lines.append("")
        for entry in by_date[date]:
            verb = verb_for(entry["action"])
            fields = entry["fields"]
            desc = transform_body(fields.get("description", ""), counters).strip()
            files = files_to_links(fields.get("file affected", ""), counters)
            source = fields.get("source", "").strip().rstrip(".")
            parts = []
            if files:
                parts.append(files)
            if desc:
                parts.append(desc)
            text_part = " — ".join(parts) if parts else "(no description recorded)"
            bullet = f"* **{verb}**: {text_part}"
            if source:
                bullet += f" (source: {source})"
            lines.append(bullet.rstrip(".") + ".")
        lines.append("")
    return "\n".join(lines).rstrip("\n") + "\n"


# --------------------------------------------------------------------------
# build + validate
# --------------------------------------------------------------------------


def build(out_dir):
    pages, logs, indexes, assets = read_tree(WIKI)
    counters = {"wikilinks": 0, "cascade_refs": 0, "rewritten": 0}
    issues = []

    if out_dir.exists():
        shutil.rmtree(out_dir)
    out_dir.mkdir(parents=True)

    for rel, page in sorted(pages.items()):
        dest = out_dir / rel
        dest.parent.mkdir(parents=True, exist_ok=True)
        dest.write_text(emit_concept(page, counters, issues), encoding="utf-8")

    directories = {Path(".")}
    for rel in list(pages) + list(logs) + list(indexes) + list(assets):
        parent = rel.parent
        while parent != Path("."):
            directories.add(parent)
            parent = parent.parent
    for directory in sorted(directories):
        dest = out_dir / directory / "index.md"
        dest.parent.mkdir(parents=True, exist_ok=True)
        dest.write_text(
            emit_index(directory, pages, logs, indexes, assets, directory == Path(".")),
            encoding="utf-8",
        )

    for rel, text in sorted(logs.items()):
        dest = out_dir / rel
        dest.parent.mkdir(parents=True, exist_ok=True)
        dest.write_text(emit_log(text, counters), encoding="utf-8")

    for rel in assets:
        dest = out_dir / rel
        dest.parent.mkdir(parents=True, exist_ok=True)
        shutil.copy2(WIKI / rel, dest)

    report = {
        "concepts": len(pages),
        "indexes": len(directories),
        "logs": len(logs),
        "assets": len(assets),
        "counters": counters,
        "issues": issues,
    }
    return report


def validate(out_dir):
    """Conformance checks against the generated bundle. Returns a list of
    strings; empty means the bundle conforms."""
    problems = []
    md_files = sorted(p for p in out_dir.rglob("*.md"))
    for path in md_files:
        rel = path.relative_to(out_dir)
        text = path.read_text(encoding="utf-8")
        fm, body = parse_frontmatter(text)

        if path.name == "index.md":
            if rel == Path("index.md"):
                if set(fm) - {"okf_version"}:
                    problems.append(f"{rel}: root index frontmatter must contain only okf_version")
            elif fm:
                problems.append(f"{rel}: nested index.md must have no frontmatter")
        elif path.name == "log.md":
            for line in body.split("\n"):
                if line.startswith("## ") and not re.match(r"^## \d{4}-\d{2}-\d{2}$", line.strip()):
                    problems.append(f"{rel}: log header is not `## YYYY-MM-DD`: {line.strip()}")
        else:
            if not fm.get("type", "").strip():
                problems.append(f"{rel}: missing or empty `type`")

        for m in re.finditer(r"\[[^\]]*\]\(([^)]+)\)", body):
            target = m.group(1).split("#", 1)[0].strip()
            if not target or re.match(r"^[a-z][a-z0-9+.-]*:", target):
                continue
            if re.match(r"^\.{0,2}/?(linked|libs)/", target):
                continue  # intentionally left unconverted
            if target.startswith("/"):
                candidate = out_dir / target.lstrip("/")
            else:
                candidate = (path.parent / target).resolve()
            if not candidate.exists():
                problems.append(f"{rel}: intra-bundle link does not resolve: {target}")
    return problems


def main():
    parser = argparse.ArgumentParser(description="Export wiki/ as an OKF v0.1 bundle.")
    parser.add_argument("--check", action="store_true", help="validate only; write nothing")
    parser.add_argument("--out", default="outputs/okf", help="output directory (default: outputs/okf)")
    args = parser.parse_args()

    if not WIKI.is_dir():
        print(f"error: no wiki/ directory under {REPO_ROOT} — run from the repository root", file=sys.stderr)
        return 2

    if args.check:
        tmp = Path(tempfile.mkdtemp(prefix="okf-check-"))
        try:
            out_dir = tmp / "okf"
            report = build(out_dir)
            problems = validate(out_dir)
            print_report(report, problems, out_dir, checked_only=True)
        finally:
            shutil.rmtree(tmp, ignore_errors=True)
    else:
        out_dir = REPO_ROOT / args.out
        report = build(out_dir)
        problems = validate(out_dir)
        print_report(report, problems, out_dir, checked_only=False)

    return 1 if problems else 0


def print_report(report, problems, out_dir, checked_only):
    counters = report["counters"]
    print(f"okf_version: {OKF_VERSION}")
    print(f"concept documents exported: {report['concepts']}")
    print(f"index.md files regenerated: {report['indexes']}")
    print(f"log.md files regenerated: {report['logs']}")
    if report["assets"]:
        print(f"non-markdown files copied: {report['assets']}")
    print(f"wikilinks stripped: {counters['wikilinks']}")
    print(f"/wiki/ links rewritten to bundle-root: {counters['rewritten']}")
    print(f"linked//libs/ cross-cascade refs left unconverted: {counters['cascade_refs']}")
    for issue in report["issues"]:
        print(f"SOURCE ISSUE: {issue}")
    if problems:
        print(f"NONCONFORMANT: {len(problems)} problem(s) in the generated bundle:")
        for p in problems:
            print(f"  - {p}")
    else:
        print("conformance: OK")
    if checked_only:
        print("output: none (--check)")
    else:
        print(f"output: {out_dir}")


if __name__ == "__main__":
    sys.exit(main())