# jobs/linkedin_service.py import uuid from urllib.parse import quote import requests import logging from django.conf import settings from urllib.parse import urlencode, quote logger = logging.getLogger(__name__) class LinkedInService: def __init__(self): self.client_id = settings.LINKEDIN_CLIENT_ID self.client_secret = settings.LINKEDIN_CLIENT_SECRET self.redirect_uri = settings.LINKEDIN_REDIRECT_URI self.access_token = None def get_auth_url(self): """Generate LinkedIn OAuth URL""" params = { 'response_type': 'code', 'client_id': self.client_id, 'redirect_uri': self.redirect_uri, 'scope': 'w_member_social openid profile', 'state': 'university_ats_linkedin' } return f"https://www.linkedin.com/oauth/v2/authorization?{urlencode(params)}" def get_access_token(self, code): """Exchange authorization code for access token""" # This function exchanges LinkedIn’s temporary authorization code for a usable access token. url = "https://www.linkedin.com/oauth/v2/accessToken" data = { 'grant_type': 'authorization_code', 'code': code, 'redirect_uri': self.redirect_uri, 'client_id': self.client_id, 'client_secret': self.client_secret } try: response = requests.post(url, data=data, timeout=60) response.raise_for_status() token_data = response.json() """ Example response:{ "access_token": "AQXq8HJkLmNpQrStUvWxYz...", "expires_in": 5184000 } """ self.access_token = token_data.get('access_token') return self.access_token except Exception as e: logger.error(f"Error getting access token: {e}") raise def get_user_profile(self): """Get user profile information""" if not self.access_token: raise Exception("No access token available") url = "https://api.linkedin.com/v2/userinfo" headers = {'Authorization': f'Bearer {self.access_token}'} try: response = requests.get(url, headers=headers, timeout=60) response.raise_for_status() # Ensure we raise an error for bad responses(4xx, 5xx) and does nothing for 2xx(success) return response.json() # returns a dict from json response (deserialize) except Exception as e: logger.error(f"Error getting user profile: {e}") raise def register_image_upload(self, person_urn): """Step 1: Register image upload with LinkedIn""" url = "https://api.linkedin.com/v2/assets?action=registerUpload" headers = { 'Authorization': f'Bearer {self.access_token}', 'Content-Type': 'application/json', 'X-Restli-Protocol-Version': '2.0.0' } payload = { "registerUploadRequest": { "recipes": ["urn:li:digitalmediaRecipe:feedshare-image"], "owner": f"urn:li:person:{person_urn}", "serviceRelationships": [{ "relationshipType": "OWNER", "identifier": "urn:li:userGeneratedContent" }] } } response = requests.post(url, headers=headers, json=payload, timeout=30) response.raise_for_status() data = response.json() return { 'upload_url': data['value']['uploadMechanism']['com.linkedin.digitalmedia.uploading.MediaUploadHttpRequest']['uploadUrl'], 'asset': data['value']['asset'] } def upload_image_to_linkedin(self, upload_url, image_file): """Step 2: Upload actual image file to LinkedIn""" # Open and read the Django ImageField image_file.open() image_content = image_file.read() image_file.close() headers = { 'Authorization': f'Bearer {self.access_token}', } response = requests.post(upload_url, headers=headers, data=image_content, timeout=60) response.raise_for_status() return True def create_job_post(self, job_posting): """Create a job announcement post on LinkedIn (with image support)""" if not self.access_token: raise Exception("Not authenticated with LinkedIn") try: # Get user profile for person URN profile = self.get_user_profile() person_urn = profile.get('sub') if not person_urn: raise Exception("Could not retrieve LinkedIn user ID") # Check if job has an image try: image_upload = job_posting.files.first() has_image = image_upload and image_upload.linkedinpost_image except Exception: has_image = False if has_image: # === POST WITH IMAGE === try: # Step 1: Register image upload upload_info = self.register_image_upload(person_urn) # Step 2: Upload image self.upload_image_to_linkedin( upload_info['upload_url'], image_upload.linkedinpost_image ) # Step 3: Create post with image return self.create_job_post_with_image( job_posting, image_upload.linkedinpost_image, person_urn, upload_info['asset'] ) except Exception as e: logger.error(f"Image upload failed: {e}") # Fall back to text-only post if image upload fails has_image = False # === FALLBACK TO URL/ARTICLE POST === # Add unique timestamp to prevent duplicates from django.utils import timezone import random unique_suffix = f"\n\nPosted: {timezone.now().strftime('%b %d, %Y at %I:%M %p')} (ID: {random.randint(1000, 9999)})" message_parts = [f"πŸš€ **We're Hiring: {job_posting.title}**"] if job_posting.department: message_parts.append(f"**Department:** {job_posting.department}") if job_posting.description: message_parts.append(f"\n{job_posting.description}") details = [] if job_posting.job_type: details.append(f"πŸ’Ό {job_posting.get_job_type_display()}") if job_posting.get_location_display() != 'Not specified': details.append(f"πŸ“ {job_posting.get_location_display()}") if job_posting.workplace_type: details.append(f"🏠 {job_posting.get_workplace_type_display()}") if job_posting.salary_range: details.append(f"πŸ’° {job_posting.salary_range}") if details: message_parts.append("\n" + " | ".join(details)) if job_posting.application_url: message_parts.append(f"\nπŸ”— **Apply now:** {job_posting.application_url}") hashtags = self.hashtags_list(job_posting.hash_tags) if job_posting.department: dept_hashtag = f"#{job_posting.department.replace(' ', '')}" hashtags.insert(0, dept_hashtag) message_parts.append("\n\n" + " ".join(hashtags)) message_parts.append(unique_suffix) message = "\n".join(message_parts) # πŸ”₯ FIX URL - REMOVE TRAILING SPACES πŸ”₯ url = "https://api.linkedin.com/v2/ugcPosts" headers = { 'Authorization': f'Bearer {self.access_token}', 'Content-Type': 'application/json', 'X-Restli-Protocol-Version': '2.0.0' } payload = { "author": f"urn:li:person:{person_urn}", "lifecycleState": "PUBLISHED", "specificContent": { "com.linkedin.ugc.ShareContent": { "shareCommentary": {"text": message}, "shareMediaCategory": "ARTICLE", "media": [{ "status": "READY", "description": {"text": f"Apply for {job_posting.title} at our university!"}, "originalUrl": job_posting.application_url, "title": {"text": job_posting.title} }] } }, "visibility": { "com.linkedin.ugc.MemberNetworkVisibility": "PUBLIC" } } response = requests.post(url, headers=headers, json=payload, timeout=60) response.raise_for_status() post_id = response.headers.get('x-restli-id', '') post_url = f"https://www.linkedin.com/feed/update/{quote(post_id)}/" if post_id else "" return { 'success': True, 'post_id': post_id, 'post_url': post_url, 'status_code': response.status_code } except Exception as e: logger.error(f"Error creating LinkedIn post: {e}") return { 'success': False, 'error': str(e), 'status_code': getattr(e.response, 'status_code', 500) if hasattr(e, 'response') else 500 } # def create_job_post_with_image(self, job_posting, image_file, person_urn, asset_urn): # """Step 3: Create post with uploaded image""" # url = "https://api.linkedin.com/v2/ugcPosts" # headers = { # 'Authorization': f'Bearer {self.access_token}', # 'Content-Type': 'application/json', # 'X-Restli-Protocol-Version': '2.0.0' # } # # Build the same message as before # message_parts = [f"πŸš€ **We're Hiring: {job_posting.title}**"] # if job_posting.department: # message_parts.append(f"**Department:** {job_posting.department}") # if job_posting.description: # message_parts.append(f"\n{job_posting.description}") # details = [] # if job_posting.job_type: # details.append(f"πŸ’Ό {job_posting.get_job_type_display()}") # if job_posting.get_location_display() != 'Not specified': # details.append(f"πŸ“ {job_posting.get_location_display()}") # if job_posting.workplace_type: # details.append(f"🏠 {job_posting.get_workplace_type_display()}") # if job_posting.salary_range: # details.append(f"πŸ’° {job_posting.salary_range}") # if details: # message_parts.append("\n" + " | ".join(details)) # if job_posting.application_url: # message_parts.append(f"\nπŸ”— **Apply now:** {job_posting.application_url}") # hashtags = self.hashtags_list(job_posting.hash_tags) # if job_posting.department: # dept_hashtag = f"#{job_posting.department.replace(' ', '')}" # hashtags.insert(0, dept_hashtag) # message_parts.append("\n\n" + " ".join(hashtags)) # message = "\n".join(message_parts) # # Create image post payload # payload = { # "author": f"urn:li:person:{person_urn}", # "lifecycleState": "PUBLISHED", # "specificContent": { # "com.linkedin.ugc.ShareContent": { # "shareCommentary": {"text": message}, # "shareMediaCategory": "IMAGE", # "media": [{ # "status": "READY", # "media": asset_urn # }] # } # }, # "visibility": { # "com.linkedin.ugc.MemberNetworkVisibility": "PUBLIC" # } # } # response = requests.post(url, headers=headers, json=payload, timeout=30) # response.raise_for_status() # post_id = response.headers.get('x-restli-id', '') # post_url = f"https://www.linkedin.com/feed/update/{quote(post_id)}/" if post_id else "" # return { # 'success': True, # 'post_id': post_id, # 'post_url': post_url, # 'status_code': response.status_code # } def hashtags_list(self,hash_tags_str): """Convert comma-separated hashtags string to list""" if not hash_tags_str: return [""] tags = [tag.strip() for tag in hash_tags_str.split(',') if tag.strip()] if not tags: return ["#HigherEd", "#Hiring", "#FacultyJobs", "#UniversityJobs"] return tags # def create_job_post(self, job_posting): # """Create a job announcement post on LinkedIn (with image support)""" # if not self.access_token: # raise Exception("Not authenticated with LinkedIn") # try: # # Get user profile for person URN # profile = self.get_user_profile() # person_urn = profile.get('sub') # if not person_urn: # raise Exception("Could not retrieve LinkedIn user ID") # # Check if job has an image # try: # image_upload = job_posting.files.first() # has_image = image_upload and image_upload.linkedinpost_image # except Exception: # has_image = False # if has_image: # # === POST WITH IMAGE === # upload_info = self.register_image_upload(person_urn) # self.upload_image_to_linkedin( # upload_info['upload_url'], # image_upload.linkedinpost_image # ) # return self.create_job_post_with_image( # job_posting, # image_upload.linkedinpost_image, # person_urn, # upload_info['asset'] # ) # else: # # === FALLBACK TO URL/ARTICLE POST === # # πŸ”₯ ADD UNIQUE TIMESTAMP TO PREVENT DUPLICATES πŸ”₯ # from django.utils import timezone # import random # unique_suffix = f"\n\nPosted: {timezone.now().strftime('%b %d, %Y at %I:%M %p')} (ID: {random.randint(1000, 9999)})" # message_parts = [f"πŸš€ **We're Hiring: {job_posting.title}**"] # if job_posting.department: # message_parts.append(f"**Department:** {job_posting.department}") # if job_posting.description: # message_parts.append(f"\n{job_posting.description}") # details = [] # if job_posting.job_type: # details.append(f"πŸ’Ό {job_posting.get_job_type_display()}") # if job_posting.get_location_display() != 'Not specified': # details.append(f"πŸ“ {job_posting.get_location_display()}") # if job_posting.workplace_type: # details.append(f"🏠 {job_posting.get_workplace_type_display()}") # if job_posting.salary_range: # details.append(f"πŸ’° {job_posting.salary_range}") # if details: # message_parts.append("\n" + " | ".join(details)) # if job_posting.application_url: # message_parts.append(f"\nπŸ”— **Apply now:** {job_posting.application_url}") # hashtags = self.hashtags_list(job_posting.hash_tags) # if job_posting.department: # dept_hashtag = f"#{job_posting.department.replace(' ', '')}" # hashtags.insert(0, dept_hashtag) # message_parts.append("\n\n" + " ".join(hashtags)) # message_parts.append(unique_suffix) # πŸ”₯ Add unique suffix # message = "\n".join(message_parts) # # πŸ”₯ FIX URL - REMOVE TRAILING SPACES πŸ”₯ # url = "https://api.linkedin.com/v2/ugcPosts" # headers = { # 'Authorization': f'Bearer {self.access_token}', # 'Content-Type': 'application/json', # 'X-Restli-Protocol-Version': '2.0.0' # } # payload = { # "author": f"urn:li:person:{person_urn}", # "lifecycleState": "PUBLISHED", # "specificContent": { # "com.linkedin.ugc.ShareContent": { # "shareCommentary": {"text": message}, # "shareMediaCategory": "ARTICLE", # "media": [{ # "status": "READY", # "description": {"text": f"Apply for {job_posting.title} at our university!"}, # "originalUrl": job_posting.application_url, # "title": {"text": job_posting.title} # }] # } # }, # "visibility": { # "com.linkedin.ugc.MemberNetworkVisibility": "PUBLIC" # } # } # response = requests.post(url, headers=headers, json=payload, timeout=60) # response.raise_for_status() # post_id = response.headers.get('x-restli-id', '') # # πŸ”₯ FIX POST URL - REMOVE TRAILING SPACES πŸ”₯ # post_url = f"https://www.linkedin.com/feed/update/{quote(post_id)}/" if post_id else "" # return { # 'success': True, # 'post_id': post_id, # 'post_url': post_url, # 'status_code': response.status_code # } # except Exception as e: # logger.error(f"Error creating LinkedIn post: {e}") # return { # 'success': False, # 'error': str(e), # 'status_code': getattr(e.response, 'status_code', 500) if hasattr(e, 'response') else 500 # } # def create_job_post(self, job_posting): # """Create a job announcement post on LinkedIn""" # if not self.access_token: # raise Exception("Not authenticated with LinkedIn") # try: # # Get user profile for person URN # profile = self.get_user_profile() # person_urn = profile.get('sub') # if not person_urn: # uniform resource name used to uniquely identify linked-id for internal systems and apis # raise Exception("Could not retrieve LinkedIn user ID") # # Build professional job post message # message_parts = [f"πŸš€ **We're Hiring: {job_posting.title}**"] # if job_posting.department: # message_parts.append(f"**Department:** {job_posting.department}") # if job_posting.description: # message_parts.append(f"\n{job_posting.description}") # # Add job details # details = [] # if job_posting.job_type: # details.append(f"πŸ’Ό {job_posting.get_job_type_display()}") # if job_posting.get_location_display() != 'Not specified': # details.append(f"πŸ“ {job_posting.get_location_display()}") # if job_posting.workplace_type: # details.append(f"🏠 {job_posting.get_workplace_type_display()}") # if job_posting.salary_range: # details.append(f"πŸ’° {job_posting.salary_range}") # if details: # message_parts.append("\n" + " | ".join(details)) # # Add application link # if job_posting.application_url: # message_parts.append(f"\nπŸ”— **Apply now:** {job_posting.application_url}") # # Add hashtags # hashtags = ["#HigherEd", "#Hiring", "#FacultyJobs", "#UniversityJobs"] # if job_posting.department: # dept_hashtag = f"#{job_posting.department.replace(' ', '')}" # hashtags.insert(0, dept_hashtag) # message_parts.append("\n\n" + " ".join(hashtags)) # message = "\n".join(message_parts) # # Create LinkedIn post # url = "https://api.linkedin.com/v2/ugcPosts" # headers = { # 'Authorization': f'Bearer {self.access_token}', # 'Content-Type': 'application/json', # 'X-Restli-Protocol-Version': '2.0.0' # } # payload = { # "author": f"urn:li:person:{person_urn}", # "lifecycleState": "PUBLISHED", # "specificContent": { # "com.linkedin.ugc.ShareContent": { # "shareCommentary": {"text": message}, # "shareMediaCategory": "ARTICLE", # "media": [{ # "status": "READY", # "description": {"text": f"Apply for {job_posting.title} at our university!"}, # "originalUrl": job_posting.application_url, # "title": {"text": job_posting.title} # }] # } # }, # "visibility": { # "com.linkedin.ugc.MemberNetworkVisibility": "PUBLIC" # } # } # response = requests.post(url, headers=headers, json=payload, timeout=60) # response.raise_for_status() # # Extract post ID from response # post_id = response.headers.get('x-restli-id', '') # post_url = f"https://www.linkedin.com/feed/update/{quote(post_id)}/" if post_id else "" # return { # 'success': True, # 'post_id': post_id, # 'post_url': post_url, # 'status_code': response.status_code # } # except Exception as e: # logger.error(f"Error creating LinkedIn post: {e}") # return { # 'success': False, # 'error': str(e), # 'status_code': getattr(e.response, 'status_code', 500) if hasattr(e, 'response') else 500 # }