Blame for backend/app/services/social/linkedin.py

"""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