"""LinkedIn job-board provider (JD-006). Posts the job as a LinkedIn share (UGC post) on the configured author (a company page or a member profile) via the LinkedIn REST API. Configuration (backend/.env): LINKEDIN_ACCESS_TOKEN OAuth 2.0 bearer token - member post -> scope w_member_social - company page -> scope w_organization_social LINKEDIN_AUTHOR_URN urn:li:person:XXXX or urn:li:organization:1234567 If LINKEDIN_ACCESS_TOKEN / LINKEDIN_AUTHOR_URN are not set, the transport falls back to a simulated post so the rest of the flow keeps working in dev. """ import json import logging import os import re import urllib.request import urllib.error import uuid from django.conf import settings from apps.jobs.tracking_tokens import build_application_url from .base import JobBoardIntegration, PostResult, IntegrationError logger = logging.getLogger(__name__) _UGC_URL = "https://api.linkedin.com/v2/ugcPosts" def _cfg(name, default=""): return getattr(settings, name, "") or os.getenv(name, default) class LinkedInIntegration(JobBoardIntegration): channel = "LINKEDIN" REQUIRED_FIELDS = ("title", "location", "work_details") def _share_text(self, job) -> str: parts = [f"We are hiring: {job.title}"] if job.location: parts.append(f"Location: {job.location}") if job.experience_band: parts.append(f"Experience: {job.experience_band}") if job.must_have_skills: parts.append(f"Key skills: {job.must_have_skills}") desc = re.sub(r"<[^>]+>", " ", job.work_details or "") desc = re.sub(r"\s+", " ", desc).strip() if desc: parts.append(desc[:600]) parts.append(f"Apply / details: {build_application_url(job, self.channel)}") # Uniqueness marker so LinkedIn does not reject a re-post as a duplicate. from django.utils import timezone parts.append(f"Ref #{job.id} | Posted {timezone.now().strftime('%d-%b-%Y %H:%M')}") return "\n\n".join(parts) def _creds(self, job=None): """Credential resolution order: 1. The JD's company (client) LinkedIn config, if set & active. 2. The global OAuth-flow token in the DB (LinkedInAuth). 3. .env values. """ # 1. Per-company config try: client_id = getattr(job, "client_id", None) if client_id: from apps.clients.models import CompanyLinkedInConfig c = CompanyLinkedInConfig.objects.filter(client_id=client_id, is_active=True).first() if c and c.access_token: return c.get_access_token(), (c.author_urn or _cfg("LINKEDIN_AUTHOR_URN")) except Exception: pass # 2. Global DB token try: from apps.jobs.models import LinkedInAuth a = LinkedInAuth.current() if a and a.access_token: return a.access_token, (a.author_urn or _cfg("LINKEDIN_AUTHOR_URN")) except Exception: pass # 3. .env return _cfg("LINKEDIN_ACCESS_TOKEN"), _cfg("LINKEDIN_AUTHOR_URN") def _post_ugc(self, job) -> dict: token, author = self._creds(job) if not token or not author: # Not connected โ€” fail clearly instead of returning a misleading fake URL. raise IntegrationError( "LinkedIn is not connected. Add a valid access token (Token Generator) " "and organization/person URN before posting." ) # Share as an ARTICLE so LinkedIn unfurls the job URL into a rich card # with the gradient banner (the careers page's Open Graph image). job_url = build_application_url(job, self.channel) card_title = f"{job.title} โ€” We're hiring"[:200] card_desc_parts = [p for p in [job.location, job.experience_band] if p] if job.must_have_skills: card_desc_parts.append(f"Key skills: {job.must_have_skills}") card_desc = " ยท ".join(card_desc_parts)[:256] or "View this opening and apply." share_content = { "shareCommentary": {"text": self._share_text(job)}, "shareMediaCategory": "ARTICLE", "media": [ { "status": "READY", "originalUrl": job_url, "title": {"text": card_title}, "description": {"text": card_desc}, } ], } payload = { "author": author, "lifecycleState": "PUBLISHED", "specificContent": {"com.linkedin.ugc.ShareContent": share_content}, "visibility": {"com.linkedin.ugc.MemberNetworkVisibility": "PUBLIC"}, } req = urllib.request.Request( _UGC_URL, data=json.dumps(payload).encode(), headers={ "Authorization": f"Bearer {token}", "Content-Type": "application/json", "X-Restli-Protocol-Version": "2.0.0", }, method="POST", ) try: with urllib.request.urlopen(req, timeout=20) as resp: post_id = resp.headers.get("x-restli-id") or resp.headers.get("X-RestLi-Id") or "" body = resp.read().decode() or "{}" data = json.loads(body) if body.strip().startswith("{") else {} ext = post_id or data.get("id") or f"li-{uuid.uuid4().hex[:10]}" share_id = ext.split(":")[-1] if ":" in ext else ext return {"id": ext, "url": f"https://www.linkedin.com/feed/update/urn:li:share:{share_id}"} except urllib.error.HTTPError as e: detail = e.read().decode(errors="ignore") if e.code == 422 and "duplicate" in detail.lower(): raise IntegrationError("LinkedIn rejected this as a duplicate โ€” the same job was already posted recently. Edit the JD or wait ~24h before re-posting.") raise IntegrationError(f"LinkedIn API error {e.code}: {detail[:300]}") except Exception as e: raise IntegrationError(f"LinkedIn request failed: {e}") def post_job(self, job) -> PostResult: self.validate(job, self.REQUIRED_FIELDS) logger.info("LinkedIn: posting JD %s (%s)", job.id, job.title) data = self._post_ugc(job) return PostResult(external_post_id=data["id"], external_url=data["url"]) def update_job(self, job) -> PostResult: # LinkedIn UGC posts are immutable; publish a fresh share. return self.post_job(job) def delete_job(self, job) -> None: logger.info("LinkedIn: delete not supported for UGC share (JD %s)", job.id)