utils/fingerprint_manager.py

"""
Minimal Privacy-Focused Fingerprint Manager
Automatically identifies unique users with minimal device data collection.
Maintains privacy through hashing and generates consistent UUIDs.
"""

import hashlib
import json
import platform
import psutil
import uuid
import os
from typing import Dict, Any, Optional
import logging
from pathlib import Path


class MinimalFingerprintManager:
    """
    Minimal device fingerprinting for automatic user identification.
    Collects only essential data needed for reliable identification.
    All sensitive data is hashed for privacy protection.
    """

    def __init__(self):
        """Initialize the fingerprint manager."""
        self.logger = logging.getLogger(__name__)
        self.cache_file = Path("user_fingerprints.json")
        self._load_cache()

    def _load_cache(self) -> None:
        """Load cached fingerprints and user mappings."""
        try:
            if self.cache_file.exists():
                with open(self.cache_file, "r") as f:
                    self.cache = json.load(f)
            else:
                self.cache = {
                    "fingerprints": {},  # fingerprint_hash -> user_uuid
                    "user_stats": {},  # user_uuid -> usage stats
                    "created_at": {},  # user_uuid -> first_seen timestamp
                }
        except Exception as e:
            self.logger.warning(f"Failed to load fingerprint cache: {e}")
            self.cache = {"fingerprints": {}, "user_stats": {}, "created_at": {}}

    def _save_cache(self) -> None:
        """Save fingerprint cache to disk."""
        try:
            with open(self.cache_file, "w") as f:
                json.dump(self.cache, f, indent=2)
        except Exception as e:
            self.logger.warning(f"Failed to save fingerprint cache: {e}")

    def _get_minimal_fingerprint(self) -> Dict[str, Any]:
        """
        Collect minimal device data for fingerprinting.
        Only essential data that's stable and privacy-safe.
        """
        try:
            # Core system information (stable across reboots)
            fingerprint = {
                # OS and architecture (stable)
                "os_system": platform.system(),
                "os_release": platform.release(),
                "architecture": platform.machine(),
                # Hardware characteristics (stable)
                "cpu_count_logical": psutil.cpu_count(logical=True),
                "cpu_count_physical": psutil.cpu_count(logical=False),
                "memory_total_gb": round(psutil.virtual_memory().total / (1024**3), 1),
                # System boot time hash (for session consistency)
                "boot_time_hash": hashlib.sha256(
                    str(int(psutil.boot_time())).encode()
                ).hexdigest()[:16],
                # User context hash (privacy-safe)
                "user_context_hash": hashlib.sha256(
                    (str(Path.home()) + os.getlogin()).encode()
                ).hexdigest()[:16],
            }

            # Add MAC address hash if available (most stable identifier)
            try:
                for interface, addrs in psutil.net_if_addrs().items():
                    for addr in addrs:
                        if (
                            addr.family == psutil.AF_LINK
                            and addr.address != "00:00:00:00:00:00"
                        ):
                            fingerprint["mac_hash"] = hashlib.sha256(
                                addr.address.encode()
                            ).hexdigest()[:16]
                            break
                    if "mac_hash" in fingerprint:
                        break
            except Exception:
                pass  # MAC address not available, continue without it

            return fingerprint

        except Exception as e:
            self.logger.error(f"Error generating fingerprint: {e}")
            # Fallback minimal fingerprint
            return {
                "os_system": platform.system(),
                "fallback": True,
                "error": str(e)[:50],
            }

    def _generate_fingerprint_hash(self, fingerprint: Dict[str, Any]) -> str:
        """Generate a consistent hash from fingerprint data."""
        # Sort keys for consistent hashing
        fingerprint_str = json.dumps(fingerprint, sort_keys=True)
        return hashlib.sha256(fingerprint_str.encode()).hexdigest()

    def get_user_uuid(self) -> str:
        """
        Get or create a consistent UUID for the current user.

        Returns:
            str: Consistent UUID for this user/device combination
        """
        # Generate current device fingerprint
        fingerprint = self._get_minimal_fingerprint()
        fingerprint_hash = self._generate_fingerprint_hash(fingerprint)

        # Check if we've seen this fingerprint before
        if fingerprint_hash in self.cache["fingerprints"]:
            user_uuid = self.cache["fingerprints"][fingerprint_hash]
            self.logger.info(f"Recognized returning user: {user_uuid[:8]}...")
        else:
            # New user - generate UUID
            user_uuid = str(uuid.uuid4())
            self.cache["fingerprints"][fingerprint_hash] = user_uuid
            self.cache["created_at"][user_uuid] = psutil.boot_time()
            self.cache["user_stats"][user_uuid] = {
                "total_operations": 0,
                "memories_stored": 0,
                "searches_performed": 0,
                "videos_built": 0,
                "first_seen": psutil.boot_time(),
                "last_seen": psutil.boot_time(),
                "device_info": {
                    "os": fingerprint.get("os_system", "unknown"),
                    "architecture": fingerprint.get("architecture", "unknown"),
                    "cpu_cores": fingerprint.get("cpu_count_logical", 0),
                    "memory_gb": fingerprint.get("memory_total_gb", 0),
                },
            }
            self._save_cache()
            self.logger.info(f"New user registered: {user_uuid[:8]}...")

        # Update last seen
        if user_uuid in self.cache["user_stats"]:
            self.cache["user_stats"][user_uuid]["last_seen"] = psutil.boot_time()
            self._save_cache()

        return user_uuid

    def update_user_stats(self, user_uuid: str, operation_type: str) -> None:
        """
        Update usage statistics for a user.

        Args:
            user_uuid (str): User's UUID
            operation_type (str): Type of operation performed
        """
        if user_uuid not in self.cache["user_stats"]:
            # Initialize stats for existing user
            self.cache["user_stats"][user_uuid] = {
                "total_operations": 0,
                "memories_stored": 0,
                "searches_performed": 0,
                "videos_built": 0,
                "first_seen": psutil.boot_time(),
                "last_seen": psutil.boot_time(),
                "device_info": {
                    "os": "unknown",
                    "architecture": "unknown",
                    "cpu_cores": 0,
                    "memory_gb": 0,
                },
            }

        # Update counters
        stats = self.cache["user_stats"][user_uuid]
        stats["total_operations"] += 1
        stats["last_seen"] = psutil.boot_time()

        # Update specific operation counters
        if operation_type in ["store_memory", "store_document"]:
            stats["memories_stored"] += 1
        elif operation_type in ["search_memory", "chat_with_memory"]:
            stats["searches_performed"] += 1
        elif operation_type == "build_memory_video":
            stats["videos_built"] += 1

        self._save_cache()

    def get_user_stats(self, user_uuid: str) -> Dict[str, Any]:
        """
        Get usage statistics for a user.

        Args:
            user_uuid (str): User's UUID

        Returns:
            Dict: User's usage statistics
        """
        if user_uuid not in self.cache["user_stats"]:
            return {"error": "User not found", "user_uuid": user_uuid}

        stats = self.cache["user_stats"][user_uuid].copy()

        # Add computed fields
        import time

        current_time = time.time()
        stats["days_since_first_seen"] = round(
            (current_time - stats["first_seen"]) / 86400, 1
        )
        stats["days_since_last_seen"] = round(
            (current_time - stats["last_seen"]) / 86400, 1
        )

        return {
            "user_uuid": user_uuid,
            "user_id_short": user_uuid[:8],
            "statistics": stats,
            "privacy_note": "All device data is hashed for privacy protection",
        }

    def get_all_users_stats(self) -> Dict[str, Any]:
        """Get aggregated statistics for all users."""
        total_users = len(self.cache["user_stats"])
        if total_users == 0:
            return {"total_users": 0, "message": "No users registered yet"}

        # Aggregate statistics
        total_operations = sum(
            stats["total_operations"] for stats in self.cache["user_stats"].values()
        )
        total_memories = sum(
            stats["memories_stored"] for stats in self.cache["user_stats"].values()
        )
        total_searches = sum(
            stats["searches_performed"] for stats in self.cache["user_stats"].values()
        )
        total_videos = sum(
            stats["videos_built"] for stats in self.cache["user_stats"].values()
        )

        # Device diversity
        os_counts = {}
        arch_counts = {}
        for stats in self.cache["user_stats"].values():
            device_info = stats.get("device_info", {})
            os_name = device_info.get("os", "unknown")
            arch_name = device_info.get("architecture", "unknown")

            os_counts[os_name] = os_counts.get(os_name, 0) + 1
            arch_counts[arch_name] = arch_counts.get(arch_name, 0) + 1

        return {
            "total_users": total_users,
            "aggregated_stats": {
                "total_operations": total_operations,
                "total_memories_stored": total_memories,
                "total_searches_performed": total_searches,
                "total_videos_built": total_videos,
                "avg_operations_per_user": round(total_operations / total_users, 1),
            },
            "device_diversity": {
                "operating_systems": os_counts,
                "architectures": arch_counts,
            },
            "privacy_note": "All statistics are aggregated and anonymized",
        }

    def get_fingerprint_info(self) -> Dict[str, Any]:
        """Get information about the current device fingerprint."""
        fingerprint = self._get_minimal_fingerprint()
        fingerprint_hash = self._generate_fingerprint_hash(fingerprint)
        user_uuid = self.get_user_uuid()

        return {
            "user_uuid": user_uuid,
            "user_id_short": user_uuid[:8],
            "fingerprint_hash": fingerprint_hash[:16],
            "device_characteristics": {
                "os": fingerprint.get("os_system", "unknown"),
                "architecture": fingerprint.get("architecture", "unknown"),
                "cpu_cores": fingerprint.get("cpu_count_logical", 0),
                "memory_gb": fingerprint.get("memory_total_gb", 0),
                "has_mac_hash": "mac_hash" in fingerprint,
            },
            "privacy_protection": {
                "data_collection": "Minimal - only essential system characteristics",
                "sensitive_data": "All identifying information is hashed",
                "storage": "Local cache only, no external transmission",
                "consistency": "Same device always generates same UUID",
            },
        }


# Global instance for easy access
_fingerprint_manager = None


def get_fingerprint_manager() -> MinimalFingerprintManager:
    """Get the global fingerprint manager instance."""
    global _fingerprint_manager
    if _fingerprint_manager is None:
        _fingerprint_manager = MinimalFingerprintManager()
    return _fingerprint_manager


def get_auto_user_uuid() -> str:
    """
    Convenience function to get automatic user UUID.

    Returns:
        str: Consistent UUID for the current user/device
    """
    return get_fingerprint_manager().get_user_uuid()


def update_user_operation_stats(user_uuid: str, operation_type: str) -> None:
    """
    Convenience function to update user operation statistics.

    Args:
        user_uuid (str): User's UUID
        operation_type (str): Type of operation performed
    """
    get_fingerprint_manager().update_user_stats(user_uuid, operation_type)


if __name__ == "__main__":
    # Test the fingerprinting system
    print("🔍 Testing Minimal Fingerprint Manager...")

    manager = MinimalFingerprintManager()

    # Test basic functionality
    user_uuid = manager.get_user_uuid()
    print(f"Generated User UUID: {user_uuid}")

    # Test fingerprint info
    info = manager.get_fingerprint_info()
    print(f"Device OS: {info['device_characteristics']['os']}")
    print(f"CPU Cores: {info['device_characteristics']['cpu_cores']}")
    print(f"Memory: {info['device_characteristics']['memory_gb']} GB")

    # Test stats
    manager.update_user_stats(user_uuid, "store_memory")
    manager.update_user_stats(user_uuid, "search_memory")

    stats = manager.get_user_stats(user_uuid)
    print(f"User Stats: {stats['statistics']['total_operations']} operations")

    print("✅ Fingerprint Manager Test Complete")