"""
Comprehensive Personal Memory System
Stores everything about the user - who they are, what they do, how they do it
"""
import sqlite3
import json
import os
from datetime import datetime
from pathlib import Path
# Database file location
MEMORY_DB_PATH = os.path.join(os.path.dirname(__file__), "..", "personal_memory.db")
def get_connection():
"""Get database connection with row factory"""
conn = sqlite3.connect(MEMORY_DB_PATH)
conn.row_factory = sqlite3.Row
return conn
def init_memory_database():
"""Initialize the comprehensive memory database"""
conn = get_connection()
cursor = conn.cursor()
# Core identity - who you are
cursor.execute(
"""
CREATE TABLE IF NOT EXISTS identity (
id INTEGER PRIMARY KEY,
category TEXT NOT NULL,
key TEXT NOT NULL,
value TEXT,
confidence TEXT DEFAULT 'confirmed',
source TEXT,
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
UNIQUE(category, key)
)
"""
)
# Behaviors - how you do things
cursor.execute(
"""
CREATE TABLE IF NOT EXISTS behaviors (
id INTEGER PRIMARY KEY AUTOINCREMENT,
behavior_type TEXT NOT NULL,
description TEXT NOT NULL,
context TEXT,
frequency TEXT,
importance TEXT DEFAULT 'medium',
examples TEXT,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
"""
)
# Preferences - what you like/dislike
cursor.execute(
"""
CREATE TABLE IF NOT EXISTS preferences (
id INTEGER PRIMARY KEY AUTOINCREMENT,
category TEXT NOT NULL,
preference TEXT NOT NULL,
value TEXT,
reason TEXT,
priority INTEGER DEFAULT 5,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
"""
)
# Patterns - recurring things you do
cursor.execute(
"""
CREATE TABLE IF NOT EXISTS patterns (
id INTEGER PRIMARY KEY AUTOINCREMENT,
pattern_type TEXT NOT NULL,
description TEXT NOT NULL,
trigger TEXT,
frequency TEXT,
time_context TEXT,
notes TEXT,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
"""
)
# Decisions - how you make decisions
cursor.execute(
"""
CREATE TABLE IF NOT EXISTS decisions (
id INTEGER PRIMARY KEY AUTOINCREMENT,
decision_area TEXT NOT NULL,
approach TEXT NOT NULL,
factors TEXT,
examples TEXT,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
"""
)
# Goals - what you want to achieve
cursor.execute(
"""
CREATE TABLE IF NOT EXISTS goals (
id INTEGER PRIMARY KEY AUTOINCREMENT,
goal_type TEXT NOT NULL,
description TEXT NOT NULL,
timeframe TEXT,
priority TEXT DEFAULT 'medium',
status TEXT DEFAULT 'active',
milestones TEXT,
progress TEXT,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
"""
)
# Interactions - things you've done/discussed
cursor.execute(
"""
CREATE TABLE IF NOT EXISTS interactions (
id INTEGER PRIMARY KEY AUTOINCREMENT,
interaction_type TEXT NOT NULL,
summary TEXT NOT NULL,
details TEXT,
outcome TEXT,
learnings TEXT,
timestamp TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
"""
)
# Facts - factual information about you
cursor.execute(
"""
CREATE TABLE IF NOT EXISTS facts (
id INTEGER PRIMARY KEY AUTOINCREMENT,
category TEXT NOT NULL,
fact TEXT NOT NULL,
context TEXT,
verified BOOLEAN DEFAULT 1,
source TEXT,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
"""
)
# Relationships - people in your life/network
cursor.execute(
"""
CREATE TABLE IF NOT EXISTS relationships (
id INTEGER PRIMARY KEY AUTOINCREMENT,
name TEXT NOT NULL,
relationship_type TEXT,
context TEXT,
importance TEXT DEFAULT 'medium',
notes TEXT,
last_interaction TEXT,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
"""
)
# Work style - how you approach work
cursor.execute(
"""
CREATE TABLE IF NOT EXISTS work_style (
id INTEGER PRIMARY KEY AUTOINCREMENT,
aspect TEXT NOT NULL,
description TEXT NOT NULL,
examples TEXT,
tips_for_ai TEXT,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
"""
)
# Communication style
cursor.execute(
"""
CREATE TABLE IF NOT EXISTS communication_style (
id INTEGER PRIMARY KEY AUTOINCREMENT,
aspect TEXT NOT NULL,
preference TEXT NOT NULL,
examples TEXT,
avoid TEXT,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
"""
)
# Learning log - things learned about user over time
cursor.execute(
"""
CREATE TABLE IF NOT EXISTS learning_log (
id INTEGER PRIMARY KEY AUTOINCREMENT,
category TEXT NOT NULL,
learned TEXT NOT NULL,
context TEXT,
confidence TEXT DEFAULT 'medium',
timestamp TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
"""
)
# Quick memory - for any ad-hoc information
cursor.execute(
"""
CREATE TABLE IF NOT EXISTS quick_memory (
id INTEGER PRIMARY KEY AUTOINCREMENT,
topic TEXT NOT NULL,
content TEXT NOT NULL,
tags TEXT,
importance TEXT DEFAULT 'normal',
expires_at TIMESTAMP,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
"""
)
conn.commit()
conn.close()
return {"success": True, "message": "Memory database initialized"}
# Initialize on import
init_memory_database()
# ==================== IDENTITY FUNCTIONS ====================
def set_identity(category, key, value, confidence="confirmed", source=None):
"""Set identity information (who you are)
Categories: personal, professional, contact, social, preferences
"""
try:
conn = get_connection()
cursor = conn.cursor()
cursor.execute(
"""
INSERT OR REPLACE INTO identity (category, key, value, confidence, source, updated_at)
VALUES (?, ?, ?, ?, ?, ?)
""",
(category, key, value, confidence, source, datetime.now()),
)
conn.commit()
conn.close()
return {"success": True, "category": category, "key": key, "value": value}
except Exception as e:
return {"success": False, "error": str(e)}
def get_identity(category=None):
"""Get identity information"""
try:
conn = get_connection()
cursor = conn.cursor()
if category:
cursor.execute("SELECT * FROM identity WHERE category = ?", (category,))
else:
cursor.execute("SELECT * FROM identity ORDER BY category, key")
rows = cursor.fetchall()
conn.close()
return {"success": True, "identity": [dict(row) for row in rows]}
except Exception as e:
return {"success": False, "error": str(e)}
# ==================== BEHAVIOR FUNCTIONS ====================
def add_behavior(
behavior_type,
description,
context=None,
frequency=None,
importance="medium",
examples=None,
):
"""Add a behavior pattern (how you do things)
Types: work, communication, decision_making, problem_solving, learning, social
"""
try:
conn = get_connection()
cursor = conn.cursor()
cursor.execute(
"""
INSERT INTO behaviors (behavior_type, description, context, frequency, importance, examples)
VALUES (?, ?, ?, ?, ?, ?)
""",
(behavior_type, description, context, frequency, importance, examples),
)
conn.commit()
conn.close()
return {"success": True, "message": f"Behavior added: {description[:50]}..."}
except Exception as e:
return {"success": False, "error": str(e)}
def get_behaviors(behavior_type=None):
"""Get behavior patterns"""
try:
conn = get_connection()
cursor = conn.cursor()
if behavior_type:
cursor.execute(
"SELECT * FROM behaviors WHERE behavior_type = ?", (behavior_type,)
)
else:
cursor.execute(
"SELECT * FROM behaviors ORDER BY importance DESC, behavior_type"
)
rows = cursor.fetchall()
conn.close()
return {"success": True, "behaviors": [dict(row) for row in rows]}
except Exception as e:
return {"success": False, "error": str(e)}
# ==================== PREFERENCE FUNCTIONS ====================
def add_preference(category, preference, value=None, reason=None, priority=5):
"""Add a preference
Categories: tools, communication, work, lifestyle, technology, food, etc.
"""
try:
conn = get_connection()
cursor = conn.cursor()
cursor.execute(
"""
INSERT INTO preferences (category, preference, value, reason, priority)
VALUES (?, ?, ?, ?, ?)
""",
(category, preference, value, reason, priority),
)
conn.commit()
conn.close()
return {"success": True, "message": f"Preference added: {preference}"}
except Exception as e:
return {"success": False, "error": str(e)}
def get_preferences(category=None):
"""Get preferences"""
try:
conn = get_connection()
cursor = conn.cursor()
if category:
cursor.execute(
"SELECT * FROM preferences WHERE category = ? ORDER BY priority DESC",
(category,),
)
else:
cursor.execute("SELECT * FROM preferences ORDER BY category, priority DESC")
rows = cursor.fetchall()
conn.close()
return {"success": True, "preferences": [dict(row) for row in rows]}
except Exception as e:
return {"success": False, "error": str(e)}
# ==================== PATTERN FUNCTIONS ====================
def add_pattern(
pattern_type,
description,
trigger=None,
frequency=None,
time_context=None,
notes=None,
):
"""Add a recurring pattern
Types: daily, weekly, work, personal, seasonal, situational
"""
try:
conn = get_connection()
cursor = conn.cursor()
cursor.execute(
"""
INSERT INTO patterns (pattern_type, description, trigger, frequency, time_context, notes)
VALUES (?, ?, ?, ?, ?, ?)
""",
(pattern_type, description, trigger, frequency, time_context, notes),
)
conn.commit()
conn.close()
return {"success": True, "message": f"Pattern added: {description[:50]}..."}
except Exception as e:
return {"success": False, "error": str(e)}
def get_patterns(pattern_type=None):
"""Get patterns"""
try:
conn = get_connection()
cursor = conn.cursor()
if pattern_type:
cursor.execute(
"SELECT * FROM patterns WHERE pattern_type = ?", (pattern_type,)
)
else:
cursor.execute("SELECT * FROM patterns ORDER BY pattern_type")
rows = cursor.fetchall()
conn.close()
return {"success": True, "patterns": [dict(row) for row in rows]}
except Exception as e:
return {"success": False, "error": str(e)}
# ==================== GOAL FUNCTIONS ====================
def add_goal(
goal_type, description, timeframe=None, priority="medium", milestones=None
):
"""Add a goal
Types: career, learning, health, financial, personal, project
"""
try:
conn = get_connection()
cursor = conn.cursor()
cursor.execute(
"""
INSERT INTO goals (goal_type, description, timeframe, priority, milestones)
VALUES (?, ?, ?, ?, ?)
""",
(goal_type, description, timeframe, priority, milestones),
)
conn.commit()
conn.close()
return {"success": True, "message": f"Goal added: {description[:50]}..."}
except Exception as e:
return {"success": False, "error": str(e)}
def get_goals(goal_type=None, status="active"):
"""Get goals"""
try:
conn = get_connection()
cursor = conn.cursor()
query = "SELECT * FROM goals WHERE 1=1"
params = []
if goal_type:
query += " AND goal_type = ?"
params.append(goal_type)
if status:
query += " AND status = ?"
params.append(status)
query += " ORDER BY priority DESC"
cursor.execute(query, params)
rows = cursor.fetchall()
conn.close()
return {"success": True, "goals": [dict(row) for row in rows]}
except Exception as e:
return {"success": False, "error": str(e)}
def update_goal(goal_id, **kwargs):
"""Update a goal"""
try:
conn = get_connection()
cursor = conn.cursor()
updates = []
values = []
for field, value in kwargs.items():
updates.append(f"{field} = ?")
values.append(value)
updates.append("updated_at = ?")
values.append(datetime.now())
values.append(goal_id)
cursor.execute(
f"""
UPDATE goals SET {', '.join(updates)} WHERE id = ?
""",
values,
)
conn.commit()
conn.close()
return {"success": True, "goal_id": goal_id}
except Exception as e:
return {"success": False, "error": str(e)}
# ==================== FACT FUNCTIONS ====================
def add_fact(category, fact, context=None, source=None):
"""Add a factual piece of information
Categories: personal, professional, education, achievement, certification, etc.
"""
try:
conn = get_connection()
cursor = conn.cursor()
cursor.execute(
"""
INSERT INTO facts (category, fact, context, source)
VALUES (?, ?, ?, ?)
""",
(category, fact, context, source),
)
conn.commit()
conn.close()
return {"success": True, "message": f"Fact added: {fact[:50]}..."}
except Exception as e:
return {"success": False, "error": str(e)}
def get_facts(category=None, search=None):
"""Get facts"""
try:
conn = get_connection()
cursor = conn.cursor()
query = "SELECT * FROM facts WHERE 1=1"
params = []
if category:
query += " AND category = ?"
params.append(category)
if search:
query += " AND (fact LIKE ? OR context LIKE ?)"
params.extend([f"%{search}%", f"%{search}%"])
cursor.execute(query, params)
rows = cursor.fetchall()
conn.close()
return {"success": True, "facts": [dict(row) for row in rows]}
except Exception as e:
return {"success": False, "error": str(e)}
# ==================== WORK STYLE FUNCTIONS ====================
def add_work_style(aspect, description, examples=None, tips_for_ai=None):
"""Add work style information
Aspects: time_management, focus, collaboration, communication, tools, environment
"""
try:
conn = get_connection()
cursor = conn.cursor()
cursor.execute(
"""
INSERT INTO work_style (aspect, description, examples, tips_for_ai)
VALUES (?, ?, ?, ?)
""",
(aspect, description, examples, tips_for_ai),
)
conn.commit()
conn.close()
return {"success": True, "message": f"Work style added: {aspect}"}
except Exception as e:
return {"success": False, "error": str(e)}
def get_work_style(aspect=None):
"""Get work style information"""
try:
conn = get_connection()
cursor = conn.cursor()
if aspect:
cursor.execute("SELECT * FROM work_style WHERE aspect = ?", (aspect,))
else:
cursor.execute("SELECT * FROM work_style")
rows = cursor.fetchall()
conn.close()
return {"success": True, "work_style": [dict(row) for row in rows]}
except Exception as e:
return {"success": False, "error": str(e)}
# ==================== COMMUNICATION STYLE FUNCTIONS ====================
def add_communication_style(aspect, preference, examples=None, avoid=None):
"""Add communication style preference
Aspects: tone, formality, length, format, frequency, channels
"""
try:
conn = get_connection()
cursor = conn.cursor()
cursor.execute(
"""
INSERT INTO communication_style (aspect, preference, examples, avoid)
VALUES (?, ?, ?, ?)
""",
(aspect, preference, examples, avoid),
)
conn.commit()
conn.close()
return {"success": True, "message": f"Communication style added: {aspect}"}
except Exception as e:
return {"success": False, "error": str(e)}
def get_communication_style(aspect=None):
"""Get communication style"""
try:
conn = get_connection()
cursor = conn.cursor()
if aspect:
cursor.execute(
"SELECT * FROM communication_style WHERE aspect = ?", (aspect,)
)
else:
cursor.execute("SELECT * FROM communication_style")
rows = cursor.fetchall()
conn.close()
return {"success": True, "communication_style": [dict(row) for row in rows]}
except Exception as e:
return {"success": False, "error": str(e)}
# ==================== QUICK MEMORY FUNCTIONS ====================
def remember(topic, content, tags=None, importance="normal", expires_at=None):
"""Quickly remember something
This is for ad-hoc information that doesn't fit other categories
"""
try:
conn = get_connection()
cursor = conn.cursor()
tags_str = ",".join(tags) if isinstance(tags, list) else tags
cursor.execute(
"""
INSERT INTO quick_memory (topic, content, tags, importance, expires_at)
VALUES (?, ?, ?, ?, ?)
""",
(topic, content, tags_str, importance, expires_at),
)
conn.commit()
conn.close()
return {"success": True, "message": f"Remembered: {topic}"}
except Exception as e:
return {"success": False, "error": str(e)}
def recall(topic=None, tags=None, search=None):
"""Recall stored memories"""
try:
conn = get_connection()
cursor = conn.cursor()
query = 'SELECT * FROM quick_memory WHERE (expires_at IS NULL OR expires_at > datetime("now"))'
params = []
if topic:
query += " AND topic LIKE ?"
params.append(f"%{topic}%")
if tags:
query += " AND tags LIKE ?"
params.append(f"%{tags}%")
if search:
query += " AND (topic LIKE ? OR content LIKE ?)"
params.extend([f"%{search}%", f"%{search}%"])
query += " ORDER BY importance DESC, created_at DESC"
cursor.execute(query, params)
rows = cursor.fetchall()
conn.close()
return {"success": True, "memories": [dict(row) for row in rows]}
except Exception as e:
return {"success": False, "error": str(e)}
# ==================== LEARNING LOG FUNCTIONS ====================
def learn_about_user(category, learned, context=None, confidence="medium"):
"""Log something learned about the user
Categories: preference, behavior, style, fact, pattern
"""
try:
conn = get_connection()
cursor = conn.cursor()
cursor.execute(
"""
INSERT INTO learning_log (category, learned, context, confidence)
VALUES (?, ?, ?, ?)
""",
(category, learned, context, confidence),
)
conn.commit()
conn.close()
return {"success": True, "message": f"Learned: {learned[:50]}..."}
except Exception as e:
return {"success": False, "error": str(e)}
def get_learnings(category=None, limit=50):
"""Get learnings about the user"""
try:
conn = get_connection()
cursor = conn.cursor()
if category:
cursor.execute(
"""
SELECT * FROM learning_log WHERE category = ?
ORDER BY timestamp DESC LIMIT ?
""",
(category, limit),
)
else:
cursor.execute(
"""
SELECT * FROM learning_log ORDER BY timestamp DESC LIMIT ?
""",
(limit,),
)
rows = cursor.fetchall()
conn.close()
return {"success": True, "learnings": [dict(row) for row in rows]}
except Exception as e:
return {"success": False, "error": str(e)}
# ==================== INTERACTION LOG FUNCTIONS ====================
def log_interaction(
interaction_type, summary, details=None, outcome=None, learnings=None
):
"""Log an interaction or activity
Types: conversation, task, decision, meeting, research, project
"""
try:
conn = get_connection()
cursor = conn.cursor()
cursor.execute(
"""
INSERT INTO interactions (interaction_type, summary, details, outcome, learnings)
VALUES (?, ?, ?, ?, ?)
""",
(interaction_type, summary, details, outcome, learnings),
)
conn.commit()
conn.close()
return {"success": True, "message": f"Interaction logged: {summary[:50]}..."}
except Exception as e:
return {"success": False, "error": str(e)}
def get_interactions(interaction_type=None, limit=50):
"""Get past interactions"""
try:
conn = get_connection()
cursor = conn.cursor()
if interaction_type:
cursor.execute(
"""
SELECT * FROM interactions WHERE interaction_type = ?
ORDER BY timestamp DESC LIMIT ?
""",
(interaction_type, limit),
)
else:
cursor.execute(
"""
SELECT * FROM interactions ORDER BY timestamp DESC LIMIT ?
""",
(limit,),
)
rows = cursor.fetchall()
conn.close()
return {"success": True, "interactions": [dict(row) for row in rows]}
except Exception as e:
return {"success": False, "error": str(e)}
# ==================== RELATIONSHIP FUNCTIONS ====================
def add_relationship(
name, relationship_type, context=None, importance="medium", notes=None
):
"""Add a relationship/contact"""
try:
conn = get_connection()
cursor = conn.cursor()
cursor.execute(
"""
INSERT INTO relationships (name, relationship_type, context, importance, notes)
VALUES (?, ?, ?, ?, ?)
""",
(name, relationship_type, context, importance, notes),
)
conn.commit()
conn.close()
return {"success": True, "message": f"Relationship added: {name}"}
except Exception as e:
return {"success": False, "error": str(e)}
def get_relationships(relationship_type=None, search=None):
"""Get relationships"""
try:
conn = get_connection()
cursor = conn.cursor()
query = "SELECT * FROM relationships WHERE 1=1"
params = []
if relationship_type:
query += " AND relationship_type = ?"
params.append(relationship_type)
if search:
query += " AND (name LIKE ? OR context LIKE ? OR notes LIKE ?)"
params.extend([f"%{search}%"] * 3)
query += " ORDER BY importance DESC, name"
cursor.execute(query, params)
rows = cursor.fetchall()
conn.close()
return {"success": True, "relationships": [dict(row) for row in rows]}
except Exception as e:
return {"success": False, "error": str(e)}
# ==================== DECISION FUNCTIONS ====================
def add_decision_style(decision_area, approach, factors=None, examples=None):
"""Add how you make decisions in a certain area
Areas: career, financial, technical, personal, work, purchases
"""
try:
conn = get_connection()
cursor = conn.cursor()
cursor.execute(
"""
INSERT INTO decisions (decision_area, approach, factors, examples)
VALUES (?, ?, ?, ?)
""",
(decision_area, approach, factors, examples),
)
conn.commit()
conn.close()
return {"success": True, "message": f"Decision style added: {decision_area}"}
except Exception as e:
return {"success": False, "error": str(e)}
def get_decision_styles(decision_area=None):
"""Get decision-making styles"""
try:
conn = get_connection()
cursor = conn.cursor()
if decision_area:
cursor.execute(
"SELECT * FROM decisions WHERE decision_area = ?", (decision_area,)
)
else:
cursor.execute("SELECT * FROM decisions")
rows = cursor.fetchall()
conn.close()
return {"success": True, "decisions": [dict(row) for row in rows]}
except Exception as e:
return {"success": False, "error": str(e)}
# ==================== COMPREHENSIVE RETRIEVAL ====================
def get_everything_about_user():
"""Get ALL stored information about the user"""
try:
conn = get_connection()
cursor = conn.cursor()
result = {"success": True, "complete_profile": {}}
# Get all tables
tables = [
("identity", "SELECT * FROM identity"),
("behaviors", "SELECT * FROM behaviors"),
("preferences", "SELECT * FROM preferences"),
("patterns", "SELECT * FROM patterns"),
("goals", 'SELECT * FROM goals WHERE status = "active"'),
("facts", "SELECT * FROM facts"),
("work_style", "SELECT * FROM work_style"),
("communication_style", "SELECT * FROM communication_style"),
("relationships", "SELECT * FROM relationships"),
("decisions", "SELECT * FROM decisions"),
(
"recent_learnings",
"SELECT * FROM learning_log ORDER BY timestamp DESC LIMIT 20",
),
(
"recent_memories",
'SELECT * FROM quick_memory WHERE expires_at IS NULL OR expires_at > datetime("now") ORDER BY created_at DESC LIMIT 20',
),
]
for table_name, query in tables:
cursor.execute(query)
rows = cursor.fetchall()
result["complete_profile"][table_name] = [dict(row) for row in rows]
conn.close()
return result
except Exception as e:
return {"success": False, "error": str(e)}
def search_all_memory(search_term):
"""Search across all memory tables"""
try:
conn = get_connection()
cursor = conn.cursor()
results = []
# Search in multiple tables
searches = [
(
"identity",
'SELECT "identity" as source, key as title, value as content FROM identity WHERE key LIKE ? OR value LIKE ?',
),
(
"behaviors",
'SELECT "behavior" as source, behavior_type as title, description as content FROM behaviors WHERE description LIKE ?',
),
(
"preferences",
'SELECT "preference" as source, preference as title, value as content FROM preferences WHERE preference LIKE ? OR value LIKE ?',
),
(
"facts",
'SELECT "fact" as source, category as title, fact as content FROM facts WHERE fact LIKE ?',
),
(
"goals",
'SELECT "goal" as source, goal_type as title, description as content FROM goals WHERE description LIKE ?',
),
(
"memories",
'SELECT "memory" as source, topic as title, content FROM quick_memory WHERE topic LIKE ? OR content LIKE ?',
),
]
term = f"%{search_term}%"
for source, query in searches:
try:
param_count = query.count("?")
cursor.execute(query, tuple([term] * param_count))
rows = cursor.fetchall()
results.extend([dict(row) for row in rows])
except Exception:
pass
conn.close()
return {"success": True, "results": results, "count": len(results)}
except Exception as e:
return {"success": False, "error": str(e)}
