225 lines
8.1 KiB
Python
225 lines
8.1 KiB
Python
"""Generate browsable module-directory artifacts from a release catalog."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from datetime import UTC, datetime
|
|
import json
|
|
from pathlib import Path
|
|
import re
|
|
from typing import Any
|
|
|
|
from .catalog import DEFAULT_PUBLIC_BASE_URL, canonical_hash
|
|
from .candidate_artifact import validate_release_channel
|
|
from .release_intelligence import compatibility_rows, module_rows, signature_rows
|
|
|
|
|
|
def write_module_directory(
|
|
*,
|
|
catalog_payload: dict[str, Any],
|
|
keyring_payload: dict[str, Any],
|
|
output_root: Path,
|
|
channel: str,
|
|
public_base_url: str = DEFAULT_PUBLIC_BASE_URL,
|
|
prune: bool = False,
|
|
) -> tuple[Path, ...]:
|
|
payloads = module_directory_payloads(
|
|
catalog_payload=catalog_payload,
|
|
keyring_payload=keyring_payload,
|
|
channel=channel,
|
|
public_base_url=public_base_url,
|
|
)
|
|
written: list[Path] = []
|
|
for relative_path, payload in payloads:
|
|
path = _prepare_output_path(output_root=output_root, relative_path=relative_path)
|
|
path.write_text(json.dumps(payload, indent=2, sort_keys=True) + "\n", encoding="utf-8")
|
|
written.append(path)
|
|
if prune:
|
|
_prune_stale_module_directory_files(
|
|
output_root=output_root,
|
|
expected={relative_path for relative_path, _payload in payloads},
|
|
)
|
|
return tuple(written)
|
|
|
|
|
|
def _prepare_output_path(*, output_root: Path, relative_path: Path) -> Path:
|
|
if output_root.is_symlink():
|
|
raise ValueError("module-directory output root must not be a symlink")
|
|
output_root.mkdir(parents=True, exist_ok=True)
|
|
current = output_root
|
|
for part in relative_path.parent.parts:
|
|
current = current / part
|
|
if current.is_symlink():
|
|
raise ValueError("module-directory path must not contain symlinks")
|
|
if current.exists():
|
|
if not current.is_dir():
|
|
raise ValueError("module-directory parent path must be a directory")
|
|
continue
|
|
current.mkdir()
|
|
path = output_root / relative_path
|
|
if path.is_symlink() or (path.exists() and not path.is_file()):
|
|
raise ValueError("module-directory output path must be a regular file")
|
|
return path
|
|
|
|
|
|
def _prune_stale_module_directory_files(
|
|
*,
|
|
output_root: Path,
|
|
expected: set[Path],
|
|
) -> None:
|
|
module_root = output_root / "modules"
|
|
if module_root.is_symlink():
|
|
raise ValueError("module-directory root must not be a symlink")
|
|
if not module_root.exists():
|
|
return
|
|
for path in module_root.rglob("*.json"):
|
|
relative_path = path.relative_to(output_root)
|
|
if relative_path not in expected:
|
|
path.unlink()
|
|
for path in sorted(
|
|
(item for item in module_root.rglob("*") if item.is_dir()),
|
|
key=lambda item: len(item.parts),
|
|
reverse=True,
|
|
):
|
|
try:
|
|
path.rmdir()
|
|
except OSError:
|
|
pass
|
|
|
|
|
|
def module_directory_payloads(
|
|
*,
|
|
catalog_payload: dict[str, Any],
|
|
keyring_payload: dict[str, Any],
|
|
channel: str,
|
|
public_base_url: str = DEFAULT_PUBLIC_BASE_URL,
|
|
) -> tuple[tuple[Path, dict[str, Any]], ...]:
|
|
channel = validate_release_channel(channel)
|
|
# Publication artifacts must be reproducible from the signed catalog so an
|
|
# interrupted push can later be reconciled against the immutable commit.
|
|
# Older/ad-hoc callers without a catalog timestamp retain the old fallback.
|
|
catalog_generated_at = catalog_payload.get("generated_at")
|
|
generated_at = (
|
|
catalog_generated_at
|
|
if isinstance(catalog_generated_at, str) and catalog_generated_at
|
|
else json_datetime(datetime.now(tz=UTC))
|
|
)
|
|
modules = module_rows(catalog_payload)
|
|
compatibility = compatibility_rows(modules)
|
|
signatures = signature_rows(catalog_payload, keyring_payload)
|
|
base_url = public_base_url.rstrip("/")
|
|
catalog_url = f"{base_url}/catalogs/v1/channels/{channel}.json"
|
|
keyring_url = f"{base_url}/catalogs/v1/keyring.json"
|
|
catalog_hash = canonical_hash(catalog_payload)
|
|
catalog_sequence = catalog_payload.get("sequence")
|
|
|
|
files: list[tuple[Path, dict[str, Any]]] = []
|
|
module_index_entries: list[dict[str, Any]] = []
|
|
for module in modules:
|
|
module_id = str(module.get("module_id") or module.get("repo") or "")
|
|
if not module_id:
|
|
continue
|
|
version = str(module.get("version") or "0.0.0")
|
|
module_slug = safe_module_id(module_id)
|
|
version_slug = safe_version(version)
|
|
module_base = f"{base_url}/catalogs/v1/modules/{module_slug}"
|
|
manifest_url = f"{module_base}/{version_slug}/manifest.json"
|
|
module_index_url = f"{module_base}/index.json"
|
|
module_compatibility = [row for row in compatibility if row.get("module_id") == module_id]
|
|
manifest = {
|
|
"manifest_version": "1",
|
|
"generated_at": generated_at,
|
|
"channel": channel,
|
|
"module": module,
|
|
"compatibility": module_compatibility,
|
|
"release": {
|
|
"catalog_url": catalog_url,
|
|
"keyring_url": keyring_url,
|
|
"catalog_hash": catalog_hash,
|
|
"catalog_sequence": catalog_sequence,
|
|
"signatures": signatures,
|
|
},
|
|
}
|
|
files.append((Path("modules") / module_slug / version_slug / "manifest.json", manifest))
|
|
files.append(
|
|
(
|
|
Path("modules") / module_slug / "index.json",
|
|
{
|
|
"index_version": "1",
|
|
"generated_at": generated_at,
|
|
"channel": channel,
|
|
"module_id": module_id,
|
|
"name": module.get("name"),
|
|
"latest_version": version,
|
|
"latest_manifest_url": manifest_url,
|
|
"versions": [
|
|
{
|
|
"version": version,
|
|
"manifest_url": manifest_url,
|
|
"python_tag": module.get("python_tag"),
|
|
"webui_tag": module.get("webui_tag"),
|
|
}
|
|
],
|
|
"release": {
|
|
"catalog_url": catalog_url,
|
|
"keyring_url": keyring_url,
|
|
"catalog_hash": catalog_hash,
|
|
"catalog_sequence": catalog_sequence,
|
|
},
|
|
},
|
|
)
|
|
)
|
|
module_index_entries.append(
|
|
{
|
|
"module_id": module_id,
|
|
"name": module.get("name"),
|
|
"latest_version": version,
|
|
"index_url": module_index_url,
|
|
"latest_manifest_url": manifest_url,
|
|
"repo": module.get("repo"),
|
|
}
|
|
)
|
|
|
|
files.append(
|
|
(
|
|
Path("modules") / "index.json",
|
|
{
|
|
"index_version": "1",
|
|
"generated_at": generated_at,
|
|
"channel": channel,
|
|
"catalog_url": catalog_url,
|
|
"keyring_url": keyring_url,
|
|
"catalog_hash": catalog_hash,
|
|
"catalog_sequence": catalog_sequence,
|
|
"module_count": len(module_index_entries),
|
|
"modules": sorted(module_index_entries, key=lambda item: str(item.get("module_id"))),
|
|
},
|
|
)
|
|
)
|
|
return tuple(files)
|
|
|
|
|
|
def safe_path_part(value: str) -> str:
|
|
if (
|
|
not isinstance(value, str)
|
|
or value in {".", ".."}
|
|
or re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9_.+!-]{0,127}", value) is None
|
|
):
|
|
raise ValueError("module-directory path part is not canonical")
|
|
return value
|
|
|
|
|
|
def safe_module_id(value: str) -> str:
|
|
if re.fullmatch(r"[a-z][a-z0-9_-]{0,63}", value) is None:
|
|
raise ValueError("module ID is not canonical")
|
|
return safe_path_part(value)
|
|
|
|
|
|
def safe_version(value: str) -> str:
|
|
if re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9._+!-]{0,127}", value) is None:
|
|
raise ValueError("module version is not canonical")
|
|
return safe_path_part(value)
|
|
|
|
|
|
def json_datetime(value: datetime) -> str:
|
|
return value.replace(microsecond=0).isoformat().replace("+00:00", "Z")
|