"""LinkedIn API client for OAuth and API operations.""" import logging import secrets from dataclasses import dataclass from datetime import UTC, datetime, timedelta from typing import Any from urllib.parse import urlencode import requests from django.conf import settings from django.core.signing import BadSignature, TimestampSigner from django.utils import timezone from integrations.exceptions import ( LinkedInAPIError, LinkedInAuthError, LinkedInInvalidStateError, LinkedInRateLimitError, LinkedInTokenExpiredError, ) logger = logging.getLogger(__name__) @dataclass class LinkedInTokenResponse: """Response from LinkedIn OAuth token exchange.""" access_token: str expires_in: int refresh_token: str refresh_token_expires_in: int scope: str @property def expires_at(self) -> datetime: """Calculate token expiration datetime.""" return timezone.now() + timedelta(seconds=self.expires_in) @dataclass class LinkedInOrganization: """LinkedIn organization data.""" organization_urn: str organization_id: str name: str vanity_name: str | None logo_url: str | None role: str state: str @dataclass class LinkedInNotification: """LinkedIn notification data from the API.""" notification_id: int organization_urn: str action: str source_post_urn: str share_urn: str | None owner_urn: str text: str landing_page_url: str media_category: str | None last_modified_at: int # Unix timestamp in milliseconds subscriber_urn: str @property def last_modified_datetime(self) -> datetime: """Convert lastModifiedAt to datetime.""" return datetime.fromtimestamp(self.last_modified_at / 1000, tz=UTC) class LinkedInClient: """Client for LinkedIn OAuth and API operations. Handles: - OAuth 2.0 authorization code flow - Token refresh - Organization Social Action Notifications API - Rate limiting and error handling """ def __init__( self, *, access_token: str | None = None, client_id: str | None = None, client_secret: str | None = None, ) -> None: """Initialize the LinkedIn client. Args: access_token: Optional access token for API calls. client_id: OAuth client ID (defaults to settings). client_secret: OAuth client secret (defaults to settings). """ self.access_token = access_token self.client_id = client_id or settings.LINKEDIN_CLIENT_ID self.client_secret = client_secret or settings.LINKEDIN_CLIENT_SECRET self.redirect_uri = settings.LINKEDIN_REDIRECT_URI self.api_version = settings.LINKEDIN_API_VERSION self.api_base_url = settings.LINKEDIN_API_BASE_URL self._signer = TimestampSigner() self._session = requests.Session() self._session.headers.update( { "Content-Type": "application/json", "X-Restli-Protocol-Version": "2.0.0", } ) # ========================================================================= # OAuth Methods # ========================================================================= def generate_state(self, *, workspace_id: str) -> str: """Generate a signed state parameter for OAuth. Args: workspace_id: The workspace ID to encode in state. Returns: Signed state string that includes workspace_id and a nonce. """ nonce = secrets.token_urlsafe(16) data = f"{workspace_id}:{nonce}" return self._signer.sign(data) def validate_state(self, *, state: str, max_age: int = 600) -> str: """Validate and extract workspace_id from OAuth state. Args: state: The state parameter from OAuth callback. max_age: Maximum age in seconds (default: 10 minutes). Returns: The workspace_id encoded in the state. Raises: LinkedInInvalidStateError: If state is invalid or expired. """ try: data = self._signer.unsign(state, max_age=max_age) workspace_id, _nonce = data.split(":", 1) return workspace_id except (BadSignature, ValueError) as e: logger.warning("Invalid LinkedIn OAuth state: %s", e) raise LinkedInInvalidStateError() from e def get_authorization_url(self, *, state: str) -> str: """Generate the LinkedIn OAuth authorization URL. Args: state: The state parameter for CSRF protection. Returns: Full authorization URL to redirect user to. """ params = { "response_type": "code", "client_id": self.client_id, "redirect_uri": self.redirect_uri, "state": state, "scope": settings.LINKEDIN_OAUTH_SCOPE, } return f"{settings.LINKEDIN_AUTHORIZATION_URL}?{urlencode(params)}" def exchange_code_for_token(self, *, code: str) -> LinkedInTokenResponse: """Exchange authorization code for access token. Args: code: The authorization code from OAuth callback. Returns: LinkedInTokenResponse with tokens and expiration. Raises: LinkedInAuthError: If token exchange fails. """ data = { "grant_type": "authorization_code", "code": code, "redirect_uri": self.redirect_uri, "client_id": self.client_id, "client_secret": self.client_secret, } try: response = self._session.post( settings.LINKEDIN_TOKEN_URL, data=data, headers={"Content-Type": "application/x-www-form-urlencoded"}, timeout=30, ) except requests.RequestException as e: logger.error("LinkedIn token exchange request failed: %s", e) raise LinkedInAuthError(f"Token exchange request failed: {e}") from e if response.status_code != 200: error_data = response.json() if response.content else {} error_code = error_data.get("error", "unknown_error") error_desc = error_data.get("error_description", response.text) logger.error( "LinkedIn token exchange failed: %s - %s", error_code, error_desc, ) raise LinkedInAuthError( f"Token exchange failed: {error_desc}", error_code=error_code, error_description=error_desc, ) data = response.json() return LinkedInTokenResponse( access_token=data["access_token"], expires_in=data["expires_in"], refresh_token=data.get("refresh_token", ""), refresh_token_expires_in=data.get("refresh_token_expires_in", 0), scope=data.get("scope", ""), ) def refresh_access_token(self, *, refresh_token: str) -> LinkedInTokenResponse: """Refresh an expired access token. Args: refresh_token: The refresh token to use. Returns: LinkedInTokenResponse with new tokens. Raises: LinkedInTokenExpiredError: If refresh token is also expired. LinkedInAuthError: If refresh fails for other reasons. """ data = { "grant_type": "refresh_token", "refresh_token": refresh_token, "client_id": self.client_id, "client_secret": self.client_secret, } try: response = self._session.post( settings.LINKEDIN_TOKEN_URL, data=data, headers={"Content-Type": "application/x-www-form-urlencoded"}, timeout=30, ) except requests.RequestException as e: logger.error("LinkedIn token refresh request failed: %s", e) raise LinkedInAuthError(f"Token refresh request failed: {e}") from e if response.status_code != 200: error_data = response.json() if response.content else {} error_code = error_data.get("error", "unknown_error") if error_code in ("invalid_grant", "expired_token"): raise LinkedInTokenExpiredError("Refresh token has expired, re-authentication required") error_desc = error_data.get("error_description", response.text) raise LinkedInAuthError( f"Token refresh failed: {error_desc}", error_code=error_code, error_description=error_desc, ) data = response.json() return LinkedInTokenResponse( access_token=data["access_token"], expires_in=data["expires_in"], refresh_token=data.get("refresh_token", refresh_token), refresh_token_expires_in=data.get("refresh_token_expires_in", 0), scope=data.get("scope", ""), ) # ========================================================================= # API Methods # ========================================================================= def _make_api_request( self, *, method: str, endpoint: str, params: dict[str, Any] | None = None, json_data: dict[str, Any] | None = None, ) -> dict[str, Any]: """Make an authenticated API request. Args: method: HTTP method (GET, POST, etc.) endpoint: API endpoint path. params: Query parameters. json_data: JSON body data. Returns: Parsed JSON response. Raises: LinkedInAPIError: If API request fails. LinkedInRateLimitError: If rate limited. LinkedInTokenExpiredError: If token is expired. """ if not self.access_token: raise LinkedInAuthError("No access token configured") url = f"{self.api_base_url}{endpoint}" headers = { "Authorization": f"Bearer {self.access_token}", "LinkedIn-Version": self.api_version, } try: response = self._session.request( method=method, url=url, params=params, json=json_data, headers=headers, timeout=30, ) except requests.RequestException as e: logger.error("LinkedIn API request failed: %s %s - %s", method, endpoint, e) raise LinkedInAPIError(f"API request failed: {e}") from e # Handle error responses if response.status_code == 401: raise LinkedInTokenExpiredError() elif response.status_code == 403: raise LinkedInAuthError( "Insufficient permissions for this operation", error_code="FORBIDDEN", ) elif response.status_code == 429: retry_after = response.headers.get("Retry-After") raise LinkedInRateLimitError( "Rate limit exceeded", retry_after=int(retry_after) if retry_after else None, ) elif response.status_code >= 400: error_data = response.json() if response.content else {} raise LinkedInAPIError( error_data.get("message", f"API error: {response.status_code}"), status_code=response.status_code, error_code=error_data.get("code"), ) return response.json() if response.content else {} def get_organizations(self) -> list[LinkedInOrganization]: """Get organizations the authenticated user can admin. Returns: List of organizations with admin access. Raises: LinkedInAPIError: If API request fails. """ # Get organization access control response = self._make_api_request( method="GET", endpoint="/organizationAcls", params={ "q": "roleAssignee", "role": "ADMINISTRATOR", "state": "APPROVED", }, ) organizations = [] for element in response.get("elements", []): org_urn = element.get("organization") or element.get("organizationalEntity") if not org_urn: continue # Extract org ID from URN org_id = org_urn.split(":")[-1] if ":" in org_urn else org_urn # Get organization details try: org_details = self._get_organization_details(organization_id=org_id) except LinkedInAPIError: # Skip orgs we can't get details for logger.warning("Failed to get details for org %s", org_urn) continue organizations.append( LinkedInOrganization( organization_urn=org_urn, organization_id=org_id, name=org_details.get("localizedName", f"Organization {org_id}"), vanity_name=org_details.get("vanityName"), logo_url=self._extract_logo_url(org_details), role=element.get("role", "ADMINISTRATOR"), state=element.get("state", "APPROVED"), ) ) return organizations def _get_organization_details(self, *, organization_id: str) -> dict[str, Any]: """Get detailed information about an organization. Args: organization_id: The organization's numeric ID. Returns: Organization details from API. """ return self._make_api_request( method="GET", endpoint=f"/organizations/{organization_id}", params={ "projection": "(id,localizedName,vanityName,logoV2)", }, ) def _extract_logo_url(self, org_details: dict[str, Any]) -> str | None: """Extract logo URL from organization details.""" logo_v2 = org_details.get("logoV2", {}) original = logo_v2.get("original") if original: return str(original) # Try cropped versions cropped = logo_v2.get("cropped") if cropped: return str(cropped) return None def get_notifications( self, *, organization_urn: str, actions: list[str] | None = None, changed_since: int | None = None, start: int = 0, count: int = 50, ) -> tuple[list[LinkedInNotification], int | None]: """Get organization social action notifications. Args: organization_urn: The organization URN to get notifications for. actions: List of action types to filter (default: all). changed_since: Unix timestamp in ms to get notifications after. start: Pagination start index. count: Number of results per page. Returns: Tuple of (notifications list, next start index or None if no more). Raises: LinkedInAPIError: If API request fails. """ if actions is None: actions = ["SHARE_MENTION", "COMMENT", "SHARE", "LIKE"] params: dict[str, Any] = { "q": "criteria", "organizationalEntity": organization_urn, "actions": f"List({','.join(actions)})", "start": start, "count": count, } if changed_since is not None: params["changedSince"] = changed_since response = self._make_api_request( method="GET", endpoint="/organizationalEntityNotifications", params=params, ) notifications = [] for elem in response.get("elements", []): decorated = elem.get("decoratedSourcePost", {}) notifications.append( LinkedInNotification( notification_id=elem["notificationId"], organization_urn=elem["organizationalEntity"], action=elem["action"], source_post_urn=elem.get("sourcePost", ""), share_urn=decorated.get("entity"), owner_urn=decorated.get("owner", ""), text=decorated.get("text", ""), landing_page_url=decorated.get("landingPageUrl", ""), media_category=decorated.get("mediaCategory"), last_modified_at=elem["lastModifiedAt"], subscriber_urn=elem.get("subscriber", ""), ) ) # Check for pagination paging = response.get("paging", {}) total = paging.get("total", len(notifications)) next_start = start + count if start + count < total else None return notifications, next_start def get_all_notifications( self, *, organization_urn: str, actions: list[str] | None = None, changed_since: int | None = None, ) -> list[LinkedInNotification]: """Get all notifications with automatic pagination. Args: organization_urn: The organization URN. actions: Action types to filter. changed_since: Timestamp to filter from. Returns: Complete list of notifications. """ all_notifications = [] start = 0 while True: notifications, next_start = self.get_notifications( organization_urn=organization_urn, actions=actions, changed_since=changed_since, start=start, ) all_notifications.extend(notifications) if next_start is None: break start = next_start return all_notifications