# Gerado por IA import sqlite3 from datetime import datetime from config import settings def get_connection() -> sqlite3.Connection: conn = sqlite3.connect(settings.DB_PATH) conn.execute("PRAGMA journal_mode=WAL") conn.row_factory = sqlite3.Row return conn def init_db() -> None: conn = get_connection() conn.execute(""" CREATE TABLE IF NOT EXISTS conversations ( id INTEGER PRIMARY KEY AUTOINCREMENT, date TEXT NOT NULL, agent_name TEXT NOT NULL, message TEXT NOT NULL, created_at TEXT NOT NULL ) """) conn.execute(""" CREATE INDEX IF NOT EXISTS idx_conversations_date ON conversations(date) """) conn.commit() conn.close() def save_message(agent_name: str, message: str) -> None: conn = get_connection() today = datetime.utcnow().strftime("%Y-%m-%d") now = datetime.utcnow().isoformat() conn.execute( "INSERT INTO conversations (date, agent_name, message, created_at) VALUES (?, ?, ?, ?)", (today, agent_name, message, now), ) conn.commit() conn.close() def get_today_messages(limit: int | None = None) -> list[dict]: conn = get_connection() today = datetime.utcnow().strftime("%Y-%m-%d") max_rows = limit or settings.MAX_CONTEXT_MESSAGES rows = conn.execute( "SELECT agent_name, message, created_at FROM conversations WHERE date = ? ORDER BY id ASC", (today,), ).fetchall() conn.close() return [{"agent_name": r["agent_name"], "message": r["message"], "created_at": r["created_at"]} for r in rows] def get_recent_messages(agent_name: str, limit: int = 10) -> list[dict]: conn = get_connection() today = datetime.utcnow().strftime("%Y-%m-%d") rows = conn.execute( "SELECT agent_name, message, created_at FROM conversations WHERE date = ? AND agent_name = ? ORDER BY id DESC LIMIT ?", (today, agent_name, limit), ).fetchall() conn.close() return [{"agent_name": r["agent_name"], "message": r["message"], "created_at": r["created_at"]} for r in reversed(rows)]