Spaces:
Runtime error
Runtime error
| import json | |
| import os | |
| import sqlite3 | |
| import uuid | |
| from datetime import datetime | |
| DB_PATH = os.path.join(os.path.dirname(__file__), "..", "conversation.db") | |
| def _conn() -> sqlite3.Connection: | |
| c = sqlite3.connect(DB_PATH) | |
| c.row_factory = sqlite3.Row | |
| return c | |
| def init_db() -> None: | |
| with _conn() as c: | |
| c.execute( | |
| """ | |
| CREATE TABLE IF NOT EXISTS conversations ( | |
| id TEXT PRIMARY KEY, | |
| title TEXT, | |
| created_at TEXT | |
| ) | |
| """ | |
| ) | |
| c.execute( | |
| """ | |
| CREATE TABLE IF NOT EXISTS messages ( | |
| id TEXT PRIMARY KEY, | |
| conversation_id TEXT NOT NULL, | |
| role TEXT NOT NULL, | |
| content TEXT NOT NULL, | |
| created_at TEXT, | |
| FOREIGN KEY (conversation_id) REFERENCES conversations(id) | |
| ) | |
| """ | |
| ) | |
| def init_users_table() -> None: | |
| with _conn() as c: | |
| c.execute( | |
| """ | |
| CREATE TABLE IF NOT EXISTS users ( | |
| id TEXT PRIMARY KEY, | |
| username TEXT NOT NULL UNIQUE, | |
| email TEXT NOT NULL UNIQUE, | |
| hashed_password TEXT NOT NULL, | |
| created_at TEXT | |
| ) | |
| """ | |
| ) | |
| def create_conversation(title: str = "New Chat") -> str: | |
| cid = str(uuid.uuid4()) | |
| with _conn() as c: | |
| c.execute( | |
| "INSERT INTO conversations (id, title, created_at) VALUES (?, ?, ?)", | |
| (cid, title, datetime.utcnow().isoformat()), | |
| ) | |
| return cid | |
| def add_message(conversation_id: str, role: str, content: str) -> str: | |
| mid = str(uuid.uuid4()) | |
| with _conn() as c: | |
| c.execute( | |
| "INSERT INTO messages (id, conversation_id, role, content, created_at) VALUES (?, ?, ?, ?, ?)", | |
| (mid, conversation_id, role, content, datetime.utcnow().isoformat()), | |
| ) | |
| return mid | |
| def get_messages(conversation_id: str) -> list: | |
| with _conn() as c: | |
| rows = c.execute( | |
| "SELECT role, content FROM messages WHERE conversation_id = ? ORDER BY created_at", | |
| (conversation_id,), | |
| ).fetchall() | |
| return [dict(r) for r in rows] | |
| def list_conversations() -> list: | |
| with _conn() as c: | |
| rows = c.execute( | |
| "SELECT id, title, created_at FROM conversations ORDER BY created_at DESC" | |
| ).fetchall() | |
| return [dict(r) for r in rows] | |
| def get_conversation(conv_id: str) -> dict | None: | |
| with _conn() as c: | |
| conv = c.execute( | |
| "SELECT id, created_at FROM conversations WHERE id = ?", (conv_id,) | |
| ).fetchone() | |
| if conv is None: | |
| return None | |
| messages = c.execute( | |
| "SELECT role, content, created_at FROM messages WHERE conversation_id = ? ORDER BY created_at ASC", | |
| (conv_id,), | |
| ).fetchall() | |
| return {"id": conv["id"], "created_at": conv["created_at"], "messages": [dict(m) for m in messages]} | |
| def init_moderation_table() -> None: | |
| """Phase W - audit trail of moderation decisions.""" | |
| with _conn() as c: | |
| c.execute( | |
| """ | |
| CREATE TABLE IF NOT EXISTS moderation_events ( | |
| id TEXT PRIMARY KEY, | |
| created_at TEXT, | |
| event_type TEXT, -- input | output | tool | injection | api | |
| source TEXT, -- chat | agent | moderations | tool:<name> | |
| identity TEXT, -- user:<id> | ip:<addr> | NULL | |
| conversation_id TEXT, | |
| severity TEXT, | |
| action TEXT, | |
| top_category TEXT, | |
| score REAL, | |
| categories TEXT, -- JSON | |
| snippet TEXT -- truncated copy of the offending text | |
| ) | |
| """ | |
| ) | |
| def record_moderation_event( | |
| event_type: str, | |
| source: str, | |
| severity: str, | |
| action: str, | |
| top_category: str | None, | |
| score: float, | |
| categories: dict | None, | |
| identity: str | None = None, | |
| conversation_id: str | None = None, | |
| snippet: str | None = None, | |
| ) -> str: | |
| eid = str(uuid.uuid4()) | |
| with _conn() as c: | |
| c.execute( | |
| """ | |
| INSERT INTO moderation_events | |
| (id, created_at, event_type, source, identity, conversation_id, | |
| severity, action, top_category, score, categories, snippet) | |
| VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) | |
| """, | |
| ( | |
| eid, | |
| datetime.utcnow().isoformat(), | |
| event_type, | |
| source, | |
| identity, | |
| conversation_id, | |
| severity, | |
| action, | |
| top_category, | |
| float(score or 0.0), | |
| json.dumps(categories or {}), | |
| (snippet or "")[:500], | |
| ), | |
| ) | |
| return eid | |
| def list_moderation_events(limit: int = 100, min_severity: str | None = None) -> list: | |
| rank = {"none": 0, "low": 1, "medium": 2, "high": 3} | |
| floor = rank.get(min_severity, 0) | |
| with _conn() as c: | |
| rows = c.execute( | |
| "SELECT * FROM moderation_events ORDER BY created_at DESC LIMIT ?", | |
| (max(1, int(limit)),), | |
| ).fetchall() | |
| out = [] | |
| for r in rows: | |
| d = dict(r) | |
| if rank.get(d.get("severity"), 0) < floor: | |
| continue | |
| try: | |
| d["categories"] = json.loads(d.get("categories") or "{}") | |
| except (TypeError, ValueError): | |
| d["categories"] = {} | |
| out.append(d) | |
| return out | |
| def create_user(username: str, email: str, hashed_password: str) -> str: | |
| uid = str(uuid.uuid4()) | |
| with _conn() as c: | |
| c.execute( | |
| "INSERT INTO users (id, username, email, hashed_password, created_at) VALUES (?, ?, ?, ?, ?)", | |
| (uid, username, email, hashed_password, datetime.utcnow().isoformat()), | |
| ) | |
| return uid | |
| def get_user_by_email(email: str) -> dict | None: | |
| with _conn() as c: | |
| row = c.execute( | |
| "SELECT id, username, email, hashed_password FROM users WHERE email = ?", | |
| (email,), | |
| ).fetchone() | |
| return dict(row) if row else None | |
| def get_user_by_id(user_id: str) -> dict | None: | |
| with _conn() as c: | |
| row = c.execute( | |
| "SELECT id, username, email FROM users WHERE id = ?", | |
| (user_id,), | |
| ).fetchone() | |
| return dict(row) if row else None | |