soleprint/soleprint/artery/plexuses/bundle/showcase.py

"""
Collect what the tools actually produce, at build time.

A page that lists tool names ages badly and proves nothing. This runs the real
tools over the real fixtures and hands the output to the template, so the
showcase cannot drift from the thing it is showcasing — the same reason rig's
`make up` applies the artifact it just generated rather than a parallel path.

Everything here is best-effort: a plexus that fails to build because a demo
panel could not be produced would be a bad trade, so each collector returns
None on failure and the template renders what it got.

Called by build.py::build_plexuses when a plexus ships a showcase.py exposing
collect().
"""

import importlib.util
import json
import sys
from pathlib import Path

HERE = Path(__file__).resolve().parent
SOLEPRINT = HERE.parents[2]
FIXTURES = SOLEPRINT / "station" / "tools" / "shuntgen" / "fixtures"


def _first_lines(path: Path, count: int) -> str:
    return "\n".join(path.read_text().splitlines()[:count])


def _extract(name: str, block: str, source: str) -> str:
    """Pull one class out of generated source, for showing a representative slice."""
    lines = source.splitlines()
    try:
        start = next(i for i, l in enumerate(lines) if l.startswith(f"{block} {name}"))
    except StopIteration:
        return source[:400]
    out = [lines[start]]
    for line in lines[start + 1:]:
        if line and not line.startswith((" ", "\t")):
            break
        out.append(line)
    return "\n".join(out).rstrip()


def _tabular(tmp: Path) -> dict | None:
    """modelgen + datagen over the sheet fixtures — the headline demo.

    Three spreadsheets become typed models, a graph schema and a generator that
    samples the real rows. Showing the input beside the output is the whole
    argument for the tool.
    """
    sheets = FIXTURES / "sheets"
    if not sheets.is_dir():
        return None

    sys.path.insert(0, str(SOLEPRINT))
    try:
        from station.tools.modelgen.loader.extract.tabular import TabularExtractor
        from station.tools.modelgen.generator import GENERATORS
    except ImportError:
        return None

    extractor = TabularExtractor(sheets)
    models, enums = extractor.extract()
    datasets = extractor.datasets()

    GENERATORS["pydantic"]().generate((models, enums), tmp / "models.py")
    GENERATORS["schema"]().generate((models, enums), tmp / "schema.json")
    GENERATORS["datagen"]().generate((models, enums, datasets), tmp / "gen.py")

    # Import the generated generator and let it produce records, so the JSON on
    # the page is genuinely what a caller would receive.
    records = []
    try:
        spec = importlib.util.spec_from_file_location("showcase_gen", tmp / "gen.py")
        module = importlib.util.module_from_spec(spec)
        spec.loader.exec_module(module)
        cls = next(
            v for k, v in vars(module).items()
            if isinstance(v, type) and k.endswith("Generator") and k != "BaseDataGenerator"
        )
        records = cls().generate("Customers", 2)
    except Exception:
        pass

    schema = json.loads((tmp / "schema.json").read_text())

    return {
        "inputs": [
            {"name": f.name, "rows": len(next((d.rows for d in datasets
                                               if d.source.startswith(f.name)), [])),
             "sample": _first_lines(f, 4) if f.suffix != ".ods" else None}
            for f in sorted(sheets.iterdir()) if f.is_file()
        ],
        "models": [
            {
                "name": name,
                "fields": [
                    {"name": fname, "type": fdef.get("type", "?"),
                     "pk": bool(fdef.get("pk")), "nullable": bool(fdef.get("nullable"))}
                    for fname, fdef in body.get("fields", {}).items()
                ],
            }
            for name, body in schema.get("models", {}).items()
        ],
        "pydantic": _extract("Customers", "class", (tmp / "models.py").read_text()),
        "records": records,
        "counts": {
            "files": sum(1 for f in sheets.iterdir() if f.is_file()),
            "models": len(models),
            "rows": sum(len(d.rows) for d in datasets),
        },
    }


def _shunt(tmp: Path) -> dict | None:
    """shuntgen: the same sheets, turned into routes that answer."""
    sheets = FIXTURES / "sheets"
    if not sheets.is_dir():
        return None

    sys.path.insert(0, str(SOLEPRINT))
    try:
        from station.tools.modelgen.loader.extract.tabular import TabularExtractor
        from station.tools.shuntgen.emit import ShuntEmitter
    except ImportError:
        return None

    extractor = TabularExtractor(sheets)
    models, enums = extractor.extract()
    datasets = extractor.datasets()

    emitter = ShuntEmitter(
        name="books", output=tmp / "shunt", models=models, enums=enums,
        datasets=datasets, source="sheets", kind="tabular",
    )
    collections = emitter._collections()
    routes = emitter._routes(collections)

    seeded = next((d.rows for d in datasets if d.model == "Customers"), [])
    return {
        "routes": [
            {"method": r["method"], "path": r["path"], "operation": r["operation"]}
            for r in routes
        ],
        "count": len(routes),
        # What GET /customers actually answers with — the imported rows.
        "response": seeded[:2],
    }


def _cabinet() -> dict | None:
    """A cabinet's compose fragment, as it gets merged into a compose file."""
    path = SOLEPRINT / "station" / "cabinets" / "postgres" / "service.yml"
    if not path.exists():
        return None
    body = "\n".join(
        l for l in path.read_text().splitlines() if not l.strip().startswith("#")
    ).strip()
    return {"name": "postgres", "fragment": body}


def collect() -> dict:
    """Everything the showcase renders. Never raises."""
    import tempfile

    out: dict = {}
    with tempfile.TemporaryDirectory(prefix="spr-showcase-") as raw:
        tmp = Path(raw)
        for key, fn in (("tabular", lambda: _tabular(tmp)),
                        ("shunt", lambda: _shunt(tmp)),
                        ("cabinet", lambda: _cabinet())):
            try:
                value = fn()
            except Exception as e:  # a demo panel is never worth failing a build
                print(f"  showcase: {key} unavailable ({type(e).__name__}: {e})")
                value = None
            if value:
                out[key] = value
    return out