import requests
import time
import random
import json
from datetime import datetime, timedelta
from typing import List, Dict, Optional
import openai
from dataclasses import dataclass
@dataclass
class LinkedInPost:
post_id: str
author_id: str
content: str
created_time: str
post_url: str
class LinkedInEngagementBot:
def __init__(self, linkedin_access_token: str, openai_api_key: str):
"""
Initialize the LinkedIn engagement bot
Args:
linkedin_access_token: Your LinkedIn API access token
openai_api_key: Your OpenAI API key for ChatGPT
"""
self.linkedin_token = linkedin_access_token
self.openai_client = openai.OpenAI(api_key=openai_api_key)
self.base_url = "https://api.linkedin.com/v2"
self.headers = {
"Authorization": f"Bearer {self.linkedin_token}",
"Content-Type": "application/json",
"X-Restli-Protocol-Version": "2.0.0"
}
# Rate limiting and organic behavior settings
self.min_delay = 30 # Minimum seconds between actions
self.max_delay = 180 # Maximum seconds between actions
self.processed_posts = set() # Track processed posts to avoid duplicates
def get_user_posts(self, user_id: str, limit: int = 10) -> List[LinkedInPost]:
"""
Retrieve recent posts from a specific LinkedIn user
Args:
user_id: LinkedIn user ID or URN
limit: Maximum number of posts to retrieve
Returns:
List of LinkedInPost objects
"""
try:
# Get user's recent posts
url = f"{self.base_url}/shares"
params = {
"q": "owners",
"owners": f"urn:li:person:{user_id}",
"count": limit,
"sortBy": "CREATED"
}
response = requests.get(url, headers=self.headers, params=params)
response.raise_for_status()
posts_data = response.json()
posts = []
for post in posts_data.get("elements", []):
post_obj = LinkedInPost(
post_id=post["id"],
author_id=user_id,
content=self._extract_post_content(post),
created_time=post.get("created", {}).get("time", ""),
post_url=f"https://www.linkedin.com/feed/update/{post['id']}"
)
posts.append(post_obj)
return posts
except requests.exceptions.RequestException as e:
print(f"Error fetching posts for user {user_id}: {e}")
return []
def _extract_post_content(self, post_data: Dict) -> str:
"""Extract text content from LinkedIn post data"""
content = ""
# Try to get text content from different possible locations
if "text" in post_data:
content = post_data["text"].get("text", "")
elif "content" in post_data:
if "contentEntities" in post_data["content"]:
for entity in post_data["content"]["contentEntities"]:
if "description" in entity:
content += entity["description"] + " "
return content.strip()
def generate_comment_with_chatgpt(self, post_content: str, author_context: str = "") -> str:
"""
Generate a thoughtful comment using ChatGPT based on post content
Args:
post_content: The content of the LinkedIn post
author_context: Additional context about the post author
Returns:
Generated comment text
"""
try:
prompt = f"""
Generate a professional, engaging, and authentic comment for this LinkedIn post.
The comment should:
- Be 1-3 sentences long
- Add value to the conversation
- Be supportive and professional
- Avoid generic responses like "Great post!"
- Feel natural and human-like
- Be relevant to the content
Post content: "{post_content}"
Author context: "{author_context}"
Generate only the comment text, no quotes or additional formatting.
"""
response = self.openai_client.chat.completions.create(
model="gpt-3.5-turbo",
messages=[
{"role": "system", "content": "You are a professional colleague providing thoughtful LinkedIn comments."},
{"role": "user", "content": prompt}
],
max_tokens=150,
temperature=0.7
)
return response.choices[0].message.content.strip()
except Exception as e:
print(f"Error generating comment with ChatGPT: {e}")
return "Thanks for sharing this insightful post!"
def like_post(self, post_id: str) -> bool:
"""
Like a LinkedIn post
Args:
post_id: The ID of the post to like
Returns:
True if successful, False otherwise
"""
try:
url = f"{self.base_url}/reactions"
data = {
"actor": "urn:li:person:YOUR_USER_ID", # Replace with your LinkedIn user ID
"object": f"urn:li:share:{post_id}",
"reactionType": "LIKE"
}
response = requests.post(url, headers=self.headers, json=data)
response.raise_for_status()
print(f"Successfully liked post {post_id}")
return True
except requests.exceptions.RequestException as e:
print(f"Error liking post {post_id}: {e}")
return False
def comment_on_post(self, post_id: str, comment_text: str) -> bool:
"""
Comment on a LinkedIn post
Args:
post_id: The ID of the post to comment on
comment_text: The comment text to post
Returns:
True if successful, False otherwise
"""
try:
url = f"{self.base_url}/socialActions/{post_id}/comments"
data = {
"actor": "urn:li:person:YOUR_USER_ID", # Replace with your LinkedIn user ID
"message": {
"text": comment_text
}
}
response = requests.post(url, headers=self.headers, json=data)
response.raise_for_status()
print(f"Successfully commented on post {post_id}: {comment_text[:50]}...")
return True
except requests.exceptions.RequestException as e:
print(f"Error commenting on post {post_id}: {e}")
return False
def engage_with_post(self, post: LinkedInPost, author_context: str = "") -> bool:
"""
Complete engagement workflow: generate comment, like, and comment on post
Args:
post: LinkedInPost object
author_context: Additional context about the post author
Returns:
True if all actions successful, False otherwise
"""
if post.post_id in self.processed_posts:
print(f"Post {post.post_id} already processed, skipping...")
return False
print(f"Processing post {post.post_id} by {post.author_id}")
print(f"Post content: {post.content[:100]}...")
# Generate comment
comment = self.generate_comment_with_chatgpt(post.content, author_context)
print(f"Generated comment: {comment}")
# Add organic delay before actions
delay = random.randint(self.min_delay, self.max_delay)
print(f"Waiting {delay} seconds for organic timing...")
time.sleep(delay)
# Like the post
like_success = self.like_post(post.post_id)
# Wait a bit before commenting
time.sleep(random.randint(5, 15))
# Comment on the post
comment_success = self.comment_on_post(post.post_id, comment)
# Mark as processed
self.processed_posts.add(post.post_id)
return like_success and comment_success
def process_user_list(self, user_configs: List[Dict[str, str]], posts_per_user: int = 3):
"""
Process engagement for multiple users
Args:
user_configs: List of dicts with 'user_id' and optional 'context'
posts_per_user: Number of recent posts to process per user
"""
print(f"Starting engagement process for {len(user_configs)} users...")
for user_config in user_configs:
user_id = user_config.get("user_id")
context = user_config.get("context", "")
print(f"\n--- Processing user: {user_id} ---")
# Get recent posts
posts = self.get_user_posts(user_id, limit=posts_per_user)
if not posts:
print(f"No posts found for user {user_id}")
continue
# Process each post
for post in posts[:posts_per_user]:
try:
success = self.engage_with_post(post, context)
if success:
print(f"Successfully engaged with post {post.post_id}")
else:
print(f"Failed to engage with post {post.post_id}")
# Random delay between posts from same user
time.sleep(random.randint(20, 60))
except Exception as e:
print(f"Error processing post {post.post_id}: {e}")
continue
# Longer delay between different users
print(f"Completed user {user_id}, waiting before next user...")
time.sleep(random.randint(120, 300))
print("\nEngagement process completed!")
# Example usage
def main():
# Initialize the bot
bot = LinkedInEngagementBot(
linkedin_access_token="YOUR_LINKEDIN_ACCESS_TOKEN",
openai_api_key="YOUR_OPENAI_API_KEY"
)
# Define users to engage with
user_configs = [
{
"user_id": "colleague1_user_id",
"context": "Software engineer colleague who posts about tech trends"
},
{
"user_id": "colleague2_user_id",
"context": "Product manager who shares insights about product development"
},
{
"user_id": "colleague3_user_id",
"context": "Data scientist who posts about AI and machine learning"
}
]
# Run the engagement process
bot.process_user_list(user_configs, posts_per_user=2)
if __name__ == "__main__":
main()