Coverage for integrations / channels / identity / preferences.py: 97.3%
297 statements
« prev ^ index » next coverage.py v7.14.0, created at 2026-05-12 04:49 +0000
1"""
2User Preferences Management for HevolveBot Integration.
4This module provides UserPreferences and PreferenceManager for managing
5user preferences across different channels.
6"""
8from dataclasses import dataclass, field, asdict
9from typing import Dict, List, Optional, Any, Callable, Set
10from datetime import datetime
11from enum import Enum
12from pathlib import Path
13import uuid
14import json
15import logging
16import os
18logger = logging.getLogger(__name__)
21# Default container-compatible paths for Docker deployment
22DEFAULT_PREFERENCES_PATH = os.environ.get(
23 'PREFERENCES_STORAGE_PATH',
24 '/app/data/preferences.json'
25)
27# Fallback for local development
28if not os.path.exists(os.path.dirname(DEFAULT_PREFERENCES_PATH)):
29 DEFAULT_PREFERENCES_PATH = os.path.join(
30 os.path.dirname(os.path.abspath(__file__)),
31 '..', '..', '..', 'data', 'preferences.json'
32 )
35class ResponseStyle(Enum):
36 """Available response styles."""
37 CONCISE = "concise"
38 BALANCED = "balanced"
39 DETAILED = "detailed"
40 TECHNICAL = "technical"
41 CASUAL = "casual"
44class Theme(Enum):
45 """Available UI themes."""
46 AUTO = "auto"
47 LIGHT = "light"
48 DARK = "dark"
51# Schema version for migrations
52SCHEMA_VERSION = 1
55@dataclass
56class UserPreferences:
57 """
58 Represents a user's preferences configuration.
60 Attributes:
61 id: Unique identifier for this preferences instance
62 user_id: The user this preferences belong to
63 language: Preferred language code (e.g., "en", "es", "fr")
64 timezone: Preferred timezone (e.g., "UTC", "America/New_York")
65 model: Preferred AI model (None uses default)
66 response_style: Preferred response style
67 notifications: Whether to receive notifications
68 theme: Preferred UI theme
69 channel_overrides: Per-channel preference overrides
70 metadata: Additional custom preferences
71 schema_version: Schema version for migrations
72 created_at: When preferences were created
73 updated_at: When preferences were last updated
74 """
76 id: str = field(default_factory=lambda: str(uuid.uuid4()))
77 user_id: str = ""
78 language: str = "en"
79 timezone: str = "UTC"
80 model: Optional[str] = None
81 response_style: str = "balanced"
82 notifications: bool = True
83 theme: str = "auto"
84 channel_overrides: Dict[str, Dict[str, Any]] = field(default_factory=dict)
85 metadata: Dict[str, Any] = field(default_factory=dict)
86 schema_version: int = SCHEMA_VERSION
87 created_at: datetime = field(default_factory=datetime.utcnow)
88 updated_at: datetime = field(default_factory=datetime.utcnow)
90 def to_dict(self) -> Dict[str, Any]:
91 """Convert preferences to dictionary."""
92 data = asdict(self)
93 data['created_at'] = self.created_at.isoformat()
94 data['updated_at'] = self.updated_at.isoformat()
95 return data
97 @classmethod
98 def from_dict(cls, data: Dict[str, Any]) -> 'UserPreferences':
99 """Create preferences from dictionary."""
100 if 'created_at' in data and isinstance(data['created_at'], str):
101 data['created_at'] = datetime.fromisoformat(data['created_at'])
102 if 'updated_at' in data and isinstance(data['updated_at'], str):
103 data['updated_at'] = datetime.fromisoformat(data['updated_at'])
104 # Handle missing fields for backwards compatibility
105 if 'channel_overrides' not in data:
106 data['channel_overrides'] = {}
107 if 'schema_version' not in data:
108 data['schema_version'] = 1
109 return cls(**data)
111 def update(self, **kwargs) -> 'UserPreferences':
112 """Update preferences fields and return self."""
113 for key, value in kwargs.items():
114 if hasattr(self, key) and key not in ('id', 'user_id', 'created_at'):
115 setattr(self, key, value)
116 self.updated_at = datetime.utcnow()
117 return self
119 def get_effective_preference(self, key: str, channel: Optional[str] = None, default: Any = None) -> Any:
120 """
121 Get effective preference value, considering channel overrides.
123 Args:
124 key: The preference key to retrieve
125 channel: Optional channel to check for overrides
126 default: Default value if not found
128 Returns:
129 The effective preference value
130 """
131 # Check channel override first
132 if channel and channel in self.channel_overrides:
133 override = self.channel_overrides[channel]
134 if key in override:
135 return override[key]
137 # Fall back to base preference
138 if hasattr(self, key):
139 return getattr(self, key)
141 # Check metadata
142 if key in self.metadata:
143 return self.metadata[key]
145 return default
147 def set_channel_override(self, channel: str, key: str, value: Any) -> None:
148 """
149 Set a channel-specific preference override.
151 Args:
152 channel: The channel identifier
153 key: The preference key
154 value: The override value
155 """
156 if channel not in self.channel_overrides:
157 self.channel_overrides[channel] = {}
158 self.channel_overrides[channel][key] = value
159 self.updated_at = datetime.utcnow()
161 def remove_channel_override(self, channel: str, key: Optional[str] = None) -> bool:
162 """
163 Remove channel override(s).
165 Args:
166 channel: The channel identifier
167 key: Optional specific key to remove. If None, removes all overrides for channel.
169 Returns:
170 True if removed, False if not found
171 """
172 if channel not in self.channel_overrides:
173 return False
175 if key is None:
176 del self.channel_overrides[channel]
177 self.updated_at = datetime.utcnow()
178 return True
180 if key in self.channel_overrides[channel]:
181 del self.channel_overrides[channel][key]
182 if not self.channel_overrides[channel]:
183 del self.channel_overrides[channel]
184 self.updated_at = datetime.utcnow()
185 return True
187 return False
189 def get_channel_overrides(self, channel: str) -> Dict[str, Any]:
190 """Get all overrides for a specific channel."""
191 return dict(self.channel_overrides.get(channel, {}))
194class PreferenceValidator:
195 """Validates preference values."""
197 # Valid language codes (subset of ISO 639-1)
198 VALID_LANGUAGES: Set[str] = {
199 'en', 'es', 'fr', 'de', 'it', 'pt', 'ru', 'zh', 'ja', 'ko',
200 'ar', 'hi', 'nl', 'pl', 'tr', 'vi', 'th', 'id', 'ms', 'sv'
201 }
203 # Valid response styles
204 VALID_RESPONSE_STYLES: Set[str] = {s.value for s in ResponseStyle}
206 # Valid themes
207 VALID_THEMES: Set[str] = {t.value for t in Theme}
209 # Common timezone prefixes (simplified validation)
210 VALID_TIMEZONE_PREFIXES: Set[str] = {
211 'UTC', 'GMT', 'America/', 'Europe/', 'Asia/', 'Africa/',
212 'Australia/', 'Pacific/', 'Atlantic/', 'Indian/', 'Etc/'
213 }
215 @classmethod
216 def validate(cls, prefs: UserPreferences) -> List[str]:
217 """
218 Validate preferences and return list of errors.
220 Args:
221 prefs: The preferences to validate
223 Returns:
224 List of validation error messages (empty if valid)
225 """
226 errors = []
228 # Validate language
229 if prefs.language not in cls.VALID_LANGUAGES:
230 errors.append(f"Invalid language code: {prefs.language}")
232 # Validate response style
233 if prefs.response_style not in cls.VALID_RESPONSE_STYLES:
234 errors.append(f"Invalid response style: {prefs.response_style}")
236 # Validate theme
237 if prefs.theme not in cls.VALID_THEMES:
238 errors.append(f"Invalid theme: {prefs.theme}")
240 # Validate timezone (simplified check)
241 if not cls._validate_timezone(prefs.timezone):
242 errors.append(f"Invalid timezone: {prefs.timezone}")
244 # Validate channel overrides
245 for channel, overrides in prefs.channel_overrides.items():
246 if 'language' in overrides and overrides['language'] not in cls.VALID_LANGUAGES:
247 errors.append(f"Invalid language in {channel} override: {overrides['language']}")
248 if 'response_style' in overrides and overrides['response_style'] not in cls.VALID_RESPONSE_STYLES:
249 errors.append(f"Invalid response_style in {channel} override: {overrides['response_style']}")
250 if 'theme' in overrides and overrides['theme'] not in cls.VALID_THEMES:
251 errors.append(f"Invalid theme in {channel} override: {overrides['theme']}")
253 return errors
255 @classmethod
256 def _validate_timezone(cls, timezone: str) -> bool:
257 """Check if timezone looks valid."""
258 if timezone in ('UTC', 'GMT'):
259 return True
260 for prefix in cls.VALID_TIMEZONE_PREFIXES:
261 if timezone.startswith(prefix):
262 return True
263 return False
265 @classmethod
266 def is_valid(cls, prefs: UserPreferences) -> bool:
267 """Check if preferences are valid."""
268 return len(cls.validate(prefs)) == 0
271class PreferenceMigrator:
272 """Handles schema migrations for preferences."""
274 # Migration functions: (from_version, to_version) -> migration_func
275 _migrations: Dict[tuple, Callable[[Dict[str, Any]], Dict[str, Any]]] = {}
277 @classmethod
278 def register_migration(
279 cls,
280 from_version: int,
281 to_version: int,
282 migration_func: Callable[[Dict[str, Any]], Dict[str, Any]]
283 ) -> None:
284 """
285 Register a migration function.
287 Args:
288 from_version: Source schema version
289 to_version: Target schema version
290 migration_func: Function that transforms the data
291 """
292 cls._migrations[(from_version, to_version)] = migration_func
294 @classmethod
295 def migrate(cls, data: Dict[str, Any], target_version: int = SCHEMA_VERSION) -> Dict[str, Any]:
296 """
297 Migrate preferences data to target version.
299 Args:
300 data: The preferences data to migrate
301 target_version: The target schema version
303 Returns:
304 Migrated data
305 """
306 current_version = data.get('schema_version', 1)
308 while current_version < target_version:
309 next_version = current_version + 1
310 migration_key = (current_version, next_version)
312 if migration_key in cls._migrations:
313 data = cls._migrations[migration_key](data)
314 data['schema_version'] = next_version
315 logger.info(f"Migrated preferences from v{current_version} to v{next_version}")
316 else:
317 # No migration needed, just update version
318 data['schema_version'] = next_version
320 current_version = next_version
322 # Ensure schema_version is always present in output
323 if 'schema_version' not in data:
324 data['schema_version'] = target_version
326 return data
329class PreferenceManager:
330 """
331 Manages user preferences across channels.
333 Supports:
334 - Preference persistence (file-based or in-memory)
335 - Default preferences configuration
336 - Channel-specific preference overrides
337 - Preference validation
338 - Schema migrations
339 """
341 def __init__(
342 self,
343 storage_path: Optional[str] = None,
344 default_preferences: Optional[Dict[str, Any]] = None,
345 auto_persist: bool = True,
346 validate_on_set: bool = True
347 ):
348 """
349 Initialize the preference manager.
351 Args:
352 storage_path: Path to preferences storage file (container-compatible)
353 default_preferences: Default preference values for new users
354 auto_persist: Automatically persist on changes
355 validate_on_set: Validate preferences when setting
356 """
357 self._storage_path = storage_path or DEFAULT_PREFERENCES_PATH
358 self._default_preferences = default_preferences or {}
359 self._auto_persist = auto_persist
360 self._validate_on_set = validate_on_set
361 self._preferences: Dict[str, UserPreferences] = {}
362 self._dirty = False
364 # Ensure storage directory exists
365 self._ensure_storage_dir()
367 # Load existing preferences
368 self._load()
370 def _ensure_storage_dir(self) -> None:
371 """Ensure the storage directory exists."""
372 storage_dir = os.path.dirname(self._storage_path)
373 if storage_dir and not os.path.exists(storage_dir):
374 try:
375 os.makedirs(storage_dir, exist_ok=True)
376 logger.info(f"Created preferences storage directory: {storage_dir}")
377 except OSError as e:
378 logger.warning(f"Could not create storage directory: {e}")
380 def _load(self) -> None:
381 """Load preferences from storage."""
382 if not os.path.exists(self._storage_path):
383 logger.info(f"No existing preferences file at {self._storage_path}")
384 return
386 try:
387 with open(self._storage_path, 'r', encoding='utf-8') as f:
388 data = json.load(f)
390 for user_id, prefs_data in data.get('preferences', {}).items():
391 # Migrate if needed
392 prefs_data = PreferenceMigrator.migrate(prefs_data)
393 prefs = UserPreferences.from_dict(prefs_data)
394 self._preferences[user_id] = prefs
396 logger.info(f"Loaded {len(self._preferences)} user preferences")
397 except Exception as e:
398 logger.error(f"Error loading preferences: {e}")
400 def _persist(self) -> None:
401 """Persist preferences to storage."""
402 try:
403 data = {
404 'schema_version': SCHEMA_VERSION,
405 'updated_at': datetime.utcnow().isoformat(),
406 'preferences': {
407 user_id: prefs.to_dict()
408 for user_id, prefs in self._preferences.items()
409 }
410 }
412 with open(self._storage_path, 'w', encoding='utf-8') as f:
413 json.dump(data, f, indent=2)
415 self._dirty = False
416 logger.debug(f"Persisted {len(self._preferences)} user preferences")
417 except Exception as e:
418 logger.error(f"Error persisting preferences: {e}")
420 def _mark_dirty(self) -> None:
421 """Mark preferences as modified."""
422 self._dirty = True
423 if self._auto_persist:
424 self._persist()
426 def get(self, user_id: str) -> UserPreferences:
427 """
428 Get preferences for a user.
430 If no preferences exist, creates default preferences.
432 Args:
433 user_id: The user identifier
435 Returns:
436 The user's preferences
437 """
438 if user_id not in self._preferences:
439 # Create default preferences
440 prefs = UserPreferences(
441 user_id=user_id,
442 **self._default_preferences
443 )
444 self._preferences[user_id] = prefs
445 self._mark_dirty()
446 logger.info(f"Created default preferences for user: {user_id}")
448 return self._preferences[user_id]
450 def get_or_none(self, user_id: str) -> Optional[UserPreferences]:
451 """
452 Get preferences for a user without creating defaults.
454 Args:
455 user_id: The user identifier
457 Returns:
458 The user's preferences or None if not found
459 """
460 return self._preferences.get(user_id)
462 def set(self, user_id: str, prefs: UserPreferences) -> None:
463 """
464 Set preferences for a user.
466 Args:
467 user_id: The user identifier
468 prefs: The preferences to set
470 Raises:
471 ValueError: If validation is enabled and preferences are invalid
472 """
473 if self._validate_on_set:
474 errors = PreferenceValidator.validate(prefs)
475 if errors:
476 raise ValueError(f"Invalid preferences: {'; '.join(errors)}")
478 prefs.user_id = user_id
479 prefs.updated_at = datetime.utcnow()
480 self._preferences[user_id] = prefs
481 self._mark_dirty()
482 logger.info(f"Set preferences for user: {user_id}")
484 def update(self, user_id: str, **kwargs) -> UserPreferences:
485 """
486 Update specific preferences for a user.
488 Args:
489 user_id: The user identifier
490 **kwargs: Preference fields to update
492 Returns:
493 The updated preferences
495 Raises:
496 ValueError: If validation is enabled and resulting preferences are invalid
497 """
498 prefs = self.get(user_id)
499 prefs.update(**kwargs)
501 if self._validate_on_set:
502 errors = PreferenceValidator.validate(prefs)
503 if errors:
504 raise ValueError(f"Invalid preferences: {'; '.join(errors)}")
506 self._mark_dirty()
507 logger.info(f"Updated preferences for user: {user_id}")
508 return prefs
510 def delete(self, user_id: str) -> bool:
511 """
512 Delete preferences for a user.
514 Args:
515 user_id: The user identifier
517 Returns:
518 True if deleted, False if not found
519 """
520 if user_id in self._preferences:
521 del self._preferences[user_id]
522 self._mark_dirty()
523 logger.info(f"Deleted preferences for user: {user_id}")
524 return True
525 return False
527 def has_preferences(self, user_id: str) -> bool:
528 """Check if a user has stored preferences."""
529 return user_id in self._preferences
531 def reset_to_defaults(self, user_id: str) -> UserPreferences:
532 """
533 Reset a user's preferences to defaults.
535 Args:
536 user_id: The user identifier
538 Returns:
539 The reset preferences
540 """
541 prefs = UserPreferences(
542 user_id=user_id,
543 **self._default_preferences
544 )
545 # Preserve ID if exists
546 if user_id in self._preferences:
547 prefs.id = self._preferences[user_id].id
548 prefs.created_at = self._preferences[user_id].created_at
550 self._preferences[user_id] = prefs
551 self._mark_dirty()
552 logger.info(f"Reset preferences to defaults for user: {user_id}")
553 return prefs
555 def get_effective_preference(
556 self,
557 user_id: str,
558 key: str,
559 channel: Optional[str] = None,
560 default: Any = None
561 ) -> Any:
562 """
563 Get effective preference value for a user, considering channel overrides.
565 Args:
566 user_id: The user identifier
567 key: The preference key
568 channel: Optional channel for override lookup
569 default: Default value if not found
571 Returns:
572 The effective preference value
573 """
574 prefs = self.get(user_id)
575 return prefs.get_effective_preference(key, channel, default)
577 def set_channel_override(
578 self,
579 user_id: str,
580 channel: str,
581 key: str,
582 value: Any
583 ) -> None:
584 """
585 Set a channel-specific preference override for a user.
587 Args:
588 user_id: The user identifier
589 channel: The channel identifier
590 key: The preference key
591 value: The override value
592 """
593 prefs = self.get(user_id)
594 prefs.set_channel_override(channel, key, value)
596 if self._validate_on_set:
597 errors = PreferenceValidator.validate(prefs)
598 if errors:
599 # Rollback the override
600 prefs.remove_channel_override(channel, key)
601 raise ValueError(f"Invalid preferences: {'; '.join(errors)}")
603 self._mark_dirty()
604 logger.info(f"Set channel override for user {user_id}, channel {channel}: {key}={value}")
606 def remove_channel_override(
607 self,
608 user_id: str,
609 channel: str,
610 key: Optional[str] = None
611 ) -> bool:
612 """
613 Remove channel override(s) for a user.
615 Args:
616 user_id: The user identifier
617 channel: The channel identifier
618 key: Optional specific key to remove
620 Returns:
621 True if removed, False if not found
622 """
623 if user_id not in self._preferences:
624 return False
626 prefs = self._preferences[user_id]
627 result = prefs.remove_channel_override(channel, key)
629 if result:
630 self._mark_dirty()
631 logger.info(f"Removed channel override for user {user_id}, channel {channel}")
633 return result
635 def list_users(self) -> List[str]:
636 """Get list of all users with stored preferences."""
637 return list(self._preferences.keys())
639 def list_preferences(self) -> List[UserPreferences]:
640 """Get all stored preferences."""
641 return list(self._preferences.values())
643 def export_preferences(self, user_id: Optional[str] = None) -> str:
644 """
645 Export preferences as JSON.
647 Args:
648 user_id: Optional specific user to export. If None, exports all.
650 Returns:
651 JSON string of preferences
652 """
653 if user_id:
654 if user_id not in self._preferences:
655 return json.dumps({})
656 return json.dumps(self._preferences[user_id].to_dict(), indent=2)
658 return json.dumps({
659 user_id: prefs.to_dict()
660 for user_id, prefs in self._preferences.items()
661 }, indent=2)
663 def import_preferences(self, json_data: str, merge: bool = True) -> int:
664 """
665 Import preferences from JSON.
667 Args:
668 json_data: JSON string of preferences
669 merge: If True, merges with existing. If False, replaces.
671 Returns:
672 Number of preferences imported
673 """
674 data = json.loads(json_data)
675 count = 0
677 if not merge:
678 self._preferences.clear()
680 # Handle single preference or dict of preferences
681 if 'user_id' in data:
682 # Single preference
683 prefs = UserPreferences.from_dict(data)
684 self._preferences[prefs.user_id] = prefs
685 count = 1
686 else:
687 # Dict of preferences
688 for user_id, prefs_data in data.items():
689 prefs_data = PreferenceMigrator.migrate(prefs_data)
690 prefs = UserPreferences.from_dict(prefs_data)
691 prefs.user_id = user_id
692 self._preferences[user_id] = prefs
693 count += 1
695 self._mark_dirty()
696 logger.info(f"Imported {count} preferences")
697 return count
699 def persist(self) -> None:
700 """Manually persist preferences to storage."""
701 self._persist()
703 def get_default_preferences(self) -> Dict[str, Any]:
704 """Get the default preference values."""
705 return dict(self._default_preferences)
707 def set_default_preferences(self, defaults: Dict[str, Any]) -> None:
708 """Set the default preference values."""
709 self._default_preferences = dict(defaults)
710 logger.info("Updated default preferences")
713# Global preference manager instance
714_preference_manager: Optional[PreferenceManager] = None
717def get_preference_manager(
718 storage_path: Optional[str] = None,
719 default_preferences: Optional[Dict[str, Any]] = None
720) -> PreferenceManager:
721 """
722 Get the global preference manager instance.
724 Args:
725 storage_path: Optional custom storage path
726 default_preferences: Optional default preferences
728 Returns:
729 The global PreferenceManager instance
730 """
731 global _preference_manager
733 if _preference_manager is None:
734 _preference_manager = PreferenceManager(
735 storage_path=storage_path,
736 default_preferences=default_preferences
737 )
739 return _preference_manager