Coverage for integrations / channels / identity / sender_mapping.py: 47.3%
222 statements
« prev ^ index » next coverage.py v7.14.0, created at 2026-05-12 04:49 +0000
1"""
2Sender Identity Mapping for HevolveBot Integration.
4This module provides SenderIdentityMapper for mapping user identities
5across different channels and platforms.
6"""
8from dataclasses import dataclass, field, asdict
9from typing import Dict, List, Optional, Any, Set
10from datetime import datetime
11from enum import Enum
12import uuid
13import logging
15logger = logging.getLogger(__name__)
18class ChannelType(Enum):
19 """Supported channel types."""
20 DISCORD = "discord"
21 TELEGRAM = "telegram"
22 SLACK = "slack"
23 TEAMS = "teams"
24 WHATSAPP = "whatsapp"
25 EMAIL = "email"
26 WEB = "web"
27 API = "api"
28 CUSTOM = "custom"
31@dataclass
32class UserIdentity:
33 """Represents a user's identity."""
35 id: str = field(default_factory=lambda: str(uuid.uuid4()))
36 username: str = ""
37 display_name: str = ""
38 email: Optional[str] = None
39 avatar_url: Optional[str] = None
40 verified: bool = False
41 roles: List[str] = field(default_factory=list)
42 metadata: Dict[str, Any] = field(default_factory=dict)
43 created_at: datetime = field(default_factory=datetime.utcnow)
44 last_seen: datetime = field(default_factory=datetime.utcnow)
46 def to_dict(self) -> Dict[str, Any]:
47 """Convert to dictionary."""
48 data = asdict(self)
49 data['created_at'] = self.created_at.isoformat()
50 data['last_seen'] = self.last_seen.isoformat()
51 return data
53 @classmethod
54 def from_dict(cls, data: Dict[str, Any]) -> 'UserIdentity':
55 """Create from dictionary."""
56 if 'created_at' in data and isinstance(data['created_at'], str):
57 data['created_at'] = datetime.fromisoformat(data['created_at'])
58 if 'last_seen' in data and isinstance(data['last_seen'], str):
59 data['last_seen'] = datetime.fromisoformat(data['last_seen'])
60 return cls(**data)
62 def update_last_seen(self) -> None:
63 """Update the last seen timestamp."""
64 self.last_seen = datetime.utcnow()
66 def has_role(self, role: str) -> bool:
67 """Check if user has a specific role."""
68 return role in self.roles
70 def add_role(self, role: str) -> None:
71 """Add a role to the user."""
72 if role not in self.roles:
73 self.roles.append(role)
75 def remove_role(self, role: str) -> bool:
76 """Remove a role from the user."""
77 if role in self.roles:
78 self.roles.remove(role)
79 return True
80 return False
83@dataclass
84class ChannelIdentity:
85 """Represents a user's identity on a specific channel."""
87 id: str = field(default_factory=lambda: str(uuid.uuid4()))
88 channel_type: ChannelType = ChannelType.CUSTOM
89 channel_id: str = "" # Platform-specific identifier
90 channel_user_id: str = "" # User's ID on that channel
91 channel_username: str = "" # User's username on that channel
92 channel_display_name: str = ""
93 channel_avatar_url: Optional[str] = None
94 is_bot: bool = False
95 is_verified: bool = False
96 permissions: List[str] = field(default_factory=list)
97 metadata: Dict[str, Any] = field(default_factory=dict)
98 linked_user_id: Optional[str] = None # Link to unified UserIdentity
99 created_at: datetime = field(default_factory=datetime.utcnow)
100 updated_at: datetime = field(default_factory=datetime.utcnow)
102 def to_dict(self) -> Dict[str, Any]:
103 """Convert to dictionary."""
104 data = asdict(self)
105 data['channel_type'] = self.channel_type.value
106 data['created_at'] = self.created_at.isoformat()
107 data['updated_at'] = self.updated_at.isoformat()
108 return data
110 @classmethod
111 def from_dict(cls, data: Dict[str, Any]) -> 'ChannelIdentity':
112 """Create from dictionary."""
113 if 'channel_type' in data and isinstance(data['channel_type'], str):
114 data['channel_type'] = ChannelType(data['channel_type'])
115 if 'created_at' in data and isinstance(data['created_at'], str):
116 data['created_at'] = datetime.fromisoformat(data['created_at'])
117 if 'updated_at' in data and isinstance(data['updated_at'], str):
118 data['updated_at'] = datetime.fromisoformat(data['updated_at'])
119 return cls(**data)
121 def has_permission(self, permission: str) -> bool:
122 """Check if channel identity has a permission."""
123 return permission in self.permissions
125 @property
126 def channel_key(self) -> str:
127 """Get unique key for this channel identity."""
128 return f"{self.channel_type.value}:{self.channel_id}:{self.channel_user_id}"
131class SenderIdentityMapper:
132 """
133 Maps sender identities across channels.
135 Supports:
136 - User identity management
137 - Channel-specific identity tracking
138 - Cross-channel identity linking
139 - Identity unification
140 """
142 def __init__(self):
143 """Initialize the sender identity mapper."""
144 self._users: Dict[str, UserIdentity] = {}
145 self._channel_identities: Dict[str, ChannelIdentity] = {} # channel_key -> identity
146 self._user_channels: Dict[str, Set[str]] = {} # user_id -> set of channel_keys
147 self._mappings: Dict[str, str] = {} # channel_key -> user_id
149 def map(
150 self,
151 channel_type: ChannelType,
152 channel_id: str,
153 channel_user_id: str,
154 **kwargs
155 ) -> ChannelIdentity:
156 """
157 Map a channel-specific identity.
159 Args:
160 channel_type: Type of channel
161 channel_id: Channel/server identifier
162 channel_user_id: User's ID on that channel
163 **kwargs: Additional channel identity attributes
165 Returns:
166 The ChannelIdentity (created or existing)
167 """
168 channel_key = f"{channel_type.value}:{channel_id}:{channel_user_id}"
170 if channel_key in self._channel_identities:
171 # Update existing
172 identity = self._channel_identities[channel_key]
173 for key, value in kwargs.items():
174 if hasattr(identity, key):
175 setattr(identity, key, value)
176 identity.updated_at = datetime.utcnow()
177 logger.debug(f"Updated channel identity: {channel_key}")
178 else:
179 # Create new
180 identity = ChannelIdentity(
181 channel_type=channel_type,
182 channel_id=channel_id,
183 channel_user_id=channel_user_id,
184 **kwargs
185 )
186 self._channel_identities[channel_key] = identity
187 logger.info(f"Created channel identity: {channel_key}")
189 return identity
191 def set_mapping(
192 self,
193 channel_identity: ChannelIdentity,
194 user: UserIdentity
195 ) -> None:
196 """
197 Link a channel identity to a unified user identity.
199 Args:
200 channel_identity: The channel-specific identity
201 user: The unified user identity
202 """
203 # Ensure user is registered
204 if user.id not in self._users:
205 self._users[user.id] = user
207 # Link channel to user
208 channel_key = channel_identity.channel_key
209 channel_identity.linked_user_id = user.id
210 self._mappings[channel_key] = user.id
212 # Track user's channels
213 if user.id not in self._user_channels:
214 self._user_channels[user.id] = set()
215 self._user_channels[user.id].add(channel_key)
217 logger.info(f"Mapped {channel_key} to user {user.id}")
219 def get_user(self, user_id: str) -> Optional[UserIdentity]:
220 """Get a user by ID."""
221 return self._users.get(user_id)
223 def get_channel_identity(
224 self,
225 channel_type: ChannelType,
226 channel_id: str,
227 channel_user_id: str
228 ) -> Optional[ChannelIdentity]:
229 """Get a channel identity."""
230 channel_key = f"{channel_type.value}:{channel_id}:{channel_user_id}"
231 return self._channel_identities.get(channel_key)
233 def get_user_from_channel(
234 self,
235 channel_type: ChannelType,
236 channel_id: str,
237 channel_user_id: str
238 ) -> Optional[UserIdentity]:
239 """
240 Get the unified user identity from a channel identity.
242 Args:
243 channel_type: Type of channel
244 channel_id: Channel identifier
245 channel_user_id: User's ID on that channel
247 Returns:
248 The linked UserIdentity or None
249 """
250 channel_key = f"{channel_type.value}:{channel_id}:{channel_user_id}"
251 user_id = self._mappings.get(channel_key)
252 if user_id:
253 return self._users.get(user_id)
254 return None
256 def get_cross_channel(self, user_id: str) -> List[ChannelIdentity]:
257 """
258 Get all channel identities for a user.
260 Args:
261 user_id: The unified user ID
263 Returns:
264 List of all channel identities linked to this user
265 """
266 if user_id not in self._user_channels:
267 return []
269 identities = []
270 for channel_key in self._user_channels[user_id]:
271 if channel_key in self._channel_identities:
272 identities.append(self._channel_identities[channel_key])
274 return identities
276 def get_cross_channel_by_channel(
277 self,
278 channel_type: ChannelType,
279 channel_id: str,
280 channel_user_id: str
281 ) -> List[ChannelIdentity]:
282 """
283 Get all channel identities for a user, starting from one channel identity.
285 Args:
286 channel_type: Starting channel type
287 channel_id: Starting channel ID
288 channel_user_id: User's ID on that channel
290 Returns:
291 List of all channel identities for this user
292 """
293 channel_key = f"{channel_type.value}:{channel_id}:{channel_user_id}"
294 user_id = self._mappings.get(channel_key)
295 if not user_id:
296 return []
297 return self.get_cross_channel(user_id)
299 def create_user(self, **kwargs) -> UserIdentity:
300 """
301 Create a new unified user identity.
303 Args:
304 **kwargs: User attributes
306 Returns:
307 The created UserIdentity
308 """
309 user = UserIdentity(**kwargs)
310 self._users[user.id] = user
311 self._user_channels[user.id] = set()
312 logger.info(f"Created user: {user.id} ({user.username})")
313 return user
315 def delete_user(self, user_id: str) -> bool:
316 """
317 Delete a user and all their channel mappings.
319 Args:
320 user_id: ID of the user to delete
322 Returns:
323 True if deleted
324 """
325 if user_id not in self._users:
326 return False
328 # Remove channel mappings
329 if user_id in self._user_channels:
330 for channel_key in self._user_channels[user_id]:
331 if channel_key in self._mappings:
332 del self._mappings[channel_key]
333 if channel_key in self._channel_identities:
334 self._channel_identities[channel_key].linked_user_id = None
335 del self._user_channels[user_id]
337 # Remove user
338 del self._users[user_id]
339 logger.info(f"Deleted user: {user_id}")
340 return True
342 def unlink_channel(
343 self,
344 channel_type: ChannelType,
345 channel_id: str,
346 channel_user_id: str
347 ) -> bool:
348 """
349 Unlink a channel identity from its user.
351 Args:
352 channel_type: Type of channel
353 channel_id: Channel identifier
354 channel_user_id: User's ID on that channel
356 Returns:
357 True if unlinked
358 """
359 channel_key = f"{channel_type.value}:{channel_id}:{channel_user_id}"
361 if channel_key not in self._mappings:
362 return False
364 user_id = self._mappings[channel_key]
365 del self._mappings[channel_key]
367 if user_id in self._user_channels:
368 self._user_channels[user_id].discard(channel_key)
370 if channel_key in self._channel_identities:
371 self._channel_identities[channel_key].linked_user_id = None
373 logger.info(f"Unlinked channel: {channel_key}")
374 return True
376 def merge_users(self, primary_id: str, secondary_id: str) -> bool:
377 """
378 Merge two users, keeping the primary and merging secondary's channels.
380 Args:
381 primary_id: ID of the user to keep
382 secondary_id: ID of the user to merge into primary
384 Returns:
385 True if merged successfully
386 """
387 if primary_id not in self._users or secondary_id not in self._users:
388 return False
390 if primary_id == secondary_id:
391 return False
393 # Move secondary's channels to primary
394 if secondary_id in self._user_channels:
395 for channel_key in self._user_channels[secondary_id]:
396 self._mappings[channel_key] = primary_id
397 if channel_key in self._channel_identities:
398 self._channel_identities[channel_key].linked_user_id = primary_id
400 if primary_id not in self._user_channels:
401 self._user_channels[primary_id] = set()
402 self._user_channels[primary_id].update(self._user_channels[secondary_id])
403 del self._user_channels[secondary_id]
405 # Remove secondary user
406 del self._users[secondary_id]
407 logger.info(f"Merged user {secondary_id} into {primary_id}")
408 return True
410 def list_users(self) -> List[UserIdentity]:
411 """Get all users."""
412 return list(self._users.values())
414 def list_channel_identities(
415 self,
416 channel_type: Optional[ChannelType] = None
417 ) -> List[ChannelIdentity]:
418 """
419 Get channel identities, optionally filtered by type.
421 Args:
422 channel_type: Optional filter by channel type
424 Returns:
425 List of channel identities
426 """
427 identities = list(self._channel_identities.values())
428 if channel_type:
429 identities = [i for i in identities if i.channel_type == channel_type]
430 return identities
432 def find_users_by_email(self, email: str) -> List[UserIdentity]:
433 """Find users by email address."""
434 return [u for u in self._users.values() if u.email == email]
436 def find_users_by_username(self, username: str) -> List[UserIdentity]:
437 """Find users by username (partial match)."""
438 username_lower = username.lower()
439 return [
440 u for u in self._users.values()
441 if username_lower in u.username.lower()
442 ]
444 def get_user_channel_count(self, user_id: str) -> int:
445 """Get the number of channels linked to a user."""
446 return len(self._user_channels.get(user_id, set()))
448 def export_mappings(self) -> Dict[str, Any]:
449 """Export all mappings as a dictionary."""
450 return {
451 'users': [u.to_dict() for u in self._users.values()],
452 'channel_identities': [c.to_dict() for c in self._channel_identities.values()],
453 'mappings': dict(self._mappings)
454 }
456 def import_mappings(self, data: Dict[str, Any]) -> int:
457 """
458 Import mappings from a dictionary.
460 Returns:
461 Number of users imported
462 """
463 count = 0
465 # Import users
466 for user_data in data.get('users', []):
467 user = UserIdentity.from_dict(user_data)
468 self._users[user.id] = user
469 self._user_channels[user.id] = set()
470 count += 1
472 # Import channel identities
473 for channel_data in data.get('channel_identities', []):
474 identity = ChannelIdentity.from_dict(channel_data)
475 self._channel_identities[identity.channel_key] = identity
477 # Restore mappings
478 for channel_key, user_id in data.get('mappings', {}).items():
479 self._mappings[channel_key] = user_id
480 if user_id in self._user_channels:
481 self._user_channels[user_id].add(channel_key)
483 return count