loading data from api yametrika referer trafic

api_yametrika_referer_trafic.py

# Developed by MikyPo
# More code for DA here: https://dzen.ru/mikypo

# === 1. 📥 Импорт библиотек
import locale
import os
import pandas as pd
import requests
from datetime import datetime, timedelta
from dotenv import load_dotenv
from io import StringIO

# Переключаем локаль на русскую
print('Кодировка:')
locale.setlocale(locale.LC_ALL, 'ru_RU.utf8')

# === 2. 💠 Окружение. Исходные данные
# Переменные окружения
load_dotenv('../.env') # по умолчанию .env в текущей директории
# В файле .env храним секреты в виде: SECRET=Значение на каждой строке новый секрет

# Для получения токена нужно пройти по адресу:
# https://oauth.yandex.ru/authorize?response_type=token&client_id=<Сюда вставить CLIENT_ID>
COUNTER_ID = os.getenv('COUNTER_ID') # ID счётчика Я.Метрики
API_TOKEN = os.getenv('API_TOKEN')

# === 3. ℹ️ Запрос API Я.Метрики
# Параметры
DATE_1 = '2026-01-01' # Дата начала периода
DATE_2 = '2026-01-31' # Дата окончания периода
LIMIT = 100000 # Сколько строк данных выгружать за 1 раз

API_URL = 'https://api-metrika.yandex.net/stat/v1/data.csv' # Эндпоинт для получения данных

# Создаём папку для сохранения
output_dir = 'from_api_load'
os.makedirs(output_dir, exist_ok=True)

# Генерация списка дат
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)]

headers = {'Authorization': f'OAuth {API_TOKEN}'} # Заголовки запроса API

# === Определение целей ===
# Создаём словарь с понятными названиями
GOALS = {
    'visits': 'ym:s:visits',  # Количество визитов (сессий)
    'demo': 'ym:s:goal_1_reaches',      # Цель №1 Демо-доступ
    'price': 'ym:s:goal_2_reaches',     # Цель №2 Заказ прайс-листа
    'buy': 'ym:s:goal1_3_reaches'       # Цель №2 Покупка системы
}

# Список целей для выгрузки
GOALS_LIST = ['demo', 'price', 'buy']

# Формируем строку для параметра metrics
metrics_str = ','.join([GOALS['visits']] + [GOALS[goal] for goal in GOALS_LIST])

# === Список измерений в запросе ===
DIMENSIONS_LIST = [
    'ym:s:deviceCategory',        # Тип устройства
    'ym:s:externalRefererDomain', # Внешняя ссылка
    'ym:s:date',                  # Дата
    'ym:s:referer',               # Полный URL ссылки
    'ym:s:UTMSource',             # UTM Source
    'ym:s:UTMMedium',             # UTM Medium
    'ym:s:UTMCampaign',           # UTM Campaign
    'ym:s:UTMContent',            # UTM Content
    'ym:s:UTMTerm'                # UTM Term
]

# Формируем строку
dimensions_str = ','.join(DIMENSIONS_LIST)

# === 4. ⬇️ Выгрузка данных
print("🚀 Начинаем выгрузку данных...")

# Проходим по каждой дате
for current_date in date_range:
    current_date_str = current_date.strftime('%Y-%m-%d')
    print(f"📥 Загружаем данные за {current_date_str}...")

    params = {
        'ids': COUNTER_ID,
        'date1': current_date_str,
        'date2': current_date_str,
        'metrics': metrics_str,
        'dimensions': dimensions_str,
        'limit': LIMIT,
        'accuracy': 'full', # Точность данных, 'medium'/'low' = быстрее, но менее точно
        'attribution': 'lastsign', # Последний значимый переход
        'include_undefined': True, # True = включать строки с пустыми/неопределёнными значениями
        'filters': "ym:s:lastsignTrafficSource=='referral'" # Источник трафика по методу 'последний значимый', только внешние ссылки
        # Возможные значения:
        # 'organic'    → органический поиск
        # 'internal'   → внутренний трафик
        # 'direct'     → прямые заходы
        # 'ad'         → реклама
        # 'social'     → социальные сети
        # 'referral'   → реферальный трафик (внешние сайты)
        # 'email'      → email-рассылки
        
    }

    all_rows_day = []
    offset = 1

    while True:
        params['offset'] = offset
        response = requests.get(API_URL, params=params, headers=headers)
        print(f"  offset = {offset}. Статус: {response.status_code}")

        if response.status_code != 200:
            print(f"❌ Ошибка при загрузке {current_date_str}, offset={offset}: {response.status_code}")
            print(response.text)
            break

        try:
            df_page = pd.read_csv(StringIO(response.text), sep=',')
        except Exception as e:
            print(f"❌ Ошибка парсинга CSV для {current_date_str}, offset={offset}: {e}")
            break

        if df_page.empty:
            print(f"✅ Данные за {current_date_str} полностью загружены.")
            break

        all_rows_day.append(df_page)
        offset += LIMIT

    # Сохраняем день в файл
    if all_rows_day:
        df_day = pd.concat(all_rows_day, ignore_index=True)
        filename = f"{current_date.strftime('%Y-%m-%d')}_yametrika.csv"
        filepath = os.path.join(output_dir, filename)
        df_day.to_csv(filepath, index=False, encoding='utf-8')
        print(f"💾 Сохранено: {filepath}")
    else:
        print(f"⚠️ Нет данных за {current_date_str}")

print("\n🔄 Объединяем все CSV-файлы в один DataFrame...")

# Собираем все файлы из папки
csv_files = [f for f in os.listdir(output_dir) if f.endswith('_yametrika.csv')]
all_dfs = []

for file in sorted(csv_files):  # сортируем по дате
    filepath = os.path.join(output_dir, file)
    df = pd.read_csv(filepath)
    all_dfs.append(df)

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

# === Постобработка ===
if not df_final.empty:

    # Преобразуем дату
    df_final['Date of visit'] = pd.to_datetime(df_final['Date of visit'], format='%Y-%m-%d')
    print("✅ Даты преобразованы к типу datetime64[ns].")

    # Убираем агрегированные строки
    df_final = df_final[df_final['Device type'] != 'Totals and averages']
    print("✅ Удалены строки 'Totals and averages'.")

else:
    print("⚠️ Итоговый DataFrame пуст.")
    
# Просмотр первых двух строк датасета
display(df_final.head(2))
df_final.info()

# === 5. 💾 Сохранение данных
# Сохраняем в CSV, т.к. большие данные не влезуть XLSX-файл
if not df_final.empty:
    current_date = datetime.now().strftime("%Y-%m-%d_%H-%M")
    start_date = df_final['Date of visit'].min().strftime("%Y-%m-%d")
    end_date = df_final['Date of visit'].max().strftime("%Y-%m-%d")
    
    # Путь сохранение данных
    csv_file = f'from_api_load/ALL_{start_date}_to_{end_date}_saved_{current_date}.csv'
    df_final.to_csv(csv_file, index=False, encoding='utf-8')
    
    print(f"✅ Данные сохранены в файл: {csv_file}")
    print(f"   Количество строк: {len(df_final)}")
    print(f"   Количество колонок: {len(df_final.columns)}")
else:
    print("❌ Данные не сохранены (пустой DataFrame).")