"""LinkedIn integration models. This module contains Django models for the LinkedIn integration: - LinkedInSyncStatus: Enum for sync status choices - LinkedInPage: Represents a connected LinkedIn Organization Page """ import logging from datetime import timedelta from django.db import models from django.utils import timezone from accounts.models import Workspace from core.utils.encryption import decrypt_value, encrypt_value from integrations.models.base import ORGANIZATION_URN_PATTERN, TimestampedModel from sources.models import Source logger = logging.getLogger(__name__) class LinkedInSyncStatus(models.TextChoices): """Sync status for LinkedIn page polling.""" PENDING = "PENDING", "Pending" # Awaiting initial sync ACTIVE = "ACTIVE", "Active" # Sync enabled, waiting for next scheduled run SYNCING = "SYNCING", "Syncing" # Currently syncing ERROR = "ERROR", "Error" # Last sync failed PAUSED = "PAUSED", "Paused" # User paused syncing EXPIRED = "EXPIRED", "Expired" # Token expired, needs reauth class LinkedInPage(TimestampedModel): """Represents a connected LinkedIn Organization Page within a workspace. Stores OAuth credentials (encrypted), sync state, and organization metadata. Each LinkedInPage is linked to a Source record for the unified activity model. """ workspace = models.ForeignKey( Workspace, on_delete=models.CASCADE, related_name="linkedin_pages", ) source = models.OneToOneField( Source, on_delete=models.SET_NULL, null=True, blank=True, related_name="linkedin_page", ) # LinkedIn Organization identifiers organization_urn = models.CharField( max_length=100, help_text="LinkedIn org URN (urn:li:organization:123)", ) organization_id = models.CharField( max_length=50, db_index=True, help_text="Numeric org ID extracted from URN", ) name = models.CharField(max_length=255, help_text="Organization display name") vanity_name = models.CharField( max_length=100, blank=True, help_text="LinkedIn vanity URL slug", ) logo_url = models.URLField(max_length=500, blank=True, help_text="Organization logo URL") # OAuth token storage (encrypted per Constitution B14) _access_token = models.TextField(blank=True, db_column="access_token") _refresh_token = models.TextField(blank=True, db_column="refresh_token") token_expires_at = models.DateTimeField(null=True, blank=True) # Who authorized this connection authorized_user_urn = models.CharField( max_length=100, blank=True, help_text="URN of user who authorized", ) # Sync configuration and state sync_status = models.CharField( max_length=20, choices=LinkedInSyncStatus.choices, default=LinkedInSyncStatus.PENDING, ) sync_error = models.TextField(blank=True, help_text="Last sync error message") last_sync_at = models.DateTimeField(null=True, blank=True) next_sync_at = models.DateTimeField(null=True, blank=True) sync_interval_minutes = models.IntegerField( default=15, help_text="Polling interval in minutes", ) sync_cursors = models.JSONField( default=dict, blank=True, help_text="Sync state and pagination cursors", ) # Soft delete for data retention is_archived = models.BooleanField(default=False) class Meta: app_label = "integrations" verbose_name = "LinkedIn Page" verbose_name_plural = "LinkedIn Pages" constraints = [ models.UniqueConstraint( fields=["workspace", "organization_urn"], condition=models.Q(is_archived=False), name="unique_active_linkedin_page_per_workspace", ), ] indexes = [ models.Index(fields=["organization_id"]), models.Index(fields=["sync_status"]), models.Index(fields=["next_sync_at"]), models.Index(fields=["is_archived"]), ] def __str__(self) -> str: return f"{self.name} ({self.organization_id})" def save(self, *args, **kwargs) -> None: """Extract organization_id from URN before saving.""" if self.organization_urn and not self.organization_id: # Extract numeric ID from urn:li:organization:123456 parts = self.organization_urn.split(":") if len(parts) == 4: self.organization_id = parts[3] super().save(*args, **kwargs) def clean(self) -> None: """Validate the model data.""" from django.core.exceptions import ValidationError if self.organization_urn and not ORGANIZATION_URN_PATTERN.match(self.organization_urn): raise ValidationError({"organization_urn": "Must match pattern urn:li:organization:"}) @property def access_token(self) -> str: """Get the decrypted access token.""" if not self._access_token: return "" try: return decrypt_value(self._access_token) except ValueError: logger.warning( "Failed to decrypt access token for LinkedIn page %s", self.organization_id, ) return "" @access_token.setter def access_token(self, value: str) -> None: """Set the access token with Fernet encryption.""" if value: self._access_token = encrypt_value(value) else: self._access_token = "" @property def refresh_token(self) -> str: """Get the decrypted refresh token.""" if not self._refresh_token: return "" try: return decrypt_value(self._refresh_token) except ValueError: logger.warning( "Failed to decrypt refresh token for LinkedIn page %s", self.organization_id, ) return "" @refresh_token.setter def refresh_token(self, value: str) -> None: """Set the refresh token with Fernet encryption.""" if value: self._refresh_token = encrypt_value(value) else: self._refresh_token = "" @property def is_token_expired(self) -> bool: """Check if the access token has expired.""" if not self.token_expires_at: return True return timezone.now() >= self.token_expires_at @property def linkedin_url(self) -> str: """Get the LinkedIn URL for this organization.""" if self.vanity_name: return f"https://www.linkedin.com/company/{self.vanity_name}" return f"https://www.linkedin.com/company/{self.organization_id}" def schedule_next_sync(self) -> None: """Schedule the next sync based on the configured interval.""" self.next_sync_at = timezone.now() + timedelta(minutes=self.sync_interval_minutes)