Keysat
75 lines
2.7 KiB
Python
75 lines
2.7 KiB
Python
"""Load human-owned seed data (conviction log, §3.1) into SQLite.
|
|
|
|
The conviction log is the highest-leverage Job B input (§3.1) and is HUMAN-OWNED:
|
|
Grant edits the YAML seed files; this loader upserts them. Re-running is idempotent.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import sqlite3
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
import yaml
|
|
|
|
_CONVICTION_COLS = (
|
|
"conviction_id",
|
|
"seam",
|
|
"thematic_proposition",
|
|
"team_conviction_note",
|
|
"conviction_level",
|
|
"current_exposure",
|
|
"exposure_note",
|
|
"disconfirming_signal",
|
|
"is_thesis_breaker",
|
|
)
|
|
|
|
|
|
def _row(c: dict[str, Any]) -> dict[str, Any]:
|
|
return {
|
|
"conviction_id": c["id"],
|
|
"seam": c.get("seam"),
|
|
"thematic_proposition": c["thematic_proposition"],
|
|
"team_conviction_note": c.get("team_conviction_note"),
|
|
"conviction_level": c.get("conviction_level"),
|
|
"current_exposure": c.get("current_exposure", "unset"),
|
|
"exposure_note": c.get("exposure_note"),
|
|
"disconfirming_signal": c.get("disconfirming_signal"),
|
|
"is_thesis_breaker": 1 if c.get("is_thesis_breaker") else 0,
|
|
}
|
|
|
|
|
|
def load_fanout(conn: sqlite3.Connection, path: Path) -> int:
|
|
"""Load a hand-written fan-out tree (§7.1 backtest). Idempotent on node_id."""
|
|
data = yaml.safe_load(Path(path).read_text()) or {}
|
|
parent = data["parent_conviction_id"]
|
|
nodes = data.get("nodes", [])
|
|
for n in nodes:
|
|
conn.execute(
|
|
"""INSERT INTO fanout_nodes
|
|
(node_id, parent_conviction_id, derivative_proposition, depth, status, distance_from_edge)
|
|
VALUES (?,?,?,?, 'hypothesis', ?)
|
|
ON CONFLICT(node_id) DO UPDATE SET derivative_proposition=excluded.derivative_proposition,
|
|
parent_conviction_id=excluded.parent_conviction_id,
|
|
distance_from_edge=excluded.distance_from_edge""",
|
|
(n["node_id"], parent, n["derivative_proposition"], n.get("depth", 1), n.get("distance_from_edge")),
|
|
)
|
|
conn.commit()
|
|
return len(nodes)
|
|
|
|
|
|
def load_convictions(conn: sqlite3.Connection, path: Path) -> int:
|
|
data = yaml.safe_load(Path(path).read_text()) or {}
|
|
rows = data.get("convictions", [])
|
|
cols = ", ".join(_CONVICTION_COLS)
|
|
placeholders = ", ".join(f":{c}" for c in _CONVICTION_COLS)
|
|
updates = ", ".join(f"{c}=excluded.{c}" for c in _CONVICTION_COLS if c != "conviction_id")
|
|
sql = (
|
|
f"INSERT INTO conviction_log ({cols}, updated_at) "
|
|
f"VALUES ({placeholders}, datetime('now')) "
|
|
f"ON CONFLICT(conviction_id) DO UPDATE SET {updates}, updated_at=datetime('now')"
|
|
)
|
|
for c in rows:
|
|
conn.execute(sql, _row(c))
|
|
conn.commit()
|
|
return len(rows)
|