478 lines
16 KiB
Python
478 lines
16 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
plane_bridge.py — Doorstop ↔ Plane issue synchronization bridge.
|
|
|
|
Reads Doorstop YAML requirement files and creates/updates corresponding
|
|
issues in Plane (plane.project.notion-like tool). Supports bi-directional
|
|
sync with status mapping and DAL-based priority assignment.
|
|
|
|
Usage:
|
|
python3 plane_bridge.py --doorstop reqs/ \\
|
|
--plane-url https://plane.portugalfuturista.org \\
|
|
--workspace universalisos --project kernel [--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
|
|
|
|
|
|
# ─── Constants ────────────────────────────────────────────────────────────────
|
|
|
|
DAL_TO_PRIORITY = {
|
|
"A": "urgent",
|
|
"B": "high",
|
|
"C": "medium",
|
|
"D": "low",
|
|
"E": "low",
|
|
"F": "low",
|
|
}
|
|
|
|
PLANE_STATUS_MAP = {
|
|
"backlog": "backlog",
|
|
"todo": "unstarted",
|
|
"in_progress": "started",
|
|
"done": "completed",
|
|
"cancelled": "cancelled",
|
|
}
|
|
|
|
|
|
# ─── 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 priority(self) -> str:
|
|
return DAL_TO_PRIORITY.get(self.dal_level, "medium")
|
|
|
|
@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_plane_title(self) -> str:
|
|
return f"[{self.uid}] {self.text[:120]}"
|
|
|
|
def to_plane_description(self) -> str:
|
|
lines = [
|
|
f"## {self.uid}",
|
|
"",
|
|
self.text,
|
|
"",
|
|
"---",
|
|
"",
|
|
f"**DAL Level:** {self.dal_level}",
|
|
f"**Category:** {self.attrs.get('category', 'N/A')}",
|
|
f"**Priority (derived):** {self.priority}",
|
|
f"**Active:** {self.active}",
|
|
f"**Derived:** {self.derived}",
|
|
]
|
|
if self.parent:
|
|
lines.append(f"**Parent:** {self.parent}")
|
|
if self.ref:
|
|
lines.append(f"**Reference:** {self.ref}")
|
|
if self.linked_files:
|
|
lines.extend(["", "### Linked Files", ""])
|
|
for f in self.linked_files:
|
|
lines.append(f"- `{f}`")
|
|
if self.linked_tests:
|
|
lines.extend(["", "### Linked Tests", ""])
|
|
for t in self.linked_tests:
|
|
lines.append(f"- {t}")
|
|
return "\n".join(lines)
|
|
|
|
def to_plane_labels(self) -> list[str]:
|
|
labels = ["requirement"]
|
|
cat = self.attrs.get("category", "")
|
|
if cat:
|
|
labels.append(cat)
|
|
labels.append(f"dal-{self.dal_level.lower()}")
|
|
return labels
|
|
|
|
|
|
@dataclass
|
|
class PlaneIssue:
|
|
id: str
|
|
name: str
|
|
description: str
|
|
state: str
|
|
priority: str
|
|
labels: list[str]
|
|
identifier: str # REQ-001 etc.
|
|
|
|
|
|
# ─── YAML reader (no PyYAML dependency) ───────────────────────────────────────
|
|
|
|
def read_doorstop_yaml(yaml_path: Path) -> DoorstopRequirement:
|
|
"""Parse a Doorstop YAML file into a 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)
|
|
|
|
|
|
# ─── Plane API client ────────────────────────────────────────────────────────
|
|
|
|
class PlaneClient:
|
|
"""Minimal Plane REST API client (v1)."""
|
|
|
|
def __init__(self, base_url: str, workspace: str, project: str, api_key: str):
|
|
self.base_url = base_url.rstrip("/")
|
|
self.workspace = workspace
|
|
self.project = project
|
|
self.api_key = api_key
|
|
|
|
def _request(self, method: str, path: str, body: dict | None = None) -> dict:
|
|
url = f"{self.base_url}/api/v1/workspaces/{self.workspace}/projects/{self.project}{path}"
|
|
headers = {
|
|
"Content-Type": "application/json",
|
|
"X-Api-Key": self.api_key,
|
|
}
|
|
data = json.dumps(body).encode() if body else None
|
|
req = urllib.request.Request(url, data=data, headers=headers, method=method)
|
|
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"Plane API error {e.code}: {error_body}") from e
|
|
|
|
def list_issues(self) -> list[dict]:
|
|
result = self._request("GET", "/issues/")
|
|
return result.get("results", result) if isinstance(result, dict) else result
|
|
|
|
def create_issue(self, name: str, description: str, priority: int,
|
|
labels: list[str], state: str = "") -> dict:
|
|
body = {
|
|
"name": name,
|
|
"description_html": description,
|
|
"priority": priority,
|
|
}
|
|
if labels:
|
|
body["labels"] = labels
|
|
if state:
|
|
body["state"] = state
|
|
return self._request("POST", "/issues/", body)
|
|
|
|
def update_issue(self, issue_id: str, **kwargs) -> dict:
|
|
return self._request("PATCH", f"/issues/{issue_id}/", kwargs)
|
|
|
|
|
|
# ─── DAL priority numeric mapping (Plane uses 0-4) ──────────────────────────
|
|
|
|
DAL_TO_PLANE_PRIORITY = {"A": 0, "B": 1, "C": 2, "D": 3, "E": 4, "F": 4}
|
|
|
|
|
|
# ─── Sync logic ───────────────────────────────────────────────────────────────
|
|
|
|
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 build_issue_index(client: PlaneClient) -> dict[str, dict]:
|
|
"""Build a mapping of issue identifier → Plane issue dict."""
|
|
index = {}
|
|
for issue in client.list_issues():
|
|
ident = issue.get("identifier", "")
|
|
if ident:
|
|
index[ident] = issue
|
|
return index
|
|
|
|
|
|
def sync_doorstop_to_plane(
|
|
reqs: list[DoorstopRequirement],
|
|
client: PlaneClient,
|
|
dry_run: bool = False,
|
|
) -> dict:
|
|
stats = {"created": 0, "updated": 0, "skipped": 0, "errors": []}
|
|
|
|
if not dry_run:
|
|
existing = build_issue_index(client)
|
|
else:
|
|
existing = {}
|
|
|
|
for req in reqs:
|
|
title = req.to_plane_title()
|
|
description = req.to_plane_description()
|
|
labels = req.to_plane_labels()
|
|
priority = DAL_TO_PLANE_PRIORITY.get(req.dal_level, 2)
|
|
|
|
if req.uid in existing:
|
|
plane_issue = existing[req.uid]
|
|
needs_update = False
|
|
updates = {}
|
|
|
|
if plane_issue.get("name") != title:
|
|
updates["name"] = title
|
|
needs_update = True
|
|
if plane_issue.get("priority") != priority:
|
|
updates["priority"] = priority
|
|
needs_update = True
|
|
|
|
if needs_update and not dry_run:
|
|
try:
|
|
client.update_issue(plane_issue["id"], **updates)
|
|
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["skipped"] += 1
|
|
if dry_run:
|
|
print(f" [dry-run] Would update: {req.uid}")
|
|
else:
|
|
if not dry_run:
|
|
try:
|
|
client.create_issue(
|
|
name=title,
|
|
description=description,
|
|
priority=priority,
|
|
labels=labels,
|
|
)
|
|
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} (DAL={req.dal_level}, priority={priority})")
|
|
|
|
return stats
|
|
|
|
|
|
def sync_plane_to_doorstop(
|
|
client: PlaneClient,
|
|
doorstop_dir: Path,
|
|
dry_run: bool = False,
|
|
) -> dict:
|
|
stats = {"updated": 0, "skipped": 0, "errors": []}
|
|
|
|
issues = client.list_issues()
|
|
existing_reqs = {r.uid: r for r in load_requirements(doorstop_dir)}
|
|
|
|
for issue in issues:
|
|
ident = issue.get("identifier", "")
|
|
if not ident or ident not in existing_reqs:
|
|
continue
|
|
|
|
req = existing_reqs[ident]
|
|
plane_state = issue.get("state", "")
|
|
|
|
# Map Plane status back to Doorstop attrs
|
|
doorstop_status = None
|
|
for d_status, p_status in PLANE_STATUS_MAP.items():
|
|
if p_status == plane_state:
|
|
doorstop_status = d_status
|
|
break
|
|
|
|
if doorstop_status and not dry_run:
|
|
req.attrs["plane_status"] = doorstop_status
|
|
# Rewrite the YAML file
|
|
_rewrite_yaml(doorstop_dir / f"{ident}.yml", req)
|
|
stats["updated"] += 1
|
|
print(f" Updated Doorstop: {ident} <- Plane status '{plane_state}'")
|
|
elif dry_run:
|
|
print(f" [dry-run] Would update Doorstop: {ident} <- Plane status '{plane_state}'")
|
|
else:
|
|
stats["skipped"] += 1
|
|
|
|
return stats
|
|
|
|
|
|
def _rewrite_yaml(path: Path, req: DoorstopRequirement) -> None:
|
|
lines = [
|
|
f"uid: {req.uid}",
|
|
f"level: {req.level}",
|
|
f"active: {'true' if req.active else 'false'}",
|
|
f"derived: {'true' if req.derived else 'false'}",
|
|
f"normative: {'true' if req.normative else 'false'}",
|
|
f'text: "{req.text}"',
|
|
f'parent: "{req.parent}"',
|
|
f'ref: "{req.ref}"',
|
|
"attrs:",
|
|
]
|
|
for key, value in req.attrs.items():
|
|
if isinstance(value, list):
|
|
lines.append(f" {key}:")
|
|
for item in value:
|
|
lines.append(f' - "{item}"')
|
|
else:
|
|
lines.append(f" {key}: {value}")
|
|
path.write_text("\n".join(lines) + "\n")
|
|
|
|
|
|
# ─── CLI ──────────────────────────────────────────────────────────────────────
|
|
|
|
def build_parser() -> argparse.ArgumentParser:
|
|
parser = argparse.ArgumentParser(
|
|
prog="plane_bridge",
|
|
description="Doorstop ↔ Plane issue synchronization bridge",
|
|
formatter_class=argparse.RawDescriptionHelpFormatter,
|
|
epilog="""
|
|
Examples:
|
|
# Sync Doorstop requirements to Plane
|
|
python3 plane_bridge.py --doorstop reqs/ \\
|
|
--plane-url https://plane.portugalfuturista.org \\
|
|
--workspace universalisos --project kernel
|
|
|
|
# Bi-directional sync (Plane status back to Doorstop)
|
|
python3 plane_bridge.py --doorstop reqs/ \\
|
|
--plane-url https://plane.portugalfuturista.org \\
|
|
--workspace universalisos --project kernel --bidirectional
|
|
|
|
# Dry run (no API calls)
|
|
python3 plane_bridge.py --doorstop reqs/ \\
|
|
--plane-url https://plane.portugalfuturista.org \\
|
|
--workspace universalisos --project kernel --dry-run
|
|
""",
|
|
)
|
|
parser.add_argument(
|
|
"--doorstop", required=True,
|
|
help="Directory containing Doorstop YAML requirement files",
|
|
)
|
|
parser.add_argument(
|
|
"--plane-url", required=True,
|
|
help="Plane server URL (e.g. https://plane.portugalfuturista.org)",
|
|
)
|
|
parser.add_argument(
|
|
"--workspace", required=True,
|
|
help="Plane workspace slug",
|
|
)
|
|
parser.add_argument(
|
|
"--project", required=True,
|
|
help="Plane project slug",
|
|
)
|
|
parser.add_argument(
|
|
"--api-key", default="",
|
|
help="Plane API key (or set PLANE_API_KEY env var)",
|
|
)
|
|
parser.add_argument(
|
|
"--bidirectional", action="store_true",
|
|
help="Also sync Plane status back to Doorstop attributes",
|
|
)
|
|
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("PLANE_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 PLANE_API_KEY env var required")
|
|
return 1
|
|
|
|
client = PlaneClient(
|
|
base_url=args.plane_url,
|
|
workspace=args.workspace,
|
|
project=args.project,
|
|
api_key=api_key,
|
|
)
|
|
|
|
print("\n=== Doorstop → Plane sync ===")
|
|
stats = sync_doorstop_to_plane(reqs, client, 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.bidirectional:
|
|
print("\n=== Plane → Doorstop sync ===")
|
|
stats2 = sync_plane_to_doorstop(client, doorstop_dir, dry_run=args.dry_run)
|
|
print(f"\nResults: {stats2['updated']} updated, {stats2['skipped']} skipped")
|
|
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|