552 lines
23 KiB
Python
552 lines
23 KiB
Python
# 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
|
||
# } |