Spaces:
Runtime error
Runtime error
| """LinkedIn member posts (Share on LinkedIn + Sign In with LinkedIn using OpenID Connect products), with an image, | |
| several images (multiImage) or a video uploaded first through the Images / Videos APIs.""" | |
| import re | |
| import time | |
| from urllib.parse import quote, urlencode | |
| import httpx | |
| from sqlalchemy.orm import Session | |
| from app.config import settings | |
| from app.models import SocialAccount | |
| from app.services.crypto import decrypt | |
| from app.services.social import SocialError, expired, save_tokens, scope_set, token | |
| AUTH_URL = "https://www.linkedin.com/oauth/v2/authorization" | |
| TOKEN_URL = "https://www.linkedin.com/oauth/v2/accessToken" | |
| SCOPES = "openid profile w_member_social" | |
| VERSION = "202609" # LinkedIn-Version header, YYYYMM; LinkedIn supports each version for a year | |
| REST = "https://api.linkedin.com/rest" | |
| PROCESSING_WAIT = 600 # seconds LinkedIn may take to process a video before we give up | |
| PART_ATTEMPTS = 4 # a video part upload is tried this often (network errors, 5xx) before the post fails | |
| def redirect_uri() -> str: | |
| return f"{settings.api_url.rstrip('/')}/api/social/linkedin/callback" | |
| def configured() -> bool: | |
| return bool(settings.linkedin_client_id and settings.linkedin_client_secret) | |
| def authorize_url(state: str) -> str: | |
| return AUTH_URL + "?" + urlencode({ | |
| "response_type": "code", "client_id": settings.linkedin_client_id, "redirect_uri": redirect_uri(), | |
| "state": state, "scope": SCOPES, | |
| }) | |
| def exchange_code(code: str) -> dict: | |
| resp = httpx.post(TOKEN_URL, data={ | |
| "grant_type": "authorization_code", "code": code, "redirect_uri": redirect_uri(), | |
| "client_id": settings.linkedin_client_id, "client_secret": settings.linkedin_client_secret, | |
| }, timeout=20) | |
| if resp.status_code >= 400: | |
| raise SocialError(f"LinkedIn token request failed: {resp.text[:300]}", permanent=True) | |
| tokens = resp.json() | |
| me = httpx.get("https://api.linkedin.com/v2/userinfo", | |
| headers={"Authorization": f"Bearer {tokens['access_token']}"}, timeout=20) | |
| me.raise_for_status() | |
| info = me.json() | |
| return {**tokens, "external_id": info["sub"], "handle": info.get("name") or info["sub"], | |
| "avatar_url": info.get("picture")} | |
| _RESERVED = re.compile(r"([\\|{}@\[\]()<>#*_~])") | |
| def little_text(text: str) -> str: | |
| """LinkedIn's commentary field uses 'little text' markup; reserved characters must be escaped.""" | |
| return _RESERVED.sub(r"\\\1", text) | |
| def _call(method: str, url: str, **kw) -> httpx.Response: | |
| """Every request to LinkedIn goes through here (tests replace it).""" | |
| return httpx.request(method, url, timeout=kw.pop("timeout", 60), **kw) | |
| def _headers(account: SocialAccount) -> dict: | |
| return {"Authorization": f"Bearer {token(account)}", "LinkedIn-Version": VERSION, | |
| "X-Restli-Protocol-Version": "2.0.0"} | |
| def can_post_media(account: SocialAccount) -> bool: | |
| return "w_member_social" in scope_set(account.scopes) | |
| def refresh(db: Session, account: SocialAccount) -> None: | |
| """Programmatic refresh, for apps LinkedIn gives refresh tokens to; without one the connection simply expires.""" | |
| if not account.refresh_token: | |
| raise SocialError("The LinkedIn connection expired. Reconnect LinkedIn.", reconnect=True) | |
| resp = _call("POST", TOKEN_URL, data={ | |
| "grant_type": "refresh_token", "refresh_token": decrypt(account.refresh_token), | |
| "client_id": settings.linkedin_client_id, "client_secret": settings.linkedin_client_secret}, timeout=20) | |
| if resp.status_code in (400, 401): | |
| raise SocialError("The LinkedIn connection expired. Reconnect LinkedIn.", reconnect=True) | |
| tokens = _check(resp, "token refresh") | |
| save_tokens(db, account, tokens["access_token"], tokens.get("refresh_token"), tokens.get("expires_in")) | |
| def check_alive(db: Session, account: SocialAccount) -> None: | |
| """One cheap call: raises SocialError(reconnect=True) when LinkedIn no longer accepts this connection.""" | |
| resp = _call("GET", "https://api.linkedin.com/v2/userinfo", headers={"Authorization": f"Bearer {token(account)}"}, | |
| timeout=20) | |
| if resp.status_code == 401: | |
| raise SocialError("LinkedIn no longer accepts this connection. Reconnect LinkedIn.", reconnect=True) | |
| def _check(resp: httpx.Response, what: str) -> dict: | |
| if resp.status_code == 426 or "NONEXISTENT_VERSION" in resp.text: | |
| raise SocialError(f"LinkedIn retired API version {VERSION}; Notestack needs an update to post.", permanent=True) | |
| if resp.status_code == 401: | |
| raise SocialError("LinkedIn rejected the connection. Reconnect LinkedIn.", reconnect=True) | |
| if resp.status_code >= 400: | |
| raise SocialError(f"LinkedIn {what} {resp.status_code}: {resp.text[:300]}", | |
| permanent=400 <= resp.status_code < 500 and resp.status_code != 429) | |
| return resp.json() if resp.content else {} | |
| def _upload_image(account: SocialAccount, m) -> str: | |
| owner = f"urn:li:person:{account.external_id}" | |
| got = _check(_call("POST", f"{REST}/images", params={"action": "initializeUpload"}, headers=_headers(account), | |
| json={"initializeUploadRequest": {"owner": owner}}), "image upload")["value"] | |
| _check(_call("PUT", got["uploadUrl"], content=m.read(), | |
| headers={"Authorization": _headers(account)["Authorization"], "Content-Type": m.content_type}), | |
| "image upload") | |
| return got["image"] | |
| def _upload_video(account: SocialAccount, m) -> str: | |
| owner = f"urn:li:person:{account.external_id}" | |
| got = _check(_call("POST", f"{REST}/videos", params={"action": "initializeUpload"}, headers=_headers(account), | |
| json={"initializeUploadRequest": {"owner": owner, "fileSizeBytes": m.size, | |
| "uploadCaptions": False, "uploadThumbnail": False}}), | |
| "video upload")["value"] | |
| etags = [_put_part(part, m) for part in got["uploadInstructions"]] | |
| _check(_call("POST", f"{REST}/videos", params={"action": "finalizeUpload"}, headers=_headers(account), | |
| json={"finalizeUploadRequest": {"video": got["video"], "uploadToken": got.get("uploadToken", ""), | |
| "uploadedPartIds": etags}}), "video upload") | |
| waited = 0 | |
| while True: # usable in a post only once LinkedIn has processed it | |
| state = _check(_call("GET", f"{REST}/videos/{quote(got['video'])}", headers=_headers(account)), | |
| "video status").get("status") | |
| if state == "AVAILABLE": | |
| return got["video"] | |
| if state == "PROCESSING_FAILED": | |
| raise SocialError("LinkedIn could not process the video.", permanent=True) | |
| if waited >= PROCESSING_WAIT: | |
| raise SocialError("LinkedIn took too long to process the video.") | |
| time.sleep(5) | |
| waited += 5 | |
| def _put_part(part: dict, m) -> str: | |
| """One part of a video, read from the file. Sent to LinkedIn's upload URL with no bearer token; network errors | |
| and 5xx are retried with backoff. Returns the part's ETag.""" | |
| data = m.read(part["firstByte"], part["lastByte"] + 1) | |
| for attempt in range(1, PART_ATTEMPTS + 1): | |
| try: | |
| resp = _call("PUT", part["uploadUrl"], content=data, headers={"Content-Type": "application/octet-stream"}, | |
| timeout=300) | |
| except httpx.HTTPError as exc: | |
| if attempt == PART_ATTEMPTS: | |
| raise SocialError(f"LinkedIn video upload failed: {exc}") from exc | |
| else: | |
| if resp.status_code in (401, 403): | |
| raise SocialError("LinkedIn's upload session expired; trying again later.") | |
| if resp.status_code < 400: | |
| return resp.headers.get("etag", "").strip('"') | |
| if resp.status_code < 500 or attempt == PART_ATTEMPTS: | |
| raise SocialError(f"LinkedIn video upload {resp.status_code}: {resp.text[:200]}", | |
| permanent=resp.status_code < 500) | |
| time.sleep(2 ** attempt) | |
| raise SocialError("LinkedIn video upload failed.") # not reached | |
| def publish(db: Session, account: SocialAccount, posts: list[str], media: list | None = None) -> tuple[str, str]: | |
| if expired(account): | |
| refresh(db, account) # without a refresh token: reconnect | |
| urns = [(_upload_video if m.kind == "video" else _upload_image)(account, m) for m in media or []] | |
| body = { | |
| "author": f"urn:li:person:{account.external_id}", | |
| "commentary": little_text("\n\n".join(posts)), | |
| "visibility": "PUBLIC", | |
| "distribution": {"feedDistribution": "MAIN_FEED", "targetEntities": [], "thirdPartyDistributionChannels": []}, | |
| "lifecycleState": "PUBLISHED", | |
| "isReshareDisabledByAuthor": False, | |
| } | |
| if len(urns) == 1: | |
| body["content"] = {"media": {"id": urns[0]}} | |
| elif urns: | |
| body["content"] = {"multiImage": {"images": [{"id": u} for u in urns]}} | |
| resp = _call("POST", f"{REST}/posts", json=body, timeout=20, headers=_headers(account)) | |
| if resp.status_code in (401, 426) or "NONEXISTENT_VERSION" in resp.text: | |
| _check(resp, "post") # reconnect, or the API version was retired | |
| if resp.status_code >= 400: | |
| raise SocialError(f"LinkedIn {resp.status_code}: {resp.text[:300]}", | |
| permanent=resp.status_code in (400, 403, 422)) | |
| urn = resp.headers.get("x-restli-id") or resp.headers.get("x-linkedin-id") or "" | |
| return urn, f"https://www.linkedin.com/feed/update/{quote(urn)}/" if urn else "" | |
| def metrics(db: Session, account: SocialAccount, external_id: str) -> dict: | |
| return {} # member post analytics need the Community Management API, which personal apps rarely have | |