jpmschweitzerandClaude Opus 4.6
63bb6ff7c7
Updated docstrings in sqlite_connector, qdrant_connector, decisions_sync, schema.sql, and two doc comments in generator.rs. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
224 lines
8.4 KiB
Python
Executable File
224 lines
8.4 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""
|
|
Settled Reach SQLite Connector — mini MCP for ticket management.
|
|
|
|
Usage:
|
|
python3 sqlite_connector.py init
|
|
python3 sqlite_connector.py query "SELECT * FROM tickets"
|
|
python3 sqlite_connector.py execute "UPDATE tickets SET status='done' WHERE id=1"
|
|
python3 sqlite_connector.py seed-decisions
|
|
python3 sqlite_connector.py --help
|
|
"""
|
|
|
|
import json
|
|
import os
|
|
import sqlite3
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Paths
|
|
# ---------------------------------------------------------------------------
|
|
|
|
SCRIPT_DIR = Path(__file__).resolve().parent
|
|
CONFIG_PATH = SCRIPT_DIR / "config.json"
|
|
WORKTREE_ROOT = (SCRIPT_DIR / ".." / "..").resolve()
|
|
SCHEMA_PATH = WORKTREE_ROOT / "db" / "schema.sql"
|
|
# Shared database lives in the parent of all worktrees (three levels up from tooling/db/).
|
|
DB_PATH = (WORKTREE_ROOT / ".." / "settledreach.db").resolve()
|
|
|
|
|
|
def load_config():
|
|
"""Load config.json and resolve the SQLite database path."""
|
|
with open(CONFIG_PATH, "r") as f:
|
|
cfg = json.load(f)
|
|
cfg["sqlite_db_resolved"] = str(DB_PATH)
|
|
return cfg
|
|
|
|
|
|
def get_connection(cfg):
|
|
"""Return an sqlite3 connection with WAL mode and foreign keys enabled."""
|
|
conn = sqlite3.connect(cfg["sqlite_db_resolved"])
|
|
conn.execute("PRAGMA journal_mode=WAL;")
|
|
conn.execute("PRAGMA foreign_keys=ON;")
|
|
conn.row_factory = sqlite3.Row
|
|
return conn
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Commands
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def cmd_init(cfg):
|
|
"""Initialise the database from schema.sql."""
|
|
if not SCHEMA_PATH.exists():
|
|
return {"ok": False, "error": f"Schema file not found: {SCHEMA_PATH}"}
|
|
|
|
schema_sql = SCHEMA_PATH.read_text()
|
|
conn = get_connection(cfg)
|
|
try:
|
|
conn.executescript(schema_sql)
|
|
conn.commit()
|
|
return {"ok": True, "message": f"Database initialised at {cfg['sqlite_db_resolved']}"}
|
|
except sqlite3.Error as exc:
|
|
return {"ok": False, "error": str(exc)}
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
def cmd_query(cfg, sql):
|
|
"""Run a SELECT query and return results as a JSON array of objects."""
|
|
conn = get_connection(cfg)
|
|
try:
|
|
cursor = conn.execute(sql)
|
|
columns = [desc[0] for desc in cursor.description] if cursor.description else []
|
|
rows = [dict(zip(columns, row)) for row in cursor.fetchall()]
|
|
return {"ok": True, "count": len(rows), "rows": rows}
|
|
except sqlite3.Error as exc:
|
|
return {"ok": False, "error": str(exc)}
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
def cmd_execute(cfg, sql):
|
|
"""Run an INSERT/UPDATE/DELETE and return affected row count."""
|
|
conn = get_connection(cfg)
|
|
try:
|
|
cursor = conn.execute(sql)
|
|
conn.commit()
|
|
return {
|
|
"ok": True,
|
|
"affected_rows": cursor.rowcount,
|
|
"last_id": cursor.lastrowid,
|
|
}
|
|
except sqlite3.Error as exc:
|
|
return {"ok": False, "error": str(exc)}
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
def cmd_seed_decisions(cfg):
|
|
"""Seed the database with initiatives derived from decisions and open questions."""
|
|
decisions = [
|
|
("initiative", "Custom game, not a mod", "backlog", "medium", "D-001"),
|
|
("initiative", "Settled Reach as first campaign", "backlog", "medium", "D-003"),
|
|
("initiative", "Single character first-person story generator", "backlog", "medium", "D-005"),
|
|
("initiative", "Prototype scenario — Institute/Armstrong City/Guardians", "backlog", "medium", "D-006"),
|
|
("initiative", "Five pillars of game design", "backlog", "medium", "D-007"),
|
|
("initiative", "Action pillar design principles", "backlog", "medium", "D-008"),
|
|
("initiative", "Multiplayer — design for it, build single-player first", "backlog", "medium", "D-009"),
|
|
("initiative", "Multiplayer-ready architectural baseline", "backlog", "medium", "D-010"),
|
|
("initiative", "Fog of perception non-negotiable", "backlog", "medium", "D-011"),
|
|
("initiative", "Chunk-based map architecture", "backlog", "medium", "D-012"),
|
|
("initiative", "Diegetic insert/POI navigation", "backlog", "medium", "D-013"),
|
|
("initiative", "v0.1 map specification", "backlog", "medium", "D-014"),
|
|
("initiative", "Camera locked to character", "backlog", "medium", "D-015"),
|
|
("initiative", "Internal monologue system", "backlog", "medium", "D-016"),
|
|
("initiative", "Perception modes as character build", "backlog", "medium", "D-017"),
|
|
("initiative", "Three-range sound model", "backlog", "medium", "D-018"),
|
|
("initiative", "Top-down with 3D cutscenes", "backlog", "medium", "D-019"),
|
|
]
|
|
|
|
questions = [
|
|
("story", "Game engine selection", "ready", "critical", "Q-001"),
|
|
("story", "v0.1 prototype scope", "backlog", "medium", "Q-002"),
|
|
("story", "Art direction", "backlog", "medium", "Q-003"),
|
|
("story", "One campaign or separate eras", "backlog", "medium", "Q-004"),
|
|
("story", "Prototype scale", "backlog", "medium", "Q-005"),
|
|
("story", "Target platforms", "backlog", "medium", "Q-007"),
|
|
("story", "Licensing/distribution", "backlog", "medium", "Q-008"),
|
|
("story", "Time system", "backlog", "medium", "Q-009"),
|
|
("story", "Storyteller AI design", "backlog", "medium", "Q-010"),
|
|
("story", "Character selection roster", "backlog", "medium", "Q-011"),
|
|
]
|
|
|
|
conn = get_connection(cfg)
|
|
inserted = 0
|
|
skipped = 0
|
|
try:
|
|
for ticket_type, title, status, priority, decision_ref in decisions + questions:
|
|
# Check if a ticket with this decision_ref already exists
|
|
existing = conn.execute(
|
|
"SELECT id FROM tickets WHERE decision_ref = ?", (decision_ref,)
|
|
).fetchone()
|
|
if existing:
|
|
skipped += 1
|
|
continue
|
|
conn.execute(
|
|
"INSERT INTO tickets (type, title, status, priority, decision_ref) "
|
|
"VALUES (?, ?, ?, ?, ?)",
|
|
(ticket_type, title, status, priority, decision_ref),
|
|
)
|
|
inserted += 1
|
|
conn.commit()
|
|
return {
|
|
"ok": True,
|
|
"inserted": inserted,
|
|
"skipped": skipped,
|
|
"message": f"Seeded {inserted} tickets ({skipped} already existed)",
|
|
}
|
|
except sqlite3.Error as exc:
|
|
conn.rollback()
|
|
return {"ok": False, "error": str(exc)}
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# CLI
|
|
# ---------------------------------------------------------------------------
|
|
|
|
HELP_TEXT = """\
|
|
Settled Reach SQLite Connector
|
|
|
|
Usage:
|
|
sqlite_connector.py init Create/update database from schema.sql
|
|
sqlite_connector.py query "<SQL>" Run a SELECT and return JSON rows
|
|
sqlite_connector.py execute "<SQL>" Run INSERT/UPDATE/DELETE, return affected rows
|
|
sqlite_connector.py seed-decisions Seed initiatives from decisions D-001..D-019 and Q-001..Q-011
|
|
sqlite_connector.py --help Show this help message
|
|
|
|
All output is JSON on stdout. Errors also use JSON with {{"ok": false, "error": "..."}}.
|
|
|
|
Config: {config}
|
|
Schema: {schema}
|
|
""".format(config=CONFIG_PATH, schema=SCHEMA_PATH)
|
|
|
|
|
|
def main():
|
|
if len(sys.argv) < 2 or sys.argv[1] in ("--help", "-h", "help"):
|
|
print(HELP_TEXT)
|
|
sys.exit(0)
|
|
|
|
cmd = sys.argv[1]
|
|
|
|
try:
|
|
cfg = load_config()
|
|
except (FileNotFoundError, json.JSONDecodeError) as exc:
|
|
print(json.dumps({"ok": False, "error": f"Config error: {exc}"}, indent=2))
|
|
sys.exit(1)
|
|
|
|
if cmd == "init":
|
|
result = cmd_init(cfg)
|
|
elif cmd == "query":
|
|
if len(sys.argv) < 3:
|
|
result = {"ok": False, "error": "query requires a SQL string argument"}
|
|
else:
|
|
result = cmd_query(cfg, sys.argv[2])
|
|
elif cmd == "execute":
|
|
if len(sys.argv) < 3:
|
|
result = {"ok": False, "error": "execute requires a SQL string argument"}
|
|
else:
|
|
result = cmd_execute(cfg, sys.argv[2])
|
|
elif cmd == "seed-decisions":
|
|
result = cmd_seed_decisions(cfg)
|
|
else:
|
|
result = {"ok": False, "error": f"Unknown command: {cmd}. Use --help for usage."}
|
|
|
|
print(json.dumps(result, indent=2))
|
|
sys.exit(0 if result.get("ok") else 1)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|