feat: seed railiance financial baseline

railiance_fabric/cli.py

@@ -15,6 +15,7 @@ from typing import Any
 from urllib.parse import quote
 
 from .connectors import ConnectorConfig
+from .financial_baseline import financial_export_from_legacy
 from .loader import declaration_files, load_yaml
 from .graph import FabricGraph, build_graph
 from .graph_explorer import fabric_graph_explorer_payload
@@ -72,9 +73,9 @@ def build_parser() -> argparse.ArgumentParser:
     blast.add_argument("interface", help="Interface type or interface declaration id.")
     blast.add_argument("paths", nargs="*", type=Path, default=[Path(".")])
 
-    export = sub.add_parser("export", help="Export graph as JSON, Mermaid, or graph-explorer payload.")
+    export = sub.add_parser("export", help="Export graph as JSON, Mermaid, graph-explorer, or financial payload.")
     export.add_argument("paths", nargs="*", type=Path, default=[Path(".")])
-    export.add_argument("--format", choices=["json", "mermaid", "graph-explorer"], default="json")
+    export.add_argument("--format", choices=["json", "mermaid", "graph-explorer", "financial"], default="json")
 
     scan = sub.add_parser("scan", help="Scan a repo for deterministic discovery candidates.")
     scan.add_argument("path", nargs="?", type=Path, default=Path("."))
@@ -310,6 +311,8 @@ def main(argv: list[str] | None = None) -> int:
             print(graph.to_mermaid())
         elif args.format == "graph-explorer":
             print(json.dumps(fabric_graph_explorer_payload(graph.to_export()), indent=2, sort_keys=True))
+        elif args.format == "financial":
+            print(json.dumps(financial_export_from_legacy(graph.to_export()), indent=2, sort_keys=True))
         else:
             print(graph.to_json())
         return 0

railiance_fabric/financial_baseline.py

@@ -0,0 +1,137 @@
+from __future__ import annotations
+
+import json
+from pathlib import Path
+from typing import Any
+
+from .financial import (
+    FINANCIAL_API_VERSION,
+    FINANCIAL_SCHEMA_VERSION,
+    financial_graph_errors,
+    materialize_financial_graph_export,
+)
+from .loader import load_yaml, repo_root
+
+DEFAULT_BASELINE_PATH = repo_root() / "fabric" / "financial" / "railiance-netkingdom.yaml"
+
+
+def load_financial_baseline(path: Path | None = None) -> dict[str, Any]:
+    baseline_path = path or DEFAULT_BASELINE_PATH
+    data = load_yaml(baseline_path)
+    if not isinstance(data, dict):
+        raise ValueError(f"financial baseline must be a mapping: {baseline_path}")
+    if data.get("kind") != "FinancialFabricBaseline":
+        raise ValueError(f"financial baseline kind must be FinancialFabricBaseline: {baseline_path}")
+    spec = data.get("spec")
+    if not isinstance(spec, dict):
+        raise ValueError(f"financial baseline spec must be a mapping: {baseline_path}")
+    return spec
+
+
+def financial_export_from_legacy(
+    legacy_graph: dict[str, Any],
+    baseline: dict[str, Any] | None = None,
+) -> dict[str, Any]:
+    baseline = json.loads(json.dumps(baseline or load_financial_baseline()))
+    defaults = baseline.get("defaults") if isinstance(baseline.get("defaults"), dict) else {}
+    containment_default = defaults.get("containment") if isinstance(defaults.get("containment"), dict) else {}
+    ownership_default = defaults.get("ownership") if isinstance(defaults.get("ownership"), dict) else {}
+    accounting_default = defaults.get("accounting") if isinstance(defaults.get("accounting"), dict) else {}
+
+    graph = {
+        "apiVersion": FINANCIAL_API_VERSION,
+        "kind": "FabricGraphExport",
+        "schema_version": FINANCIAL_SCHEMA_VERSION,
+        "source": {
+            **_object(legacy_graph.get("source")),
+            "producer": "railiance-fabric",
+            "generation_reason": "baseline_projection",
+        },
+        "compatibility": {
+            "legacy_v1alpha1_supported": True,
+            "breaking_reset": False,
+            "projected_from_apiVersion": legacy_graph.get("apiVersion"),
+        },
+        "netkingdom": baseline.get("netkingdom", {}),
+        "actors": baseline.get("actors", []),
+        "fabrics": baseline.get("fabrics", []),
+        "nodes": [
+            _financial_node_from_legacy(node, containment_default, ownership_default, accounting_default)
+            for node in legacy_graph.get("nodes", [])
+            if isinstance(node, dict)
+        ],
+        "edges": [
+            _financial_edge_from_legacy(edge)
+            for edge in legacy_graph.get("edges", [])
+            if isinstance(edge, dict)
+        ],
+        "unresolved": [],
+    }
+    if legacy_graph.get("generated_at"):
+        graph["generated_at"] = legacy_graph["generated_at"]
+    materialized = materialize_financial_graph_export(graph)
+    errors = financial_graph_errors(materialized)
+    if errors:
+        raise ValueError(f"invalid financial baseline projection: {errors[0]}")
+    return materialized
+
+
+def _financial_node_from_legacy(
+    node: dict[str, Any],
+    containment_default: dict[str, Any],
+    ownership_default: dict[str, Any],
+    accounting_default: dict[str, Any],
+) -> dict[str, Any]:
+    result: dict[str, Any] = {
+        "id": node.get("id", ""),
+        "kind": _node_kind(node),
+        "name": node.get("name") or node.get("id", ""),
+        "repo": node.get("repo", ""),
+        "domain": node.get("domain", ""),
+        "lifecycle": node.get("lifecycle", ""),
+        "containment": json.loads(json.dumps(containment_default)),
+        "ownership": json.loads(json.dumps(ownership_default)),
+        "evidence_state": node.get("evidence_state", "declared"),
+        "attributes": _object(node.get("attributes")),
+    }
+    accounting = _object(result["attributes"].get("accounting")) or accounting_default
+    if _has_value(accounting):
+        result["accounting"] = json.loads(json.dumps(accounting))
+    for key in ("canon_category", "canon_anchor", "mapping_fit"):
+        if node.get(key):
+            result[key] = node[key]
+    return result
+
+
+def _financial_edge_from_legacy(edge: dict[str, Any]) -> dict[str, Any]:
+    result = {
+        "from": edge.get("from", ""),
+        "to": edge.get("to", ""),
+        "type": edge.get("type", ""),
+        "attributes": _object(edge.get("attributes")),
+    }
+    for key in ("canonical_type", "canon_anchor", "mapping_fit", "display_only", "evidence_state"):
+        if key in edge:
+            result[key] = edge[key]
+    return result
+
+
+def _node_kind(node: dict[str, Any]) -> str:
+    kind = str(node.get("kind") or "")
+    if kind == "ServiceDeclaration":
+        return "Service"
+    if kind == "InterfaceDeclaration":
+        return "UtilityInterface"
+    return kind or "DiscoveredEntity"
+
+
+def _object(value: Any) -> dict[str, Any]:
+    return value if isinstance(value, dict) else {}
+
+
+def _has_value(value: Any) -> bool:
+    if isinstance(value, dict):
+        return any(_has_value(item) for item in value.values())
+    if isinstance(value, list):
+        return any(_has_value(item) for item in value)
+    return value not in (None, "")