database_manager.py

"""
Database Manager Module
Handles user registration data and verification logs
"""

import json
import sqlite3
import numpy as np
import logging
from datetime import datetime
from typing import Optional, Dict, Any, List
from pathlib import Path

logger = logging.getLogger(__name__)


class DatabaseManager:
    """
    Manages user face data and verification logs using SQLite
    """
    
    def __init__(self, db_path: str = "face_verification.db"):
        """
        Initialize database manager
        
        Args:
            db_path: Path to SQLite database file
        """
        self.db_path = db_path
        self.conn = None
    
    def initialize(self):
        """Initialize database and create tables"""
        try:
            self.conn = sqlite3.connect(self.db_path, check_same_thread=False)
            self._create_tables()
            logger.info(f"✓ Database initialized: {self.db_path}")
        except Exception as e:
            logger.error(f"Database initialization error: {e}")
            raise
    
    def _create_tables(self):
        """Create necessary database tables"""
        cursor = self.conn.cursor()
        
        # Users table
        cursor.execute("""
            CREATE TABLE IF NOT EXISTS users (
                user_id TEXT PRIMARY KEY,
                embedding BLOB NOT NULL,
                image_path TEXT NOT NULL,
                registered_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
                updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
            )
        """)
        
        # Verification logs table
        cursor.execute("""
            CREATE TABLE IF NOT EXISTS verification_logs (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                user_id TEXT NOT NULL,
                verified_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
                is_match BOOLEAN NOT NULL,
                confidence REAL NOT NULL,
                is_live BOOLEAN,
                FOREIGN KEY (user_id) REFERENCES users(user_id)
            )
        """)
        
        # Create indexes
        cursor.execute("""
            CREATE INDEX IF NOT EXISTS idx_user_id 
            ON verification_logs(user_id)
        """)
        
        cursor.execute("""
            CREATE INDEX IF NOT EXISTS idx_verified_at 
            ON verification_logs(verified_at)
        """)
        
        self.conn.commit()
        logger.info("✓ Database tables created/verified")
    
    def register_user(self, user_id: str, embedding: np.ndarray, image_path: str) -> bool:
        """
        Register a new user with their face embedding
        
        Args:
            user_id: Unique user identifier
            embedding: Face embedding vector
            image_path: Path to stored profile image
        
        Returns:
            True if successful, False otherwise
        """
        try:
            cursor = self.conn.cursor()
            
            # Convert embedding to bytes
            embedding_bytes = embedding.tobytes()
            
            # Insert or replace user
            cursor.execute("""
                INSERT OR REPLACE INTO users (user_id, embedding, image_path, updated_at)
                VALUES (?, ?, ?, ?)
            """, (user_id, embedding_bytes, image_path, datetime.now()))
            
            self.conn.commit()
            logger.info(f"✓ User {user_id} registered successfully")
            return True
            
        except Exception as e:
            logger.error(f"User registration error: {e}")
            self.conn.rollback()
            return False
    
    def get_user(self, user_id: str) -> Optional[Dict[str, Any]]:
        """
        Get user data by user_id
        
        Args:
            user_id: User identifier
        
        Returns:
            User data dictionary or None
        """
        try:
            cursor = self.conn.cursor()
            
            cursor.execute("""
                SELECT user_id, embedding, image_path, registered_at, updated_at
                FROM users
                WHERE user_id = ?
            """, (user_id,))
            
            row = cursor.fetchone()
            
            if row is None:
                return None
            
            # Convert embedding bytes back to numpy array
            embedding = np.frombuffer(row[1], dtype=np.float64)
            
            return {
                'user_id': row[0],
                'embedding': embedding,
                'image_path': row[2],
                'registered_at': row[3],
                'updated_at': row[4]
            }
            
        except Exception as e:
            logger.error(f"Get user error: {e}")
            return None
    
    def user_exists(self, user_id: str) -> bool:
        """
        Check if user exists in database
        
        Args:
            user_id: User identifier
        
        Returns:
            True if user exists, False otherwise
        """
        try:
            cursor = self.conn.cursor()
            cursor.execute("SELECT 1 FROM users WHERE user_id = ?", (user_id,))
            return cursor.fetchone() is not None
        except Exception as e:
            logger.error(f"User exists check error: {e}")
            return False
    
    def record_verification(self, user_id: str, is_match: bool, 
                          confidence: float, is_live: Optional[bool] = None) -> bool:
        """
        Record a verification attempt
        
        Args:
            user_id: User identifier
            is_match: Whether faces matched
            confidence: Confidence score
            is_live: Liveness detection result
        
        Returns:
            True if successful, False otherwise
        """
        try:
            cursor = self.conn.cursor()
            
            cursor.execute("""
                INSERT INTO verification_logs (user_id, is_match, confidence, is_live)
                VALUES (?, ?, ?, ?)
            """, (user_id, is_match, confidence, is_live))
            
            self.conn.commit()
            logger.debug(f"✓ Verification logged for user {user_id}")
            return True
            
        except Exception as e:
            logger.error(f"Verification logging error: {e}")
            self.conn.rollback()
            return False
    
    def get_user_verification_history(self, user_id: str, limit: int = 10) -> List[Dict[str, Any]]:
        """
        Get verification history for a user
        
        Args:
            user_id: User identifier
            limit: Maximum number of records to return
        
        Returns:
            List of verification records
        """
        try:
            cursor = self.conn.cursor()
            
            cursor.execute("""
                SELECT id, verified_at, is_match, confidence, is_live
                FROM verification_logs
                WHERE user_id = ?
                ORDER BY verified_at DESC
                LIMIT ?
            """, (user_id, limit))
            
            rows = cursor.fetchall()
            
            history = []
            for row in rows:
                history.append({
                    'id': row[0],
                    'verified_at': row[1],
                    'is_match': bool(row[2]),
                    'confidence': row[3],
                    'is_live': bool(row[4]) if row[4] is not None else None
                })
            
            return history
            
        except Exception as e:
            logger.error(f"Get verification history error: {e}")
            return []
    
    def get_statistics(self) -> Dict[str, Any]:
        """
        Get system statistics
        
        Returns:
            Dictionary with statistics
        """
        try:
            cursor = self.conn.cursor()
            
            # Total users
            cursor.execute("SELECT COUNT(*) FROM users")
            total_users = cursor.fetchone()[0]
            
            # Total verifications
            cursor.execute("SELECT COUNT(*) FROM verification_logs")
            total_verifications = cursor.fetchone()[0]
            
            # Successful verifications
            cursor.execute("SELECT COUNT(*) FROM verification_logs WHERE is_match = 1")
            successful_verifications = cursor.fetchone()[0]
            
            # Success rate
            success_rate = (successful_verifications / total_verifications * 100) if total_verifications > 0 else 0
            
            # Liveness detection stats
            cursor.execute("SELECT COUNT(*) FROM verification_logs WHERE is_live = 1")
            live_detections = cursor.fetchone()[0]
            
            cursor.execute("SELECT COUNT(*) FROM verification_logs WHERE is_live = 0")
            spoof_detections = cursor.fetchone()[0]
            
            # Recent verifications (last 24 hours)
            cursor.execute("""
                SELECT COUNT(*) FROM verification_logs 
                WHERE verified_at >= datetime('now', '-1 day')
            """)
            recent_verifications = cursor.fetchone()[0]
            
            # Average confidence
            cursor.execute("SELECT AVG(confidence) FROM verification_logs WHERE is_match = 1")
            avg_confidence = cursor.fetchone()[0] or 0
            
            return {
                'total_users': total_users,
                'total_verifications': total_verifications,
                'successful_verifications': successful_verifications,
                'success_rate': round(success_rate, 2),
                'live_detections': live_detections,
                'spoof_detections': spoof_detections,
                'recent_verifications_24h': recent_verifications,
                'average_confidence': round(avg_confidence, 4)
            }
            
        except Exception as e:
            logger.error(f"Get statistics error: {e}")
            return {
                'total_users': 0,
                'total_verifications': 0,
                'successful_verifications': 0,
                'success_rate': 0,
                'live_detections': 0,
                'spoof_detections': 0,
                'recent_verifications_24h': 0,
                'average_confidence': 0
            }
    
    def delete_user(self, user_id: str) -> bool:
        """
        Delete a user and their verification logs
        
        Args:
            user_id: User identifier
        
        Returns:
            True if successful, False otherwise
        """
        try:
            cursor = self.conn.cursor()
            
            # Delete verification logs
            cursor.execute("DELETE FROM verification_logs WHERE user_id = ?", (user_id,))
            
            # Delete user
            cursor.execute("DELETE FROM users WHERE user_id = ?", (user_id,))
            
            self.conn.commit()
            logger.info(f"✓ User {user_id} deleted successfully")
            return True
            
        except Exception as e:
            logger.error(f"Delete user error: {e}")
            self.conn.rollback()
            return False
    
    def get_all_users(self) -> List[Dict[str, Any]]:
        """
        Get all registered users
        
        Returns:
            List of user data dictionaries
        """
        try:
            cursor = self.conn.cursor()
            
            cursor.execute("""
                SELECT user_id, image_path, registered_at, updated_at
                FROM users
                ORDER BY registered_at DESC
            """)
            
            rows = cursor.fetchall()
            
            users = []
            for row in rows:
                users.append({
                    'user_id': row[0],
                    'image_path': row[1],
                    'registered_at': row[2],
                    'updated_at': row[3]
                })
            
            return users
            
        except Exception as e:
            logger.error(f"Get all users error: {e}")
            return []
    
    def close(self):
        """Close database connection"""
        if self.conn:
            self.conn.close()
            logger.info("✓ Database connection closed")
    
    def __del__(self):
        """Cleanup on deletion"""
        self.close()