Интеграция с LinkedIn API через Python: возможности, ограничения и решения

import os
from dotenv import load_dotenv

# Загружаем переменные окружения из файла .env
load_dotenv()

# Получаем ключи из переменных окружения
CLIENT_ID = os.getenv('LINKEDIN_CLIENT_ID')
CLIENT_SECRET = os.getenv('LINKEDIN_CLIENT_SECRET')
REDIRECT_URI = os.getenv('LINKEDIN_REDIRECT_URI')

from requests_oauthlib import OAuth2Session
from requests_oauthlib.compliance_fixes import linkedin_compliance_fix

# Определение параметров OAuth 2.0
client_id = "ваш_client_id"
client_secret = "ваш_client_secret"
redirect_uri = "ваш_redirect_uri"

# Определение областей доступа (scopes)
scope = ["r_liteprofile", "r_emailaddress", "w_member_social"]

# Создание сессии OAuth
oauth = OAuth2Session(client_id, redirect_uri=redirect_uri, scope=scope)
oauth = linkedin_compliance_fix(oauth)

# Генерация URL для авторизации
authorization_url, state = oauth.authorization_url(
"https://www.linkedin.com/oauth/v2/authorization"
)

print(f"Перейдите по ссылке для авторизации: {authorization_url}")

# После перехода по ссылке пользователь получит код авторизации
# Введите полученный код:
authorization_code = input("Введите полученный код: ")

# Обмен кода авторизации на токен доступа
token = oauth.fetch_token(
"https://www.linkedin.com/oauth/v2/accessToken",
authorization_response=f"{redirect_uri}?code={authorization_code}",
client_secret=client_secret
)

# Теперь у нас есть токен доступа для работы с API
access_token = token['access_token']
print(f"Токен доступа получен: {access_token[:10]}...")

import requests

# Формирование заголовков с токеном доступа
headers = {
'Authorization': f'Bearer {access_token}',
'cache-control': 'no-cache',
'X-Restli-Protocol-Version': '2.0.0'
}

# Запрос базовой информации о профиле
response = requests.get(
'https://api.linkedin.com/v2/me',
headers=headers
)

# Проверка и вывод результата
if response.status_code == 200:
profile_data = response.json()
print(f"Профиль получен: {profile_data['localizedFirstName']} {profile_data['localizedLastName']}")
else:
print(f"Ошибка {response.status_code}: {response.text}")

def make_api_request(url, access_token, params=None):
"""Базовая функция для выполнения запросов к LinkedIn API"""
headers = {
'Authorization': f'Bearer {access_token}',
'cache-control': 'no-cache',
'X-Restli-Protocol-Version': '2.0.0'
}

response = requests.get(url, headers=headers, params=params)

if response.status_code == 200:
return response.json()
else:
print(f"Ошибка {response.status_code}: {response.text}")
return None

def get_profile_info(access_token):
"""Получение расширенной информации о профиле"""
# Запрашиваем базовую информацию
base_profile = make_api_request(
'https://api.linkedin.com/v2/me',
access_token
)

if not base_profile:
return None

# Запрашиваем email (требует дополнительных разрешений)
email_info = make_api_request(
'https://api.linkedin.com/v2/emailAddress?q=members&projection=(elements*(handle~))',
access_token
)

# Запрашиваем информацию о текущей должности
positions = make_api_request(
f'https://api.linkedin.com/v2/positions?q=members&vanityName={base_profile["vanityName"]}',
access_token
)

# Собираем все данные в единую структуру
profile_data = {
'id': base_profile.get('id'),
'firstName': base_profile.get('localizedFirstName'),
'lastName': base_profile.get('localizedLastName'),
'profilePicture': base_profile.get('profilePicture', {}).get('displayImage', '')
}

# Добавляем email, если доступен
if email_info and 'elements' in email_info:
profile_data['email'] = email_info['elements'][0]['handle~']['emailAddress']

# Добавляем информацию о текущей должности
if positions and 'elements' in positions:
current_position = positions['elements'][0]
profile_data['position'] = {
'title': current_position.get('title', {}).get('localized', {}).get('en_US', ''),
'company': current_position.get('company', {}).get('name', {}).get('localized', {}).get('en_US', '')
}

return profile_data

def get_company_info(access_token, company_id):
"""Получение информации о компании по ID"""
company_data = make_api_request(
f'https://api.linkedin.com/v2/organizations/{company_id}',
access_token
)

if not company_data:
return None

# Получаем дополнительные данные: последние публикации компании
company_posts = make_api_request(
f'https://api.linkedin.com/v2/ugcPosts?q=authors&authors=List(urn:li:organization:{company_id})',
access_token
)

# Формируем структуру данных компании
result = {
'id': company_data.get('id'),
'name': company_data.get('localizedName'),
'description': company_data.get('localizedDescription'),
'website': company_data.get('localizedWebsite'),
'industry': company_data.get('localizedIndustry'),
'logoUrl': company_data.get('logoV2', {}).get('original', ''),
'followerCount': company_data.get('followerCount', 0)
}

# Добавляем данные о публикациях
if company_posts and 'elements' in company_posts:
result['recentPosts'] = [
{
'id': post.get('id'),
'text': post.get('specificContent', {}).get('com.linkedin.ugc.ShareContent', {}).get('text', ''),
'createdAt': post.get('created', {}).get('time')
}
for post in company_posts['elements'][:5] # Берем только 5 последних постов
]

return result

def get_recent_connections(access_token, count=10):
"""Получение списка последних подключений пользователя"""
connections = make_api_request(
'https://api.linkedin.com/v2/connections?q=viewer&start=0&count=' + str(count),
access_token
)

if not connections or 'elements' not in connections:
return []

result = []
for connection in connections['elements']:
# Получаем данные профиля для каждого подключения
mini_profile = make_api_request(
f"https://api.linkedin.com/v2/people/{connection['miniProfile']['id']}",
access_token
)

if mini_profile:
result.append({
'id': mini_profile.get('id'),
'firstName': mini_profile.get('localizedFirstName'),
'lastName': mini_profile.get('localizedLastName'),
'headline': mini_profile.get('headline', {}).get('localized', {}).get('en_US', ''),
'connectionDate': connection.get('createdAt', 0)
})

return result

def get_all_company_followers(access_token, company_id):
"""Получение всех подписчиков компании с пагинацией"""
all_followers = []
start = 0
count = 50 # Максимальное количество результатов на страницу

while True:
followers_page = make_api_request(
f'https://api.linkedin.com/v2/organizationalEntityFollowers?q=organizationalEntity&organizationalEntity=urn:li:organization:{company_id}&start={start}&count={count}',
access_token
)

if not followers_page or 'elements' not in followers_page:
break

elements = followers_page['elements']
all_followers.extend(elements)

# Проверяем, есть ли еще данные
if len(elements) < count:
break

# Увеличиваем смещение для следующей страницы
start += count

# Небольшая задержка для соблюдения ограничений API
time.sleep(1)

return all_followers

def create_text_post(access_token, text_content, visibility="PUBLIC"):
"""Создание простого текстового поста в LinkedIn"""
# Определяем URL и заголовки запроса
url = "https://api.linkedin.com/v2/ugcPosts"
headers = {
'Authorization': f'Bearer {access_token}',
'Content-Type': 'application/json',
'X-Restli-Protocol-Version': '2.0.0'
}

# Получаем URN пользователя (person URN)
person_data = requests.get(
'https://api.linkedin.com/v2/me',
headers=headers
).json()

person_urn = f"urn:li:person:{person_data['id']}"

# Формируем данные для поста
post_data = {
"author": person_urn,
"lifecycleState": "PUBLISHED",
"specificContent": {
"com.linkedin.ugc.ShareContent": {
"shareCommentary": {
"text": text_content
},
"shareMediaCategory": "NONE"
}
},
"visibility": {
"com.linkedin.ugc.MemberNetworkVisibility": visibility
}
}

# Отправляем запрос
response = requests.post(
url,
headers=headers,
json=post_data
)

if response.status_code == 201:
return response.json()
else:
print(f"Ошибка создания поста: {response.status_code} – {response.text}")
return None

def create_image_post(access_token, text_content, image_path, title="", visibility="PUBLIC"):
"""Создание поста с изображением в LinkedIn"""
# Определяем заголовки запроса
headers = {
'Authorization': f'Bearer {access_token}',
'X-Restli-Protocol-Version': '2.0.0'
}

# Получаем URN пользователя
person_data = requests.get(
'https://api.linkedin.com/v2/me',
headers=headers
).json()

person_urn = f"urn:li:person:{person_data['id']}"

# Шаг 1: Инициализация загрузки изображения
init_upload_url = "https://api.linkedin.com/v2/assets?action=registerUpload"

init_upload_data = {
"registerUploadRequest": {
"recipes": ["urn:li:digitalmediaRecipe:feedshare-image"],
"owner": person_urn,
"serviceRelationships": [
{
"relationshipType": "OWNER",
"identifier": "urn:li:userGeneratedContent"
}
]
}
}

init_upload_headers = headers.copy()
init_upload_headers['Content-Type'] = 'application/json'

init_upload_response = requests.post(
init_upload_url,
json=init_upload_data,
headers=init_upload_headers
)

if init_upload_response.status_code != 200:
print(f"Ошибка инициализации загрузки: {init_upload_response.status_code} – {init_upload_response.text}")
return None

# Извлекаем данные для загрузки
upload_data = init_upload_response.json()
upload_url = upload_data['value']['uploadMechanism']['com.linkedin.digitalmedia.uploading.MediaUploadHttpRequest']['uploadUrl']
asset_id = upload_data['value']['asset']

# Шаг 2: Загрузка изображения
with open(image_path, 'rb') as image_file:
image_data = image_file.read()

upload_response = requests.put(
upload_url,
data=image_data,
headers={'Authorization': f'Bearer {access_token}'}
)

if upload_response.status_code != 201:
print(f"Ошибка загрузки изображения: {upload_response.status_code} – {upload_response.text}")
return None

# Шаг 3: Создание поста с изображением
post_url = "https://api.linkedin.com/v2/ugcPosts"

post_data = {
"author": person_urn,
"lifecycleState": "PUBLISHED",
"specificContent": {
"com.linkedin.ugc.ShareContent": {
"shareCommentary": {
"text": text_content
},
"shareMediaCategory": "IMAGE",
"media": [
{
"status": "READY",
"description": {
"text": title
},
"media": asset_id,
"title": {
"text": title
}
}
]
}
},
"visibility": {
"com.linkedin.ugc.MemberNetworkVisibility": visibility
}
}

post_headers = headers.copy()
post_headers['Content-Type'] = 'application/json'

post_response = requests.post(
post_url,
json=post_data,
headers=post_headers
)

if post_response.status_code == 201:
return post_response.json()
else:
print(f"Ошибка создания поста: {post_response.status_code} – {post_response.text}")
return None

import schedule
import time
import json
from datetime import datetime

def schedule_posts(access_token, content_file):
"""Планировщик постов из JSON файла"""
# Загружаем данные о постах из файла
with open(content_file, 'r', encoding='utf-8') as file:
posts = json.load(file)

# Настраиваем расписание для каждого поста
for post in posts:
post_type = post.get('type', 'text')
post_time = post.get('time') # Формат: "HH:MM"

if post_type == 'text':
# Планируем текстовый пост
schedule.every().day.at(post_time).do(
create_text_post,
access_token=access_token,
text_content=post['content'],
visibility=post.get('visibility', 'PUBLIC')
)
elif post_type == 'image':
# Планируем пост с изображением
schedule.every().day.at(post_time).do(
create_image_post,
access_token=access_token,
text_content=post['content'],
image_path=post['image_path'],
title=post.get('title', ''),
visibility=post.get('visibility', 'PUBLIC')
)

# Запускаем планировщик
print(f"Планировщик запущен в {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
while True:
schedule.run_pending()
time.sleep(60) # Проверка каждую минуту

def get_post_analytics(access_token, post_urn):
"""Получение аналитики по конкретному посту"""
# Кодируем URN для использования в URL
import urllib.parse
encoded_urn = urllib.parse.quote(post_urn)

# Определяем URL и заголовки
url = f"https://api.linkedin.com/v2/socialActions/{encoded_urn}"
headers = {
'Authorization': f'Bearer {access_token}',
'X-Restli-Protocol-Version': '2.0.0'
}

# Отправляем запрос
response = requests.get(url, headers=headers)

if response.status_code != 200:
print(f"Ошибка получения аналитики: {response.status_code} – {response.text}")
return None

# Обрабатываем ответ
analytics_data = response.json()

# Возвращаем структурированные данные
result = {
'likes': analytics_data.get('likesSummary', {}).get('totalLikes', 0),
'comments': analytics_data.get('commentsSummary', {}).get('totalComments', 0),
'shares': analytics_data.get('sharesSummary', {}).get('totalShares', 0)
}

return result

import time
import functools
import random
from requests.exceptions import RequestException, ConnectionError, Timeout

def retry_on_api_error(max_retries=3, base_delay=2):
"""Декоратор для повторных попыток при ошибках API"""
def decorator(func):
@functools.wraps(func)
def wrapper(*args, **kwargs):
retries = 0
while retries <= max_retries:
try:
return func(*args, **kwargs)
except ConnectionError as e:
# Проблемы с сетью
retries += 1
if retries > max_retries:
raise
wait_time = base_delay * (2 ** retries) + random.uniform(0, 1)
print(f"Ошибка соединения. Повторная попытка через {wait_time:.2f} сек...")
time.sleep(wait_time)
except Timeout as e:
# Превышено время ожидания
retries += 1
if retries > max_retries:
raise
wait_time = base_delay * (2 ** retries) + random.uniform(0, 1)
print(f"Таймаут. Повторная попытка через {wait_time:.2f} сек...")
time.sleep(wait_time)
except RequestException as e:
# Обработка кодов ответа HTTP
status_code = getattr(e.response, 'status_code', None)

# 429 – превышение лимита запросов
if status_code == 429:
retries += 1
if retries > max_retries:
raise
# Проверяем заголовок Retry-After
retry_after = int(e.response.headers.get('Retry-After', base_delay * 5))
print(f"Превышен лимит запросов. Ожидание {retry_after} сек...")
time.sleep(retry_after)

# 401, 403 – проблемы с авторизацией
elif status_code in (401, 403):
print("Ошибка авторизации. Проверьте токен доступа.")
raise

# 5xx – проблемы на стороне сервера
elif status_code and 500 <= status_code < 600:
retries += 1
if retries > max_retries:
raise
wait_time = base_delay * (2 ** retries) + random.uniform(0, 1)
print(f"Ошибка сервера ({status_code}). Повторная попытка через {wait_time:.2f} сек...")
time.sleep(wait_time)
else:
# Другие ошибки HTTP
raise
return wrapper
return decorator

@retry_on_api_error(max_retries=3, base_delay=2)
def make_api_request(url, access_token, params=None, method='GET', data=None):
"""Улучшенная функция для выполнения запросов к LinkedIn API"""
headers = {
'Authorization': f'Bearer {access_token}',
'X-Restli-Protocol-Version': '2.0.0',
'Content-Type': 'application/json'
}

try:
if method.upper() == 'GET':
response = requests.get(url, headers=headers, params=params, timeout=10)
elif method.upper() == 'POST':
response = requests.post(url, headers=headers, json=data, timeout=10)
elif method.upper() == 'PUT':
response = requests.put(url, headers=headers, json=data, timeout=10)
elif method.upper() == 'DELETE':
response = requests.delete(url, headers=headers, timeout=10)
else:
raise ValueError(f"Неподдерживаемый метод HTTP: {method}")

# Принудительно вызываем исключение для ошибок HTTP
response.raise_for_status()

# Проверяем, есть ли в ответе JSON
if response.headers.get('Content-Type', '').startswith('application/json'):
return response.json()
else:
return response.text

except ValueError as e:
# Ошибки JSON или параметров
print(f"Ошибка обработки данных: {str(e)}")
return None

class RateLimiter:
"""Класс для контроля частоты запросов к API"""
def __init__(self, calls_limit=100, period=60):
self.calls_limit = calls_limit # Максимальное число запросов
self.period = period # Период в секундах
self.timestamps = [] # История временных меток запросов

def wait_if_needed(self):
"""Ожидание, если достигнут лимит запросов"""
now = time.time()

# Удаляем устаревшие временные метки
self.timestamps = [ts for ts in self.timestamps if now – ts < self.period]

# Если достигнут лимит, ожидаем
if len(self.timestamps) >= self.calls_limit:
oldest = self.timestamps[0]
sleep_time = self.period – (now – oldest)
if sleep_time > 0:
print(f"Достигнут лимит запросов. Ожидание {sleep_time:.2f} секунд...")
time.sleep(sleep_time)
# После ожидания обновляем текущее время
now = time.time()

# Добавляем новую временную метку
self.timestamps.append(now)

# Создаем лимитер на уровне модуля
limiter = RateLimiter(calls_limit=90, period=60) # 90 запросов в минуту для запаса

@retry_on_api_error(max_retries=3, base_delay=2)
def make_api_request_with_rate_limit(url, access_token, params=None, method='GET', data=None):
"""Функция для запросов с контролем частоты"""
# Проверяем и ждем, если нужно
limiter.wait_if_needed()

# Выполняем запрос с помощью базовой функции
return make_api_request(url, access_token, params, method, data)

def get_all_connections_generator(access_token):
"""Генератор для эффективного перебора всех подключений"""
start = 0
count = 50 # Размер страницы

while True:
url = f'https://api.linkedin.com/v2/connections?q=viewer&start={start}&count={count}'
connections_page = make_api_request_with_rate_limit(url, access_token)

if not connections_page or 'elements' not in connections_page:
break

elements = connections_page['elements']
if not elements:
break

# Возвращаем элементы по одному
for element in elements:
yield element

# Если элементов меньше размера страницы, это последняя страница
if len(elements) < count:
break

# Увеличиваем смещение для следующей страницы
start += count

class LinkedInTokenManager:
"""Класс для управления токеном доступа LinkedIn"""
def __init__(self, client_id, client_secret, redirect_uri, refresh_token=None):
self.client_id = client_id
self.client_secret = client_secret
self.redirect_uri = redirect_uri
self.access_token = None
self.refresh_token = refresh_token
self.expiry_time = None

def is_token_valid(self):
"""Проверка действительности токена"""
if not self.access_token or not self.expiry_time:
return False

# Добавляем буфер 5 минут для безопасности
return time.time() < (self.expiry_time – 300)

def refresh_access_token(self):
"""Обновление токена доступа с помощью refresh token"""
if not self.refresh_token:
raise ValueError("Отсутствует refresh token для обновления доступа")

url = "https://www.linkedin.com/oauth/v2/accessToken"
data = {
'grant_type': 'refresh_token',
'refresh_token': self.refresh_token,
'client_id': self.client_id,
'client_secret': self.client_secret
}

response = requests.post(url, data=data)

if response.status_code == 200:
token_data = response.json()
self.access_token = token_data['access_token']

# Обновляем refresh token, если он был предоставлен
if 'refresh_token' in token_data:
self.refresh_token = token_data['refresh_token']

# Устанавливаем время истечения
self.expiry_time = time.time() + token_data['expires_in']
return True
else:
print(f"Ошибка обновления токена: {response.status_code} – {response.text}")
return False

def get_token(self):
"""Получение действительного токена доступа"""
if not self.is_token_valid():
success = self.refresh_access_token()
if not success:
raise Exception("Не удалось получить действительный токен доступа")

return self.access_token

import hashlib
import json
import os
from datetime import datetime, timedelta

class APICache:
"""Простой кэш для результатов API запросов"""
def __init__(self, cache_dir='api_cache', expiry_hours=24):
self.cache_dir = cache_dir
self.expiry_hours = expiry_hours

# Создаем директорию кэша, если она не существует
if not os.path.exists(cache_dir):
os.makedirs(cache_dir)

def _get_cache_key(self, url, params=None):
"""Создание уникального ключа кэша на основе URL и параметров"""
key_parts = [url]
if params:
key_parts.append(json.dumps(params, sort_keys=True))

key_string = '|'.join(key_parts)
return hashlib.md5(key_string.encode()).hexdigest()

def get(self, url, params=None):
"""Получение кэшированного результата"""
cache_key = self._get_cache_key(url, params)
cache_file = os.path.join(self.cache_dir, cache_key)

if os.path.exists(cache_file):
# Проверяем возраст кэша
file_mod_time = datetime.fromtimestamp(os.path.getmtime(cache_file))
if datetime.now() – file_mod_time < timedelta(hours=self.expiry_hours):
try:
with open(cache_file, 'r', encoding='utf-8') as f:
return json.load(f)
except:
# Если файл поврежден, игнорируем кэш
pass

return None

def set(self, url, data, params=None):
"""Сохранение результата в кэш"""
cache_key = self._get_cache_key(url, params)
cache_file = os.path.join(self.cache_dir, cache_key)

try:
with open(cache_file, 'w', encoding='utf-8') as f:
json.dump(data, f)
return True
except:
return False

# Пример использования кэша в запросах
api_cache = APICache(expiry_hours=12)

def cached_api_request(url, access_token, params=None, force_refresh=False):
"""Функция с кэшированием результатов API запросов"""
if not force_refresh:
# Пытаемся получить данные из кэша
cached_result = api_cache.get(url, params)
if cached_result:
return cached_result

# Если данных в кэше нет или требуется обновление
result = make_api_request_with_rate_limit(url, access_token, params)

# Кэшируем результат, если запрос успешен
if result:
api_cache.set(url, result, params)

return result