"""LinkedIn sync service for fetching notifications and creating Signals. This service handles: - Fetching notifications from the LinkedIn API - Processing notifications into Signal records - Mapping LinkedIn authors to Member records - Managing sync status transitions """ from __future__ import annotations import logging from datetime import UTC, datetime from typing import TYPE_CHECKING, Any from django.db import IntegrityError, transaction from django.utils import timezone as django_timezone from accounts.models import Workspace from integrations.exceptions import LinkedInAPIError, LinkedInTokenExpiredError from integrations.models import LinkedInPage, LinkedInSyncStatus from integrations.services.linkedin_client import LinkedInClient from members.models import Identity, Member from messages.models import Signal from sources.models import Source if TYPE_CHECKING: from integrations.contracts import LinkedInMentionMeta logger = logging.getLogger(__name__) class LinkedInSyncService: """Service for syncing LinkedIn notifications to Signal records.""" def __init__(self, linkedin_page: LinkedInPage) -> None: """Initialize the sync service. Args: linkedin_page: The LinkedInPage to sync notifications for. """ self.linkedin_page = linkedin_page # Initialize client with the page's access token self.client = LinkedInClient(access_token=linkedin_page.access_token) self._stats: dict[str, Any] = { "notifications_fetched": 0, "signals_created": 0, "signals_skipped": 0, "errors": [], } def sync_page(self, full_sync: bool = False) -> dict[str, Any]: """Sync notifications for the LinkedIn page. Args: full_sync: If True, fetch historical notifications (60 days). If False, fetch only new notifications since last sync. Returns: Dict with sync statistics. Raises: LinkedInAPIError: If the LinkedIn API returns an error. """ # Reset stats for this sync run self._stats = { "notifications_fetched": 0, "signals_created": 0, "signals_skipped": 0, "errors": [], } logger.info( f"Starting sync for LinkedIn page {self.linkedin_page.name} " f"(org: {self.linkedin_page.organization_urn})" ) # Set status to SYNCING self.linkedin_page.sync_status = LinkedInSyncStatus.SYNCING self.linkedin_page.sync_error = "" self.linkedin_page.save(update_fields=["sync_status", "sync_error", "updated_at"]) try: # Get or create the source for this page source = self._get_or_create_source() # Fetch notifications from LinkedIn API notifications = self._fetch_notifications(full_sync=full_sync) self._stats["notifications_fetched"] = len(notifications) # Process each notification into a Signal with transaction.atomic(): for notification in notifications: try: signal = self._process_notification(notification, source) if signal: self._stats["signals_created"] += 1 else: self._stats["signals_skipped"] += 1 except Exception as e: logger.warning(f"Error processing notification {notification.get('notificationId')}: {e}") self._stats["errors"].append(str(e)) # Update sync timestamps self.linkedin_page.last_sync_at = django_timezone.now() self.linkedin_page.sync_status = LinkedInSyncStatus.ACTIVE self.linkedin_page.save(update_fields=["last_sync_at", "sync_status", "updated_at"]) logger.info( f"Sync complete for {self.linkedin_page.name}: " f"{self._stats['signals_created']} signals created, " f"{self._stats['signals_skipped']} skipped" ) return self._stats except LinkedInTokenExpiredError: # Try to refresh the token and retry once logger.info(f"Token expired for {self.linkedin_page.name}, attempting refresh") if self._refresh_token_and_retry(full_sync): return self._stats # If refresh failed, set to EXPIRED status logger.error(f"Token refresh failed for {self.linkedin_page.name}") self.linkedin_page.sync_status = LinkedInSyncStatus.EXPIRED self.linkedin_page.sync_error = "Token expired and refresh failed. Please reconnect." self.linkedin_page.save(update_fields=["sync_status", "sync_error", "updated_at"]) raise except LinkedInAPIError as e: logger.error(f"LinkedIn API error during sync: {e}") self.linkedin_page.sync_status = LinkedInSyncStatus.ERROR self.linkedin_page.sync_error = str(e)[:500] # Truncate long errors self.linkedin_page.save(update_fields=["sync_status", "sync_error", "updated_at"]) raise except Exception as e: logger.exception(f"Unexpected error during sync: {e}") self.linkedin_page.sync_status = LinkedInSyncStatus.ERROR self.linkedin_page.sync_error = f"Unexpected error: {str(e)[:450]}" self.linkedin_page.save(update_fields=["sync_status", "sync_error", "updated_at"]) raise def _refresh_token_and_retry(self, full_sync: bool = False) -> bool: """Try to refresh the access token and retry the sync. Args: full_sync: Whether to perform a full sync on retry. Returns: True if refresh and retry succeeded, False otherwise. """ if not self.linkedin_page.refresh_token: logger.warning(f"No refresh token available for {self.linkedin_page.name}") return False try: # Create a client for token refresh (doesn't need access token) refresh_client = LinkedInClient() token_response = refresh_client.refresh_access_token(refresh_token=self.linkedin_page.refresh_token) # Update the page with new tokens self.linkedin_page.access_token = token_response.access_token self.linkedin_page.token_expires_at = token_response.expires_at if token_response.refresh_token: self.linkedin_page.refresh_token = token_response.refresh_token self.linkedin_page.save( update_fields=[ "_access_token", "token_expires_at", "_refresh_token", "updated_at", ] ) # Update the client with the new token self.client = LinkedInClient(access_token=token_response.access_token) logger.info(f"Token refreshed successfully for {self.linkedin_page.name}") # Retry the sync # Note: We don't call sync_page() recursively to avoid infinite loops # Instead, we manually fetch and process notifications here source = self._get_or_create_source() notifications = self._fetch_notifications(full_sync=full_sync) self._stats["notifications_fetched"] = len(notifications) with transaction.atomic(): for notification in notifications: try: signal = self._process_notification(notification, source) if signal: self._stats["signals_created"] += 1 else: self._stats["signals_skipped"] += 1 except Exception as e: logger.warning(f"Error processing notification {notification.get('notificationId')}: {e}") self._stats["errors"].append(str(e)) # Update sync timestamps self.linkedin_page.last_sync_at = django_timezone.now() self.linkedin_page.sync_status = LinkedInSyncStatus.ACTIVE self.linkedin_page.save(update_fields=["last_sync_at", "sync_status", "updated_at"]) return True except LinkedInTokenExpiredError: logger.error(f"Refresh token also expired for {self.linkedin_page.name}") return False except Exception as e: logger.error(f"Failed to refresh token for {self.linkedin_page.name}: {e}") return False def _get_or_create_source(self) -> Source: """Get or create the Source record for this LinkedIn page.""" # Check if page already has a source linked if self.linkedin_page.source: return self.linkedin_page.source # Create new source and link it to the page source, created = Source.objects.get_or_create( kind="linkedin_page", external_id=self.linkedin_page.organization_urn, defaults={ "workspace": self.linkedin_page.workspace, "name": self.linkedin_page.name, }, ) if created: logger.info(f"Created source for LinkedIn page {self.linkedin_page.name}") # Link the source to the page self.linkedin_page.source = source self.linkedin_page.save(update_fields=["source", "updated_at"]) return source def _fetch_notifications(self, full_sync: bool = False) -> list[dict[str, Any]]: """Fetch notifications from the LinkedIn API. Args: full_sync: If True, fetch all notifications (for backfill). If False, fetch only notifications since last sync. Returns: List of notification dictionaries converted from LinkedInNotification dataclasses. """ # Calculate changed_since for incremental sync changed_since = None if not full_sync and self.linkedin_page.last_sync_at: # Convert to milliseconds timestamp changed_since = int(self.linkedin_page.last_sync_at.timestamp() * 1000) # Use the client's get_all_notifications for automatic pagination notifications = self.client.get_all_notifications( organization_urn=self.linkedin_page.organization_urn, changed_since=changed_since, ) # Convert LinkedInNotification dataclasses to dicts for processing return [self._notification_to_dict(n) for n in notifications] def _notification_to_dict(self, notification: Any) -> dict[str, Any]: """Convert LinkedInNotification dataclass to dict for processing. Args: notification: LinkedInNotification dataclass instance. Returns: Dictionary representation for _process_notification. """ from integrations.services.linkedin_client import LinkedInNotification if isinstance(notification, LinkedInNotification): return { "notificationId": notification.notification_id, "action": notification.action, "sourcePost": notification.source_post_urn, "lastModifiedAt": notification.last_modified_at, "subscriber": notification.subscriber_urn, "decoratedSourcePost": { "text": notification.text, "landingPageUrl": notification.landing_page_url, "owner": notification.owner_urn, "entity": notification.share_urn, "mediaCategory": notification.media_category, }, } # If already a dict (e.g., from tests), return as-is return dict(notification) if isinstance(notification, dict) else {"notificationId": 0} def _process_notification(self, notification: dict[str, Any], source: Source) -> Signal | None: """Process a single notification into a Signal record. Args: notification: Notification dict from LinkedIn API. source: The Source record to link the Signal to. Returns: The created Signal, or None if skipped (duplicate). """ notification_id: int = notification.get("notificationId", 0) action = notification.get("action", "UNKNOWN") decorated_post = notification.get("decoratedSourcePost", {}) # Build external_id from notification ID external_id = f"linkedin_notification_{notification_id}" # Check for duplicate if Signal.objects.filter(source=source, external_id=external_id).exists(): return None # Extract data from notification text = decorated_post.get("text", "") landing_url = decorated_post.get("landingPageUrl", "") author_urn = notification.get("subscriber", decorated_post.get("owner", "")) source_post = notification.get("sourcePost", "") # Convert timestamp from milliseconds last_modified_ms = notification.get("lastModifiedAt", 0) occurred_at = datetime.fromtimestamp(last_modified_ms / 1000, tz=UTC) # Build title based on action type title = self._build_title(action, text) # Build metadata metadata: LinkedInMentionMeta = { "action": action, "notification_id": notification_id, "source_post": source_post, "author_urn": author_urn, "organization_urn": self.linkedin_page.organization_urn, } # Add reaction type for LIKE actions if action == "LIKE" and "reactionType" in notification: metadata["reaction_type"] = notification["reactionType"] # Map author to member (if possible) actor = self._map_author_to_member( person_urn=author_urn, workspace=source.workspace, ) if actor: metadata["author_member_id"] = str(actor.pk) # Create the Signal signal = Signal.objects.create( workspace=source.workspace, source=source, signal_type="mention", occurred_at=occurred_at, title=title, external_url=landing_url or "https://www.linkedin.com/feed/", external_id=external_id, body=text, actor=actor, metadata={"linkedin": metadata}, ) return signal def _build_title(self, action: str, text: str) -> str: """Build a title for the signal based on action type. Args: action: The LinkedIn action type (SHARE_MENTION, COMMENT, etc.) text: The notification text content. Returns: A human-readable title for the signal. """ action_labels = { "SHARE_MENTION": "Mentioned in post", "COMMENT": "Comment on content", "SHARE": "Content shared", "LIKE": "Reaction received", } label = action_labels.get(action, f"LinkedIn {action}") # Add text preview if available if text: preview = text[:50] + "..." if len(text) > 50 else text return f"{label}: {preview}" return label def _map_author_to_member( self, person_urn: str, workspace: Workspace, person_name: str | None = None, ) -> Member | None: """Map a LinkedIn person URN to a Member record. Args: person_urn: The LinkedIn person URN (e.g., urn:li:person:ABC123). workspace: The workspace to find/create the member in. person_name: Optional name for creating new member. Returns: The Member record, or None if not found/created. """ if not person_urn: return None # Try to find existing member via Identity try: identity = Identity.objects.select_related("member").get( provider="linkedin", external_id=person_urn, member__workspace=workspace, ) return identity.member except Identity.DoesNotExist: pass # Extract ID from URN for display person_id = person_urn.split(":")[-1] if ":" in person_urn else person_urn display_name = person_name or f"LinkedIn User {person_id[:8]}" # Create new member and identity try: with transaction.atomic(): member = Member.objects.create( workspace=workspace, display_name=display_name, ) Identity.objects.create( member=member, provider="linkedin", external_id=person_urn, handle=person_id, # Use URN ID as handle ) return member except IntegrityError: # Race condition - identity already created try: identity = Identity.objects.select_related("member").get( provider="linkedin", external_id=person_urn, member__workspace=workspace, ) return identity.member except Identity.DoesNotExist: logger.warning(f"Could not create member for {person_urn}") return None