Интеграция с API фото-сервисов на Python: руководство для разработчиков

pip install requests instagrapi python-dotenv

CLIENT_ID=ваш_client_id
CLIENT_SECRET=ваш_client_secret
REDIRECT_URI=https://localhost:8000/auth

/project_root
.env (файл с конфиденциальными данными)
auth.py (модуль авторизации)
media.py (модуль работы с медиа)
analytics.py (модуль аналитики)
main.py (основной скрипт)

import os
import requests
from dotenv import load_dotenv

load_dotenv()

def get_auth_url():
client_id = os.getenv("CLIENT_ID")
redirect_uri = os.getenv("REDIRECT_URI")
scope = "user_profile,user_media" # запрашиваемые разрешения

auth_url = f"https://api.instagram.com/oauth/authorize?client_id={client_id}&redirect_uri={redirect_uri}&scope={scope}&response_type=code"

return auth_url

# Выводим URL, по которому нужно перейти пользователю
print(get_auth_url())

def get_access_token(auth_code):
client_id = os.getenv("CLIENT_ID")
client_secret = os.getenv("CLIENT_SECRET")
redirect_uri = os.getenv("REDIRECT_URI")

url = "https://api.instagram.com/oauth/access_token"
payload = {
"client_id": client_id,
"client_secret": client_secret,
"grant_type": "authorization_code",
"redirect_uri": redirect_uri,
"code": auth_code
}

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

# Возвращаем access_token и user_id
return data.get("access_token"), data.get("user_id")

def refresh_long_lived_token(access_token):
client_secret = os.getenv("CLIENT_SECRET")

url = "https://graph.instagram.com/access_token"
params = {
"grant_type": "ig_refresh_token",
"access_token": access_token
}

response = requests.get(url, params=params)
data = response.json()

return data.get("access_token"), data.get("expires_in")

import json
import time

class InstagramAuth:
def __init__(self, token_file="tokens.json"):
self.token_file = token_file
self.access_token = None
self.user_id = None
self.expires_at = 0
self.load_tokens()

def load_tokens(self):
try:
with open(self.token_file, "r") as f:
data = json.load(f)
self.access_token = data.get("access_token")
self.user_id = data.get("user_id")
self.expires_at = data.get("expires_at", 0)
except FileNotFoundError:
pass

def save_tokens(self):
with open(self.token_file, "w") as f:
json.dump({
"access_token": self.access_token,
"user_id": self.user_id,
"expires_at": self.expires_at
}, f)

def get_valid_token(self):
if not self.access_token:
return None

# Если токен скоро истечет, обновляем его
current_time = int(time.time())
if current_time > self.expires_at – 86400: # За 1 день до истечения
self.refresh_token()

return self.access_token

def set_token_from_code(self, auth_code):
token, user_id = get_access_token(auth_code)
if token and user_id:
self.access_token = token
self.user_id = user_id
self.expires_at = int(time.time()) + 3600 # По умолчанию 1 час
self.refresh_token() # Сразу получаем long-lived token
self.save_tokens()
return True
return False

def refresh_token(self):
if not self.access_token:
return False

token, expires_in = refresh_long_lived_token(self.access_token)
if token:
self.access_token = token
self.expires_at = int(time.time()) + expires_in
self.save_tokens()
return True
return False

# Создаем экземпляр класса авторизации
auth = InstagramAuth()

# Если нет действующего токена, запрашиваем новый
if not auth.get_valid_token():
auth_url = get_auth_url()
print(f"Перейдите по ссылке для авторизации: {auth_url}")
auth_code = input("Введите полученный код: ")
if auth.set_token_from_code(auth_code):
print("Авторизация успешна!")
else:
print("Ошибка авторизации")
else:
print("Используем существующий токен")

pip install instagrapi

from instagrapi import Client
import os
from PIL import Image
from datetime import datetime

class InstagramMediaManager:
def __init__(self, username=None, password=None):
self.client = Client()
if username and password:
self.login(username, password)

def login(self, username, password):
try:
self.client.login(username, password)
return True
except Exception as e:
print(f"Ошибка входа: {e}")
return False

def get_user_info(self, username):
user_id = self.client.user_id_from_username(username)
return self.client.user_info(user_id).dict()

def get_user_media(self, username, amount=10):
user_id = self.client.user_id_from_username(username)
medias = self.client.user_medias(user_id, amount)
return [media.dict() for media in medias]

def download_media(self, media_id, folder="downloads"):
if not os.path.exists(folder):
os.makedirs(folder)

media_info = self.client.media_info(media_id).dict()
media_type = media_info["media_type"]

if media_type == 1: # Photo
path = self.client.photo_download(media_id, folder)
elif media_type == 2: # Video
path = self.client.video_download(media_id, folder)
elif media_type == 8: # Album
path = self.client.album_download(media_id, folder)
else:
raise ValueError(f"Неизвестный тип медиа: {media_type}")

return path

def upload_photo(self, photo_path, caption=""):
# Проверяем и оптимизируем изображение перед загрузкой
img = Image.open(photo_path)
if img.mode != 'RGB':
img = img.convert('RGB')
optimized_path = f"{os.path.splitext(photo_path)[0]}_optimized.jpg"
img.save(optimized_path, quality=95)
photo_path = optimized_path

# Загружаем фото
media = self.client.photo_upload(photo_path, caption)
return media.dict()

def upload_video(self, video_path, thumbnail_path=None, caption=""):
if not thumbnail_path:
# Если миниатюра не указана, создаем временную
import tempfile
import cv2

video = cv2.VideoCapture(video_path)
success, frame = video.read()
if success:
temp_dir = tempfile.gettempdir()
thumbnail_path = os.path.join(temp_dir, f"thumb_{int(datetime.now().timestamp())}.jpg")
cv2.imwrite(thumbnail_path, frame)

media = self.client.video_upload(video_path, thumbnail_path, caption)
return media.dict()

def comment_media(self, media_id, text):
comment = self.client.media_comment(media_id, text)
return comment.dict()

def like_media(self, media_id):
return self.client.media_like(media_id)

# Создаем менеджер медиа и входим в аккаунт
media_manager = InstagramMediaManager()
username = os.getenv("INSTA_USERNAME")
password = os.getenv("INSTA_PASSWORD")

if media_manager.login(username, password):
# Получаем информацию о пользователе
user_info = media_manager.get_user_info("targetusername")
print(f"Информация о пользователе: {user_info['full_name']}")

# Получаем последние публикации пользователя
user_media = media_manager.get_user_media("targetusername", 5)
print(f"Получено {len(user_media)} публикаций")

# Скачиваем первую публикацию
if user_media:
media_path = media_manager.download_media(user_media[0]["id"])
print(f"Медиа скачано в {media_path}")

# Публикуем новое фото
new_media = media_manager.upload_photo("path/to/photo.jpg", "Тестовая публикация через Python #python #api")
print(f"Опубликовано новое фото: {new_media['code']}")

import os
import time
import schedule
import random
from datetime import datetime
import pandas as pd

def post_scheduler(media_folder, schedule_file):
# Инициализируем менеджер медиа
manager = InstagramMediaManager()
username = os.getenv("INSTA_USERNAME")
password = os.getenv("INSTA_PASSWORD")

if not manager.login(username, password):
print("Ошибка входа в аккаунт")
return

# Загружаем расписание из Excel или CSV
schedule_df = pd.read_excel(schedule_file) if schedule_file.endswith('.xlsx') else pd.read_csv(schedule_file)

# Помечаем, что задача выполняется
print(f"Запускаем планировщик публикаций. Время: {datetime.now()}")

# Функция для публикации контента
def post_content():
# Получаем текущую дату для поиска в расписании
current_date = datetime.now().strftime('%Y-%m-%d')

# Фильтруем записи для текущей даты
today_posts = schedule_df[schedule_df['date'] == current_date]

if today_posts.empty:
print(f"Нет запланированных публикаций на {current_date}")
return

# Для каждой запланированной публикации
for index, row in today_posts.iterrows():
filename = row['filename']
caption = row['caption']
post_time = row['time']

# Проверяем, наступило ли время публикации
current_time = datetime.now().strftime('%H:%M')
if current_time >= post_time:
file_path = os.path.join(media_folder, filename)

# Проверяем существование файла
if not os.path.exists(file_path):
print(f"Файл не найден: {file_path}")
continue

# Определяем тип медиа по расширению
if file_path.lower().endswith(('.jpg', '.jpeg', '.png')):
result = manager.upload_photo(file_path, caption)
print(f"Опубликовано фото: {filename}, URL: instagram.com/p/{result['code']}")
elif file_path.lower().endswith(('.mp4', '.mov')):
result = manager.upload_video(file_path, caption=caption)
print(f"Опубликовано видео: {filename}, URL: instagram.com/p/{result['code']}")
else:
print(f"Неподдерживаемый формат файла: {filename}")
continue

# Отмечаем публикацию как выполненную
schedule_df.loc[index, 'published'] = True

# Сохраняем обновленное расписание
if schedule_file.endswith('.xlsx'):
schedule_df.to_excel(schedule_file, index=False)
else:
schedule_df.to_csv(schedule_file, index=False)

# Пауза между публикациями для снижения риска блокировки
time.sleep(random.randint(60, 180))

# Запускаем проверку каждые 10 минут
schedule.every(10).minutes.do(post_content)

# Бесконечный цикл проверки расписания
while True:
schedule.run_pending()
time.sleep(60)

def auto_engagement(username, keywords_replies, check_interval=3600):
manager = InstagramMediaManager()
if not manager.login(os.getenv("INSTA_USERNAME"), os.getenv("INSTA_PASSWORD")):
print("Ошибка входа в аккаунт")
return

# Функция проверки и ответа на комментарии
def check_and_reply():
print(f"Проверяем комментарии в {datetime.now()}")

# Получаем последние публикации
user_media = manager.get_user_media(username, 5)

for media in user_media:
media_id = media["id"]

# Получаем комментарии
comments = manager.client.media_comments(media_id)

# Обрабатываем каждый комментарий
for comment in comments:
comment_text = comment.text.lower()
comment_user = comment.user.username

# Пропускаем свои комментарии
if comment_user == os.getenv("INSTA_USERNAME"):
continue

# Проверяем наличие ключевых слов и отвечаем
for keyword, replies in keywords_replies.items():
if keyword in comment_text:
# Выбираем случайный ответ из списка возможных
reply = random.choice(replies)

try:
# Отвечаем на комментарий
manager.comment_media(media_id, f"@{comment_user} {reply}")
print(f"Ответили на комментарий от {comment_user} под постом {media['code']}")

# Пауза между ответами
time.sleep(random.randint(30, 90))
except Exception as e:
print(f"Ошибка при ответе на комментарий: {e}")

# Запускаем проверку по расписанию
schedule.every(check_interval).seconds.do(check_and_reply)

while True:
schedule.run_pending()
time.sleep(60)

def track_account_growth(username, output_file, interval_days=1):
manager = InstagramMediaManager()
if not manager.login(os.getenv("INSTA_USERNAME"), os.getenv("INSTA_PASSWORD")):
print("Ошибка входа в аккаунт")
return

# Загружаем существующие данные или создаем новый файл
try:
stats_df = pd.read_csv(output_file)
except FileNotFoundError:
stats_df = pd.DataFrame(columns=[
'date', 'followers_count', 'following_count',
'media_count', 'total_likes', 'total_comments',
'engagement_rate'
])

def collect_stats():
# Получаем информацию о пользователе
user_info = manager.get_user_info(username)

# Получаем последние публикации (до 10)
user_media = manager.get_user_media(username, 10)

# Считаем общее количество лайков и комментариев
total_likes = sum(media['like_count'] for media in user_media)
total_comments = sum(media['comment_count'] for media in user_media)

# Рассчитываем коэффициент вовлеченности
engagement_rate = (total_likes + total_comments) / (user_info['follower_count'] * len(user_media)) * 100 if user_info['follower_count'] > 0 else 0

# Создаем новую запись
new_row = {
'date': datetime.now().strftime('%Y-%m-%d'),
'followers_count': user_info['follower_count'],
'following_count': user_info['following_count'],
'media_count': user_info['media_count'],
'total_likes': total_likes,
'total_comments': total_comments,
'engagement_rate': engagement_rate
}

# Добавляем запись в DataFrame
nonlocal stats_df
stats_df = pd.concat([stats_df, pd.DataFrame([new_row])], ignore_index=True)

# Сохраняем данные
stats_df.to_csv(output_file, index=False)

print(f"Статистика сохранена на {new_row['date']}")
print(f"Подписчики: {new_row['followers_count']}, Вовлеченность: {new_row['engagement_rate']:.2f}%")

# Запускаем сбор статистики по расписанию
schedule.every(interval_days).days.at("00:01").do(collect_stats)

# Собираем статистику при первом запуске
collect_stats()

while True:
schedule.run_pending()
time.sleep(3600) # Проверяем расписание каждый час

import pandas as pd
import numpy as np
import matplotlib.pyplot as plt
import seaborn as sns
from datetime import datetime, timedelta
import nltk
from nltk.corpus import stopwords
from collections import Counter
from wordcloud import WordCloud

# Загружаем необходимые данные для анализа текста
nltk.download('stopwords')
nltk.download('punkt')

class InstagramAnalytics:
def __init__(self, media_manager):
self.media_manager = media_manager

def analyze_profile(self, username):
# Получаем информацию о профиле
user_info = self.media_manager.get_user_info(username)

# Получаем последние публикации (до 50)
media_items = self.media_manager.get_user_media(username, 50)

# Конвертируем в DataFrame для удобства анализа
media_df = pd.DataFrame(media_items)

# Преобразуем timestamp в datetime
media_df['taken_at'] = pd.to_datetime(media_df['taken_at'], unit='s')

# Общие статистики профиля
profile_stats = {
'username': user_info['username'],
'full_name': user_info['full_name'],
'followers_count': user_info['follower_count'],
'following_count': user_info['following_count'],
'media_count': user_info['media_count'],
'average_likes': media_df['like_count'].mean(),
'average_comments': media_df['comment_count'].mean(),
'engagement_rate': (media_df['like_count'].sum() + media_df['comment_count'].sum()) / 
(user_info['follower_count'] * len(media_df)) * 100 if user_info['follower_count'] > 0 else 0,
}

return profile_stats, media_df

def post_frequency_analysis(self, media_df):
# Группируем по дате и считаем количество постов
media_df['date'] = media_df['taken_at'].dt.date
post_frequency = media_df.groupby('date').size().reset_index(name='count')

# Добавляем дни без постов
date_range = pd.date_range(post_frequency['date'].min(), post_frequency['date'].max(), freq='D')
date_range_df = pd.DataFrame({'date': date_range.date})

# Объединяем с существующими данными
post_frequency = pd.merge(date_range_df, post_frequency, on='date', how='left').fillna(0)

# Статистика по дням недели
media_df['day_of_week'] = media_df['taken_at'].dt.day_name()
day_of_week_stats = media_df.groupby('day_of_week').size().reset_index(name='count')

# Переупорядочиваем дни недели
days_order = ['Monday', 'Tuesday', 'Wednesday', 'Thursday', 'Friday', 'Saturday', 'Sunday']
day_of_week_stats['day_of_week'] = pd.Categorical(day_of_week_stats['day_of_week'], categories=days_order, ordered=True)
day_of_week_stats = day_of_week_stats.sort_values('day_of_week')

# Статистика по времени суток
media_df['hour'] = media_df['taken_at'].dt.hour
time_of_day_stats = media_df.groupby('hour').size().reset_index(name='count')

return post_frequency, day_of_week_stats, time_of_day_stats

def content_performance_analysis(self, media_df):
# Копируем DataFrame для анализа
performance_df = media_df.copy()

# Добавляем расчет вовлеченности для каждой публикации
performance_df['engagement'] = performance_df['like_count'] + performance_df['comment_count']

# Сортируем по вовлеченности
best_posts = performance_df.sort_values('engagement', ascending=False).head(10)
worst_posts = performance_df.sort_values('engagement', ascending=True).head(10)

# Определяем наилучшее время для публикации (по среднему engagement)
time_performance = performance_df.groupby('hour')['engagement'].mean().reset_index()
best_time = time_performance.loc[time_performance['engagement'].idxmax()]['hour']

# Определяем наилучший день для публикации
day_performance = performance_df.groupby('day_of_week')['engagement'].mean().reset_index()
day_performance['day_of_week'] = pd.Categorical(day_performance['day_of_week'], categories=days_order, ordered=True)
day_performance = day_performance.sort_values('day_of_week')
best_day = day_performance.loc[day_performance['engagement'].idxmax()]['day_of_week']

# Анализ типов контента
content_type_performance = performance_df.groupby('media_type')['engagement'].agg(['count', 'mean', 'sum']).reset_index()

return best_posts, worst_posts, time_performance, day_performance, content_type_performance, best_time, best_day

def hashtag_analysis(self, media_df):
# Объединяем все описания
all_captions = ' '.join(media_df['caption_text'].dropna())

# Извлекаем хештеги
hashtags = [tag.strip('#') for tag in all_captions.split() if tag.startswith('#')]
hashtag_counts = Counter(hashtags)

# Преобразуем в DataFrame
hashtag_df = pd.DataFrame(hashtag_counts.items(), columns=['hashtag', 'count'])
hashtag_df = hashtag_df.sort_values('count', ascending=False)

# Анализируем производительность хештегов
hashtag_performance = {}

for idx, row in media_df.iterrows():
if pd.isna(row['caption_text']):
continue

caption_hashtags = [tag.strip('#') for tag in row['caption_text'].split() if tag.startswith('#')]

for tag in caption_hashtags:
if tag not in hashtag_performance:
hashtag_performance[tag] = {
'count': 0,
'total_likes': 0,
'total_comments': 0
}

hashtag_performance[tag]['count'] += 1
hashtag_performance[tag]['total_likes'] += row['like_count']
hashtag_performance[tag]['total_comments'] += row['comment_count']

# Преобразуем в DataFrame и рассчитываем средние показатели
hashtag_perf_df = pd.DataFrame.from_dict(hashtag_performance, orient='index').reset_index()
hashtag_perf_df.columns = ['hashtag', 'count', 'total_likes', 'total_comments']

hashtag_perf_df['avg_likes'] = hashtag_perf_df['total_likes'] / hashtag_perf_df['count']
hashtag_perf_df['avg_comments'] = hashtag_perf_df['total_comments'] / hashtag_perf_df['count']
hashtag_perf_df['avg_engagement'] = (hashtag_perf_df['total_likes'] + hashtag_perf_df['total_comments']) / hashtag_perf_df['count']

# Сортируем по среднему engagement
hashtag_perf_df = hashtag_perf_df.sort_values('avg_engagement', ascending=False)

return hashtag_df, hashtag_perf_df

def generate_report(self, username, output_file):
# Получаем все необходимые данные для анализа
profile_stats, media_df = self.analyze_profile(username)
post_frequency, day_stats, time_stats = self.post_frequency_analysis(media_df)
best_posts, worst_posts, time_perf, day_perf, content_perf, best_time, best_day = self.content_performance_analysis(media_df)
hashtag_counts, hashtag_perf = self.hashtag_analysis(media_df)

# Создаем отчет в HTML формате
html_content = f"""
<html>
<head>
<title>Instagram Analytics Report for @{username}</title>
<style>
body {{ font-family: Arial, sans-serif; margin: 20px; }}
h1, h2 {{ color: #405DE6; }}
.stat-container {{ display: flex; flex-wrap: wrap; }}
.stat-box {{ background: #f8f9fa; border-radius: 10px; padding: 20px; margin: 10px; flex: 1; min-width: 200px; }}
table {{ border-collapse: collapse; width: 100%; margin-bottom: 20px; }}
th, td {{ text-align: left; padding: 8px; border-bottom: 1px solid #ddd; }}
th {{ background-color: #f2f2f2; }}
</style>
</head>
<body>
<h1>Instagram Analytics Report for @{username}</h1>
<p>Generated on {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}</p>
<h2>Profile Overview</h2>
<div class="stat-container">
<div class="stat-box">
<h3>Followers</h3>
<p style="font-size: 24px;">{profile_stats['followers_count']:,}</p>
</div>
<div class="stat-box">
<h3>Following</h3>
<p style="font-size: 24px;">{profile_stats['following_count']:,}</p>
</div>
<div class="stat-box">
<h3>Posts</h3>
<p style="font-size: 24px;">{profile_stats['media_count']:,}</p>
</div>
<div class="stat-box">
<h3>Engagement Rate</h3>
<p style="font-size: 24px;">{profile_stats['engagement_rate']:.2f}%</p>
</div>
</div>
<h2>Content Performance</h2>
<p>Average likes per post: {profile_stats['average_likes']:.1f}</p>
<p>Average comments per post: {profile_stats['average_comments']:.1f}</p>
<h3>Best Performing Posts</h3>
<table>
<tr><th>Date</th><th>Likes</th><th>Comments</th><th>Engagement</th><th>Link</th></tr>
"""

for _, row in best_posts.head(5).iterrows():
html_content += f"""
<tr>
<td>{row['taken_at'].strftime('%Y-%m-%d')}</td>
<td>{row['like_count']:,}</td>
<td>{row['comment_count']:,}</td>
<td>{row['engagement']:,}</td>
<td><a href="https://instagram.com/p/{row['code']}" target="_blank">View Post</a></td>
</tr>
"""

html_content += """
</table>
<h3>Best Time to Post</h3>
<p>Based on your historical data, the best time to post is: <strong>"""
html_content += f"{best_time}:00" + "</strong></p>"
html_content += f"<p>The best day to post is: <strong>{best_day}</strong></p>"
html_content += """
<h3>Top Hashtags by Engagement</h3>
<table>
<tr><th>Hashtag</th><th>Occurrences</th><th>Avg. Likes</th><th>Avg. Comments</th><th>Avg. Engagement</th></tr>
"""

for _, row in hashtag_perf.head(10).iterrows():
html_content += f"""
<tr>
<td>#{row['hashtag']}</td>
<td>{row['count']}</td>
<td>{row['avg_likes']:.1f}</td>
<td>{row['avg_comments']:.1f}</td>
<td>{row['avg_engagement']:.1f}</td>
</tr>
"""

html_content += """
</table>
<h2>Posting Frequency</h2>
<p>Your posting pattern by day of week:</p>
<table>
<tr><th>Day</th><th>Number of Posts</th></tr>
"""

# Добавляем вывод статистики по дням недели
for _, row in day_stats.iterrows():
html_content += f"""
<tr>
<td>{row['day_of_week']}</td>
<td>{int(row['count'])}</td>
</tr>
"""

html_content += """
</table>
</body></html>
"""

# Сохраняем отчет в файл
with open(output_file, 'w', encoding='utf-8') as f:
f.write(html_content)

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

def generate_instagram_analytics(username, output_folder="reports"):
# Создаем папку для отчетов, если она не существует
if not os.path.exists(output_folder):
os.makedirs(output_folder)

# Инициализируем менеджер медиа
media_manager = InstagramMediaManager()
if not media_manager.login(os.getenv("INSTA_USERNAME"), os.getenv("INSTA_PASSWORD")):
print("Ошибка входа в аккаунт")
return

# Создаем аналитический модуль
analytics = InstagramAnalytics(media_manager)

# Генерируем HTML-отчет
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
html_report_path = os.path.join(output_folder, f"instagram_report_{username}_{timestamp}.html")
analytics.generate_report(username, html_report_path)

# Получаем данные для визуализаций
profile_stats, media_df = analytics.analyze_profile(username)
post_frequency, day_stats, time_stats = analytics.post_frequency_analysis(media_df)
_, _, time_perf, day_perf, content_perf, _, _ = analytics.content_performance_analysis(media_df)
hashtag_counts, hashtag_perf = analytics.hashtag_analysis(media_df)

# Создаем визуализации
plt.figure(figsize=(15, 10))

# 1. График активности по дням недели
plt.subplot(2, 2, 1)
sns.barplot(x='day_of_week', y='count', data=day_stats)
plt.title('Posts by Day of Week')
plt.xticks(rotation=45)
plt.tight_layout()

# 2. График активности по времени суток
plt.subplot(2, 2, 2)
sn