Эффективная автоматизация работы с YouTube API на Python: советы

pip install google-api-python-client
pip install google-auth google-auth-oauthlib google-auth-httplib2

python -m venv youtube_api_env
source youtube_api_env/bin/activate # для Linux/Mac
youtube_api_env\Scripts\activate # для Windows

pip freeze > requirements.txt

from googleapiclient.discovery import build
print("Окружение настроено корректно!")

from googleapiclient.discovery import build

# Инициализация сервиса с API-ключом
API_KEY = 'ваш_API_ключ'
youtube = build('youtube', 'v3', developerKey=API_KEY)

# Теперь можно делать запросы к API
request = youtube.channels().list(
part='snippet,contentDetails,statistics',
forUsername='GoogleDevelopers'
)
response = request.execute()

print('Канал: ', response['items'][0]['snippet']['title'])
print('Подписчиков: ', response['items'][0]['statistics']['subscriberCount'])

import os
import pickle
from google_auth_oauthlib.flow import InstalledAppFlow
from google.auth.transport.requests import Request
from googleapiclient.discovery import build

# Области доступа, которые требуются вашему приложению
SCOPES = ['https://www.googleapis.com/auth/youtube.force-ssl']

def get_authenticated_service():
credentials = None

# Проверяем, есть ли сохраненный токен
if os.path.exists('token.pickle'):
with open('token.pickle', 'rb') as token:
credentials = pickle.load(token)

# Если нет валидных учетных данных, получаем новые
if not credentials or not credentials.valid:
if credentials and credentials.expired and credentials.refresh_token:
credentials.refresh(Request())
else:
flow = InstalledAppFlow.from_client_secrets_file(
'client_secret.json', SCOPES)
credentials = flow.run_local_server(port=8080)

# Сохраняем учетные данные для следующего запуска
with open('token.pickle', 'wb') as token:
pickle.dump(credentials, token)

return build('youtube', 'v3', credentials=credentials)

# Использование функции
youtube = get_authenticated_service()

def search_videos(youtube, query, max_results=10):
"""
Поиск видео по ключевым словам
"""
search_response = youtube.search().list(
q=query,
part='id,snippet',
maxResults=max_results,
type='video' # Только видео, без каналов и плейлистов
).execute()

videos = []
for item in search_response['items']:
video = {
'id': item['id']['videoId'],
'title': item['snippet']['title'],
'channelTitle': item['snippet']['channelTitle'],
'publishedAt': item['snippet']['publishedAt'],
'thumbnail': item['snippet']['thumbnails']['high']['url']
}
videos.append(video)

return videos

# Пример использования
results = search_videos(youtube, 'Python Tutorial', max_results=5)
for video in results:
print(f"Название: {video['title']}")
print(f"Канал: {video['channelTitle']}")
print(f"Дата публикации: {video['publishedAt']}")
print(f"ID видео: {video['id']}")
print("---")

def get_video_details(youtube, video_id):
"""
Получение подробной информации о видео
"""
video_response = youtube.videos().list(
part='snippet,contentDetails,statistics',
id=video_id
).execute()

if not video_response['items']:
return None

video = video_response['items'][0]
details = {
'title': video['snippet']['title'],
'description': video['snippet']['description'],
'publishedAt': video['snippet']['publishedAt'],
'channelTitle': video['snippet']['channelTitle'],
'tags': video['snippet'].get('tags', []),
'viewCount': video['statistics'].get('viewCount', 0),
'likeCount': video['statistics'].get('likeCount', 0),
'commentCount': video['statistics'].get('commentCount', 0),
'duration': video['contentDetails']['duration'] # Формат ISO 8601
}

return details

# Пример использования
video_details = get_video_details(youtube, 'dQw4w9WgXcQ') # Известный ID видео
print(f"Название: {video_details['title']}")
print(f"Просмотры: {video_details['viewCount']}")
print(f"Лайки: {video_details['likeCount']}")

def get_video_comments(youtube, video_id, max_results=100):
"""
Получение комментариев к видео
"""
comments = []
next_page_token = None

while True:
comment_response = youtube.commentThreads().list(
part='snippet',
videoId=video_id,
maxResults=min(100, max_results – len(comments)), # API ограничивает 100 комментариев за запрос
pageToken=next_page_token
).execute()

for item in comment_response['items']:
comment = item['snippet']['topLevelComment']['snippet']
comments.append({
'authorDisplayName': comment['authorDisplayName'],
'text': comment['textDisplay'],
'likeCount': comment['likeCount'],
'publishedAt': comment['publishedAt']
})

next_page_token = comment_response.get('nextPageToken')

if next_page_token is None or len(comments) >= max_results:
break

return comments

# Пример использования
comments = get_video_comments(youtube, 'dQw4w9WgXcQ', max_results=5)
for comment in comments:
print(f"Автор: {comment['authorDisplayName']}")
print(f"Текст: {comment['text'][:100]}...") # Ограничиваем вывод для читаемости
print(f"Лайков: {comment['likeCount']}")
print("---")

def get_channel_info(youtube, channel_id=None, username=None):
"""
Получение информации о YouTube канале по ID или имени пользователя
"""
if channel_id:
request = youtube.channels().list(
part='snippet,contentDetails,statistics',
id=channel_id
)
elif username:
request = youtube.channels().list(
part='snippet,contentDetails,statistics',
forUsername=username
)
else:
return None

response = request.execute()

if not response['items']:
return None

channel = response['items'][0]
info = {
'id': channel['id'],
'title': channel['snippet']['title'],
'description': channel['snippet']['description'],
'subscriberCount': channel['statistics'].get('subscriberCount', 0),
'videoCount': channel['statistics'].get('videoCount', 0),
'viewCount': channel['statistics'].get('viewCount', 0),
'uploadsPlaylist': channel['contentDetails']['relatedPlaylists']['uploads']
}

return info

# Пример использования
channel_info = get_channel_info(youtube, username='GoogleDevelopers')
print(f"Канал: {channel_info['title']}")
print(f"Подписчиков: {channel_info['subscriberCount']}")
print(f"Всего просмотров: {channel_info['viewCount']}")

def get_playlist_videos(youtube, playlist_id, max_results=50):
"""
Получение видео из плейлиста
"""
videos = []
next_page_token = None

while True:
playlist_response = youtube.playlistItems().list(
part='snippet',
playlistId=playlist_id,
maxResults=min(50, max_results – len(videos)), # API ограничивает 50 видео за запрос
pageToken=next_page_token
).execute()

for item in playlist_response['items']:
video = {
'id': item['snippet']['resourceId']['videoId'],
'title': item['snippet']['title'],
'position': item['snippet']['position'],
'publishedAt': item['snippet']['publishedAt'],
'channelTitle': item['snippet']['channelTitle']
}
videos.append(video)

next_page_token = playlist_response.get('nextPageToken')

if next_page_token is None or len(videos) >= max_results:
break

return videos

# Пример использования (получение последних загруженных видео канала)
channel_info = get_channel_info(youtube, username='GoogleDevelopers')
uploads_playlist = channel_info['uploadsPlaylist']
recent_videos = get_playlist_videos(youtube, uploads_playlist, max_results=5)

for video in recent_videos:
print(f"Название: {video['title']}")
print(f"Дата публикации: {video['publishedAt']}")
print(f"Позиция в плейлисте: {video['position']}")
print("---")

from googleapiclient.http import MediaFileUpload

def upload_video(youtube, file_path, title, description, category_id='22',
keywords=None, privacy_status='private'):
"""
Загрузка видео на YouTube

Категории (category_id):
1 – Фильмы и анимация
2 – Автомобили и транспорт
10 – Музыка
15 – Животные
17 – Спорт
22 – Люди и блоги
23 – Комедия
26 – Развлечения
27 – Образование
28 – Наука и техника
"""
body = {
'snippet': {
'title': title,
'description': description,
'categoryId': category_id
},
'status': {
'privacyStatus': privacy_status
}
}

if keywords:
body['snippet']['tags'] = keywords

media = MediaFileUpload(file_path, 
chunksize=1024*1024,
resumable=True)

# Вызов API для загрузки видео
insert_request = youtube.videos().insert(
part=','.join(body.keys()),
body=body,
media_body=media
)

response = None
while response is None:
status, response = insert_request.next_chunk()
if status:
print(f"Загружено {int(status.progress() * 100)}%")

print(f"Видео успешно загружено! ID: {response['id']}")
return response['id']

# Пример использования
video_id = upload_video(
youtube,
'my_video.mp4',
'Мое тестовое видео',
'Это описание моего тестового видео, загруженного через YouTube API',
keywords=['python', 'youtube api', 'tutorial'],
privacy_status='unlisted' # Может быть 'private', 'unlisted' или 'public'
)

import csv
import time

def batch_upload_videos(youtube, csv_file_path):
"""
Массовая загрузка видео из CSV-файла

Формат CSV:
file_path,title,description,category_id,keywords,privacy_status,publish_at
"""
with open(csv_file_path, 'r', encoding='utf-8') as file:
reader = csv.DictReader(file)
for row in reader:
keywords = row.get('keywords', '').split(',') if row.get('keywords') else None

try:
video_id = upload_video(
youtube,
row['file_path'],
row['title'],
row['description'],
category_id=row.get('category_id', '22'),
keywords=keywords,
privacy_status=row.get('privacy_status', 'private')
)

# Если указана дата публикации, устанавливаем ее
if 'publish_at' in row and row['publish_at']:
youtube.videos().update(
part='status',
body={
'id': video_id,
'status': {
'privacyStatus': 'private',
'publishAt': row['publish_at'] # Формат: 2023-12-31T12:00:00Z
}
}
).execute()
print(f"Видео {video_id} будет опубликовано в {row['publish_at']}")

print(f"Успешно обработано видео: {row['title']}")
time.sleep(2) # Пауза между загрузками для избежания превышения квоты
except Exception as e:
print(f"Ошибка при обработке {row['file_path']}: {str(e)}")

return "Массовая загрузка завершена"

def update_video_metadata(youtube, video_id, title=None, description=None, 
tags=None, category_id=None, privacy_status=None):
"""
Обновление метаданных видео
"""
# Получаем текущие данные видео
video_response = youtube.videos().list(
part='snippet,status',
id=video_id
).execute()

if not video_response['items']:
return f"Видео с ID {video_id} не найдено"

video_snippet = video_response['items'][0]['snippet']
video_status = video_response['items'][0]['status']

# Обновляем только те поля, которые переданы в функцию
if title:
video_snippet['title'] = title
if description:
video_snippet['description'] = description
if tags:
video_snippet['tags'] = tags
if category_id:
video_snippet['categoryId'] = category_id

body = {
'id': video_id,
'snippet': video_snippet
}

if privacy_status:
body['status'] = {'privacyStatus': privacy_status}

# Вызов API для обновления видео
update_response = youtube.videos().update(
part='snippet,status' if privacy_status else 'snippet',
body=body
).execute()

return f"Видео {update_response['id']} успешно обновлено"

def create_playlist(youtube, title, description='', privacy_status='private'):
"""
Создание нового плейлиста
"""
playlists_insert_response = youtube.playlists().insert(
part='snippet,status',
body={
'snippet': {
'title': title,
'description': description
},
'status': {
'privacyStatus': privacy_status
}
}
).execute()

return playlists_insert_response['id']

def add_video_to_playlist(youtube, playlist_id, video_id, position=0):
"""
Добавление видео в плейлист
"""
add_video_response = youtube.playlistItems().insert(
part='snippet',
body={
'snippet': {
'playlistId': playlist_id,
'resourceId': {
'kind': 'youtube#video',
'videoId': video_id
},
'position': position
}
}
).execute()

return add_video_response['id']

def add_comment(youtube, video_id, text):
"""
Добавление комментария к видео
"""
comment_insert_response = youtube.commentThreads().insert(
part='snippet',
body={
'snippet': {
'videoId': video_id,
'topLevelComment': {
'snippet': {
'textOriginal': text
}
}
}
}
).execute()

return comment_insert_response['id']

def reply_to_comment(youtube, comment_id, text):
"""
Ответ на существующий комментарий
"""
reply_insert_response = youtube.comments().insert(
part='snippet',
body={
'snippet': {
'parentId': comment_id,
'textOriginal': text
}
}
).execute()

return reply_insert_response['id']

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

def get_channel_statistics(youtube, channel_id):
"""
Получение основных статистических данных канала
"""
request = youtube.channels().list(
part='statistics,snippet',
id=channel_id
)
response = request.execute()

if not response['items']:
return None

stats = response['items'][0]['statistics']
return {
'title': response['items'][0]['snippet']['title'],
'subscriberCount': int(stats.get('subscriberCount', 0)),
'viewCount': int(stats.get('viewCount', 0)),
'videoCount': int(stats.get('videoCount', 0)),
'commentCount': int(stats.get('commentCount', 0))
}

def get_video_performance(youtube, video_ids):
"""
Получение статистики производительности для списка видео
"""
if isinstance(video_ids, str):
video_ids = [video_ids]

chunks = [video_ids[i:i+50] for i in range(0, len(video_ids), 50)] # Разбиваем на части по 50 ID

all_videos_data = []
for chunk in chunks:
request = youtube.videos().list(
part='snippet,contentDetails,statistics',
id=','.join(chunk)
)
response = request.execute()

for video in response['items']:
stats = video['statistics']
snippet = video['snippet']

# Преобразуем ISO 8601 длительность в секунды
duration = video['contentDetails']['duration']
duration_seconds = parse_duration(duration)

# Получаем основные метрики
video_data = {
'id': video['id'],
'title': snippet['title'],
'publishedAt': snippet['publishedAt'],
'viewCount': int(stats.get('viewCount', 0)),
'likeCount': int(stats.get('likeCount', 0)),
'commentCount': int(stats.get('commentCount', 0)),
'duration': duration_seconds,
'durationText': format_duration(duration_seconds)
}

# Рассчитываем производные метрики
if video_data['duration'] > 0:
# Процент просмотра (лайки / просмотры)
video_data['engagementRate'] = video_data['likeCount'] / video_data['viewCount'] if video_data['viewCount'] > 0 else 0

# Время публикации в днях
publish_date = datetime.strptime(snippet['publishedAt'], '%Y-%m-%dT%H:%M:%SZ')
days_since_publish = (datetime.now() – publish_date).days

# Просмотры в день
video_data['viewsPerDay'] = video_data['viewCount'] / max(1, days_since_publish)

# CTR и удержание требуют YouTube Analytics API

all_videos_data.append(video_data)

return pd.DataFrame(all_videos_data)

def parse_duration(duration_str):
"""Преобразует строку ISO 8601 в секунды"""
hours, minutes, seconds = 0, 0, 0

# Находим часы, минуты и секунды
if 'H' in duration_str:
hours = int(duration_str.split('H')[0].split('T')[-1])
if 'M' in duration_str:
if 'H' in duration_str:
minutes = int(duration_str.split('H')[1].split('M')[0])
else:
minutes = int(duration_str.split('T')[1].split('M')[0])
if 'S' in duration_str:
if 'M' in duration_str:
seconds = int(duration_str.split('M')[1].split('S')[0])
elif 'H' in duration_str:
seconds = int(duration_str.split('H')[1].split('S')[0])
else:
seconds = int(duration_str.split('T')[1].split('S')[0])

return hours * 3600 + minutes * 60 + seconds

def format_duration(seconds):
"""Форматирует секунды в читаемый формат времени"""
hours = seconds // 3600
minutes = (seconds % 3600) // 60
seconds = seconds % 60

if hours > 0:
return f"{hours}:{minutes:02d}:{seconds:02d}"
else:
return f"{minutes}:{seconds:02d}"

# Получаем ID канала
channel_info = get_channel_info(youtube, username='GoogleDevelopers')
channel_id = channel_info['id']

# Получаем список видео канала
uploads_playlist = channel_info['uploadsPlaylist']
videos = get_playlist_videos(youtube, uploads_playlist, max_results=50)
video_ids = [video['id'] for video in videos]

# Получаем данные производительности
performance_df = get_video_performance(youtube, video_ids)

# Базовый анализ
print("Самые популярные видео по просмотрам:")
print(performance_df.sort_values(by='viewCount', ascending=False)[['title', 'viewCount', 'likeCount']].head(5))

print("\nСамые вовлекающие видео (по рейтингу лайков):")
print(performance_df.sort_values(by='engagementRate', ascending=False)[['title', 'viewCount', 'likeCount', 'engagementRate']].head(5))

# Визуализация данных
plt.figure(figsize=(12, 6))
plt.scatter(performance_df['duration'], performance_df['viewCount'])
plt.title('Зависимость просмотров от длительности видео')
plt.xlabel('Длительность (сек)')
plt.ylabel('Количество просмотров')
plt.grid(True)
plt.show()

# Анализ тенденций
performance_df['publishDate'] = pd.to_datetime(performance_df['publishedAt'])
performance_df = performance_df.sort_values('publishDate')

plt.figure(figsize=(14, 7))
plt.plot(performance_df['publishDate'], performance_df['viewsPerDay'])
plt.title('Просмотры в день в зависимости от даты публикации')
plt.xlabel('Дата публикации')
plt.ylabel('Просмотры в день')
plt.grid(True)
plt.tight_layout()
plt.show()

def get_channel_analytics(youtube_analytics, channel_id, metrics, dimensions=None, 
start_date=None, end_date=None, filters=None, sort=None, max_results=1000):
"""
Получение аналитических данных канала из YouTube Analytics API

metrics: список метрик, например ['views', 'likes', 'subscribersGained']
dimensions: список измерений, например ['day', 'video']
start_date: начальная дата в формате YYYY-MM-DD
end_date: конечная дата в формате YYYY-MM-DD
filters: строка фильтров, например 'video==VIDEO_ID'
sort: строка сортировки, например '-views'
"""
if not start_date:
start_date = (datetime.now() – timedelta(days=30)).strftime('%Y-%m-%d')
if not end_date:
end_date = datetime.now().strftime('%Y-%m-%d')

request = youtube_analytics.reports().query(
ids=f'channel=={channel_id}',
startDate=start_date,
endDate=end_date,
metrics=','.join(metrics),
dimensions=','.join(dimensions) if dimensions else None,
filters=filters,
sort=sort,
maxResults=max_results
)

response = request.execute()

# Преобразуем результаты в DataFrame
column_headers = response['columnHeaders']
header_names = [header['name'] for header in column_headers]

df = pd.DataFrame(response['rows'], columns=header_names)

# Преобразуем типы данных
for header in column_headers:
if header['dataType'] == 'INTEGER':
df[header['name']] = pd.to_numeric(df[header['name']])
elif header['dataType'] == 'FLOAT':
df[header['name']] = pd.to_numeric(df[header['name']], errors='coerce')
elif header['dataType'] == 'STRING' and header['name'] == 'day':
df[header['name']] = pd.to_datetime(df[header['name']])

return df

# Настройка YouTube Analytics API (требует отдельной аутентификации)
youtube_analytics = build('youtubeAnalytics', 'v2', credentials=credentials)

# Анализ просмотров по дням
views_by_day = get_channel_analytics(
youtube_analytics, 
channel_id, 
metrics=['views', 'estimatedMinutesWatched', 'averageViewDuration'],
dimensions=['day'],
start_date=(datetime.now() – timedelta(days=90)).strftime('%Y-%m-%d')
)

# Анализ по источникам трафика
traffic_sources = get_channel_analytics(
youtube_analytics, 
channel_id, 
metrics=['views', 'estimatedMinutesWatched'],
dimensions=['insightTrafficSourceType'],
sort='-views'
)

# Анализ географии просмотров
geography = get_channel_analytics(
youtube_analytics, 
channel_id, 
metrics=['views', 'estimatedMinutesWatched'],
dimensions=['country'],
sort='-views'
)

# Визуализация результатов
plt.figure(figsize=(15, 7))
plt.plot(views_by_day['day'], views_by_day['views'], 'b-', linewidth=2)
plt.fill_between(views_by_day['day'], views_by_day['views'], alpha=0.3)
plt.title('Просмотры канала за последние 90 дней')
plt.xlabel('Дата')
plt.ylabel('Просмотры')
plt.grid(True)
plt.tight_layout()
plt.show()

# Визуализация источников трафика
plt.figure(figsize=(12, 8))
plt.barh(traffic_sources['insightTrafficSourceType'], traffic_sources['views'])
plt.title('Источники трафика')
plt.xlabel('Просмотры')
plt.tight_layout()
plt.show()

# Визуализация географии просмотров (топ-10 стран)
top_countries = geography.head(10)
plt.figure(figsize=(12, 8))
plt.barh(top_countries['country'], top_countries['views'])
plt.title('Топ-10 стран по просмотрам')
plt.xlabel('Просмотры')
plt.tight_layout()
plt.show()

def generate_weekly_report(youtube, youtube_analytics, channel_id, email=None):
"""
Генерация и отправка еженедельного отчета
"""
from io import BytesIO
import smtplib
from email.mime.multipart import MIMEMultipart
from email.mime.text import MIMEText
from email.mime.application import MIMEApplication

# Получение данных
channel_info = get_channel_info(youtube, channel_id=channel_id)

start_date = (datetime.now() – ti