ourdigital/our-claude-skills
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(okf): add okf_validate; harden YAML parser for block lists + folded scalars

Browse Source 
Verified conformant against Google reference bundles (crypto_bitcoin, ga4, stackoverflow).

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
Andrew Yim
2026-06-16 19:38:31 +09:00
parent 9762ee97ab
commit 7b239dda8f
4 changed files with 223 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files

59
custom-skills/97-ourdigital-okf/code/scripts/okf_common.py
View File

@@ -6,6 +6,7 @@ from pathlib import Path
RESERVED = {"index.md", "log.md"}
_LINK_RE = re.compile(r"\[[^\]]*\]\(([^)]+)\)")
_BLOCK_INDICATORS = {">", "|", ">-", "|-", ">+", "|+"}
class FrontmatterError(ValueError):
@@ -33,24 +34,74 @@ def _scalar(value):
def parse_yaml_subset(raw):
"""Parse the small YAML subset OKF uses: `key: value` and `key: [a, b]`."""
"""Parse the small YAML subset OKF bundles use in practice.
Handles: ``key: value``; inline lists ``key: [a, b]``; block lists
(``key:`` followed by ``- item`` lines); folded multi-line scalars
(a value continued on following indented lines); and ``>``/``|`` block
scalars. Surrounding quotes are stripped from scalar values. This is a
deliberately small parser — not a full YAML implementation.
"""
lines = raw.splitlines()
n = len(lines)
meta = {}
for line in raw.splitlines():
i = 0
while i < n:
line = lines[i]
stripped = line.strip()
if not stripped or stripped.startswith("#"):
i += 1
continue
if ":" not in line:
raise FrontmatterError("unparseable frontmatter line: %r" % line)
key, _, value = line.partition(":")
key, value = key.strip(), value.strip()
i += 1
# Inline list: key: [a, b]
if value.startswith("[") and value.endswith("]"):
inner = value[1:-1].strip()
meta[key] = [s for s in (_scalar(x) for x in inner.split(",")) if s] if inner else []
else:
meta[key] = _scalar(value)
continue
# Empty value: either a block list (- item lines) or a continued scalar.
if value == "":
items = []
while i < n and lines[i].strip().startswith("-"):
items.append(_scalar(lines[i].strip()[1:].strip()))
i += 1
if items:
meta[key] = items
else:
cont = []
i = _collect_indented(lines, i, n, cont)
meta[key] = " ".join(cont)
continue
# Explicit block scalar indicator (> folded, | literal).
if value in _BLOCK_INDICATORS:
cont = []
i = _collect_indented(lines, i, n, cont)
meta[key] = ("\n" if value[0] == "|" else " ").join(cont)
continue
# Plain scalar, possibly folded across following indented lines.
parts = [value]
i = _collect_indented(lines, i, n, parts, skip_list_items=True)
meta[key] = _scalar(" ".join(parts))
return meta
def _collect_indented(lines, i, n, out, skip_list_items=False):
"""Append stripped indented continuation lines to ``out``; return new index."""
while i < n and lines[i][:1].isspace() and lines[i].strip():
if skip_list_items and lines[i].strip().startswith("-"):
break
out.append(lines[i].strip())
i += 1
return i
def parse_frontmatter(text):
"""Return (meta, body). meta is None when no frontmatter block is present."""
raw, body = split_frontmatter(text)

89
custom-skills/97-ourdigital-okf/code/scripts/okf_validate.py Normal file
View File

@@ -0,0 +1,89 @@
#!/usr/bin/env python3
"""OKF v0.1 conformance + broken-link validator (Python standard library only).
Conformance (SPEC.md section 9): every non-reserved .md has a parseable
frontmatter block with a non-empty `type`. Broken cross-links are reported as
warnings, never errors (consumers MUST tolerate them).
"""
from __future__ import annotations
import argparse
import json
import sys
from pathlib import Path
sys.path.insert(0, str(Path(__file__).resolve().parent))
from okf_common import (FrontmatterError, concept_id, extract_links,
iter_concepts, parse_frontmatter)
def _resolve_link(bundle, path, target):
"""Return a Path for an internal .md link, or None for external/anchor/non-md."""
t = target.split("#", 1)[0].strip()
if not t or "://" in t or t.startswith("mailto:") or not t.endswith(".md"):
return None
if t.startswith("/"):
return bundle / t.lstrip("/")
return path.parent / t
def validate_bundle(bundle_dir):
bundle = Path(bundle_dir)
errors, warnings = [], []
concept_files = list(iter_concepts(bundle))
existing = {p.resolve() for p in bundle.rglob("*.md")}
for path in concept_files:
cid = concept_id(bundle, path)
text = path.read_text(encoding="utf-8")
try:
meta, body = parse_frontmatter(text)
except FrontmatterError as exc:
errors.append({"concept": cid, "rule": "frontmatter", "message": str(exc)})
continue
if meta is None:
errors.append({"concept": cid, "rule": "frontmatter",
"message": "missing YAML frontmatter block"})
continue
if not str(meta.get("type", "")).strip():
errors.append({"concept": cid, "rule": "type",
"message": "missing or empty required 'type' field"})
for target in extract_links(body):
resolved = _resolve_link(bundle, path, target)
if resolved is not None and resolved.resolve() not in existing:
warnings.append({"concept": cid, "rule": "broken_link",
"message": "link target not found: %s" % target})
return {
"bundle": str(bundle),
"concepts": len(concept_files),
"conformant": len(errors) == 0,
"errors": errors,
"warnings": warnings,
}
def format_report(report):
status = "CONFORMANT" if report["conformant"] else "NON-CONFORMANT"
lines = [
"OKF v0.1 validation: %s" % report["bundle"],
" concepts: %d status: %s" % (report["concepts"], status),
" errors: %d warnings: %d" % (len(report["errors"]), len(report["warnings"])),
]
for e in report["errors"]:
lines.append(" ERROR [%s] %s: %s" % (e["concept"], e["rule"], e["message"]))
for w in report["warnings"]:
lines.append(" WARN [%s] %s: %s" % (w["concept"], w["rule"], w["message"]))
return "\n".join(lines)
def main(argv=None):
ap = argparse.ArgumentParser(description="Validate an OKF v0.1 bundle.")
ap.add_argument("bundle", help="Path to the bundle directory")
ap.add_argument("--json", action="store_true", help="Emit a JSON report")
args = ap.parse_args(argv)
report = validate_bundle(args.bundle)
print(json.dumps(report, indent=2) if args.json else format_report(report))
return 0 if report["conformant"] else 1
if __name__ == "__main__":
sys.exit(main())

15
custom-skills/97-ourdigital-okf/code/scripts/tests/test_okf_common.py
View File

@@ -24,6 +24,21 @@ class TestFrontmatter(unittest.TestCase):
def test_empty_list(self):
self.assertEqual(parse_yaml_subset("tags: []")["tags"], [])
def test_block_list(self):
raw = "type: BigQuery Dataset\ntags:\n- cryptocurrency\n- bitcoin\n- public data\ntimestamp: '2026-05-28T22:44:47+00:00'"
meta = parse_yaml_subset(raw)
self.assertEqual(meta["tags"], ["cryptocurrency", "bitcoin", "public data"])
self.assertEqual(meta["timestamp"], "2026-05-28T22:44:47+00:00")
def test_folded_multiline_scalar(self):
raw = "description: This dataset contains a complete history of the Bitcoin\n blockchain and updates every 10 minutes.\ntype: BigQuery Dataset"
meta = parse_yaml_subset(raw)
self.assertEqual(
meta["description"],
"This dataset contains a complete history of the Bitcoin blockchain and updates every 10 minutes.",
)
self.assertEqual(meta["type"], "BigQuery Dataset")
def test_no_frontmatter(self):
meta, body = parse_frontmatter("# Just markdown\n")
self.assertIsNone(meta)

64
custom-skills/97-ourdigital-okf/code/scripts/tests/test_okf_validate.py Normal file
View File

@@ -0,0 +1,64 @@
import sys, tempfile, unittest
from pathlib import Path
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
from okf_validate import validate_bundle
FIX = Path(__file__).resolve().parent / "fixtures" / "mini_bundle"
GOOGLE = Path.home() / "Documents/reference-library/open-knowledge-format/okf/bundles"
def write_bundle(root, files):
for rel, content in files.items():
p = Path(root) / rel
p.parent.mkdir(parents=True, exist_ok=True)
p.write_text(content, encoding="utf-8")
class TestValidate(unittest.TestCase):
def test_mini_fixture_conformant(self):
report = validate_bundle(FIX)
self.assertTrue(report["conformant"], report["errors"])
self.assertGreaterEqual(report["concepts"], 3)
def test_missing_type_is_error(self):
with tempfile.TemporaryDirectory() as d:
write_bundle(d, {"tables/x.md": "---\ntitle: X\n---\nbody"})
report = validate_bundle(d)
self.assertFalse(report["conformant"])
self.assertTrue(any(e["rule"] == "type" for e in report["errors"]))
def test_missing_frontmatter_is_error(self):
with tempfile.TemporaryDirectory() as d:
write_bundle(d, {"tables/x.md": "# no frontmatter\n"})
self.assertFalse(validate_bundle(d)["conformant"])
def test_reserved_files_not_required_to_have_type(self):
with tempfile.TemporaryDirectory() as d:
write_bundle(d, {"index.md": "# Index\n* [x](/tables/x.md)\n",
"tables/x.md": "---\ntype: T\n---\nb"})
self.assertTrue(validate_bundle(d)["conformant"])
def test_broken_link_is_warning_not_error(self):
with tempfile.TemporaryDirectory() as d:
write_bundle(d, {"tables/x.md": "---\ntype: T\n---\nSee [y](/tables/y.md)."})
report = validate_bundle(d)
self.assertTrue(report["conformant"])
self.assertTrue(any(w["rule"] == "broken_link" for w in report["warnings"]))
class TestGoogleBundles(unittest.TestCase):
@unittest.skipUnless(GOOGLE.exists(), "Google reference bundles not present")
def test_google_sample_bundles_conformant(self):
for name in ("crypto_bitcoin", "ga4", "stackoverflow"):
bundle = GOOGLE / name
if not bundle.exists():
continue
report = validate_bundle(bundle)
self.assertTrue(
report["conformant"],
"%s non-conformant; first errors: %s" % (name, report["errors"][:3]),
)
if __name__ == "__main__":
unittest.main()