Create services/database_service.py

src/core/services/database_service.py

@@ -0,0 +1,196 @@
+"""Database service for managing application data"""
+import sqlite3
+from contextlib import contextmanager
+from pathlib import Path
+from typing import Dict, List, Any, Optional
+import json
+from datetime import datetime
+
+class DatabaseService:
+    def __init__(self, db_path: str = "data/robata.db"):
+        """Initialize database service"""
+        self.db_path = db_path
+        Path(db_path).parent.mkdir(parents=True, exist_ok=True)
+        self.setup_database()
+    
+    @contextmanager
+    def get_db(self):
+        """Context manager for database connections"""
+        conn = sqlite3.connect(self.db_path)
+        try:
+            yield conn
+        finally:
+            conn.close()
+    
+    def setup_database(self):
+        """Create database tables if they don't exist"""
+        with self.get_db() as conn:
+            c = conn.cursor()
+            c.executescript('''
+                CREATE TABLE IF NOT EXISTS users (
+                    id TEXT PRIMARY KEY,
+                    email TEXT UNIQUE,
+                    name TEXT,
+                    role TEXT,
+                    department TEXT,
+                    title TEXT,
+                    company TEXT,
+                    created_at TEXT
+                );
+                
+                CREATE TABLE IF NOT EXISTS accounts (
+                    id TEXT PRIMARY KEY,
+                    name TEXT,
+                    industry TEXT,
+                    status TEXT,
+                    website TEXT,
+                    annual_revenue REAL,
+                    employee_count INTEGER,
+                    created_at TEXT
+                );
+                
+                CREATE TABLE IF NOT EXISTS contacts (
+                    id TEXT PRIMARY KEY,
+                    account_id TEXT,
+                    first_name TEXT,
+                    last_name TEXT,
+                    email TEXT,
+                    title TEXT,
+                    department TEXT,
+                    influence_level TEXT,
+                    created_at TEXT,
+                    FOREIGN KEY (account_id) REFERENCES accounts (id)
+                );
+                
+                CREATE TABLE IF NOT EXISTS interactions (
+                    id TEXT PRIMARY KEY,
+                    type TEXT,
+                    account_id TEXT,
+                    owner_id TEXT,
+                    transcript TEXT,
+                    summary TEXT,
+                    sentiment_score REAL,
+                    metadata TEXT,
+                    created_at TEXT,
+                    FOREIGN KEY (account_id) REFERENCES accounts (id),
+                    FOREIGN KEY (owner_id) REFERENCES users (id)
+                );
+            ''')
+    
+    def get_user_by_email(self, email: str) -> Optional[Dict]:
+        """Get user by email address"""
+        with self.get_db() as conn:
+            cursor = conn.execute(
+                "SELECT * FROM users WHERE email = ?",
+                (email,)
+            )
+            row = cursor.fetchone()
+            if row:
+                return {
+                    'id': row[0],
+                    'email': row[1],
+                    'name': row[2],
+                    'role': row[3],
+                    'department': row[4],
+                    'title': row[5],
+                    'company': row[6]
+                }
+            return None
+    
+    def get_user_accounts(self, user_id: str) -> List[Dict]:
+        """Get accounts associated with user"""
+        with self.get_db() as conn:
+            cursor = conn.execute("""
+                SELECT DISTINCT a.*
+                FROM accounts a
+                JOIN interactions i ON a.id = i.account_id
+                WHERE i.owner_id = ?
+                ORDER BY a.name
+            """, (user_id,))
+            return [
+                {
+                    'id': row[0],
+                    'name': row[1],
+                    'industry': row[2],
+                    'status': row[3],
+                    'website': row[4],
+                    'annual_revenue': row[5],
+                    'employee_count': row[6]
+                }
+                for row in cursor.fetchall()
+            ]
+    
+    def get_account_metrics(self, account_id: str) -> Dict:
+        """Get key metrics for an account"""
+        with self.get_db() as conn:
+            # Get total interactions
+            interaction_count = conn.execute(
+                "SELECT COUNT(*) FROM interactions WHERE account_id = ?",
+                (account_id,)
+            ).fetchone()[0]
+            
+            # Get contact count
+            contact_count = conn.execute(
+                "SELECT COUNT(*) FROM contacts WHERE account_id = ?",
+                (account_id,)
+            ).fetchone()[0]
+            
+            # Get average sentiment
+            avg_sentiment = conn.execute(
+                "SELECT AVG(sentiment_score) FROM interactions WHERE account_id = ?",
+                (account_id,)
+            ).fetchone()[0]
+            
+            return {
+                'interaction_count': interaction_count,
+                'contact_count': contact_count,
+                'avg_sentiment': avg_sentiment or 0
+            }
+    
+    def save_interaction(self, interaction_data: Dict) -> str:
+        """Save new interaction"""
+        with self.get_db() as conn:
+            cursor = conn.execute("""
+                INSERT INTO interactions (
+                    id, type, account_id, owner_id, transcript,
+                    summary, sentiment_score, metadata, created_at
+                ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
+            """, (
+                interaction_data['id'],
+                interaction_data['type'],
+                interaction_data['account_id'],
+                interaction_data['owner_id'],
+                interaction_data['transcript'],
+                interaction_data.get('summary'),
+                interaction_data.get('sentiment_score'),
+                json.dumps(interaction_data.get('metadata', {})),
+                datetime.now().isoformat()
+            ))
+            conn.commit()
+            return interaction_data['id']
+    
+    def get_recent_interactions(self, user_id: str, limit: int = 10) -> List[Dict]:
+        """Get recent interactions for user"""
+        with self.get_db() as conn:
+            cursor = conn.execute("""
+                SELECT i.*, a.name as account_name
+                FROM interactions i
+                JOIN accounts a ON i.account_id = a.id
+                WHERE i.owner_id = ?
+                ORDER BY i.created_at DESC
+                LIMIT ?
+            """, (user_id, limit))
+            return [
+                {
+                    'id': row[0],
+                    'type': row[1],
+                    'account_id': row[2],
+                    'account_name': row[9],
+                    'transcript': row[4],
+                    'summary': row[5],
+                    'sentiment_score': row[6],
+                    'metadata': json.loads(row[7]) if row[7] else {},
+                    'created_at': row[8]
+                }
+                for row in cursor.fetchall()
+            ]