419 lines
14 KiB
Python
419 lines
14 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Second Brain FastAPI Dashboard
|
|
|
|
Goals:
|
|
- "Release-ready" defaults (no hardcoded absolute paths)
|
|
- Minimal config via env vars
|
|
- Serves the existing static dashboard (templates/dashboard.html + static/)
|
|
"""
|
|
|
|
import json
|
|
import os
|
|
import sqlite3
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
|
|
from fastapi import FastAPI, Form, Query, Request
|
|
from fastapi.responses import HTMLResponse, JSONResponse, PlainTextResponse
|
|
from fastapi.staticfiles import StaticFiles
|
|
|
|
# ─── Config ──────────────────────────────────────────────────────────────────
|
|
REPO_ROOT = Path(__file__).resolve().parent
|
|
WORKSPACE = Path(os.environ.get("SECOND_BRAIN_WORKSPACE", str(REPO_ROOT))).resolve()
|
|
DB_PATH = Path(os.environ.get("SECOND_BRAIN_DB_PATH", str(WORKSPACE / "data" / "brain.sqlite"))).resolve()
|
|
|
|
PORT = int(os.environ.get("SECOND_BRAIN_PORT", os.environ.get("PORT", "8501")))
|
|
HOST = os.environ.get("SECOND_BRAIN_HOST", "0.0.0.0")
|
|
|
|
|
|
def create_app() -> FastAPI:
|
|
app = FastAPI(title="Second Brain Dashboard")
|
|
|
|
static_dir = WORKSPACE / "static"
|
|
if static_dir.is_dir():
|
|
app.mount("/static", StaticFiles(directory=str(static_dir)), name="static")
|
|
|
|
return app
|
|
|
|
|
|
app = create_app()
|
|
|
|
# ─── Helpers ─────────────────────────────────────────────────────────────────
|
|
def get_db():
|
|
if not DB_PATH.exists():
|
|
raise FileNotFoundError(f"DB not found: {DB_PATH}")
|
|
conn = sqlite3.connect(str(DB_PATH))
|
|
conn.row_factory = sqlite3.Row
|
|
return conn
|
|
|
|
|
|
def parse_engram(row: sqlite3.Row) -> dict:
|
|
meta = json.loads(row["metadata_json"] or "{}")
|
|
correctness = json.loads(row["correctness_json"] or "{}")
|
|
return {
|
|
"id": row["id"],
|
|
"content": row["content"],
|
|
"confidence": meta.get("confidence", 0.0),
|
|
"confirmed": correctness.get("confirmed", False),
|
|
"confirmations": correctness.get("confirmations", 0),
|
|
"rejections": correctness.get("rejections", 0),
|
|
"tags": meta.get("tags", []),
|
|
"created": meta.get("created", row["created_at"]),
|
|
"modified": meta.get("modified", row["modified_at"]),
|
|
"last_reviewed": correctness.get("last_reviewed"),
|
|
"review_history": correctness.get("review_history", []),
|
|
"source": meta.get("source", "unknown"),
|
|
"access_count": meta.get("access_count", 0),
|
|
"grounding": meta.get("grounding", 0),
|
|
}
|
|
|
|
|
|
# ─── API Endpoints ───────────────────────────────────────────────────────────
|
|
|
|
@app.get("/healthz", response_class=PlainTextResponse)
|
|
def healthz():
|
|
return "ok"
|
|
|
|
|
|
@app.get("/api/config")
|
|
def api_config():
|
|
return {
|
|
"workspace": str(WORKSPACE),
|
|
"db_path": str(DB_PATH),
|
|
}
|
|
|
|
|
|
@app.exception_handler(FileNotFoundError)
|
|
def handle_file_not_found(request: Request, exc: FileNotFoundError):
|
|
return JSONResponse(
|
|
status_code=503,
|
|
content={
|
|
"error": str(exc),
|
|
"hint": "Set SECOND_BRAIN_DB_PATH or SECOND_BRAIN_WORKSPACE to a valid location.",
|
|
},
|
|
)
|
|
|
|
|
|
@app.exception_handler(sqlite3.Error)
|
|
def handle_sqlite_error(request: Request, exc: sqlite3.Error):
|
|
return JSONResponse(status_code=500, content={"error": f"sqlite error: {exc}"})
|
|
|
|
|
|
@app.get("/api/stats")
|
|
def api_stats():
|
|
conn = get_db()
|
|
c = conn.cursor()
|
|
total = c.execute("SELECT COUNT(*) FROM engrams").fetchone()[0]
|
|
confirmed = c.execute(
|
|
"SELECT COUNT(*) FROM engrams WHERE json_extract(correctness_json, '$.confirmed') = 1"
|
|
).fetchone()[0]
|
|
pending = total - confirmed
|
|
errors = c.execute(
|
|
"SELECT COUNT(*) FROM engrams WHERE json_extract(metadata_json, '$.tags') LIKE '%error%'"
|
|
).fetchone()[0]
|
|
avg_conf = c.execute(
|
|
"SELECT AVG(json_extract(metadata_json, '$.confidence')) FROM engrams"
|
|
).fetchone()[0] or 0.0
|
|
conn.close()
|
|
return {
|
|
"total": total,
|
|
"confirmed": confirmed,
|
|
"pending": pending,
|
|
"errors": errors,
|
|
"avg_confidence": round(avg_conf, 2),
|
|
}
|
|
|
|
|
|
@app.get("/api/engrams")
|
|
def api_engrams(
|
|
limit: int = Query(20, ge=1, le=100),
|
|
offset: int = Query(0, ge=0),
|
|
tag: str = Query(None),
|
|
confirmed: bool = Query(None),
|
|
search: str = Query(None),
|
|
min_confidence: float = Query(0.0),
|
|
):
|
|
conn = get_db()
|
|
c = conn.cursor()
|
|
where_clauses = ["json_extract(metadata_json, '$.confidence') >= ?"]
|
|
params = [min_confidence]
|
|
|
|
if tag:
|
|
where_clauses.append("json_extract(metadata_json, '$.tags') LIKE ?")
|
|
params.append(f'%"{tag}"%')
|
|
|
|
if confirmed is not None:
|
|
where_clauses.append(
|
|
f"json_extract(correctness_json, '$.confirmed') = {int(confirmed)}"
|
|
)
|
|
|
|
if search:
|
|
# Use FTS
|
|
try:
|
|
ids = [
|
|
r[0] for r in c.execute(
|
|
"SELECT rowid FROM engrams_fts WHERE content MATCH ? LIMIT 200",
|
|
(search,)
|
|
).fetchall()
|
|
]
|
|
if ids:
|
|
placeholders = ",".join("?" * len(ids))
|
|
where_clauses.append(f"id IN ({placeholders})")
|
|
params.extend(ids)
|
|
else:
|
|
# Full-text fallback on content
|
|
where_clauses.append("content LIKE ?")
|
|
params.append(f"%{search}%")
|
|
except Exception:
|
|
where_clauses.append("content LIKE ?")
|
|
params.append(f"%{search}%")
|
|
|
|
where_sql = " AND ".join(where_clauses)
|
|
rows = c.execute(
|
|
f"""
|
|
SELECT * FROM engrams
|
|
WHERE {where_sql}
|
|
ORDER BY created_at DESC
|
|
LIMIT ? OFFSET ?
|
|
""",
|
|
params + [limit, offset],
|
|
).fetchall()
|
|
|
|
result = [parse_engram(r) for r in rows]
|
|
conn.close()
|
|
return {"items": result, "limit": limit, "offset": offset}
|
|
|
|
|
|
@app.get("/api/engrams/{engram_id}")
|
|
def api_engram_detail(engram_id: str):
|
|
conn = get_db()
|
|
c = conn.cursor()
|
|
row = c.execute("SELECT * FROM engrams WHERE id = ?", (engram_id,)).fetchone()
|
|
if not row:
|
|
conn.close()
|
|
return JSONResponse({"error": "Not found"}, status_code=404)
|
|
|
|
# Links
|
|
links = c.execute(
|
|
"SELECT to_id FROM engrams_links WHERE from_id = ?", (engram_id,)
|
|
).fetchall()
|
|
result = parse_engram(row)
|
|
result["links"] = [r[0] for r in links]
|
|
conn.close()
|
|
return result
|
|
|
|
|
|
@app.post("/api/engrams/{engram_id}/confirm")
|
|
def api_confirm(engram_id: str, reason: str = Form("")):
|
|
conn = get_db()
|
|
c = conn.cursor()
|
|
row = c.execute(
|
|
"SELECT correctness_json FROM engrams WHERE id = ?", (engram_id,)
|
|
).fetchone()
|
|
if not row:
|
|
conn.close()
|
|
return JSONResponse({"error": "Not found"}, status_code=404)
|
|
|
|
correctness = json.loads(row["correctness_json"] or "{}")
|
|
correctness["confirmed"] = True
|
|
correctness["confirmations"] = correctness.get("confirmations", 0) + 1
|
|
correctness["last_reviewed"] = datetime.now(timezone.utc).isoformat()
|
|
review_history = correctness.get("review_history", [])
|
|
review_history.append({
|
|
"by": "web",
|
|
"action": "confirm",
|
|
"at": datetime.now(timezone.utc).isoformat(),
|
|
"note": reason or "confirmed via dashboard",
|
|
})
|
|
correctness["review_history"] = review_history
|
|
|
|
c.execute(
|
|
"UPDATE engrams SET correctness_json = ?, modified_at = ? WHERE id = ?",
|
|
(json.dumps(correctness), datetime.now(timezone.utc).isoformat(), engram_id),
|
|
)
|
|
conn.commit()
|
|
conn.close()
|
|
return {"success": True, "engram_id": engram_id}
|
|
|
|
|
|
@app.post("/api/engrams/{engram_id}/reject")
|
|
def api_reject(engram_id: str, reason: str = Form("")):
|
|
conn = get_db()
|
|
c = conn.cursor()
|
|
row = c.execute(
|
|
"SELECT correctness_json FROM engrams WHERE id = ?", (engram_id,)
|
|
).fetchone()
|
|
if not row:
|
|
conn.close()
|
|
return JSONResponse({"error": "Not found"}, status_code=404)
|
|
|
|
correctness = json.loads(row["correctness_json"] or "{}")
|
|
correctness["confirmed"] = False
|
|
correctness["rejections"] = correctness.get("rejections", 0) + 1
|
|
correctness["last_reviewed"] = datetime.now(timezone.utc).isoformat()
|
|
review_history = correctness.get("review_history", [])
|
|
review_history.append({
|
|
"by": "web",
|
|
"action": "reject",
|
|
"at": datetime.now(timezone.utc).isoformat(),
|
|
"note": reason or "rejected via dashboard",
|
|
})
|
|
correctness["review_history"] = review_history
|
|
|
|
c.execute(
|
|
"UPDATE engrams SET correctness_json = ?, modified_at = ? WHERE id = ?",
|
|
(json.dumps(correctness), datetime.now(timezone.utc).isoformat(), engram_id),
|
|
)
|
|
conn.commit()
|
|
conn.close()
|
|
return {"success": True, "engram_id": engram_id}
|
|
|
|
|
|
@app.post("/api/engrams/{engram_id}/refresh")
|
|
def api_refresh(engram_id: str):
|
|
conn = get_db()
|
|
c = conn.cursor()
|
|
row = c.execute(
|
|
"SELECT metadata_json, correctness_json FROM engrams WHERE id = ?", (engram_id,)
|
|
).fetchone()
|
|
if not row:
|
|
conn.close()
|
|
return JSONResponse({"error": "Not found"}, status_code=404)
|
|
|
|
meta = json.loads(row["metadata_json"] or "{}")
|
|
correctness = json.loads(row["correctness_json"] or "{}")
|
|
|
|
# Simple heuristic: confidence based on confirmations vs rejections
|
|
conf = 0.5
|
|
conf += 0.1 * correctness.get("confirmations", 0)
|
|
conf -= 0.15 * correctness.get("rejections", 0)
|
|
conf = max(0.1, min(1.0, conf))
|
|
|
|
meta["confidence"] = round(conf, 2)
|
|
meta["modified"] = datetime.now(timezone.utc).isoformat()
|
|
|
|
c.execute(
|
|
"UPDATE engrams SET metadata_json = ?, modified_at = ? WHERE id = ?",
|
|
(json.dumps(meta), datetime.now(timezone.utc).isoformat(), engram_id),
|
|
)
|
|
conn.commit()
|
|
conn.close()
|
|
return {"success": True, "new_confidence": round(conf, 2)}
|
|
|
|
|
|
@app.post("/api/engrams")
|
|
def api_create_engram(content: str = Form(...), tags: str = Form(""), source: str = Form("web")):
|
|
engram_id = f"web-{datetime.now(timezone.utc).strftime('%Y%m%d-%H%M%S-%f')[:20]}"
|
|
now = datetime.now(timezone.utc).isoformat()
|
|
meta = {
|
|
"source": source,
|
|
"confidence": 0.5,
|
|
"created": now,
|
|
"modified": now,
|
|
"access_count": 0,
|
|
"last_accessed": now,
|
|
"tags": [t.strip() for t in tags.split(",") if t.strip()] or ["web"],
|
|
"session_id": None,
|
|
"agent_id": None,
|
|
"grounding": 0,
|
|
"hash": "",
|
|
}
|
|
correctness = {
|
|
"confirmed": False,
|
|
"confirmations": 0,
|
|
"rejections": 0,
|
|
"last_reviewed": None,
|
|
"review_history": [],
|
|
}
|
|
conn = get_db()
|
|
c = conn.cursor()
|
|
c.execute(
|
|
"""
|
|
INSERT INTO engrams (id, content, metadata_json, correctness_json, links_json, hierarchy_json, created_at, modified_at)
|
|
VALUES (?, ?, ?, ?, ?, ?, ?, ?)
|
|
""",
|
|
(engram_id, content, json.dumps(meta), json.dumps(correctness), "[]", '{"parent": null, "children": [], "depth": 0}', now, now),
|
|
)
|
|
conn.commit()
|
|
conn.close()
|
|
return {"success": True, "engram_id": engram_id}
|
|
|
|
|
|
@app.get("/api/pending")
|
|
def api_pending(limit: int = Query(20, ge=1, le=100), offset: int = Query(0, ge=0)):
|
|
conn = get_db()
|
|
c = conn.cursor()
|
|
rows = c.execute(
|
|
"""
|
|
SELECT * FROM engrams
|
|
WHERE json_extract(correctness_json, '$.confirmed') = 0
|
|
ORDER BY created_at DESC
|
|
LIMIT ? OFFSET ?
|
|
""",
|
|
(limit, offset),
|
|
).fetchall()
|
|
result = [parse_engram(r) for r in rows]
|
|
conn.close()
|
|
return {"items": result, "limit": limit, "offset": offset}
|
|
|
|
|
|
@app.get("/api/search")
|
|
def api_search(
|
|
q: str = Query(..., min_length=1),
|
|
min_confidence: float = Query(0.0),
|
|
limit: int = Query(20, ge=1, le=100),
|
|
):
|
|
conn = get_db()
|
|
c = conn.cursor()
|
|
try:
|
|
ids = [
|
|
r[0] for r in c.execute(
|
|
"SELECT rowid FROM engrams_fts WHERE content MATCH ? LIMIT 200",
|
|
(q,)
|
|
).fetchall()
|
|
]
|
|
if ids:
|
|
placeholders = ",".join("?" * len(ids))
|
|
rows = c.execute(
|
|
f"""
|
|
SELECT * FROM engrams
|
|
WHERE id IN ({placeholders})
|
|
AND json_extract(metadata_json, '$.confidence') >= ?
|
|
ORDER BY created_at DESC
|
|
LIMIT ?
|
|
""",
|
|
ids + [min_confidence, limit],
|
|
).fetchall()
|
|
else:
|
|
rows = []
|
|
except Exception:
|
|
rows = c.execute(
|
|
"""
|
|
SELECT * FROM engrams
|
|
WHERE content LIKE ? AND json_extract(metadata_json, '$.confidence') >= ?
|
|
ORDER BY created_at DESC
|
|
LIMIT ?
|
|
""",
|
|
(f"%{q}%", min_confidence, limit),
|
|
).fetchall()
|
|
|
|
result = [parse_engram(r) for r in rows]
|
|
conn.close()
|
|
return {"items": result, "query": q}
|
|
|
|
|
|
# ─── Frontend ────────────────────────────────────────────────────────────────
|
|
|
|
@app.get("/", response_class=HTMLResponse)
|
|
def dashboard(request: Request):
|
|
with open(WORKSPACE / "templates" / "dashboard.html", "r", encoding="utf-8") as f:
|
|
html = f.read()
|
|
return HTMLResponse(content=html)
|
|
|
|
|
|
# ─── Main ────────────────────────────────────────────────────────────────────
|
|
if __name__ == "__main__":
|
|
import uvicorn
|
|
uvicorn.run("fastapi_app:app", host=HOST, port=PORT)
|