11e799188d
- main.py: FastAPI app with profile CRUD, start/stop, send message endpoints - profiles.py: asyncio bot lifecycle using simplex-chat Python SDK - db.py: SQLite registry tracking profiles, types, config, addresses - templates/: Jinja2 + HTMX web UI - login.html: token-based auth - index.html: profile list with live status polling, create dialog - profile.html: per-bot dashboard with QR code, contacts/groups, event log, send form - requirements.txt: fastapi, uvicorn, jinja2, simplex-chat - start.sh: one-command startup with venv bootstrap Bot types: echo, broadcast, support (business address), directory, deadmans Run: cd manager && MANAGER_TOKEN=secret ./start.sh Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
67 lines
2.1 KiB
Python
67 lines
2.1 KiB
Python
"""Profile registry — SQLite database for manager state."""
|
|
|
|
import sqlite3
|
|
import json
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
DB_PATH = Path(__file__).parent / "data" / "manager.db"
|
|
|
|
|
|
def get_conn() -> sqlite3.Connection:
|
|
DB_PATH.parent.mkdir(parents=True, exist_ok=True)
|
|
conn = sqlite3.connect(DB_PATH)
|
|
conn.row_factory = sqlite3.Row
|
|
conn.execute("PRAGMA journal_mode=WAL")
|
|
return conn
|
|
|
|
|
|
def init_db() -> None:
|
|
with get_conn() as conn:
|
|
conn.executescript("""
|
|
CREATE TABLE IF NOT EXISTS profiles (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
name TEXT NOT NULL UNIQUE,
|
|
bot_type TEXT NOT NULL,
|
|
db_prefix TEXT NOT NULL UNIQUE,
|
|
config TEXT NOT NULL DEFAULT '{}',
|
|
created_at TEXT NOT NULL DEFAULT (datetime('now')),
|
|
address TEXT
|
|
);
|
|
""")
|
|
|
|
|
|
def list_profiles() -> list[dict]:
|
|
with get_conn() as conn:
|
|
rows = conn.execute("SELECT * FROM profiles ORDER BY id").fetchall()
|
|
return [dict(r) for r in rows]
|
|
|
|
|
|
def get_profile(profile_id: int) -> dict | None:
|
|
with get_conn() as conn:
|
|
row = conn.execute("SELECT * FROM profiles WHERE id=?", (profile_id,)).fetchone()
|
|
return dict(row) if row else None
|
|
|
|
|
|
def create_profile(name: str, bot_type: str, config: dict) -> dict:
|
|
safe = name.lower().replace(" ", "_")
|
|
db_prefix = f"data/bots/{safe}"
|
|
Path(db_prefix).parent.mkdir(parents=True, exist_ok=True)
|
|
with get_conn() as conn:
|
|
conn.execute(
|
|
"INSERT INTO profiles (name, bot_type, db_prefix, config) VALUES (?,?,?,?)",
|
|
(name, bot_type, db_prefix, json.dumps(config)),
|
|
)
|
|
row = conn.execute("SELECT * FROM profiles WHERE name=?", (name,)).fetchone()
|
|
return dict(row)
|
|
|
|
|
|
def update_address(profile_id: int, address: str) -> None:
|
|
with get_conn() as conn:
|
|
conn.execute("UPDATE profiles SET address=? WHERE id=?", (address, profile_id))
|
|
|
|
|
|
def delete_profile(profile_id: int) -> None:
|
|
with get_conn() as conn:
|
|
conn.execute("DELETE FROM profiles WHERE id=?", (profile_id,))
|