433 lines
15 KiB
Python
433 lines
15 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
outline_sync.py — Doorstop ↔ Outline wiki synchronization.
|
|
|
|
Reads Doorstop YAML requirement files and creates/updates corresponding
|
|
pages in an Outline wiki. Organizes requirements in a dedicated collection
|
|
with sub-pages for linked test cases.
|
|
|
|
Usage:
|
|
python3 outline_sync.py --doorstop reqs/ \\
|
|
--outline-url https://wiki.portugalfuturista.org \\
|
|
--api-key <key> [--dry-run]
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import sys
|
|
import urllib.error
|
|
import urllib.request
|
|
from dataclasses import dataclass, field
|
|
from pathlib import Path
|
|
from typing import Optional
|
|
|
|
|
|
# ─── Data model ───────────────────────────────────────────────────────────────
|
|
|
|
@dataclass
|
|
class DoorstopRequirement:
|
|
uid: str
|
|
level: int
|
|
active: bool
|
|
derived: bool
|
|
normative: bool
|
|
text: str
|
|
parent: str
|
|
ref: str
|
|
attrs: dict = field(default_factory=dict)
|
|
|
|
@property
|
|
def dal_level(self) -> str:
|
|
return self.attrs.get("dal_level", "B")
|
|
|
|
@property
|
|
def linked_files(self) -> list[str]:
|
|
return self.attrs.get("linked_files", [])
|
|
|
|
@property
|
|
def linked_tests(self) -> list[str]:
|
|
return self.attrs.get("linked_tests", [])
|
|
|
|
def to_outline_markdown(self) -> str:
|
|
lines = [
|
|
f"# {self.uid}",
|
|
"",
|
|
self.text,
|
|
"",
|
|
"## Attributes",
|
|
"",
|
|
f"| Field | Value |",
|
|
f"|-------|-------|",
|
|
f"| DAL Level | {self.dal_level} |",
|
|
f"| Category | {self.attrs.get('category', 'N/A')} |",
|
|
f"| Priority | {self.attrs.get('priority', 'N/A')} |",
|
|
f"| Active | {self.active} |",
|
|
f"| Derived | {self.derived} |",
|
|
f"| Normative | {self.normative} |",
|
|
]
|
|
if self.parent:
|
|
lines.append(f"| Parent | {self.parent} |")
|
|
if self.ref:
|
|
lines.append(f"| Reference | {self.ref} |")
|
|
lines.append("")
|
|
|
|
if self.linked_files:
|
|
lines.extend(["## Linked Files", ""])
|
|
for f in self.linked_files:
|
|
lines.append(f"- `{f}`")
|
|
lines.append("")
|
|
|
|
if self.linked_tests:
|
|
lines.extend(["## Linked Tests", ""])
|
|
for t in self.linked_tests:
|
|
lines.append(f"- {t}")
|
|
lines.append("")
|
|
|
|
return "\n".join(lines)
|
|
|
|
|
|
# ─── YAML reader ──────────────────────────────────────────────────────────────
|
|
|
|
def read_doorstop_yaml(yaml_path: Path) -> DoorstopRequirement:
|
|
content = yaml_path.read_text()
|
|
result = {
|
|
"uid": "", "level": 0, "active": True, "derived": False,
|
|
"normative": True, "text": "", "parent": "", "ref": "", "attrs": {},
|
|
}
|
|
current_section = None
|
|
current_list_key = None
|
|
|
|
for line in content.split("\n"):
|
|
stripped = line.strip()
|
|
if not stripped or stripped.startswith("#"):
|
|
continue
|
|
if ":" in stripped and not stripped.startswith("-") and not stripped.startswith(" "):
|
|
key, _, value = stripped.partition(":")
|
|
key, value = key.strip(), value.strip().strip('"').strip("'")
|
|
if key == "attrs":
|
|
current_section = "attrs"
|
|
current_list_key = None
|
|
continue
|
|
current_section = None
|
|
current_list_key = None
|
|
if key in ("uid", "text", "parent", "ref"):
|
|
result[key] = value
|
|
elif key in ("level",):
|
|
result[key] = int(value) if value.isdigit() else 0
|
|
elif key in ("active", "derived", "normative"):
|
|
result[key] = value.lower() in ("true", "yes", "1")
|
|
elif current_section == "attrs":
|
|
if stripped.startswith("- ") and current_list_key:
|
|
item = stripped[2:].strip().strip('"').strip("'")
|
|
result["attrs"].setdefault(current_list_key, []).append(item)
|
|
elif ":" in stripped and not stripped.startswith("-"):
|
|
key, _, value = stripped.partition(":")
|
|
key, value = key.strip(), value.strip().strip('"').strip("'")
|
|
if value:
|
|
result["attrs"][key] = value
|
|
else:
|
|
current_list_key = key
|
|
result["attrs"][key] = []
|
|
else:
|
|
current_list_key = None
|
|
|
|
return DoorstopRequirement(**result)
|
|
|
|
|
|
# ─── Outline API client ──────────────────────────────────────────────────────
|
|
|
|
class OutlineClient:
|
|
"""Minimal Outline API client (self-hosted or cloud)."""
|
|
|
|
def __init__(self, base_url: str, api_key: str):
|
|
self.base_url = base_url.rstrip("/")
|
|
self.api_key = api_key
|
|
|
|
def _request(self, endpoint: str, body: dict) -> dict:
|
|
url = f"{self.base_url}/api/v1{endpoint}"
|
|
headers = {
|
|
"Content-Type": "application/json",
|
|
"Authorization": f"Bearer {self.api_key}",
|
|
}
|
|
data = json.dumps(body).encode()
|
|
req = urllib.request.Request(url, data=data, headers=headers, method="POST")
|
|
try:
|
|
with urllib.request.urlopen(req, timeout=30) as resp:
|
|
return json.loads(resp.read().decode())
|
|
except urllib.error.HTTPError as e:
|
|
error_body = e.read().decode() if e.fp else ""
|
|
raise RuntimeError(f"Outline API error {e.code}: {error_body}") from e
|
|
|
|
def list_collections(self) -> list[dict]:
|
|
result = self._request("/collections.list", {})
|
|
return result.get("data", [])
|
|
|
|
def create_collection(self, name: str, description: str = "") -> dict:
|
|
return self._request("/collections.create", {
|
|
"name": name,
|
|
"description": description,
|
|
})
|
|
|
|
def list_documents(self, collection_id: str) -> list[dict]:
|
|
result = self._request("/documents.list", {"collectionId": collection_id})
|
|
return result.get("data", [])
|
|
|
|
def create_document(self, title: str, text: str,
|
|
collection_id: str, parent_document_id: str = "") -> dict:
|
|
body = {
|
|
"title": title,
|
|
"text": text,
|
|
"collectionId": collection_id,
|
|
}
|
|
if parent_document_id:
|
|
body["parentDocumentId"] = parent_document_id
|
|
return self._request("/documents.create", body)
|
|
|
|
def update_document(self, document_id: str, title: str, text: str) -> dict:
|
|
return self._request("/documents.update", {
|
|
"id": document_id,
|
|
"title": title,
|
|
"text": text,
|
|
})
|
|
|
|
def search_documents(self, query: str) -> list[dict]:
|
|
result = self._request("/documents.search", {"query": query})
|
|
return result.get("data", [])
|
|
|
|
|
|
# ─── Sync logic ───────────────────────────────────────────────────────────────
|
|
|
|
COLLECTION_NAME = "UniversalisOS Requirements"
|
|
|
|
|
|
def load_requirements(doorstop_dir: Path) -> list[DoorstopRequirement]:
|
|
reqs = []
|
|
for yml in sorted(doorstop_dir.glob("*.yml")):
|
|
reqs.append(read_doorstop_yaml(yml))
|
|
return reqs
|
|
|
|
|
|
def load_markdown_docs(docs_dir: Path) -> list[tuple[str, str]]:
|
|
"""Load existing Markdown docs as (title, content) pairs."""
|
|
docs = []
|
|
if not docs_dir.exists():
|
|
return docs
|
|
for md in sorted(docs_dir.glob("**/*.md")):
|
|
content = md.read_text()
|
|
title = md.stem.replace("_", " ").replace("-", " ").title()
|
|
docs.append((title, content))
|
|
return docs
|
|
|
|
|
|
def get_or_create_collection(client: OutlineClient, dry_run: bool) -> str | None:
|
|
if dry_run:
|
|
print(f" [dry-run] Would get/create collection: {COLLECTION_NAME}")
|
|
return "dry-run-collection-id"
|
|
|
|
collections = client.list_collections()
|
|
for coll in collections:
|
|
if coll.get("name") == COLLECTION_NAME:
|
|
print(f" Found existing collection: {COLLECTION_NAME} ({coll['id']})")
|
|
return coll["id"]
|
|
|
|
result = client.create_collection(
|
|
name=COLLECTION_NAME,
|
|
description="Doorstop requirements for UniversalisOS hypervisor",
|
|
)
|
|
coll_id = result.get("data", {}).get("id", "")
|
|
print(f" Created collection: {COLLECTION_NAME} ({coll_id})")
|
|
return coll_id
|
|
|
|
|
|
def sync_doorstop_to_outline(
|
|
reqs: list[DoorstopRequirement],
|
|
client: OutlineClient,
|
|
collection_id: str,
|
|
dry_run: bool = False,
|
|
) -> dict:
|
|
stats = {"created": 0, "updated": 0, "skipped": 0, "errors": []}
|
|
|
|
if not dry_run:
|
|
existing = {doc["title"]: doc for doc in client.list_documents(collection_id)}
|
|
else:
|
|
existing = {}
|
|
|
|
for req in reqs:
|
|
title = req.uid
|
|
markdown = req.to_outline_markdown()
|
|
|
|
if title in existing:
|
|
doc = existing[title]
|
|
if not dry_run:
|
|
try:
|
|
client.update_document(doc["id"], title=title, text=markdown)
|
|
stats["updated"] += 1
|
|
print(f" Updated: {req.uid}")
|
|
except Exception as e:
|
|
stats["errors"].append(f"{req.uid}: {e}")
|
|
print(f" ERROR updating {req.uid}: {e}")
|
|
else:
|
|
stats["updated"] += 1
|
|
print(f" [dry-run] Would update: {req.uid}")
|
|
else:
|
|
if not dry_run:
|
|
try:
|
|
client.create_document(
|
|
title=title,
|
|
text=markdown,
|
|
collection_id=collection_id,
|
|
)
|
|
stats["created"] += 1
|
|
print(f" Created: {req.uid}")
|
|
except Exception as e:
|
|
stats["errors"].append(f"{req.uid}: {e}")
|
|
print(f" ERROR creating {req.uid}: {e}")
|
|
else:
|
|
stats["created"] += 1
|
|
print(f" [dry-run] Would create: {req.uid}")
|
|
|
|
return stats
|
|
|
|
|
|
def sync_markdown_docs_to_outline(
|
|
docs: list[tuple[str, str]],
|
|
client: OutlineClient,
|
|
collection_id: str,
|
|
dry_run: bool = False,
|
|
) -> dict:
|
|
stats = {"created": 0, "skipped": 0, "errors": []}
|
|
|
|
if not dry_run:
|
|
existing = {doc["title"] for doc in client.list_documents(collection_id)}
|
|
else:
|
|
existing = set()
|
|
|
|
for title, content in docs:
|
|
if title in existing:
|
|
stats["skipped"] += 1
|
|
continue
|
|
|
|
if not dry_run:
|
|
try:
|
|
client.create_document(
|
|
title=title,
|
|
text=content,
|
|
collection_id=collection_id,
|
|
)
|
|
stats["created"] += 1
|
|
print(f" Imported doc: {title}")
|
|
except Exception as e:
|
|
stats["errors"].append(f"{title}: {e}")
|
|
print(f" ERROR importing {title}: {e}")
|
|
else:
|
|
stats["created"] += 1
|
|
print(f" [dry-run] Would import: {title}")
|
|
|
|
return stats
|
|
|
|
|
|
# ─── CLI ──────────────────────────────────────────────────────────────────────
|
|
|
|
def build_parser() -> argparse.ArgumentParser:
|
|
parser = argparse.ArgumentParser(
|
|
prog="outline_sync",
|
|
description="Doorstop ↔ Outline wiki synchronization",
|
|
formatter_class=argparse.RawDescriptionHelpFormatter,
|
|
epilog="""
|
|
Examples:
|
|
# Sync requirements to Outline
|
|
python3 outline_sync.py --doorstop reqs/ \\
|
|
--outline-url https://wiki.portugalfuturista.org \\
|
|
--api-key ok_xxxxx
|
|
|
|
# Include existing Markdown docs
|
|
python3 outline_sync.py --doorstop reqs/ \\
|
|
--outline-url https://wiki.portugalfuturista.org \\
|
|
--api-key ok_xxxxx --docs-dir ../../docs/
|
|
|
|
# Dry run
|
|
python3 outline_sync.py --doorstop reqs/ \\
|
|
--outline-url https://wiki.portugalfuturista.org \\
|
|
--api-key ok_xxxxx --dry-run
|
|
""",
|
|
)
|
|
parser.add_argument(
|
|
"--doorstop", required=True,
|
|
help="Directory containing Doorstop YAML requirement files",
|
|
)
|
|
parser.add_argument(
|
|
"--outline-url", required=True,
|
|
help="Outline server URL (e.g. https://wiki.portugalfuturista.org)",
|
|
)
|
|
parser.add_argument(
|
|
"--api-key", default="",
|
|
help="Outline API key (or set OUTLINE_API_KEY env var)",
|
|
)
|
|
parser.add_argument(
|
|
"--docs-dir",
|
|
help="Optional: import existing Markdown docs from this directory",
|
|
)
|
|
parser.add_argument(
|
|
"--dry-run", action="store_true",
|
|
help="Show what would be synced without making API calls",
|
|
)
|
|
return parser
|
|
|
|
|
|
def main(argv: list[str] | None = None) -> int:
|
|
parser = build_parser()
|
|
args = parser.parse_args(argv)
|
|
|
|
import os
|
|
api_key = args.api_key or os.environ.get("OUTLINE_API_KEY", "")
|
|
|
|
doorstop_dir = Path(args.doorstop)
|
|
if not doorstop_dir.exists():
|
|
print(f"Error: doorstop directory not found: {doorstop_dir}")
|
|
return 1
|
|
|
|
reqs = load_requirements(doorstop_dir)
|
|
if not reqs:
|
|
print(f"No requirement files found in {doorstop_dir}")
|
|
return 1
|
|
|
|
print(f"Loaded {len(reqs)} requirements from {doorstop_dir}")
|
|
|
|
if args.dry_run:
|
|
print("Mode: DRY RUN (no API calls)")
|
|
else:
|
|
if not api_key:
|
|
print("Error: --api-key or OUTLINE_API_KEY env var required")
|
|
return 1
|
|
|
|
client = OutlineClient(
|
|
base_url=args.outline_url,
|
|
api_key=api_key,
|
|
)
|
|
|
|
print("\n=== Getting/creating Outline collection ===")
|
|
collection_id = get_or_create_collection(client, dry_run=args.dry_run)
|
|
|
|
print("\n=== Doorstop → Outline sync ===")
|
|
stats = sync_doorstop_to_outline(reqs, client, collection_id, dry_run=args.dry_run)
|
|
print(f"\nResults: {stats['created']} created, {stats['updated']} updated, {stats['skipped']} skipped")
|
|
if stats["errors"]:
|
|
print(f"Errors: {len(stats['errors'])}")
|
|
for err in stats["errors"]:
|
|
print(f" - {err}")
|
|
|
|
if args.docs_dir:
|
|
docs_dir = Path(args.docs_dir)
|
|
docs = load_markdown_docs(docs_dir)
|
|
if docs:
|
|
print(f"\n=== Importing {len(docs)} Markdown docs ===")
|
|
stats2 = sync_markdown_docs_to_outline(docs, client, collection_id, dry_run=args.dry_run)
|
|
print(f"\nDoc import: {stats2['created']} created, {stats2['skipped']} skipped")
|
|
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|