638 lines
20 KiB
Python
638 lines
20 KiB
Python
"""
|
|
Soleprint - Bare-metal single-port development server.
|
|
|
|
Serves everything on a single port for basic testing without docker/nginx.
|
|
Routes /artery/*, /atlas/*, /station/* internally instead of redirecting.
|
|
|
|
Usage:
|
|
python run.py # Serves on :12000 with all subsystems
|
|
PORT=8080 python run.py # Custom port
|
|
|
|
This is for soleprint development only, not for managed rooms (use docker for those).
|
|
"""
|
|
|
|
import importlib.util
|
|
import json
|
|
import os
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
# Add current directory to path for imports (artery, atlas, station)
|
|
sys.path.insert(0, str(Path(__file__).parent))
|
|
|
|
from fastapi import FastAPI, Request
|
|
from fastapi.responses import HTMLResponse, JSONResponse
|
|
from fastapi.staticfiles import StaticFiles
|
|
from fastapi.templating import Jinja2Templates
|
|
|
|
app = FastAPI(title="Soleprint (dev)", version="0.1.0")
|
|
|
|
templates = Jinja2Templates(directory=Path(__file__).parent)
|
|
|
|
# Base path for systems (gen/ directory where this runs from)
|
|
SPR_ROOT = Path(__file__).parent
|
|
DATA_DIR = SPR_ROOT / "data"
|
|
CFG_DIR = SPR_ROOT / "cfg"
|
|
|
|
|
|
# ============================================================================
|
|
# Vein Loading
|
|
# ============================================================================
|
|
|
|
# Preload pip packages that share names with veins to prevent shadowing
|
|
# These must be imported BEFORE any vein that might shadow them
|
|
_preloaded_packages = []
|
|
for _pkg in ["google.auth", "jira", "slack_sdk"]:
|
|
try:
|
|
__import__(_pkg)
|
|
_preloaded_packages.append(_pkg)
|
|
except ImportError:
|
|
pass # Package not installed, vein will fail gracefully
|
|
|
|
|
|
def load_vein(vein_name: str):
|
|
"""
|
|
Load a vein's router dynamically.
|
|
|
|
Veins use relative imports for their internal modules (..core, ..models)
|
|
and absolute imports for shared utilities (oauth, base from veins/).
|
|
|
|
IMPORTANT: Vein folder names may shadow pip packages (e.g., 'google', 'jira').
|
|
We ONLY register under prefixed names (vein_google) to avoid shadowing.
|
|
Relative imports work because we set __package__ correctly on each module.
|
|
"""
|
|
vein_path = SPR_ROOT / "artery" / "veins" / vein_name
|
|
if not vein_path.exists():
|
|
raise FileNotFoundError(f"Vein not found: {vein_path}")
|
|
|
|
routes_file = vein_path / "api" / "routes.py"
|
|
if not routes_file.exists():
|
|
raise FileNotFoundError(f"Vein routes not found: {routes_file}")
|
|
|
|
# Use prefixed name to avoid shadowing pip packages
|
|
vein_prefix = f"vein_{vein_name}"
|
|
|
|
# Clear any previously loaded vein modules to avoid conflicts
|
|
for mod_name in list(sys.modules.keys()):
|
|
if mod_name == vein_prefix or mod_name.startswith(f"{vein_prefix}."):
|
|
del sys.modules[mod_name]
|
|
# Clear shared modules that might have stale references
|
|
for mod_name in ["oauth", "base"]:
|
|
if mod_name in sys.modules:
|
|
del sys.modules[mod_name]
|
|
|
|
# Add veins directory to path for shared modules (oauth.py, base.py)
|
|
veins_path = vein_path.parent
|
|
if str(veins_path) not in sys.path:
|
|
sys.path.insert(0, str(veins_path))
|
|
|
|
# Create the vein package module
|
|
vein_init = vein_path / "__init__.py"
|
|
spec = importlib.util.spec_from_file_location(
|
|
vein_prefix,
|
|
vein_init if vein_init.exists() else None,
|
|
submodule_search_locations=[str(vein_path)],
|
|
)
|
|
vein_pkg = importlib.util.module_from_spec(spec)
|
|
sys.modules[vein_prefix] = vein_pkg
|
|
|
|
if spec.loader:
|
|
spec.loader.exec_module(vein_pkg)
|
|
|
|
# Load subpackages (core, api, models)
|
|
for subpkg in ["core", "models", "api"]:
|
|
subpkg_path = vein_path / subpkg
|
|
if subpkg_path.exists():
|
|
_load_vein_subpackage(vein_pkg, vein_prefix, subpkg, subpkg_path)
|
|
|
|
# Load individual modules in core/ and models/ that routes.py needs
|
|
for subpkg in ["core", "models"]:
|
|
subpkg_path = vein_path / subpkg
|
|
if subpkg_path.exists():
|
|
for py_file in subpkg_path.glob("*.py"):
|
|
if py_file.name.startswith("_"):
|
|
continue
|
|
module_name = py_file.stem
|
|
_load_vein_module(vein_pkg, vein_prefix, subpkg, module_name, py_file)
|
|
|
|
# Now load routes.py with all dependencies available
|
|
routes_mod = _load_vein_module(vein_pkg, vein_prefix, "api", "routes", routes_file)
|
|
|
|
return routes_mod.router
|
|
|
|
|
|
def _load_vein_subpackage(vein_pkg, vein_prefix, subpkg, subpkg_path):
|
|
"""Load a vein subpackage (core, api, models)."""
|
|
subpkg_init = subpkg_path / "__init__.py"
|
|
sub_spec = importlib.util.spec_from_file_location(
|
|
f"{vein_prefix}.{subpkg}",
|
|
subpkg_init if subpkg_init.exists() else None,
|
|
submodule_search_locations=[str(subpkg_path)],
|
|
)
|
|
sub_mod = importlib.util.module_from_spec(sub_spec)
|
|
sys.modules[f"{vein_prefix}.{subpkg}"] = sub_mod
|
|
setattr(vein_pkg, subpkg, sub_mod)
|
|
|
|
if sub_spec.loader:
|
|
sub_spec.loader.exec_module(sub_mod)
|
|
|
|
return sub_mod
|
|
|
|
|
|
def _load_vein_module(vein_pkg, vein_prefix, subpkg, module_name, file_path):
|
|
"""Load a specific module within a vein subpackage."""
|
|
mod_spec = importlib.util.spec_from_file_location(
|
|
f"{vein_prefix}.{subpkg}.{module_name}",
|
|
file_path,
|
|
)
|
|
mod = importlib.util.module_from_spec(mod_spec)
|
|
|
|
# Set __package__ so relative imports resolve via vein_prefix
|
|
mod.__package__ = f"{vein_prefix}.{subpkg}"
|
|
|
|
sys.modules[f"{vein_prefix}.{subpkg}.{module_name}"] = mod
|
|
|
|
# Also set on parent package
|
|
parent_pkg = getattr(vein_pkg, subpkg, None)
|
|
if parent_pkg:
|
|
setattr(parent_pkg, module_name, mod)
|
|
|
|
mod_spec.loader.exec_module(mod)
|
|
return mod
|
|
|
|
|
|
def mount_veins(app):
|
|
"""Auto-discover and mount all veins from artery/veins/."""
|
|
veins_dir = SPR_ROOT / "artery" / "veins"
|
|
if not veins_dir.exists():
|
|
return
|
|
|
|
for vein_path in sorted(veins_dir.iterdir()):
|
|
if not vein_path.is_dir():
|
|
continue
|
|
if vein_path.name.startswith(("_", ".")):
|
|
continue
|
|
# Skip non-vein directories (no api/routes.py)
|
|
if not (vein_path / "api" / "routes.py").exists():
|
|
continue
|
|
|
|
vein_name = vein_path.name
|
|
try:
|
|
router = load_vein(vein_name)
|
|
app.include_router(router, prefix=f"/artery/{vein_name}", tags=[vein_name])
|
|
print(f"Vein mounted: /artery/{vein_name}")
|
|
except Exception as e:
|
|
print(f"Warning: Could not load vein '{vein_name}': {e}")
|
|
|
|
|
|
# ============================================================================
|
|
# Authentication Setup (optional, based on room config)
|
|
# ============================================================================
|
|
|
|
|
|
def setup_auth(app, config: dict):
|
|
"""
|
|
Configure authentication if enabled in room config.
|
|
|
|
Auth is optional - rooms without auth config run without authentication.
|
|
"""
|
|
try:
|
|
from common.auth.config import load_auth_config
|
|
|
|
auth_config = load_auth_config(config)
|
|
if not auth_config:
|
|
return
|
|
|
|
print(f"Auth: enabled for domains {auth_config.allowed_domains or ['*']}")
|
|
|
|
# Get session secret
|
|
session_secret = auth_config.session_secret
|
|
if session_secret.startswith("ENV:"):
|
|
session_secret = os.getenv(session_secret[4:], "dev-secret-change-in-prod")
|
|
|
|
# Add session middleware
|
|
from starlette.middleware.sessions import SessionMiddleware
|
|
|
|
app.add_middleware(
|
|
SessionMiddleware,
|
|
secret_key=session_secret,
|
|
session_cookie="soleprint_session",
|
|
max_age=auth_config.session_timeout_hours * 3600,
|
|
same_site="lax",
|
|
https_only=os.getenv("HTTPS_ONLY", "").lower() == "true",
|
|
)
|
|
|
|
# Initialize auth routes
|
|
from common.auth.routes import init_auth
|
|
from common.auth.routes import router as auth_router
|
|
|
|
init_auth(auth_config)
|
|
app.include_router(auth_router)
|
|
print("Auth: routes mounted at /auth")
|
|
|
|
# Add auth middleware
|
|
from common.auth.middleware import AuthMiddleware
|
|
|
|
app.add_middleware(AuthMiddleware, auth_config=auth_config)
|
|
except ImportError:
|
|
# common.auth not available (standalone without auth)
|
|
pass
|
|
except Exception as e:
|
|
print(f"Auth setup error: {e}")
|
|
|
|
|
|
def load_config() -> dict:
|
|
"""Load config.json from cfg/ directory."""
|
|
config_path = CFG_DIR / "config.json"
|
|
if config_path.exists():
|
|
return json.loads(config_path.read_text())
|
|
return {}
|
|
|
|
|
|
def get_system_config(system_key: str) -> dict:
|
|
"""Get system configuration by key (data_flow, documentation, execution)."""
|
|
config = load_config()
|
|
for system in config.get("systems", []):
|
|
if system.get("key") == system_key:
|
|
return system
|
|
return {}
|
|
|
|
|
|
def get_components(system_key: str) -> dict:
|
|
"""Get component definitions for a system."""
|
|
config = load_config()
|
|
return config.get("components", {}).get(system_key, {})
|
|
|
|
|
|
def load_data(filename: str) -> list[dict]:
|
|
"""Load data from JSON file in data/ directory."""
|
|
path = DATA_DIR / filename
|
|
if path.exists():
|
|
data = json.loads(path.read_text())
|
|
return data.get("items", [])
|
|
return []
|
|
|
|
|
|
def scan_directory(base_path: Path, pattern: str = "*") -> list[dict]:
|
|
"""Scan a directory and return list of items with metadata."""
|
|
items = []
|
|
if base_path.exists():
|
|
for item in sorted(base_path.iterdir()):
|
|
if item.is_dir() and not item.name.startswith(("_", ".")):
|
|
readme = item / "README.md"
|
|
description = ""
|
|
if readme.exists():
|
|
# Get first non-empty, non-header line
|
|
for line in readme.read_text().split("\n"):
|
|
line = line.strip()
|
|
if line and not line.startswith("#"):
|
|
description = line[:100]
|
|
break
|
|
items.append(
|
|
{
|
|
"name": item.name,
|
|
"path": str(item.relative_to(SPR_ROOT)),
|
|
"description": description,
|
|
}
|
|
)
|
|
return items
|
|
|
|
|
|
# ============================================================================
|
|
# Initialize: Load config, setup auth, mount veins
|
|
# ============================================================================
|
|
|
|
_config = load_config()
|
|
setup_auth(app, _config)
|
|
mount_veins(app)
|
|
|
|
|
|
@app.get("/health")
|
|
def health():
|
|
return {
|
|
"status": "ok",
|
|
"service": "soleprint",
|
|
"mode": "bare-metal",
|
|
}
|
|
|
|
|
|
# === Artery ===
|
|
|
|
|
|
@app.get("/artery", response_class=HTMLResponse)
|
|
@app.get("/artery/", response_class=HTMLResponse)
|
|
def artery_index(request: Request):
|
|
"""Artery landing page."""
|
|
html_path = SPR_ROOT / "artery" / "index.html"
|
|
if html_path.exists():
|
|
# Get system config and components
|
|
system = get_system_config("data_flow")
|
|
components = get_components("data_flow")
|
|
|
|
# Load from data files
|
|
veins = load_data("veins.json")
|
|
pulses = load_data("pulses.json")
|
|
shunts_data = load_data("shunts.json")
|
|
plexuses = load_data("plexuses.json")
|
|
|
|
# Scan directories for shunts not in data files
|
|
shunts = (
|
|
shunts_data
|
|
if shunts_data
|
|
else scan_directory(SPR_ROOT / "artery" / "shunts")
|
|
)
|
|
for s in shunts:
|
|
if "slug" not in s:
|
|
s["slug"] = s.get("name", "")
|
|
if "title" not in s:
|
|
s["title"] = s.get("name", "").replace("-", " ").title()
|
|
if "status" not in s:
|
|
s["status"] = "ready"
|
|
|
|
from jinja2 import Template
|
|
|
|
template = Template(html_path.read_text())
|
|
return HTMLResponse(
|
|
template.render(
|
|
request=request,
|
|
system=system,
|
|
components=components,
|
|
veins=veins,
|
|
pulses=pulses,
|
|
shunts=shunts,
|
|
plexuses=plexuses,
|
|
soleprint_url="/",
|
|
)
|
|
)
|
|
veins_path = SPR_ROOT / "artery" / "veins"
|
|
veins = scan_directory(veins_path)
|
|
system = get_system_config("data_flow")
|
|
return JSONResponse(
|
|
{
|
|
"system": system.get("name", "artery"),
|
|
"tagline": system.get("tagline", ""),
|
|
"veins": veins,
|
|
}
|
|
)
|
|
|
|
|
|
@app.get("/artery/{path:path}")
|
|
def artery_route(path: str):
|
|
"""Artery sub-routes."""
|
|
return {"system": "artery", "path": path}
|
|
|
|
|
|
# === Atlas ===
|
|
|
|
|
|
@app.get("/atlas", response_class=HTMLResponse)
|
|
@app.get("/atlas/", response_class=HTMLResponse)
|
|
def atlas_index(request: Request):
|
|
"""Atlas landing page."""
|
|
html_path = SPR_ROOT / "atlas" / "index.html"
|
|
if html_path.exists():
|
|
# Get system config and components
|
|
system = get_system_config("documentation")
|
|
components = get_components("documentation")
|
|
|
|
# Load books from data file (includes template info for templated vs original)
|
|
books = load_data("books.json")
|
|
templates = load_data("templates.json")
|
|
depots = load_data("depots.json")
|
|
|
|
from jinja2 import Template
|
|
|
|
template = Template(html_path.read_text())
|
|
return HTMLResponse(
|
|
template.render(
|
|
request=request,
|
|
system=system,
|
|
components=components,
|
|
books=books,
|
|
templates=templates,
|
|
depots=depots,
|
|
soleprint_url="/",
|
|
)
|
|
)
|
|
system = get_system_config("documentation")
|
|
return JSONResponse(
|
|
{
|
|
"system": system.get("name", "atlas"),
|
|
"tagline": system.get("tagline", ""),
|
|
"books": [],
|
|
}
|
|
)
|
|
|
|
|
|
@app.get("/atlas/book/{book_name:path}")
|
|
def atlas_book(book_name: str):
|
|
"""Serve atlas books."""
|
|
book_name = book_name.rstrip("/")
|
|
book_path = SPR_ROOT / "atlas" / "books" / book_name
|
|
|
|
if not book_path.exists():
|
|
return JSONResponse({"detail": "Book not found"}, status_code=404)
|
|
|
|
index_html = book_path / "index.html"
|
|
if index_html.exists():
|
|
return HTMLResponse(index_html.read_text())
|
|
|
|
readme = book_path / "README.md"
|
|
if readme.exists():
|
|
return HTMLResponse(f"<pre>{readme.read_text()}</pre>")
|
|
|
|
return JSONResponse(
|
|
{"book": book_name, "files": [f.name for f in book_path.iterdir()]}
|
|
)
|
|
|
|
|
|
@app.get("/atlas/{path:path}")
|
|
def atlas_route(path: str):
|
|
"""Atlas sub-routes."""
|
|
return {"system": "atlas", "path": path}
|
|
|
|
|
|
# === Station ===
|
|
|
|
|
|
@app.get("/station", response_class=HTMLResponse)
|
|
@app.get("/station/", response_class=HTMLResponse)
|
|
def station_index(request: Request):
|
|
"""Station landing page."""
|
|
html_path = SPR_ROOT / "station" / "index.html"
|
|
if html_path.exists():
|
|
# Get system config and components
|
|
system = get_system_config("execution")
|
|
components = get_components("execution")
|
|
|
|
tools = scan_directory(SPR_ROOT / "station" / "tools")
|
|
for t in tools:
|
|
t["slug"] = t["name"]
|
|
t["title"] = t["name"].replace("-", " ").title()
|
|
t["status"] = "ready"
|
|
monitors = scan_directory(SPR_ROOT / "station" / "monitors")
|
|
for m in monitors:
|
|
m["slug"] = m["name"]
|
|
m["title"] = m["name"].replace("-", " ").title()
|
|
m["status"] = "ready"
|
|
desks = scan_directory(SPR_ROOT / "station" / "desks")
|
|
for d in desks:
|
|
d["slug"] = d["name"]
|
|
d["title"] = d["name"].replace("-", " ").title()
|
|
d["status"] = "ready"
|
|
from jinja2 import Template
|
|
|
|
template = Template(html_path.read_text())
|
|
return HTMLResponse(
|
|
template.render(
|
|
request=request,
|
|
system=system,
|
|
components=components,
|
|
tools=tools,
|
|
monitors=monitors,
|
|
desks=desks,
|
|
soleprint_url="/",
|
|
)
|
|
)
|
|
tools_path = SPR_ROOT / "station" / "tools"
|
|
tools = scan_directory(tools_path)
|
|
system = get_system_config("execution")
|
|
return JSONResponse(
|
|
{
|
|
"system": system.get("name", "station"),
|
|
"tagline": system.get("tagline", ""),
|
|
"tools": tools,
|
|
}
|
|
)
|
|
|
|
|
|
@app.get("/station/{path:path}")
|
|
def station_route(path: str):
|
|
"""Station sub-routes."""
|
|
return {"system": "station", "path": path}
|
|
|
|
|
|
# === Sidebar Wrapper (served at /spr/* when proxied) ===
|
|
|
|
|
|
@app.get("/sidebar.css")
|
|
def sidebar_css():
|
|
"""Serve sidebar CSS for injection."""
|
|
css_path = SPR_ROOT / "station" / "tools" / "sbwrapper" / "sidebar.css"
|
|
if css_path.exists():
|
|
from fastapi.responses import Response
|
|
|
|
return Response(content=css_path.read_text(), media_type="text/css")
|
|
return Response(content="/* sidebar.css not found */", media_type="text/css")
|
|
|
|
|
|
@app.get("/sidebar.js")
|
|
def sidebar_js():
|
|
"""Serve sidebar JS for injection."""
|
|
js_path = SPR_ROOT / "station" / "tools" / "sbwrapper" / "sidebar.js"
|
|
if js_path.exists():
|
|
from fastapi.responses import Response
|
|
|
|
return Response(
|
|
content=js_path.read_text(), media_type="application/javascript"
|
|
)
|
|
return Response(
|
|
content="/* sidebar.js not found */", media_type="application/javascript"
|
|
)
|
|
|
|
|
|
@app.get("/api/sidebar/config")
|
|
def sidebar_config(request: Request):
|
|
"""Return sidebar configuration for the current room."""
|
|
config = load_config()
|
|
managed = config.get("managed", {})
|
|
auth = config.get("auth", {})
|
|
|
|
# Get soleprint URL (where tools are)
|
|
host = request.headers.get("host", "localhost")
|
|
host_no_port = host.split(":")[0]
|
|
scheme = request.headers.get("x-forwarded-proto", "http")
|
|
|
|
# Soleprint tools are at /spr/* when accessed via <room>.spr.<domain>
|
|
soleprint_base = "/spr"
|
|
|
|
return {
|
|
"room": managed.get("name", "standalone"),
|
|
"soleprint_base": soleprint_base,
|
|
"auth_enabled": auth.get("enabled", False),
|
|
"tools": {
|
|
"artery": f"{soleprint_base}/artery",
|
|
"atlas": f"{soleprint_base}/atlas",
|
|
"station": f"{soleprint_base}/station",
|
|
},
|
|
"auth": {
|
|
"login_url": f"{soleprint_base}/artery/google/oauth/login",
|
|
"status_url": f"{soleprint_base}/artery/google/oauth/status",
|
|
"logout_url": f"{soleprint_base}/artery/google/oauth/logout",
|
|
},
|
|
}
|
|
|
|
|
|
# === Main ===
|
|
|
|
|
|
def get_managed_url(request: Request, managed: dict) -> str | None:
|
|
"""
|
|
Derive managed app URL from current host.
|
|
|
|
Pattern: <room>.spr.<domain> -> <room>.<domain>
|
|
localhost:port -> None (no managed URL for direct port access)
|
|
"""
|
|
if not managed:
|
|
return None
|
|
|
|
host = request.headers.get("host", "localhost")
|
|
host_no_port = host.split(":")[0]
|
|
|
|
# Check if host matches pattern: <name>.spr.<domain>
|
|
if ".spr." in host_no_port:
|
|
# Replace .spr. with . to get managed app domain
|
|
managed_host = host_no_port.replace(".spr.", ".")
|
|
scheme = request.headers.get("x-forwarded-proto", "http")
|
|
return f"{scheme}://{managed_host}"
|
|
|
|
return None
|
|
|
|
|
|
@app.get("/")
|
|
def index(request: Request):
|
|
"""Landing page with links to subsystems."""
|
|
config = load_config()
|
|
managed = config.get("managed", {})
|
|
managed_url = get_managed_url(request, managed)
|
|
|
|
return templates.TemplateResponse(
|
|
"index.html",
|
|
{
|
|
"request": request,
|
|
"artery": "/artery",
|
|
"atlas": "/atlas",
|
|
"station": "/station",
|
|
"managed": managed,
|
|
"managed_url": managed_url,
|
|
},
|
|
)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
import uvicorn
|
|
|
|
port = int(os.getenv("PORT", "12000"))
|
|
print(f"Soleprint bare-metal dev server starting on http://localhost:{port}")
|
|
print(" /artery - Connectors (veins)")
|
|
print(" /atlas - Documentation (templates)")
|
|
print(" /station - Tools")
|
|
print()
|
|
|
|
uvicorn.run(
|
|
"run:app",
|
|
host="0.0.0.0",
|
|
port=port,
|
|
reload=True,
|
|
)
|