westfarn
1f7d78de64
Unignore site/ (was blocked by mkdocs /site rule), add compose/Docker/uv tooling, and split deploys so push to main goes to beta while prod stays manual.
49 lines
1.6 KiB
Python
49 lines
1.6 KiB
Python
"""LinkedIn API connector."""
|
|
|
|
import json
|
|
import logging
|
|
|
|
import requests
|
|
|
|
from social.crypto import decrypt_tokens
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class LinkedInConnector:
|
|
platform = "linkedin"
|
|
|
|
def publish(self, post, target) -> str:
|
|
tokens = json.loads(decrypt_tokens(target.account.encrypted_tokens) or "{}")
|
|
access_token = tokens.get("access_token")
|
|
author_urn = target.account.external_id or tokens.get("author_urn")
|
|
if not access_token or not author_urn:
|
|
raise RuntimeError("LinkedIn account missing access_token/author_urn")
|
|
|
|
payload = {
|
|
"author": author_urn,
|
|
"lifecycleState": "PUBLISHED",
|
|
"specificContent": {
|
|
"com.linkedin.ugc.ShareContent": {
|
|
"shareCommentary": {"text": post.body},
|
|
"shareMediaCategory": "NONE",
|
|
}
|
|
},
|
|
"visibility": {"com.linkedin.ugc.MemberNetworkVisibility": "PUBLIC"},
|
|
}
|
|
response = requests.post(
|
|
"https://api.linkedin.com/v2/ugcPosts",
|
|
json=payload,
|
|
headers={
|
|
"Authorization": f"Bearer {access_token}",
|
|
"X-Restli-Protocol-Version": "2.0.0",
|
|
"Content-Type": "application/json",
|
|
},
|
|
timeout=30,
|
|
)
|
|
response.raise_for_status()
|
|
return str(response.headers.get("x-restli-id") or response.json().get("id") or "")
|
|
|
|
def refresh_token(self, account) -> None:
|
|
logger.info("LinkedIn token refresh stub for %s", account.pk)
|