Spaces:
Sleeping
Sleeping
File size: 4,720 Bytes
639b959 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 | import os
import json
import requests
import time
from pathlib import Path
from dotenv import load_dotenv
env_path = Path(__file__).resolve().parent.parent / ".env"
load_dotenv(dotenv_path=env_path)
CLIENT_ID = os.getenv("LINKEDIN_CLIENT_ID", "")
CLIENT_SECRET = os.getenv("LINKEDIN_CLIENT_SECRET", "")
REDIRECT_URI = os.getenv("LINKEDIN_REDIRECT_URI", "http://localhost:8000/linkedin/callback/")
SCOPE = "w_member_social profile email openid"
API_BASE = "https://api.linkedin.com"
AUTH_BASE = "https://www.linkedin.com/oauth/v2"
def _log(msg):
print(f"[LinkedInAPI] {msg}")
def get_oauth_url(state=""):
params = {
"response_type": "code",
"client_id": CLIENT_ID,
"redirect_uri": REDIRECT_URI,
"scope": SCOPE,
}
if state:
params["state"] = state
qs = "&".join(f"{k}={requests.utils.quote(str(v))}" for k, v in params.items())
return f"{AUTH_BASE}/authorization?{qs}"
def exchange_code(code):
if not CLIENT_ID or not CLIENT_SECRET:
return None, "LinkedIn app not configured. Set LINKEDIN_CLIENT_ID and LINKEDIN_CLIENT_SECRET."
data = {
"grant_type": "authorization_code",
"code": code,
"client_id": CLIENT_ID,
"client_secret": CLIENT_SECRET,
"redirect_uri": REDIRECT_URI,
}
try:
resp = requests.post(f"{AUTH_BASE}/accessToken", data=data, timeout=30)
if resp.status_code != 200:
_log(f"Token exchange failed ({resp.status_code}): {resp.text[:200]}")
return None, f"Failed to get access token: {resp.text[:100]}"
body = resp.json()
token = body.get("access_token")
expires = body.get("expires_in", 86400)
if not token:
return None, "No access_token in response"
_log("Token exchange successful")
return {"access_token": token, "expires_at": time.time() + expires}, None
except requests.exceptions.RequestException as e:
_log(f"Token exchange error: {e}")
return None, str(e)
def get_user_info(access_token):
headers = {"Authorization": f"Bearer {access_token}"}
try:
resp = requests.get(f"{API_BASE}/v2/userinfo", headers=headers, timeout=15)
if resp.status_code != 200:
_log(f"User info failed ({resp.status_code}): {resp.text[:200]}")
return None, "Failed to get user info"
body = resp.json()
sub = body.get("sub", "")
name = body.get("name", "")
picture = body.get("picture", "")
_log(f"User info: {name} ({sub})")
return {"urn": f"urn:li:person:{sub}", "name": name, "picture": picture, "sub": sub}, None
except requests.exceptions.RequestException as e:
_log(f"User info error: {e}")
return None, str(e)
def create_post(access_token, author_urn, text, hashtags=None, visibility="PUBLIC"):
if not access_token or not author_urn:
return None, "Missing access_token or author_urn"
body = {"author": author_urn, "lifecycleState": "PUBLISHED", "visibility": visibility}
full_text = text
if hashtags:
tag_str = " ".join(f"#{h.lstrip('#')}" for h in hashtags[:5])
full_text = f"{text}\n\n{tag_str}"
body["specificContent"] = {
"com.linkedin.ugc.ShareContent": {
"shareCommentary": {"text": full_text},
"shareMediaCategory": "NONE",
}
}
headers = {
"Authorization": f"Bearer {access_token}",
"Content-Type": "application/json",
"X-Restli-Protocol-Version": "2.0.0",
}
try:
resp = requests.post(f"{API_BASE}/v2/ugcPosts", json=body, headers=headers, timeout=30)
if resp.status_code in (200, 201):
post_id = resp.headers.get("X-RestLi-Id", "")
post_url = f"https://www.linkedin.com/feed/update/{post_id}" if post_id else ""
_log(f"Post created: {post_id}")
return {"post_id": post_id, "post_url": post_url}, None
_log(f"Post failed ({resp.status_code}): {resp.text[:300]}")
if resp.status_code == 401:
return None, "Access token expired. Please reconnect LinkedIn."
if resp.status_code == 403:
return None, "Missing permissions. Re-authenticate with w_member_social scope."
if resp.status_code == 429:
return None, "Rate limited. Try again later."
detail = ""
try:
detail = resp.json().get("message", resp.text[:100])
except Exception:
detail = resp.text[:100]
return None, f"LinkedIn API error: {detail}"
except requests.exceptions.RequestException as e:
_log(f"Post request error: {e}")
return None, str(e)
|