"""Service for rendering one DSW project with a document template."""
from __future__ import annotations
import json
import shutil
import uuid
from dataclasses import dataclass
from pathlib import Path
from typing import Any
from dsw_document_template_tool.api import DSWApiClient
from dsw_document_template_tool.tdk import (
TemplateToolError,
put_template_dir,
read_local_template_coordinates,
read_local_template_package_coordinates,
stage_local_template_dir,
stage_local_template_package,
verify_template_dir,
)
[docs]
@dataclass(frozen=True)
class ResolvedProject:
"""Project UUID resolved from an existing reference or created from events."""
project_uuid: str
created_by_tool: bool = False
[docs]
def render_project(
*,
project_uuid: str | None,
project_ref: Path,
template_dir: Path,
template_package: Path | None,
output_path: Path,
format_uuid: str,
stage_id: str | None,
api_url: str,
api_key: str | None,
email: str,
password: str,
tdk_executable: str,
timeout_seconds: int,
poll_seconds: float,
verify_ssl: bool,
keep_created_project: bool = False,
) -> Path:
"""Render one project with a draft source tree or released package zip."""
template_dir = template_dir.resolve()
template_package = template_package.resolve() if template_package is not None else None
output_path = output_path.resolve()
if template_package is None and not template_dir.is_dir():
raise TemplateToolError(f"Template directory does not exist: {template_dir}")
if template_package is not None and not template_package.is_file():
raise TemplateToolError(f"Template package does not exist: {template_package}")
client = DSWApiClient(api_url=api_url, verify_ssl=verify_ssl)
staged_path: Path | None = None
resolved_project: ResolvedProject | None = None
try:
if api_key:
client.set_token(api_key)
else:
client.login(email=email, password=password)
if client.token is None:
raise TemplateToolError("Authentication did not produce a DSW bearer token.")
user = client.get_current_user()
print(f"INFO: Authenticated as {user.get('name') or user.get('email')}")
resolved_project = _resolve_or_create_project(
client=client,
project_uuid=project_uuid,
project_ref=project_ref,
)
if template_package is not None:
source_coordinates = read_local_template_package_coordinates(template_package)
staged_package, staged_coordinates = stage_local_template_package(
source_package=template_package,
)
staged_path = staged_package
document, render_info = _render_released_template_package(
client=client,
package_path=staged_package,
project_uuid=resolved_project.project_uuid,
format_uuid=format_uuid,
timeout_seconds=timeout_seconds,
poll_seconds=poll_seconds,
)
render_info.update(
{
"template_package": str(template_package),
"source_template_id": source_coordinates.full_id,
"staged_template_id": staged_coordinates.full_id,
}
)
else:
document, render_info, staged_path = _render_draft_template_dir(
client=client,
template_dir=template_dir,
project_uuid=resolved_project.project_uuid,
format_uuid=format_uuid,
stage_id=stage_id,
api_url=api_url,
tdk_executable=tdk_executable,
timeout_seconds=timeout_seconds,
poll_seconds=poll_seconds,
)
_write_rendered_document(
output_path=output_path,
document=document,
metadata={
"api_url": api_url,
"project_uuid": resolved_project.project_uuid,
"project_created_by_tool": resolved_project.created_by_tool,
"format_uuid": format_uuid,
"output": str(output_path),
"bytes": len(document),
**render_info,
},
)
return output_path
finally:
if (
resolved_project is not None
and resolved_project.created_by_tool
and not keep_created_project
):
try:
client.delete_project(resolved_project.project_uuid)
except Exception as exc:
print(
"WARNING: Failed to clean up created project "
f"{resolved_project.project_uuid}: {exc}"
)
client.close()
if staged_path is not None:
shutil.rmtree(staged_path.parent, ignore_errors=True)
def _render_draft_template_dir(
*,
client: DSWApiClient,
template_dir: Path,
project_uuid: str,
format_uuid: str,
stage_id: str | None,
api_url: str,
tdk_executable: str,
timeout_seconds: int,
poll_seconds: float,
) -> tuple[bytes, dict[str, object], Path]:
staged_dir: Path | None = None
try:
source_coordinates = read_local_template_coordinates(template_dir)
staged_dir, staged_coordinates = stage_local_template_dir(
source_dir=template_dir,
subject_label="project-render",
stage_id=stage_id,
)
print(f"INFO: Verifying staged template {staged_coordinates.full_id}")
verify_template_dir(executable=tdk_executable, template_dir=staged_dir)
print(f"INFO: Uploading staged draft {staged_coordinates.full_id}")
put_template_dir(
executable=tdk_executable,
template_dir=staged_dir,
api_url=api_url,
api_key=client.token,
)
draft_uuid = client.find_draft_uuid_by_id(staged_coordinates.full_id)
if draft_uuid is None:
raise TemplateToolError(
f"Could not resolve uploaded draft UUID for {staged_coordinates.full_id}"
)
print(f"INFO: Rendering project {project_uuid} with draft {draft_uuid}")
client.put_draft_preview_settings(
draft_uuid=draft_uuid,
format_uuid=format_uuid,
project_uuid=project_uuid,
)
download_url = client.poll_draft_preview_url(
draft_uuid=draft_uuid,
timeout_seconds=timeout_seconds,
poll_seconds=poll_seconds,
)
return (
client.download_url_bytes(download_url),
{
"mode": "draft_preview",
"source_template_id": source_coordinates.full_id,
"staged_template_id": staged_coordinates.full_id,
"draft_uuid": draft_uuid,
},
staged_dir,
)
except Exception:
if staged_dir is not None:
shutil.rmtree(staged_dir.parent, ignore_errors=True)
raise
def _render_released_template_package(
*,
client: DSWApiClient,
package_path: Path,
project_uuid: str,
format_uuid: str,
timeout_seconds: int,
poll_seconds: float,
) -> tuple[bytes, dict[str, object]]:
print(f"INFO: Importing released template package {package_path}")
template_ref = client.upload_document_template_bundle_reference(package_path)
template_label = template_ref.uuid or template_ref.template_id
print(f"INFO: Rendering project {project_uuid} with released template {template_label}")
# Bulk fixture imports can give every event the same timestamp. Rendering the
# current project avoids selecting an arbitrary, incomplete event snapshot.
created_document = client.create_document(
name=f"Render {package_path.stem}",
project_uuid=project_uuid,
document_template=template_ref,
format_uuid=format_uuid,
project_event_uuid=None,
)
document_uuid = created_document.get("uuid")
if not isinstance(document_uuid, str) or not document_uuid:
raise TemplateToolError("DSW did not return a valid document UUID")
client.poll_document_ready(
project_uuid=project_uuid,
document_uuid=document_uuid,
timeout_seconds=timeout_seconds,
poll_seconds=poll_seconds,
)
download_url = client.get_document_download_url(document_uuid)
return (
client.download_url_bytes(download_url),
{
"mode": "released_package",
"template_package": str(package_path),
"document_template_id": template_ref.template_id,
"document_template_uuid": template_ref.uuid,
"document_uuid": document_uuid,
"project_event_uuid": None,
},
)
def _write_rendered_document(
*,
output_path: Path,
document: bytes,
metadata: dict[str, object],
) -> None:
output_path.parent.mkdir(parents=True, exist_ok=True)
output_path.write_bytes(document)
info_path = output_path.with_suffix(output_path.suffix + ".json")
# Sidecars may be published as CI artifacts. Keep them useful for identifying
# the render kind and validating the payload without exposing DSW identifiers,
# instance URLs, or resolved local paths contained in the internal metadata.
public_metadata = {key: metadata[key] for key in ("mode", "bytes") if key in metadata}
info_path.write_text(
json.dumps(public_metadata, indent=2, ensure_ascii=False) + "\n",
encoding="utf-8",
)
print(f"INFO: Wrote render metadata to {info_path}")
def _resolve_or_create_project(
*,
client: DSWApiClient,
project_uuid: str | None,
project_ref: Path,
) -> ResolvedProject:
if project_uuid:
return ResolvedProject(project_uuid=project_uuid)
if project_ref.is_file():
payload = json.loads(project_ref.read_text(encoding="utf-8"))
if isinstance(payload, dict):
ref_uuid = payload.get("project_uuid") or payload.get("uuid")
if isinstance(ref_uuid, str) and ref_uuid:
return ResolvedProject(project_uuid=ref_uuid)
created_uuid = _create_project_from_ref_payload(
client=client,
project_ref=project_ref,
payload=payload,
)
return ResolvedProject(project_uuid=created_uuid, created_by_tool=True)
raise TemplateToolError(
"Set --project-uuid, DSW_PROJECT_UUID, or a valid project reference at "
f"{project_ref}. A project reference may contain `project_uuid`, or "
"`knowledge_model_package_id` plus optional `events_file`."
)
def _create_project_from_ref_payload(
*,
client: DSWApiClient,
project_ref: Path,
payload: dict[str, object],
) -> str:
project_payload = payload.get("project")
if isinstance(project_payload, dict):
seed = {
**project_payload,
**{k: v for k, v in payload.items() if k != "project"},
}
else:
seed = payload
knowledge_model_package_id = seed.get("knowledge_model_package_id")
if not isinstance(knowledge_model_package_id, str) or not knowledge_model_package_id:
raise TemplateToolError(
f"Project reference {project_ref} must define `knowledge_model_package_id` "
"when it does not point at an existing project UUID."
)
project_ref_dir = project_ref.resolve().parent
project_name = _optional_str(seed, "name") or project_ref.stem
events_file = _optional_path(seed, "events_file", base_dir=project_ref_dir)
knowledge_model_package_id = _resolve_path_value(
knowledge_model_package_id,
base_dir=project_ref_dir,
)
question_tag_uuids = seed.get("question_tag_uuids") or []
if not isinstance(question_tag_uuids, list) or not all(
isinstance(item, str) for item in question_tag_uuids
):
raise TemplateToolError(
f"Project reference {project_ref} has invalid `question_tag_uuids`."
)
print(f"INFO: Creating render project {project_name}")
created_project = client.create_project_from_package(
name=project_name,
knowledge_model_package_id=knowledge_model_package_id,
question_tag_uuids=question_tag_uuids,
visibility=_optional_str(seed, "visibility") or "PrivateProjectVisibility",
sharing=_optional_str(seed, "sharing") or "RestrictedProjectSharing",
)
created_uuid = created_project.get("uuid")
if not isinstance(created_uuid, str) or not created_uuid:
raise TemplateToolError(f"DSW did not return a valid project UUID for {project_ref}")
if events_file is not None:
try:
events = _load_project_events(events_file)
print(f"INFO: Applying {len(events)} project events from {events_file}")
client.put_project_content(project_uuid=created_uuid, events=events)
except Exception:
client.delete_project(created_uuid)
raise
return created_uuid
def _load_project_events(events_file: Path) -> list[dict[str, Any]]:
payload = json.loads(events_file.read_text(encoding="utf-8"))
events = payload.get("events") if isinstance(payload, dict) else payload
if not isinstance(events, list) or not all(isinstance(item, dict) for item in events):
raise TemplateToolError(
"Project events file must contain a JSON list or an object with an "
f"`events` list: {events_file}"
)
return events
def _optional_str(payload: dict[str, object], key: str) -> str | None:
value = payload.get(key)
return value if isinstance(value, str) and value else None
def _optional_path(payload: dict[str, object], key: str, *, base_dir: Path) -> Path | None:
value = _optional_str(payload, key)
if value is None:
return None
path = _contained_fixture_path(value, base_dir=base_dir, key=key)
if not path.is_file():
raise TemplateToolError(f"Project reference points at missing `{key}` file: {path}")
return path
def _resolve_path_value(value: str, *, base_dir: Path) -> str:
path = Path(value)
looks_like_path = path.suffix == ".km" or path.is_absolute() or value.startswith("~")
looks_like_path = looks_like_path or len(path.parts) > 1
if looks_like_path:
if path.suffix != ".km":
raise TemplateToolError(
"Local knowledge model package references must use a `.km` bundle."
)
bundle_path = _contained_fixture_path(
value,
base_dir=base_dir,
key="knowledge_model_package_id",
)
if not bundle_path.is_file():
raise TemplateToolError(
f"Project reference points at missing knowledge model bundle: {bundle_path}"
)
return str(bundle_path)
if value.count(":") == 2:
return value
try:
uuid.UUID(value)
except ValueError as exc:
raise TemplateToolError(
"Knowledge model package references must be a UUID, `org:km:version`, "
"or a relative `.km` bundle path."
) from exc
return value
def _fixture_root(base_dir: Path) -> Path:
"""Return the nearest repository fixture root, or the reference directory."""
resolved_base = base_dir.resolve()
for candidate in (resolved_base, *resolved_base.parents):
if candidate.name == "fixtures":
return candidate
return resolved_base
def _contained_fixture_path(value: str, *, base_dir: Path, key: str) -> Path:
"""Resolve a repository fixture without allowing filesystem traversal."""
path = Path(value)
if path.is_absolute() or value.startswith("~"):
raise TemplateToolError(f"Project reference `{key}` must use a relative path.")
fixture_root = _fixture_root(base_dir)
resolved_path = (base_dir.resolve() / path).resolve()
if not resolved_path.is_relative_to(fixture_root):
raise TemplateToolError(f"Project reference `{key}` must stay within {fixture_root}.")
return resolved_path