Как автоматизировать Pinterest: полное руководство по Python API

# Создание виртуального окружения
python -m venv pinterest_env

# Активация окружения
# На Windows:
pinterest_env\Scripts\activate
# На macOS/Linux:
source pinterest_env/bin/activate

pip install requests oauthlib requests-oauthlib python-dotenv

# Содержимое файла .env
PINTEREST_CLIENT_ID=ваш_client_id
PINTEREST_CLIENT_SECRET=ваш_client_secret
PINTEREST_REDIRECT_URI=ваш_redirect_uri

import os
from dotenv import load_dotenv
from requests_oauthlib import OAuth2Session
from oauthlib.oauth2 import BackendApplicationClient

load_dotenv()

# Загрузка конфигурации из .env файла
CLIENT_ID = os.getenv('PINTEREST_CLIENT_ID')
CLIENT_SECRET = os.getenv('PINTEREST_CLIENT_SECRET')
REDIRECT_URI = os.getenv('PINTEREST_REDIRECT_URI')

# Определение разрешений (скоупов)
SCOPE = ['boards:read', 'pins:read', 'pins:write', 'user_accounts:read']

# Конфигурация OAuth сессии
oauth = OAuth2Session(
CLIENT_ID,
redirect_uri=REDIRECT_URI,
scope=SCOPE
)

# URL для авторизации и токена
AUTH_URL = 'https://www.pinterest.com/oauth/'
TOKEN_URL = 'https://api.pinterest.com/v5/oauth/token'

def get_authorization_url():
authorization_url, state = oauth.authorization_url(AUTH_URL)
print(f"Перейдите по следующей ссылке для авторизации: {authorization_url}")
return authorization_url, state

def get_access_token(redirect_response):
# Полный URL после перенаправления
# Например: http://localhost:8000/callback/?code=123456789&state=abcdef
token = oauth.fetch_token(
TOKEN_URL,
authorization_response=redirect_response,
client_secret=CLIENT_SECRET
)
return token

from http.server import HTTPServer, BaseHTTPRequestHandler
import webbrowser
from urllib.parse import parse_qs, urlparse
import json

class CallbackHandler(BaseHTTPRequestHandler):
def do_GET(self):
self.send_response(200)
self.send_header('Content-type', 'text/html')
self.end_headers()

# Получаем callback с кодом
full_callback_url = f"{REDIRECT_URI}{self.path}"
token = get_access_token(full_callback_url)

# Сохраняем токен
with open('token.json', 'w') as f:
json.dump(token, f)

# Отображаем успешное сообщение
self.wfile.write(b"Авторизация успешна! Можете закрыть это окно.")

# Останавливаем сервер
server.shutdown()

def start_auth_flow():
# Получаем URL авторизации
auth_url, _ = get_authorization_url()

# Открываем браузер для авторизации
webbrowser.open(auth_url)

# Запускаем локальный сервер для перехвата callback
global server
server = HTTPServer(('localhost', 8000), CallbackHandler)
print("Ожидаем ответа от Pinterest...")
server.serve_forever()

def create_pinterest_client():
try:
# Пытаемся загрузить существующий токен
with open('token.json', 'r') as f:
token = json.load(f)

client = OAuth2Session(CLIENT_ID, token=token)
return client
except FileNotFoundError:
# Если токена нет, запускаем процесс авторизации
start_auth_flow()
return create_pinterest_client()

def get_user_boards():
client = create_pinterest_client()
response = client.get('https://api.pinterest.com/v5/boards')

if response.status_code == 200:
return response.json()
else:
print(f"Ошибка при получении досок: {response.status_code}")
print(response.text)
return None

def create_board(name, description=''):
client = create_pinterest_client()
data = {
'name': name,
'description': description
}

response = client.post(
'https://api.pinterest.com/v5/boards',
json=data
)

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

def create_pin(board_id, title, description, image_path, link=None):
client = create_pinterest_client()

# Сначала загрузим изображение
with open(image_path, 'rb') as image_file:
files = {'image': image_file}
data = {
'board_id': board_id,
'title': title,
'description': description
}

if link:
data['link'] = link

response = client.post(
'https://api.pinterest.com/v5/pins',
data=data,
files=files
)

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

def get_pin_analytics(pin_id, start_date, end_date, metrics=None):
client = create_pinterest_client()

if metrics is None:
metrics = ['IMPRESSION', 'PIN_CLICK', 'SAVE']

params = {
'pin_id': pin_id,
'start_date': start_date,
'end_date': end_date,
'metrics': ','.join(metrics)
}

response = client.get(
'https://api.pinterest.com/v5/pins/analytics',
params=params
)

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

def update_pin(pin_id, title=None, description=None, board_id=None, link=None):
client = create_pinterest_client()
data = {}

# Добавляем в запрос только те поля, которые нужно обновить
if title is not None:
data['title'] = title
if description is not None:
data['description'] = description
if board_id is not None:
data['board_id'] = board_id
if link is not None:
data['link'] = link

# Если нечего обновлять, выходим
if not data:
return None

response = client.patch(
f'https://api.pinterest.com/v5/pins/{pin_id}',
json=data
)

if response.status_code == 200:
return response.json()
else:
print(f"Ошибка при обновлении пина: {response.status_code}")
print(response.text)
return None

def delete_pin(pin_id):
client = create_pinterest_client()

response = client.delete(
f'https://api.pinterest.com/v5/pins/{pin_id}'
)

if response.status_code == 204:
return True
else:
print(f"Ошибка при удалении пина: {response.status_code}")
print(response.text)
return False

import os
import time
from datetime import datetime

def bulk_upload_pins(board_id, images_dir, title_template, description_template, link_template=None, delay=5):
"""
Загружает все изображения из указанной директории как пины

:param board_id: ID доски для публикации
:param images_dir: Путь к директории с изображениями
:param title_template: Шаблон заголовка (например, "Продукт {index}: {filename}")
:param description_template: Шаблон описания
:param link_template: Шаблон ссылки (если нужен)
:param delay: Задержка между загрузками в секундах
:return: Список созданных пинов
"""
created_pins = []

# Получаем список всех изображений в директории
image_files = [f for f in os.listdir(images_dir) if f.lower().endswith(('.png', '.jpg', '.jpeg', '.gif'))]

for index, image_file in enumerate(image_files, 1):
image_path = os.path.join(images_dir, image_file)

# Формируем заголовок и описание, заменяя переменные в шаблоне
title = title_template.format(
index=index,
filename=os.path.splitext(image_file)[0],
date=datetime.now().strftime('%Y-%m-%d')
)

description = description_template.format(
index=index,
filename=os.path.splitext(image_file)[0],
date=datetime.now().strftime('%Y-%m-%d')
)

# Формируем ссылку, если задан шаблон
link = None
if link_template:
link = link_template.format(
index=index,
filename=os.path.splitext(image_file)[0],
date=datetime.now().strftime('%Y-%m-%d')
)

# Создаем пин
pin = create_pin(board_id, title, description, image_path, link)

if pin:
created_pins.append(pin)
print(f"Создан пин {index}/{len(image_files)}: {title}")

# Делаем паузу между запросами, чтобы не превысить лимиты API
if index < len(image_files):
print(f"Ожидаем {delay} секунд перед следующей загрузкой...")
time.sleep(delay)

print(f"Загрузка завершена. Создано {len(created_pins)} пинов из {len(image_files)} изображений.")
return created_pins

import schedule
import threading
import time
from datetime import datetime, timedelta

def schedule_pin_publishing(board_id, image_path, title, description, link=None, publish_time=None):
"""
Планирует публикацию пина на определенное время

:param board_id: ID доски
:param image_path: Путь к изображению
:param title: Заголовок пина
:param description: Описание пина
:param link: Ссылка (опционально)
:param publish_time: Время публикации в формате "HH:MM" (если None, то через 1 час)
:return: None
"""
def publish_job():
pin = create_pin(board_id, title, description, image_path, link)
if pin:
print(f"Пин успешно опубликован в {datetime.now().strftime('%H:%M:%S')}")
print(f"ID пина: {pin.get('id')}")
else:
print(f"Ошибка при публикации пина в {datetime.now().strftime('%H:%M:%S')}")

if publish_time:
# Запланировать на конкретное время
schedule.every().day.at(publish_time).do(publish_job)
print(f"Публикация запланирована на {publish_time}")
else:
# Запланировать через час
future_time = datetime.now() + timedelta(hours=1)
schedule_time = future_time.strftime("%H:%M")
schedule.every().day.at(schedule_time).do(publish_job)
print(f"Публикация запланирована на {schedule_time}")

# Запустить планировщик в отдельном потоке
def run_scheduler():
while True:
schedule.run_pending()
time.sleep(1)

scheduler_thread = threading.Thread(target=run_scheduler)
scheduler_thread.daemon = True
scheduler_thread.start()

import feedparser
import requests
from io import BytesIO
from PIL import Image
import tempfile
import os

def publish_pins_from_rss(board_id, rss_url, max_pins=5, image_min_width=600):
"""
Создает пины на основе контента из RSS-ленты

:param board_id: ID доски Pinterest
:param rss_url: URL RSS-ленты
:param max_pins: Максимальное количество пинов для публикации
:param image_min_width: Минимальная ширина изображения в пикселях
:return: Список созданных пинов
"""
# Парсим RSS-ленту
feed = feedparser.parse(rss_url)

created_pins = []
count = 0

for entry in feed.entries:
if count >= max_pins:
break

title = entry.title
description = entry.summary if hasattr(entry, 'summary') else ""
link = entry.link

# Ищем изображение в записи
image_url = None

# Проверяем наличие медиа-вложений
if hasattr(entry, 'media_content'):
for media in entry.media_content:
if 'url' in media:
image_url = media['url']
break

# Альтернативный поиск изображения в тексте записи
if not image_url and hasattr(entry, 'content'):
from bs4 import BeautifulSoup
for content in entry.content:
soup = BeautifulSoup(content.value, 'html.parser')
images = soup.find_all('img')
if images:
image_url = images[0].get('src')
break

if not image_url:
print(f"Пропуск записи '{title}': изображение не найдено")
continue

# Скачиваем и проверяем размер изображения
try:
response = requests.get(image_url)
img = Image.open(BytesIO(response.content))

# Проверка размера изображения
if img.width < image_min_width:
print(f"Пропуск записи '{title}': изображение слишком маленькое ({img.width}x{img.height})")
continue

# Сохраняем изображение во временный файл
with tempfile.NamedTemporaryFile(suffix='.jpg', delete=False) as temp_file:
temp_path = temp_file.name
img.save(temp_path)

# Создаем пин
pin = create_pin(board_id, title, description[:500], temp_path, link)

# Удаляем временный файл
os.unlink(temp_path)

if pin:
created_pins.append(pin)
count += 1
print(f"Создан пин {count}: '{title}'")

except Exception as e:
print(f"Ошибка при обработке записи '{title}': {str(e)}")

print(f"Публикация из RSS завершена. Создано {len(created_pins)} пинов.")
return created_pins

from datetime import datetime, timedelta
import pandas as pd
import matplotlib.pyplot as plt

def get_account_analytics(start_date, end_date, metrics=None):
"""
Получает аналитические данные для аккаунта за указанный период

:param start_date: Начальная дата в формате "YYYY-MM-DD"
:param end_date: Конечная дата в формате "YYYY-MM-DD"
:param metrics: Список метрик для запроса (по умолчанию основные метрики)
:return: Словарь с аналитическими данными
"""
client = create_pinterest_client()

if metrics is None:
metrics = ['IMPRESSION', 'PIN_CLICK', 'OUTBOUND_CLICK', 'SAVE', 'ENGAGEMENT', 'ENGAGEMENT_RATE']

params = {
'start_date': start_date,
'end_date': end_date,
'metrics': ','.join(metrics),
'granularity': 'DAY'
}

response = client.get(
'https://api.pinterest.com/v5/user_account/analytics',
params=params
)

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

def get_boards_analytics(board_ids, start_date, end_date, metrics=None):
"""
Получает аналитические данные для нескольких досок

:param board_ids: Список ID досок
:param start_date: Начальная дата в формате "YYYY-MM-DD"
:param end_date: Конечная дата в формате "YYYY-MM-DD"
:param metrics: Список метрик для запроса
:return: Словарь с аналитическими данными по доскам
"""
client = create_pinterest_client()

if metrics is None:
metrics = ['IMPRESSION', 'PIN_CLICK', 'SAVE']

all_data = {}

for board_id in board_ids:
params = {
'board_id': board_id,
'start_date': start_date,
'end_date': end_date,
'metrics': ','.join(metrics)
}

response = client.get(
'https://api.pinterest.com/v5/boards/analytics',
params=params
)

if response.status_code == 200:
all_data[board_id] = response.json()
else:
print(f"Ошибка при получении аналитики для доски {board_id}: {response.status_code}")
print(response.text)

return all_data

def create_account_performance_report(start_date, end_date, output_file='pinterest_report.html'):
"""
Создает HTML-отчет о производительности аккаунта Pinterest

:param start_date: Начальная дата в формате "YYYY-MM-DD"
:param end_date: Конечная дата в формате "YYYY-MM-DD"
:param output_file: Путь к выходному HTML-файлу
:return: None
"""
# Получаем аналитические данные
analytics = get_account_analytics(start_date, end_date)

if not analytics:
print("Не удалось получить аналитические данные")
return

# Преобразуем данные в DataFrame
data_rows = []

for item in analytics.get('items', []):
date = item.get('date')
metrics = item.get('metrics', {})

row = {'Дата': date}
for metric_name, metric_value in metrics.items():
row[metric_name] = metric_value

data_rows.append(row)

df = pd.DataFrame(data_rows)

# Преобразуем дату в datetime
df['Дата'] = pd.to_datetime(df['Дата'])
df = df.sort_values('Дата')

# Создаем графики
plt.figure(figsize=(12, 18))

# График просмотров
plt.subplot(3, 1, 1)
plt.plot(df['Дата'], df['IMPRESSION'], marker='o', linestyle='-')
plt.title('Просмотры по дням')
plt.grid(True)
plt.xticks(rotation=45)

# График кликов
plt.subplot(3, 1, 2)
plt.plot(df['Дата'], df['PIN_CLICK'], marker='s', linestyle='-', color='green')
plt.title('Клики по дням')
plt.grid(True)
plt.xticks(rotation=45)

# График сохранений
plt.subplot(3, 1, 3)
plt.plot(df['Дата'], df['SAVE'], marker='^', linestyle='-', color='red')
plt.title('Сохранения по дням')
plt.grid(True)
plt.xticks(rotation=45)

plt.tight_layout()

# Сохраняем график во временный файл
temp_image_path = 'temp_chart.png'
plt.savefig(temp_image_path)

# Создаем HTML-отчет
html_content = f"""
<!DOCTYPE html>
<html>
<head>
<title>Отчет Pinterest: {start_date} – {end_date}</title>
<style>
body {{ font-family: Arial, sans-serif; margin: 20px; }}
h1 {{ color: #E60023; }}
.summary {{ background-color: #f9f9f9; padding: 15px; border-radius: 5px; margin-bottom: 20px; }}
table {{ border-collapse: collapse; width: 100%; margin-top: 20px; }}
th, td {{ border: 1px solid #ddd; padding: 8px; text-align: left; }}
th {{ background-color: #E60023; color: white; }}
tr:nth-child(even) {{ background-color: #f2f2f2; }}
.chart {{ width: 100%; margin-top: 30px; }}
</style>
</head>
<body>
<h1>Отчет о производительности Pinterest</h1>
<p>Период: {start_date} – {end_date}</p>

<div class="summary">
<h2>Сводка</h2>
<p>Всего просмотров: {df['IMPRESSION'].sum()}</p>
<p>Всего кликов: {df['PIN_CLICK'].sum()}</p>
<p>Всего сохранений: {df['SAVE'].sum()}</p>
<p>Средний коэффициент вовлеченности: {df['ENGAGEMENT_RATE'].mean():.2f}%</p>
</div>

<div class="chart">
<h2>Динамика показателей</h2>
<img src="data:image/png;base64,{encode_image_to_base64(temp_image_path)}" width="100%">
</div>

<h2>Детальные данные</h2>
{df.to_html(index=False)}
</body>
</html>
"""

# Записываем HTML в файл
with open(output_file, 'w', encoding='utf-8') as f:
f.write(html_content)

# Удаляем временный файл с графиком
os.remove(temp_image_path)

print(f"Отчет сохранен в файле: {output_file}")

import base64

def encode_image_to_base64(image_path):
"""
Кодирует изображение в строку base64 для встраивания в HTML

:param image_path: Путь к изображению
:return: Строка base64
"""
with open(image_path, 'rb') as image_file:
encoded_string = base64.b64encode(image_file.read()).decode('utf-8')
return encoded_string

def analyze_top_performing_pins(board_id=None, limit=10, time_range=30):
"""
Анализирует и выявляет наиболее эффективные пины

:param board_id: ID доски (если None, то для всего аккаунта)
:param limit: Количество пинов для анализа
:param time_range: Временной диапазон в днях
:return: DataFrame с лучшими пинами
"""
client = create_pinterest_client()

# Вычисляем даты
end_date = datetime.now().strftime('%Y-%m-%d')
start_date = (datetime.now() – timedelta(days=time_range)).strftime('%Y-%m-%d')

# Получаем список пинов
params = {
'page_size': 100, # Максимальный размер страницы
}

if board_id:
endpoint = f'https://api.pinterest.com/v5/boards/{board_id}/pins'
else:
endpoint = 'https://api.pinterest.com/v5/pins'

response = client.get(endpoint, params=params)

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

pins_data = response.json()
pins = pins_data.get('items', [])

# Получаем аналитику для каждого пина
pins_analytics = []

for pin in pins:
pin_id = pin.get('id')
analytics = get_pin_analytics(pin_id, start_date, end_date)

if analytics and 'items' in analytics:
# Суммируем метрики за весь период
total_impressions = sum(item.get('metrics', {}).get('IMPRESSION', 0) for item in analytics['items'])
total_clicks = sum(item.get('metrics', {}).get('PIN_CLICK', 0) for item in analytics['items'])
total_saves = sum(item.get('metrics', {}).get('SAVE', 0) for item in analytics['items'])

# Вычисляем CTR (Click-Through Rate)
ctr = (total_clicks / total_impressions * 100) if total_impressions > 0 else 0

# Вычисляем коэффициент сохранений
save_rate = (total_saves / total_impressions * 100) if total_impressions > 0 else 0

pins_analytics.append({
'pin_id': pin_id,
'title': pin.get('title', ''),
'link': pin.get('link', ''),
'created_at': pin.get('created_at', ''),
'impressions': total_impressions,
'clicks': total_clicks,
'saves': total_saves,
'ctr': ctr,
'save_rate': save_rate,
'engagement': total_clicks + total_saves,
})

# Создаем DataFrame и сортируем по вовлеченности
df = pd.DataFrame(pins_analytics)
df = df.sort_values('engagement', ascending=False).head(limit)

return df