app-platform: generate catalog from app manifests
This commit is contained in:
@@ -0,0 +1,173 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Report drift between app-catalog/catalog.json and apps/*/manifest.yml."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import json
|
||||
import sys
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
import yaml
|
||||
|
||||
INTERNAL_MANIFEST_IDS = {
|
||||
"aiui",
|
||||
"archy-btcpay-db",
|
||||
"archy-mempool-db",
|
||||
"archy-mempool-web",
|
||||
"archy-nbxplorer",
|
||||
"bitcoin-ui",
|
||||
"core-lightning",
|
||||
"did-wallet",
|
||||
"electrs-ui",
|
||||
"lightning-stack",
|
||||
"lnd-ui",
|
||||
"mempool-api",
|
||||
"morphos-server",
|
||||
"onlyoffice",
|
||||
"router",
|
||||
"strfry",
|
||||
"web5-dwn",
|
||||
}
|
||||
|
||||
LEGACY_STACK_CATALOG_IDS = {
|
||||
"immich",
|
||||
"netbird",
|
||||
"saleor",
|
||||
"tailscale",
|
||||
}
|
||||
|
||||
|
||||
def load_catalog(path: Path) -> dict[str, dict[str, Any]]:
|
||||
with path.open("r", encoding="utf-8") as fh:
|
||||
data = json.load(fh)
|
||||
apps = data.get("apps", [])
|
||||
if not isinstance(apps, list):
|
||||
raise ValueError(f"{path}: expected .apps to be a list")
|
||||
return {str(app.get("id", "")): app for app in apps if isinstance(app, dict) and app.get("id")}
|
||||
|
||||
|
||||
def load_manifests(apps_dir: Path) -> dict[str, dict[str, Any]]:
|
||||
manifests: dict[str, dict[str, Any]] = {}
|
||||
for path in sorted(apps_dir.glob("*/manifest.yml")):
|
||||
with path.open("r", encoding="utf-8") as fh:
|
||||
data = yaml.safe_load(fh)
|
||||
if not isinstance(data, dict) or not isinstance(data.get("app"), dict):
|
||||
continue
|
||||
app = data["app"]
|
||||
app_id = app.get("id")
|
||||
if app_id:
|
||||
manifests[str(app_id)] = {"path": str(path), "app": app}
|
||||
return manifests
|
||||
|
||||
|
||||
def metadata(app: dict[str, Any]) -> dict[str, Any]:
|
||||
value = app.get("metadata")
|
||||
return value if isinstance(value, dict) else {}
|
||||
|
||||
|
||||
def manifest_value(app: dict[str, Any], field: str) -> Any:
|
||||
meta = metadata(app)
|
||||
container = app.get("container") if isinstance(app.get("container"), dict) else {}
|
||||
match field:
|
||||
case "title":
|
||||
return app.get("name")
|
||||
case "version":
|
||||
return str(app.get("version", ""))
|
||||
case "description":
|
||||
return app.get("description")
|
||||
case "dockerImage":
|
||||
return container.get("image")
|
||||
case "category":
|
||||
return app.get("category") or meta.get("category")
|
||||
case "tier":
|
||||
return meta.get("tier")
|
||||
case "icon":
|
||||
return meta.get("icon")
|
||||
case "repoUrl":
|
||||
return meta.get("repo") or meta.get("repoUrl")
|
||||
case _:
|
||||
return None
|
||||
|
||||
|
||||
def normalize(value: Any) -> str:
|
||||
if value is None:
|
||||
return ""
|
||||
return str(value).strip()
|
||||
|
||||
|
||||
def main() -> int:
|
||||
parser = argparse.ArgumentParser(description=__doc__)
|
||||
parser.add_argument("--catalog", default="app-catalog/catalog.json")
|
||||
parser.add_argument("--apps-dir", default="apps")
|
||||
parser.add_argument(
|
||||
"--strict",
|
||||
action="store_true",
|
||||
help="exit non-zero when missing entries or metadata drift are found",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--release",
|
||||
action="store_true",
|
||||
help="suppress known internal/legacy-stack entries so output is release-actionable",
|
||||
)
|
||||
args = parser.parse_args()
|
||||
|
||||
catalog = load_catalog(Path(args.catalog))
|
||||
manifests = load_manifests(Path(args.apps_dir))
|
||||
|
||||
catalog_ids = set(catalog)
|
||||
manifest_ids = set(manifests)
|
||||
missing_manifests = sorted(catalog_ids - manifest_ids)
|
||||
missing_catalog = sorted(manifest_ids - catalog_ids)
|
||||
if args.release:
|
||||
missing_manifests = [app_id for app_id in missing_manifests if app_id not in LEGACY_STACK_CATALOG_IDS]
|
||||
missing_catalog = [app_id for app_id in missing_catalog if app_id not in INTERNAL_MANIFEST_IDS]
|
||||
|
||||
compared_fields = [
|
||||
"title",
|
||||
"version",
|
||||
"description",
|
||||
"dockerImage",
|
||||
"category",
|
||||
"tier",
|
||||
"icon",
|
||||
"repoUrl",
|
||||
]
|
||||
drift: list[str] = []
|
||||
for app_id in sorted(catalog_ids & manifest_ids):
|
||||
catalog_app = catalog[app_id]
|
||||
manifest_app = manifests[app_id]["app"]
|
||||
for field in compared_fields:
|
||||
catalog_val = normalize(catalog_app.get(field))
|
||||
manifest_val = normalize(manifest_value(manifest_app, field))
|
||||
if catalog_val and manifest_val and catalog_val != manifest_val:
|
||||
drift.append(f"{app_id}: {field}: catalog={catalog_val!r} manifest={manifest_val!r}")
|
||||
|
||||
print(
|
||||
json.dumps(
|
||||
{
|
||||
"catalog_apps": len(catalog),
|
||||
"manifest_apps": len(manifests),
|
||||
"missing_manifests": len(missing_manifests),
|
||||
"missing_catalog": len(missing_catalog),
|
||||
"metadata_drift": len(drift),
|
||||
},
|
||||
sort_keys=True,
|
||||
)
|
||||
)
|
||||
|
||||
for app_id in missing_manifests:
|
||||
print(f"MISSING_MANIFEST {app_id}")
|
||||
for app_id in missing_catalog:
|
||||
print(f"MISSING_CATALOG {app_id}")
|
||||
for item in drift:
|
||||
print(f"DRIFT {item}")
|
||||
|
||||
if args.strict and (missing_manifests or missing_catalog or drift):
|
||||
return 1
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main())
|
||||
Reference in New Issue
Block a user