调用LinkedIn API发布内容遇422错误,求助URN格式规范
LinkedIn API 发布内容时422错误排查与解决方案
问题描述
调用LinkedIn API的ugcPosts接口发布内容时返回422错误,错误提示author字段URN格式不符合要求,要求匹配格式urn:li:company:\d+|urn:li:member:\d+,但当前使用的是urn:li:person:aba2390d-7792-4cb2-ad6b-e116d9783606格式。
请求详情
- 请求类型:POST
- API地址:
https://api.linkedin.com/v2/ugcPosts
环境信息
- 开发语言:Python
- HTTP请求库:requests
代码片段
def linkedin_login(request): auth_url = f"https://www.linkedin.com/oauth/v2/authorization?response_type=code&client_id={settings.LINKEDIN_CLIENT_ID}&redirect_uri={settings.LINKEDIN_REDIRECT_URI}&scope=w_member_social" return redirect(auth_url) def linkedin_callback(request): if 'error' in request.GET: error = request.GET.get('error') error_description = request.GET.get('error_description') return render(request, 'error.html', {'error': error, 'description': error_description}) code = request.GET.get('code') if not code: return render(request, 'error.html', {'error': 'No code provided'}) # Exchange the code for a token token_url = 'https://www.linkedin.com/oauth/v2/accessToken' data = { 'grant_type': 'authorization_code', 'code': code, 'redirect_uri': settings.LINKEDIN_REDIRECT_URI, 'client_id': settings.LINKEDIN_CLIENT_ID, 'client_secret': settings.LINKEDIN_CLIENT_SECRET } response = requests.post(token_url, data=data) if response.status_code != 200: return render(request, 'error.html', {'error': 'Failed to obtain access token', 'description': response.text}) token_data = response.json() access_token = token_data.get('access_token') expires_in = token_data.get('expires_in', 3600) # default to 1 hour if not provided # Generate a unique identifier for this token linkedin_id = str(uuid.uuid4()) # If the user is not authenticated, create a new user or log them in if request.user.is_anonymous: # Create a new user with a unique username username = f"linkedin_user_{linkedin_id[:8]}" user = User.objects.create_user(username=username, password=None) login(request, user) else: user = request.user # Create or update the LinkedIn account LinkedInProfile.objects.update_or_create( user=user, defaults={ 'linkedin_id': linkedin_id, 'access_token': access_token, 'expires_at': timezone.now() + timezone.timedelta(seconds=expires_in) } ) return redirect('dashboard') def dashboard(request): if not request.user.is_authenticated: return redirect('linkedin_login_page') scheduled_posts = LinkedInPost.objects.filter(user=request.user).order_by('scheduled_time') return render(request, 'dashboard.html', {'scheduled_posts': scheduled_posts}) def linkedin_login_page(request): return render(request, 'linkedin_login.html') def error(requests): return render(requests, "error.html") def create_linkedin_post(request): if request.method == 'POST': content = request.POST.get('content') scheduled_time_str = request.POST.get('scheduled_time') # Convert the date/time string to a datetime object if scheduled_time_str: scheduled_time = parse_datetime(scheduled_time_str) if scheduled_time: # Check if the date is naive or already timezone aware if timezone.is_naive(scheduled_time): scheduled_time = timezone.make_aware(scheduled_time, timezone.get_current_timezone()) # If it's already timezone aware, we don't need to modify it else: # Handle the case where parse_datetime returns None (invalid format) return redirect('error_page') # Or another appropriate page else: scheduled_time = None try: linkedin_profile = LinkedInProfile.objects.get(user=request.user) except LinkedInProfile.DoesNotExist: # Handle the case where the user doesn't have a LinkedIn profile return redirect('error_page') # Or another appropriate page post = LinkedInPost.objects.create( user=request.user, content=content, scheduled_time=scheduled_time, linkedin_profile=linkedin_profile, is_published=True # or False, depending on your business logic ) # Publish immediately if no future date is specified now = timezone.now() if not scheduled_time or scheduled_time <= now: post.publish_to_linkedin() return redirect('dashboard') # My models class LinkedInProfile(models.Model): user = models.OneToOneField(User, on_delete=models.CASCADE) access_token = models.CharField(max_length=255) expires_at = models.DateTimeField() linkedin_id = models.CharField(max_length=20, unique=True) # Changed to 20 to accommodate numeric IDs def __str__(self): return f"LinkedIn Profile of {self.user.username}" class LinkedInPost(models.Model): user = models.ForeignKey(User, on_delete=models.CASCADE) content = models.TextField() scheduled_time = models.DateTimeField(auto_now=True) posted = models.BooleanField(default=False) aware_datetime = models.DateTimeField(null=True, blank=True) is_published = models.BooleanField(default=False) linkedin_profile = models.ForeignKey(LinkedInProfile, on_delete=models.CASCADE) def save(self, *args, **kwargs): if self.scheduled_time: if timezone.is_naive(self.scheduled_time): self.aware_datetime = timezone.make_aware(self.scheduled_time, timezone.get_current_timezone()) else: self.aware_datetime = self.scheduled_time super(LinkedInPost, self).save(*args, **kwargs) def publish_to_linkedin(self): if self.is_published and not self.posted: try: url = "https://api.linkedin.com/v2/ugcPosts" headers = { "X-Restli-Protocol-Version": "2.0.0", "Content-Type": "application/json", "Authorization": f"Bearer {self.linkedin_profile.access_token}" } # Use the correct format for the LinkedIn identifier author = f"urn:li:person:{self.linkedin_profile.linkedin_id}" payload = { "author": author, "lifecycleState": "PUBLISHED", "specificContent": { "com.linkedin.ugc.ShareContent": { "shareCommentary": { "text": self.content }, "shareMediaCategory": "NONE" } }, "visibility": { "com.linkedin.ugc.MemberNetworkVisibility": "PUBLIC" } } response = requests.post(url, headers=headers, json=payload) if response.status_code == 201: self.posted = True self.aware_datetime = timezone.now() self.save() return True else: print(f"Error when publishing on LinkedIn. Code: {response.status_code}, Response: {response.text}") return False except Exception as e: print(f"Error when publishing on LinkedIn: {str(e)}") return False return False def __str__(self): return f"LinkedIn Post by {self.user.username} - {self.scheduled_time}"
咨询问题
- 发布内容时author字段的正确URN格式是什么?
- 使用LinkedIn API发布内容时,URN还有哪些特殊注意事项?
解决方案与说明
1. 正确的author URN格式
根据LinkedIn v2 API规范,发布用户内容时,author字段必须使用成员URN格式:urn:li:member:{numeric_member_id}
而非当前使用的urn:li:person:{uuid}格式。
你的代码核心问题在于:linkedin_callback函数中生成的随机UUID并非LinkedIn官方返回的用户真实ID,必须通过me接口获取用户的数字ID。
2. URN的特殊注意事项
- 资源类型匹配:不同资源对应不同URN前缀,用户用
urn:li:member、公司用urn:li:company、内容用urn:li:ugcPost,不能混用。 - 真实ID要求:URN中的ID必须是LinkedIn返回的官方唯一标识,禁止自定义或使用随机生成的UUID。
- 大小写严格:URN前缀(如
urn:li:member)必须严格遵循规范大小写,不能随意修改。 - 版本兼容:旧版API可能使用
urn:li:person格式,但v2的ugcPosts接口已废弃该格式,统一使用urn:li:member。
代码修正要点
在linkedin_callback函数中,获取access_token后新增获取用户真实ID的逻辑:
# 新增:调用LinkedIn me接口获取用户真实数字ID user_info_url = "https://api.linkedin.com/v2/me" headers = {"Authorization": f"Bearer {access_token}"} user_response = requests.get(user_info_url, headers=headers) if user_response.status_code != 200: return render(request, 'error.html', {'error': 'Failed to get user info', 'description': user_response.text}) user_data = user_response.json() real_linkedin_id = user_data.get('id') # 这是LinkedIn返回的数字ID
将real_linkedin_id存入LinkedInProfile的linkedin_id字段,替代原随机UUID。
同时修改publish_to_linkedin函数中的author格式:
author = f"urn:li:member:{self.linkedin_profile.linkedin_id}"
内容的提问来源于stack exchange,提问作者usrmetamask
相关产品推荐
相关产品推荐

