82 lines
2.5 KiB
Python
82 lines
2.5 KiB
Python
import os
|
|
import requests
|
|
from dotenv import load_dotenv
|
|
|
|
# 1. Load the environment variables
|
|
load_dotenv()
|
|
|
|
ACCESS_TOKEN = os.getenv('LINKEDIN_ACCESS_TOKEN')
|
|
ORGANIZATION_ID = os.getenv('LINKEDIN_ORG_ID')
|
|
|
|
# LinkedIn requires a version header (Format: YYYYMM)
|
|
API_VERSION = '202602'
|
|
|
|
def get_all_linkedin_posts(access_token, org_id):
|
|
"""
|
|
Fetches ALL posts from a specific LinkedIn Organization Page using pagination.
|
|
"""
|
|
if not access_token or not org_id:
|
|
print("🚨 Error: Missing credentials. Please check your .env file.")
|
|
return []
|
|
|
|
url = "https://api.linkedin.com/rest/posts"
|
|
|
|
headers = {
|
|
"Authorization": f"Bearer {access_token}",
|
|
"LinkedIn-Version": API_VERSION,
|
|
"X-Restli-Protocol-Version": "2.0.0"
|
|
}
|
|
|
|
all_posts = []
|
|
start = 0
|
|
count = 100 # Maximum allowed by LinkedIn per request
|
|
|
|
print(f"📥 Starting to fetch posts for Organization ID: {org_id}...")
|
|
|
|
while True:
|
|
params = {
|
|
"q": "author",
|
|
"author": f"urn:li:organization:{org_id}",
|
|
"count": count,
|
|
"start": start
|
|
}
|
|
|
|
try:
|
|
response = requests.get(url, headers=headers, params=params)
|
|
response.raise_for_status()
|
|
|
|
data = response.json()
|
|
elements = data.get('elements', [])
|
|
|
|
# If no more posts are returned, we've reached the end!
|
|
if not elements:
|
|
break
|
|
|
|
all_posts.extend(elements)
|
|
print(f"✅ Fetched posts {start + 1} to {start + len(elements)}...")
|
|
|
|
# Increment the starting point for the next page
|
|
start += count
|
|
|
|
except requests.exceptions.RequestException as e:
|
|
print(f"❌ Error fetching posts at offset {start}: {e}")
|
|
if response.text:
|
|
print(f"LinkedIn API Response: {response.text}")
|
|
break
|
|
|
|
print(f"\n🎉 Finished! Successfully retrieved a total of {len(all_posts)} posts.")
|
|
return all_posts
|
|
|
|
# --- Run the application ---
|
|
if __name__ == "__main__":
|
|
posts = get_all_linkedin_posts(ACCESS_TOKEN, ORGANIZATION_ID)
|
|
|
|
# Print a quick preview of the first 3 posts
|
|
if posts:
|
|
print("\n--- Preview of latest 3 posts ---")
|
|
for post in posts[:3]:
|
|
# Safely extract the text content
|
|
text = post.get('commentary', {}).get('text', 'No text content')
|
|
print(f"ID: {post.get('id')}")
|
|
print(f"Content: {text[:100]}...\n")
|