Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
114 changes: 114 additions & 0 deletions adapters/selfhosted/redmine.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,114 @@
"""Redmine adapter — creates projects and seeds issue trackers via the Redmine REST API.

IMPORTANT (verified API limit): custom field DEFINITIONS cannot be created via the API.
Field definitions must be set up manually once. See docs/redmine-setup.md.
This adapter only seeds field VALUES on issues.
"""

from __future__ import annotations

import os
from typing import Any

import requests

from src.core.config import CoreConfig, RecordType


class RedmineAdapter:
"""Seeds QMS record structure into a Redmine instance.

Authentication: REDMINE_API_KEY env var.
API reference: https://www.redmine.org/projects/redmine/wiki/Rest_api
"""

def __init__(self, config: CoreConfig) -> None:
# TODO: extract Redmine settings from overlay config once overlay schema is wired
self._base_url = "http://localhost:3000"
self._api_key = os.environ.get("REDMINE_API_KEY", "")
self._project_key = "qms"
self._config = config

@property
def _headers(self) -> dict[str, str]:
return {"X-Redmine-API-Key": self._api_key, "Content-Type": "application/json"}

def _get(self, path: str) -> Any:
resp = requests.get(f"{self._base_url}{path}", headers=self._headers, timeout=10)
resp.raise_for_status()
return resp.json()

def _post(self, path: str, payload: dict) -> Any:
resp = requests.post(
f"{self._base_url}{path}", json=payload, headers=self._headers, timeout=30
)
resp.raise_for_status()
return resp.json()

def project_exists(self) -> bool:
try:
self._get(f"/projects/{self._project_key}.json")
return True
except requests.HTTPError:
return False

def create_project(self) -> None:
"""Idempotent: skips creation if project already exists."""
if self.project_exists():
print(f" Redmine: project '{self._project_key}' already exists, skipping")
return
payload = {
"project": {
"name": self._config.organisation.name,
"identifier": self._project_key,
"description": f"QMS records for {self._config.organisation.name}",
}
}
self._post("/projects.json", payload)
print(f" Redmine: created project '{self._project_key}'")

def get_tracker_id(self, tracker_name: str) -> int | None:
"""Look up tracker ID by name. Returns None if not found."""
data = self._get("/trackers.json")
for tracker in data.get("trackers", []):
if tracker["name"] == tracker_name:
return int(tracker["id"])
return None

def seed_record_type(self, record_type: RecordType, tracker_name: str) -> None:
"""Create a placeholder issue for a record type to confirm the tracker is wired."""
tracker_id = self.get_tracker_id(tracker_name)
if tracker_id is None:
print(
f" Redmine: tracker '{tracker_name}' not found — "
f"run manual setup first (docs/redmine-setup.md)"
)
return
payload = {
"issue": {
"project_id": self._project_key,
"tracker_id": tracker_id,
"subject": f"[SCAFFOLD] {record_type.label} tracker ready",
"description": (
f"This issue confirms the '{tracker_name}' tracker is configured.\n"
f"ISO 9001 clause: {record_type.clause}\n"
f"Expected fields: {', '.join(record_type.fields)}"
),
}
}
self._post("/issues.json", payload)
print(f" Redmine: seeded tracker '{tracker_name}' with scaffold issue")

def deploy(self, tracker_mapping: dict[str, str]) -> None:
"""Create project and seed one scaffold issue per record type.

Args:
tracker_mapping: maps record type ID -> Redmine tracker name
"""
self.create_project()
for record_type in self._config.record_types:
tracker_name = tracker_mapping.get(record_type.id)
if not tracker_name:
print(f" Redmine: no tracker mapping for '{record_type.id}', skipping")
continue
self.seed_record_type(record_type, tracker_name)
53 changes: 53 additions & 0 deletions adapters/selfhosted/xwiki.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,53 @@
"""XWiki adapter — creates spaces and pages via the XWiki REST API."""

from __future__ import annotations

import os

import requests

from src.core.config import CoreConfig


class XWikiAdapter:
"""Seeds QMS document structure into an XWiki instance.

Authentication: username + XWIKI_PASSWORD env var.
API reference: https://www.xwiki.org/xwiki/bin/view/Documentation/UserGuide/Features/XWikiRESTfulAPI
"""

def __init__(self, config: CoreConfig) -> None:
# TODO: extract XWiki settings from overlay config once overlay schema is wired
self._base_url = "http://localhost:8080"
self._space_key = "QMS"
self._username = "Admin"
self._password = os.environ.get("XWIKI_PASSWORD", "")
self._config = config

@property
def _auth(self) -> tuple[str, str]:
return (self._username, self._password)

def _page_url(self, page_name: str) -> str:
return f"{self._base_url}/rest/wikis/xwiki/spaces/{self._space_key}/pages/{page_name}"

def page_exists(self, page_name: str) -> bool:
resp = requests.get(self._page_url(page_name), auth=self._auth, timeout=10)
return resp.status_code == 200

def create_or_update_page(self, page_name: str, title: str, content: str) -> None:
"""Idempotent: creates the page if absent, updates content if present."""
url = self._page_url(page_name)
payload = {"title": title, "content": content, "syntax": "markdown/1.2"}
resp = requests.put(url, json=payload, auth=self._auth, timeout=30)
resp.raise_for_status()

def deploy(self, rendered_pages: dict[str, tuple[str, str]]) -> None:
"""Deploy all rendered pages.

Args:
rendered_pages: mapping of page_name -> (title, rendered_markdown)
"""
for page_name, (title, content) in rendered_pages.items():
self.create_or_update_page(page_name, title, content)
print(f" XWiki: upserted page '{page_name}'")
106 changes: 106 additions & 0 deletions cli/main.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,106 @@
"""qms-kit CLI — entry point for deploying the QMS scaffold."""

from __future__ import annotations

from pathlib import Path

import click

from src.core.config import load_config
from src.core.renderer import render_template

TEMPLATES_DIR = Path(__file__).parent.parent / "templates"
CONFIG_DIR = Path(__file__).parent.parent / "config"


@click.group()
def cli() -> None:
"""qms-kit: deploy an ISO 9001 QMS scaffold into your target platform."""


@cli.command()
@click.option(
"--target",
required=True,
type=click.Choice(["selfhosted", "atlassian", "m365"]),
help="Deployment target platform.",
)
@click.option(
"--config",
"config_path",
required=True,
type=click.Path(exists=True, path_type=Path),
help="Path to the client overlay config (e.g. config/clients/acme.yaml).",
)
@click.option(
"--dry-run",
is_flag=True,
default=False,
help="Render templates and validate config without writing to the target.",
)
def deploy(target: str, config_path: Path, dry_run: bool) -> None:
"""Deploy the QMS scaffold to the specified target platform."""
core_path = CONFIG_DIR / "core.yaml"

click.echo(f"Loading config: core={core_path}, overlay={config_path}")
config = load_config(core_path, config_path)
click.echo(
f"Config loaded: {config.organisation.name} "
f"({config.meta.standard}, {len(config.documents)} documents)"
)

# Render all templates that have a template key set
rendered: dict[str, tuple[str, str]] = {}
for doc in config.documents:
if doc.template is None:
continue
template_file = f"{doc.template}.md.j2"
content = render_template(
template_name=template_file,
config=config,
templates_dir=TEMPLATES_DIR,
extra={
"procedure_title": doc.title,
"doc_id": f"QMS-{doc.clause}-{doc.id.upper()[:8]}",
"clause": doc.clause,
"owner_role": "Quality Management Officer",
"approver": config.organisation.management,
"author": config.organisation.quality_officer,
"date": "TODO",
},
)
rendered[doc.id] = (doc.title, content)
click.echo(f" Rendered: {doc.id} ({template_file})")

if dry_run:
click.echo(f"\nDry run complete — {len(rendered)} template(s) rendered, nothing deployed.")
return

if target == "selfhosted":
_deploy_selfhosted(config, rendered)
elif target == "atlassian":
raise click.ClickException("Atlassian adapter not yet implemented (Phase 2).")
elif target == "m365":
raise click.ClickException("M365 adapter not yet implemented (Phase 3).")


def _deploy_selfhosted(config: object, rendered: dict[str, tuple[str, str]]) -> None:
from adapters.selfhosted.redmine import RedmineAdapter
from adapters.selfhosted.xwiki import XWikiAdapter

click.echo("\nDeploying to self-hosted (XWiki + Redmine)...")

xwiki = XWikiAdapter(config) # type: ignore[arg-type]
xwiki.deploy(rendered)

# TODO: load tracker_mapping from overlay config
tracker_mapping = {
"nc": "Nonconformity",
"capa": "CAPA",
"audit": "Internal Audit",
"kpi": "KPI Measurement",
}
redmine = RedmineAdapter(config) # type: ignore[arg-type]
redmine.deploy(tracker_mapping)

click.echo("\nDone.")