Как выгружать статистику из Яндекс.Метрики по скачиванию конкретного файла

Файлы

Если на сайте есть какой-либо файл, например анкета в формате PDF для заполнения, и требуется узнать как часто скачивают этот файл, то можно воспользоваться отчётом в кабинете Яндекс.Метрики, который называется «Загрузки файлов».

Я.Метрика. Загрузки файлов

Но этот отчёт может не отобразить все данные из-за семплирования, особенно, если таких скачивания мало. Тогда вооружаемся ноутбуком Jupyter, API Яндекса и пишем Python-код.

Исходные данные

• Аккаунт в Яндексе с доступом к Яндекс.Метрике
• Доступ к счётчику Я.Метрики с номером COUNTER_ID
• OAuth-токен для доступа к API (документация Яндекса)
• Файл .env лежит в папки вместе с python-скриптом

Код с комментариями

Импорт библиотек и настройка окружения

import locale # Для корректного отображения русских символов в консоли
import os # Работа с файловой системой
import pandas as pd # Библиотека для работы с таблицами
import plotly.express as px # Для построения графиков
import plotly.graph_objects as go
import requests # Отправка HTTP-запросов
import time # Паузы

# Обработка исключений
from requests.exceptions import RequestException, Timeout, ConnectionError
from io import StringIO # Преобразование текста ответа API в CSV-формат
from datetime import datetime, timedelta # Работа с датами
from dotenv import load_dotenv # Загрузка секретов из .env-файла

# Вкл русской локали для корректного вывода дат и чисел в консоли
print('Кодировка:')
locale.setlocale(locale.LC_ALL, 'ru_RU.utf8')

Загрузка конфиденциальных данных из файла .env

# Файл .env должен содержать:
# COUNTER_ID=123456789
# API_TOKEN=your_oauth_token_here

load_dotenv('.env') # Загружаем переменные из файла .env в окружение
COUNTER_ID = os.getenv('COUNTER_ID') # ID счётчика Яндекс.Метрики
API_TOKEN = os.getenv('API_TOKEN') # OAuth-токен для авторизации в API

Параметры выгрузки данных

DATE_1 = '2026-01-01' # Начальная дата периода выгрузки
DATE_2 = '2026-03-31' # Конечная дата периода
LIMIT = 10000 # Максимальное число строк в одном ответе API
API_URL = 'https://api-metrika.yandex.net/stat/v1/data.csv' # Эндпоинт API
output_dir = 'loads' # Папка для сохранения скачанных CSV-файлов
FILE_URL = 'Полный путь к файлу на сайте с расширением'

# Создаём папку для выгрузки, если она ещё не существует
os.makedirs(output_dir, exist_ok=True)

# Генерируем список всех дат в диапазоне [DATE_1, DATE_2]
start_date = datetime.strptime(DATE_1, '%Y-%m-%d')
end_date = datetime.strptime(DATE_2, '%Y-%m-%d')
date_range = [start_date + timedelta(days=x) for x in range((end_date - start_date).days + 1)]

# Формируем заголовки для авторизации в запросах к API
headers = {'Authorization': f'OAuth {API_TOKEN}'}

Подготовка к выгрузке. Загрузка и сохранение в CSV

print("🚀 Начинаем выгрузку данных...")
# Настройки механизма повторов при ошибках
MAX_RETRIES = 3 # Максимальное число попыток запроса
RETRY_DELAY = 5 # Начальная пауза между попытками (секунды)
BACKOFF_MULTIPLIER = 2 # Множитель паузы: 5с → 10с → 20с

# Проходим по каждой дате из указанного периода
for current_date in date_range:
····current_date_str = current_date.strftime('%Y-%m-%d') # дата запроса
····filename = f"{current_date_str}.csv" # Имя файла для сохранения
····filepath = os.path.join(output_dir, filename) # Полный путь к файлу
····# 🔍 Пропускаем дату, если файл уже существует (идемпотентность)
····if os.path.exists(filepath):
········print(f"⏭️ Файл {filename} уже существует, пропускаем {current_date_str}")
········continue

····print(f"📥 Загружаем данные за {current_date_str}...")
····# Параметры запроса к API Яндекс.Метрики
····params = {
········'ids': COUNTER_ID, # ID счётчика
········'date1': current_date_str, # Дата начала (включительно)
········'date2': current_date_str, # Дата окончания (включительно)
········'metrics': 'ym:dl:pageviews', # Метрика: количество просмотров
········'dimensions': 'ym:dl:date,ym:dl:URL', # Дата и URL страницы
········'limit': LIMIT, # Лимит строк в ответе
········'accuracy': 'full', # Полная точность расчёта метрик
········'include_undefined': True, # Включать строки с неопределёнными значениями
········'filters': f"ym:dl:URL=='{FILE_URL}'"} # Фильтр только нужный файл

····all_rows_day = [] # Список DataFrame'ов для накопления всех страниц (offset)
····offset = 1 # Начальный смещение для пагинации ответа API
····day_failed = False # Флаг: произошла ли критическая ошибка при загрузке дня

····# Цикл запросов с пагинацией и механизмом повтора при ошибках
····while True:
········params['offset'] = offset # Устанавливаем смещение для текущей страницы
········attempt = 0 # Счётчик попыток для текущего offset
········delay = RETRY_DELAY # Сбрасываем задержку для новой серии попыток
········success = False # Флаг успешного получения данных

········# 🔄 Внутренний цикл: повторные попытки при временных ошибках
········while attempt < MAX_RETRIES:
············try:
················print(f" 🔹 Попытка №{attempt + 1}/{MAX_RETRIES} (offset={offset})...")
················# Отправляем GET-запрос к API с таймаутом 30 секунд
················response = requests.get(
····················API_URL, params=params, headers=headers,timeout=30)

················# ✅ Успешный ответ: статус 200
················if response.status_code == 200:
····················success = True
····················print(f" ✅ Успешно получено (offset={offset})")
····················break # Выходим из цикла повторов, переходим к парсингу

················# ⚠️ Временные ошибки сервера или лимиты: повторяем с задержкой
················elif response.status_code in [429, 500, 502, 503, 504]:
····················print(f" ⚠️ Сервер вернул {response.status_code}, жду {delay}с...")
····················time.sleep(delay) # Пауза перед следующей попыткой
····················delay *= BACKOFF_MULTIPLIER # Увеличение задержки
····················attempt += 1 # Увеличиваем счётчик попыток
····················continue

················# ❌ Клиентские ошибки (400, 401, 403): не повторяем, логируем
················else:
····················print(f"❌ Критическая ошибка {response.status_code}: {response.text[:150]}")
····················break # Выходим из цикла повторов
············# 🌐 Обработка сетевых исключений: таймауты, разрывы соединения
············except RequestException as e:
················print(f"⚠️ Сетевая ошибка: {type(e).__name__}, жду {delay}с...")
················time.sleep(delay)
················delay *= BACKOFF_MULTIPLIER
················attempt += 1
················continue

············# ❗ Любые другие неожиданные исключения
············except Exception as e:
················print(f"❌ Неожиданная ошибка: {type(e).__name__}: {e}")
················break # Прерываем попытки

········# ❌ Обработка случая, когда все попытки исчерпаны
········if not success:
············print(f"❌ Не удалось загрузить offset={offset} за {current_date_str} после {MAX_RETRIES} попыток.")
············day_failed = True # Помечаем день как неудачный
············break # Выходим из основного цикла while True, переходим к сохранению

········# Парсинг успешного ответа
········try:
············# Преобразуем текст ответа в CSV и читаем в DataFrame
············df_page = pd.read_csv(StringIO(response.text), sep=',')
········except Exception as e:
············print(f"❌ Ошибка парсинга CSV: {e}")
············day_failed = True
············break # Ошибка парсинга = критическая, прерываем загрузку дня
········# 🎯 Если страница пуста — значит, все данные за день получены
········if df_page.empty:
············print(f"✅ Все данные за {current_date_str} загружены.")
············break

········# Добавляем полученную страницу в список для последующего объединения
········all_rows_day.append(df_page)
········offset += LIMIT # Переходим к следующей странице (пагинация)

····# 📊 Формирование итогового датафрейма
····if all_rows_day and not day_failed:
········# ✅ Данные получены: объединяем все страницы в один DataFrame
········df_day = pd.concat(all_rows_day, ignore_index=True)
····else:
········# ⚠️ Нет данных или ошибка: создаём файл с нулевыми значениями
········print(f"⚠️ Создаю файл с нулями (данные не получены или ошибка)")
········df_day = pd.DataFrame({
············'ym:dl:date': [current_date_str], # Колонка даты в формате API
············'ym:dl:URL': [FILE_URL], # Колонка URL в формате API
············'ym:dl:pageviews': [0]}) # Колонка метрики с нулевым значением

····# Привидение имён колонок к человекочитаемому виду
····df_output = df_day.rename(columns={
········'ym:dl:date': 'Page view date', # ym:dl:date → "Page view date"
········'ym:dl:URL': 'URL', # ym:dl:URL → "URL"
········'ym:dl:pageviews': 'Pageviews'}) # ym:dl:pageviews → "Pageviews"
····# Сохранение результата в CSV-файл (если данных нет, то значения = 0)
····df_output.to_csv(filepath, index=False, encoding='utf-8')
····print(f"💾 Сохранено: {filepath}\n")

Объединение всех скачанных CSV-файлов в единый датафрейм

print("\n🔄 Объединяем все CSV-файлы в один DataFrame...")
# Собираем список всех CSV-файлов в папке выгрузки
csv_files = [f for f in os.listdir(output_dir) if f.endswith('.csv')]
all_dfs = [] # Список для накопления DataFrame'ов из файлов

# Читаем каждый файл в порядке сортировки (по имени = по дате)
for file in sorted(csv_files):
····filepath = os.path.join(output_dir, file)
····df = pd.read_csv(filepath)
····all_dfs.append(df)

# Объединяем все данные или создаём пустой DataFrame, если файлов нет
if all_dfs:
····df_final = pd.concat(all_dfs, ignore_index=True)
····print(f"✅ Всего загружено строк: {len(df_final)}")
else:
····print("❌ Не найдено ни одного CSV-файла.")
····df_final = pd.DataFrame()

# Удаляем служебную строку "Totals and averages", если она есть в выгрузке
df_final = df_final[df_final['Page view date'] != 'Totals and averages']

# Конвертируем колонку дат в тип datetime для корректной сортировки и визуализации
df_final['Page view date'] = pd.to_datetime(df_final['Page view date'], format='%Y-%m-%d')

Построение интерактивного графика

fig = px.line(
····df_final,
····x='Page view date', # Ось X: даты
····y='Pageviews', # Ось Y: количество просмотров
····title='📊 Динамика просмотров файла', # Заголовок графика
····labels={ # Подписи осей
········'Page view date': 'Дата',
········'Pageviews': 'Количество просмотров'},
····markers=True, # 🔵 Показывать точки на линии
····line_shape='linear') # Тип линии: прямая между точками

# 🎨 Дополнительные настройки стиля линии и маркеров
fig.update_traces(
····line=dict(width=3, color='#4fbbff'), # Толщина и цвет линии
····marker=dict(size=4, color='#3ea2e0')) # Размер и цвет точек

# 🧭 Настройка осей, сетки и интерактивности
fig.update_layout(
····hovermode='x unified', # Показывать значения
····xaxis=dict(
········tickformat='%d.%m.%Y', # Формат отображения дат
········tickangle=45, # Поворот подписей для удобства чтения
········showgrid=True), # Показать вертикальную сетку
····yaxis=dict(
········showgrid=True, # Показать горизонтальную сетку
········zeroline=True, # Показать линию нуля
········zerolinewidth=1,
········zerolinecolor='gray'),
····plot_bgcolor='white', # Белый фон области графика
····height=500) # Высота графика в пикселях

# Отобразить график
fig.show()

# 💾 Сохранить график в отдельный HTML-файл
fig.write_html('pageviews.html')

График

Весь код на GitHub.