Files
soleprint/build.py
2026-08-11 07:30:27 -03:00

842 lines
29 KiB
Python

#!/usr/bin/env python3
"""
Soleprint Build Tool
Generates soleprint instances from source + room config.
Usage:
python build.py # Build gen/standalone/
python build.py --cfg <room> # Build gen/<room>/
python build.py --all # Build all rooms
python build.py --output /path/ # Build to custom path
python build.py --models # Only regenerate models
Generated structure for standalone rooms:
gen/standalone/
artery/, atlas/, station/, main.py, ...
Generated structure for managed rooms:
gen/<room>/
<managed_name>/ # Copied repos + ctrl
link/ # Bridge code
soleprint/ # Soleprint instance
"""
import argparse
import importlib.util
import json
import logging
import shutil
import stat
import subprocess
import sys
from pathlib import Path
SPR_ROOT = Path(__file__).resolve().parent
logging.basicConfig(level=logging.INFO, format="%(message)s")
log = logging.getLogger(__name__)
def load_config(cfg_name: str | None) -> dict:
"""Load room config.json."""
room = cfg_name or "standalone"
config_path = SPR_ROOT / "cfg" / room / "config.json"
if config_path.exists():
return json.loads(config_path.read_text())
return {}
def ensure_dir(path: Path):
path.mkdir(parents=True, exist_ok=True)
def _rmtree_resilient(path: Path):
"""Remove path tree, tolerating root-owned files written by containers.
Docker containers that mount gen/ as a volume sometimes write files as
root (e.g. __pycache__). A plain shutil.rmtree then fails with EACCES.
We first try shutil.rmtree; if that hits a PermissionError we fall back
to deleting the offending files from inside an ephemeral alpine container.
"""
def _chmod_and_retry(func, target, exc_info):
try:
Path(target).chmod(stat.S_IWUSR | stat.S_IRUSR | stat.S_IXUSR)
func(target)
except Exception:
raise
try:
shutil.rmtree(path, onerror=_chmod_and_retry)
return
except PermissionError:
pass
log.info(" (falling back to docker-based cleanup)")
subprocess.run(
["docker", "run", "--rm", "-v", f"{path.parent}:/work",
"alpine:3", "sh", "-c", f"rm -rf /work/{path.name}"],
check=True,
)
def copy_path(source: Path, target: Path, quiet: bool = False):
"""Copy file or directory, resolving symlinks."""
if target.is_symlink():
target.unlink()
elif target.exists():
if target.is_dir():
shutil.rmtree(target)
else:
target.unlink()
if source.is_dir():
shutil.copytree(source, target, symlinks=False)
if not quiet:
log.info(f" {target.name}/")
else:
shutil.copy2(source, target)
if not quiet:
log.info(f" {target.name}")
def count_files(path: Path) -> int:
return sum(1 for _ in path.rglob("*") if _.is_file())
def merge_into(source: Path, target: Path):
"""Merge source directory into target (additive, overwrites conflicts)."""
if not source.exists():
return
for item in source.rglob("*"):
if item.is_file():
rel = item.relative_to(source)
dest = target / rel
dest.parent.mkdir(parents=True, exist_ok=True)
shutil.copy2(item, dest)
def parse_gitignore(gitignore_path: Path) -> set[str]:
"""Parse .gitignore and return set of patterns to ignore."""
patterns = set()
if not gitignore_path.exists():
return patterns
for line in gitignore_path.read_text().splitlines():
line = line.strip()
# Skip comments and empty lines
if not line or line.startswith("#"):
continue
# Remove trailing slashes (directory indicators)
pattern = line.rstrip("/")
# Skip negation patterns (we don't support them)
if pattern.startswith("!"):
continue
patterns.add(pattern)
return patterns
def copy_repo(source: Path, target: Path):
"""Copy a repo directory, excluding .git and gitignored files."""
if not source.exists():
log.warning(f"Repo not found: {source}")
return False
# Always ignore these
always_ignore = {".git", "__pycache__", "node_modules", ".venv", "venv", ".env"}
# Parse .gitignore from repo root
gitignore_patterns = parse_gitignore(source / ".gitignore")
def ignore_patterns(directory, files):
ignored = set()
rel_dir = Path(directory).relative_to(source)
for f in files:
# Always ignore these
if f in always_ignore:
ignored.add(f)
continue
# Check gitignore patterns
rel_path = rel_dir / f if str(rel_dir) != "." else Path(f)
for pattern in gitignore_patterns:
# Simple pattern matching (filename or extension)
if pattern.startswith("*."):
# Extension pattern like *.pyc
if f.endswith(pattern[1:]):
ignored.add(f)
break
elif pattern == f or pattern == str(rel_path):
# Exact match
ignored.add(f)
break
elif "/" not in pattern and f == pattern:
# Simple name match anywhere
ignored.add(f)
break
return ignored
shutil.copytree(source, target, ignore=ignore_patterns, symlinks=False)
return True
def build_managed(output_dir: Path, cfg_name: str, config: dict):
"""Build managed folder with repos + ctrl."""
managed = config.get("managed", {})
managed_name = managed.get("name", cfg_name)
repos = managed.get("repos", {})
managed_dir = output_dir / managed_name
ensure_dir(managed_dir)
log.info(f"Building managed ({managed_name})...")
# Copy repos (relative paths resolve from SPR_ROOT)
for repo_name, repo_path in repos.items():
source = Path(repo_path)
if not source.is_absolute():
source = SPR_ROOT / source
target = managed_dir / repo_name
if copy_repo(source, target):
log.info(f" {repo_name}/")
room_cfg = SPR_ROOT / "cfg" / cfg_name
# Docker files from room root -> managed root
for item in room_cfg.iterdir():
if item.is_file() and (
item.name.startswith("Dockerfile") or item.name.startswith("docker-compose")
):
copy_path(item, managed_dir / item.name)
# Copy managed app config from cfg/<room>/<managed_name>/ (e.g., .env, dumps/)
room_managed_cfg = room_cfg / managed_name
if room_managed_cfg.exists():
log.info(f" Copying {managed_name} config...")
for item in room_managed_cfg.iterdir():
if item.is_file():
copy_path(item, managed_dir / item.name, quiet=True)
elif item.is_dir():
target = managed_dir / item.name
if target.exists():
# Merge into existing repo directory
merge_into(item, target)
else:
copy_path(item, target)
# Scripts from ctrl/ -> output_dir/ctrl/ (sibling of managed, link, soleprint)
room_ctrl = room_cfg / "ctrl"
if room_ctrl.exists():
ctrl_dir = output_dir / "ctrl"
ensure_dir(ctrl_dir)
for item in room_ctrl.iterdir():
if item.is_file():
copy_path(item, ctrl_dir / item.name)
def build_link(output_dir: Path, cfg_name: str):
"""Build link folder."""
room_cfg = SPR_ROOT / "cfg" / cfg_name
link_source = room_cfg / "link"
if not link_source.exists():
return
log.info("Building link...")
link_dir = output_dir / "link"
copy_path(link_source, link_dir)
def generate_models(output_dir: Path, room: str):
"""Generate models using modelgen tool."""
from soleprint.station.tools.modelgen import ModelGenerator, load_config
config_path = SPR_ROOT / "cfg" / room / "config.json"
if not config_path.exists():
log.warning(f"Config not found: {config_path}")
return False
models_file = output_dir / "models" / "pydantic" / "__init__.py"
models_file.parent.mkdir(parents=True, exist_ok=True)
try:
config = load_config(config_path)
generator = ModelGenerator(
config=config,
output_path=models_file,
output_format="pydantic",
)
generator.generate()
return True
except Exception as e:
log.error(f"Model generation failed: {e}")
return False
def copy_cfg(output_dir: Path, room: str):
"""Copy room configuration files to output directory."""
room_cfg = SPR_ROOT / "cfg" / room
if not room_cfg.exists():
log.warning(f"Room config not found: {room}")
return
log.info(f"Copying {room} config...")
# config.json -> cfg/
cfg_dir = output_dir / "cfg"
ensure_dir(cfg_dir)
if (room_cfg / "config.json").exists():
copy_path(room_cfg / "config.json", cfg_dir / "config.json")
# data/ -> data/
if (room_cfg / "data").exists():
copy_path(room_cfg / "data", output_dir / "data")
# .env.example
if (room_cfg / ".env.example").exists():
copy_path(room_cfg / ".env.example", output_dir / ".env.example")
# Room-specific soleprint config (docker-compose.yml, etc)
# Now in cfg/<room>/soleprint/
room_soleprint = room_cfg / "soleprint"
if room_soleprint.exists():
systems = {"artery", "atlas", "station"}
for item in room_soleprint.iterdir():
if item.name in systems:
# Merge system dirs into already-copied framework code
log.info(f" Merging {room} {item.name}...")
merge_into(item, output_dir / item.name)
elif item.is_file():
copy_path(item, output_dir / item.name)
elif item.is_dir():
# Copy non-system dirs as-is (nginx/, etc.)
copy_path(item, output_dir / item.name)
def load_cabinets(room: str) -> list[dict]:
"""The dependency containers a room asked for, in the order it listed them.
Read from cfg/<room>/data/cabinets.json — the same shape and the same place
as its sibling data/*.json files, so nothing new has to know about it.
Entries are {"name": "postgres"} and may carry an "env" override.
"""
path = SPR_ROOT / "cfg" / room / "data" / "cabinets.json"
if not path.exists():
return []
try:
entries = json.loads(path.read_text())
except ValueError as e:
log.warning(f" cabinets.json is not valid JSON, ignoring: {e}")
return []
out = []
for entry in entries if isinstance(entries, list) else []:
if isinstance(entry, str):
entry = {"name": entry}
if isinstance(entry, dict) and entry.get("name"):
out.append(entry)
return out
def resolve_cabinets(requested: list[dict]) -> list[dict]:
"""Expand each request into its definition, pulling in what it depends on.
Airflow without postgres is a container that exits on boot, so a cabinet's
depends_on is added for you rather than left as something to remember.
"""
cabinets_dir = SPR_ROOT / "soleprint" / "station" / "cabinets"
resolved: dict[str, dict] = {}
def add(name: str, overrides: dict) -> None:
if name in resolved:
# Already pulled in as somebody's dependency. The room asking for it
# by name is the more specific statement, so its env still applies —
# otherwise declaring airflow before postgres would silently drop
# postgres's settings.
if overrides.get("env"):
resolved[name]["env"] = {
**resolved[name].get("env", {}),
**overrides["env"],
}
return
definition_path = cabinets_dir / name / "cabinet.json"
if not definition_path.exists():
available = sorted(
p.name for p in cabinets_dir.iterdir() if p.is_dir()
) if cabinets_dir.exists() else []
log.warning(f" no such cabinet: {name} (available: {', '.join(available) or 'none'})")
return
try:
definition = json.loads(definition_path.read_text())
except ValueError as e:
log.warning(f" cabinet {name} has invalid cabinet.json: {e}")
return
# Mark it claimed before recursing, so a dependency cycle terminates.
resolved[name] = definition
for dependency in definition.get("depends_on", []) or []:
add(dependency, {})
definition["env"] = {**definition.get("env", {}), **overrides.get("env", {})}
for entry in requested:
add(entry["name"], entry)
# Dependencies first, so compose reads in the order things start.
ordered: list[dict] = []
seen: set[str] = set()
def emit(name: str) -> None:
if name in seen or name not in resolved:
return
seen.add(name)
for dependency in resolved[name].get("depends_on", []) or []:
emit(dependency)
ordered.append(resolved[name])
for name in resolved:
emit(name)
return ordered
def compose_cabinets(output_dir: Path, room: str):
"""Merge the room's cabinets into its docker-compose.yml and .env.example.
This is the compile step for dependencies: a room declares postgres, and the
built instance comes out with postgres in its compose file rather than with
instructions for adding it.
"""
requested = load_cabinets(room)
if not requested:
return
try:
import yaml
except ImportError:
log.warning(
" cabinets need PyYAML to merge into docker-compose.yml "
"(pip install pyyaml) — skipping"
)
return
cabinets = resolve_cabinets(requested)
if not cabinets:
return
compose_path = output_dir / "docker-compose.yml"
if not compose_path.exists():
log.warning(
f" no docker-compose.yml in {output_dir.name}, "
f"so there is nothing to merge {len(cabinets)} cabinet(s) into"
)
return
original = compose_path.read_text()
# A YAML round-trip drops every comment, and the room's compose file leads
# with the one that says how to run it. Keep the header block; the rest is
# generated anyway.
header = []
for line in original.splitlines():
if line.startswith("#") or not line.strip():
header.append(line)
else:
break
while header and not header[-1].strip():
header.pop()
compose = yaml.safe_load(original) or {}
services = compose.setdefault("services", {})
volumes = compose.setdefault("volumes", {}) or {}
cabinets_dir = SPR_ROOT / "soleprint" / "station" / "cabinets"
added, skipped = [], []
for cabinet in cabinets:
name = cabinet["name"]
service_name = cabinet.get("service", name)
# The room's own compose file is the authority. A room that already
# declares `db` has arranged it deliberately, and silently replacing it
# would be the worst possible outcome of switching a cabinet on.
if service_name in services:
skipped.append(service_name)
continue
fragment_path = cabinets_dir / name / "service.yml"
if not fragment_path.exists():
log.warning(f" cabinet {name} has no service.yml")
continue
fragment = yaml.safe_load(fragment_path.read_text()) or {}
for key, value in fragment.items():
if key in services:
skipped.append(key)
continue
services[key] = value
added.append(key)
for volume in cabinet.get("volumes", []) or []:
volumes.setdefault(volume, None)
if volumes:
compose["volumes"] = volumes
rendered = yaml.safe_dump(compose, sort_keys=False, default_flow_style=False)
banner = f"# Cabinets merged in by build.py: {', '.join(c['name'] for c in cabinets)}.\n"
preamble = ("\n".join(header) + "\n" + banner + "\n") if header else banner + "\n"
compose_path.write_text(preamble + rendered)
if added:
log.info(f" cabinets: {', '.join(added)}")
if skipped:
log.info(f" cabinets already declared by the room, left alone: {', '.join(skipped)}")
_append_cabinet_env(output_dir, cabinets)
def _append_cabinet_env(output_dir: Path, cabinets: list[dict]):
"""Add each cabinet's settings to .env.example, without touching .env."""
example = output_dir / ".env.example"
existing = example.read_text() if example.exists() else ""
# Match whole settings, not substrings: `POSTGRES_DB=` appears inside
# `MY_POSTGRES_DB=`, and a substring test would decide the setting was
# already there and skip it.
declared = {
line.split("=", 1)[0].strip()
for line in existing.splitlines()
if "=" in line and not line.lstrip().startswith("#")
}
lines = []
for cabinet in cabinets:
env = cabinet.get("env", {})
if not env:
continue
block = [f"\n# ── {cabinet.get('title', cabinet['name'])} (cabinet) ──"]
for note in cabinet.get("notes", []) or []:
block.append(f"# {note}")
wrote = False
for key, value in env.items():
if key in declared:
continue
block.append(f"{key}={value}")
declared.add(key)
wrote = True
if wrote:
lines.extend(block)
if lines:
example.write_text(existing.rstrip("\n") + "\n" + "\n".join(lines) + "\n")
def load_plexuses(room: str) -> list[dict]:
"""The plexuses a room asked for. Same shape as its sibling data/*.json."""
path = SPR_ROOT / "cfg" / room / "data" / "plexuses.json"
if not path.exists():
return []
try:
raw = json.loads(path.read_text())
except ValueError as e:
log.warning(f" plexuses.json is not valid JSON, ignoring: {e}")
return []
entries = raw.get("items", raw) if isinstance(raw, dict) else raw
out = []
for entry in entries if isinstance(entries, list) else []:
if isinstance(entry, str):
entry = {"name": entry}
if isinstance(entry, dict) and entry.get("name"):
out.append(entry)
return out
def _theme_css(theme: str) -> str:
"""The token contract plus one theme, flattened for inlining.
Only the named theme ships alongside the others it can switch to, because
the export has to work with no server: there is no /theme.css to fetch.
"""
theme_dir = SPR_ROOT / "soleprint" / "common" / "theme"
parts = []
tokens = theme_dir / "tokens.css"
if tokens.exists():
parts.append(tokens.read_text())
# Every theme, so the switcher in the page has something to switch to.
for sheet in sorted((theme_dir / "themes").glob("*.css")):
parts.append(sheet.read_text())
return "\n".join(parts)
def _inline_svg(name: str, theme: str) -> str:
"""A rendered graph, stripped of its XML prolog so it can sit in HTML.
Inlined rather than <img>-linked so the page's CSS can recolour it when the
theme switches — graphviz writes class="node accent" into the SVG, and CSS
outranks the presentation attributes it bakes in.
"""
graphs = SPR_ROOT / "docs" / "graphs"
for candidate in (graphs / f"{name}.{theme}.svg", graphs / f"{name}.svg"):
if candidate.exists():
svg = candidate.read_text()
start = svg.find("<svg")
return svg[start:] if start >= 0 else svg
log.warning(f" no rendered graph '{name}' — run docs/graphs/render.sh")
return "<p>diagram not rendered</p>"
def build_plexuses(output_dir: Path, room: str):
"""Export each plexus the room declared to a single self-contained file.
A plexus is exported, not served. The output is one index.html carrying its
theme, its data and its diagram, so it survives a locked-down machine, a zip
attachment and a double-click — which is the whole point of the format.
"""
requested = load_plexuses(room)
if not requested:
return
source_root = SPR_ROOT / "soleprint" / "artery" / "plexuses"
built = []
for entry in requested:
name = entry["name"]
source = source_root / name
manifest_path = source / "plexus.json"
if not manifest_path.exists():
available = sorted(
p.name for p in source_root.iterdir() if p.is_dir()
) if source_root.exists() else []
log.warning(
f" no such plexus: {name} (available: {', '.join(available) or 'none'})"
)
continue
try:
manifest = json.loads(manifest_path.read_text())
except ValueError as e:
log.warning(f" plexus {name} has invalid plexus.json: {e}")
continue
# The room may override anything the plexus declares — theme first.
manifest.update({k: v for k, v in entry.items() if k != "name"})
template_path = source / "app" / "index.html"
if not template_path.exists():
log.warning(f" plexus {name} has no app/index.html")
continue
theme = manifest.get("theme", "soleprint")
data = {k: v for k, v in manifest.items() if not k.startswith("_")}
# A plexus may ship a showcase.py exposing collect(): anything it returns
# is merged into the page's data. The bundle uses it to run the real
# tools over the real fixtures at build time, so what the page shows
# cannot drift from what the tools do.
collector = source / "showcase.py"
if collector.exists():
try:
spec = importlib.util.spec_from_file_location(
f"plexus_{name}_showcase", collector
)
module = importlib.util.module_from_spec(spec)
spec.loader.exec_module(module)
data["showcase"] = module.collect()
except Exception as e:
log.warning(f" {name}: showcase.py failed ({type(e).__name__}: {e})")
page = template_path.read_text()
for token, value in (
("%%TITLE%%", manifest.get("title", name)),
("%%DESCRIPTION%%", manifest.get("description", "")),
("%%DEFAULT_THEME%%", theme),
("%%BUILT%%", f"{room} · built by soleprint build.py"),
("%%THEME_CSS%%", _theme_css(theme)),
("%%GRAPH%%", _inline_svg(manifest.get("graph", "system_overview"), theme)),
("%%BUNDLE%%", json.dumps(data, indent=2)),
):
page = page.replace(token, value)
target = output_dir / "plexuses" / name
ensure_dir(target)
(target / "index.html").write_text(page)
# Anything else in app/ rides along, for a plexus that outgrows one file.
for extra in (source / "app").iterdir():
if extra.name != "index.html":
copy_path(extra, target / extra.name, quiet=True)
built.append(f"{name} ({theme})")
if built:
log.info(f" plexuses: {', '.join(built)}")
def build_soleprint(output_dir: Path, room: str):
"""Build soleprint folder with core + room config merged."""
soleprint = SPR_ROOT / "soleprint"
# Soleprint core files
log.info("Copying soleprint core...")
for name in [
"main.py",
"run.py",
"index.html",
"requirements.txt",
"Dockerfile",
]:
if (soleprint / name).exists():
copy_path(soleprint / name, output_dir / name)
copy_path(soleprint / "dataloader", output_dir / "dataloader")
# System directories
log.info("Copying systems...")
for system in ["artery", "atlas", "station"]:
source = soleprint / system
if source.exists():
copy_path(source, output_dir / system)
# Common modules (auth, etc)
if (soleprint / "common").exists():
copy_path(soleprint / "common", output_dir / "common")
# Room config (includes merging room-specific artery/atlas/station)
copy_cfg(output_dir, room)
# Dependency containers the room asked for, merged into its compose file.
# After copy_cfg, because the compose file being merged into is the room's.
log.info("Composing cabinets...")
compose_cabinets(output_dir, room)
# Plexuses are exported rather than served, so this is a compile step like
# the cabinet merge above — not something run.py does at request time.
log.info("Exporting plexuses...")
build_plexuses(output_dir, room)
# Generate models
log.info("Generating models...")
if not generate_models(output_dir, room):
log.warning("Model generation failed")
def build(output_dir: Path, cfg_name: str | None = None, clean: bool = True):
"""Build complete room instance."""
room = cfg_name or "standalone"
config = load_config(cfg_name)
managed = config.get("managed")
log.info(f"\n=== Building {room} ===")
# Clean output directory first
if clean and output_dir.exists():
log.info(f"Cleaning {output_dir}...")
_rmtree_resilient(output_dir)
ensure_dir(output_dir)
if managed:
# 3-folder structure: <managed>/, link/, soleprint/
build_managed(output_dir, room, config)
build_link(output_dir, room)
soleprint_name = config.get("framework", {}).get("name", "soleprint")
soleprint_dir = output_dir / soleprint_name
ensure_dir(soleprint_dir)
build_soleprint(soleprint_dir, room)
else:
# Standalone: everything in output_dir
build_soleprint(output_dir, room)
# Layer 7 (optional): render kind-cluster manifests
try:
from soleprint.ctrl.k8s import render_k8s
from soleprint.ctrl.k8s.render import k8s_enabled
if k8s_enabled(config):
render_k8s(room=room, config=config, gen_dir=output_dir)
except ImportError as e:
log.warning(f"k8s rendering unavailable: {e}")
log.info(f"\n✓ Built: {output_dir}")
def build_models_only():
"""Only regenerate models."""
log.info("Generating models...")
if generate_models(SPR_ROOT / "gen"):
log.info("✓ Models generated")
else:
log.error("Model generation failed")
sys.exit(1)
def build_plexuses_only(room: str):
"""Compile just the plexus UIs, without rebuilding the room around them.
The equivalent of `vite build` for this repo: the iteration loop when you
are working on the UI itself is edit, compile, reopen the file — and a full
room build to see a CSS change is a slow way to do that.
"""
output_dir = SPR_ROOT / "gen" / room
if not output_dir.exists():
log.error(f"Room '{room}' is not built — run: python build.py --cfg {room}")
sys.exit(1)
log.info(f"Compiling plexus UIs for {room}...")
build_plexuses(output_dir, room)
built = sorted((output_dir / "plexuses").glob("*/index.html"))
if not built:
log.warning(
f" nothing compiled — does cfg/{room}/data/plexuses.json list one?"
)
return
for page in built:
log.info(f" {page.relative_to(SPR_ROOT)} ({page.stat().st_size // 1024} KB)")
log.info("\n✓ Open directly — no server needed:")
log.info(f" xdg-open {built[0]}")
def main():
parser = argparse.ArgumentParser(description="Soleprint Build Tool")
parser.add_argument("--output", "-o", type=Path, help="Output directory")
parser.add_argument("--cfg", "-c", type=str, help="Room config name")
parser.add_argument("--all", action="store_true", help="Build all rooms")
parser.add_argument("--models", action="store_true", help="Only regenerate models")
parser.add_argument(
"--plexuses",
action="store_true",
help="Only compile the plexus UIs into an already-built room",
)
args = parser.parse_args()
if args.plexuses:
build_plexuses_only(args.cfg or "standalone")
elif args.models:
build_models_only()
elif args.all:
build(SPR_ROOT / "gen" / "standalone", None)
for room in (SPR_ROOT / "cfg").iterdir():
# cfg/ is itself a git repo and rooms may carry dot-dirs — skip them,
# or --all tries to build ".git" as a room.
if room.name.startswith(".") or room.name == "__pycache__":
continue
if room.is_dir() and room.name != "standalone":
build(SPR_ROOT / "gen" / room.name, room.name)
else:
if args.output:
output_dir = args.output.resolve()
elif args.cfg:
output_dir = SPR_ROOT / "gen" / args.cfg
else:
output_dir = SPR_ROOT / "gen" / "standalone"
build(output_dir, args.cfg)
if __name__ == "__main__":
main()