Парсинг 300к+ видое на Ютуб без пап мам и бабушек

Передо мною стояла амбициозная задача — собрать базу видео для нескольких тысяч страниц на сайте, каждая страница имеет свой уникальный ключ, по которому она продвигается и ранжируется, так что все что фактически требовалось — определить, как мы будем взаимодействовать с Ютубом, чтобы он нам отдал указанное количество урлов и не попросил за это денег!

Все хотят денег, а монополист в лице Гугла (кому, кстати, и принадлежит Ютуб) хочет их более яростнее остальных. Правда за парсинг он денег не просил, он просто ограничил квоту на ежедневное количество обрабатываемых ключей, так для одного проекта это 10000 запросов, а так как в ответе необходимо получить не одно видео, а не менее 10, да еще и с дополнительными параметрами, вроде кто автор, длительность, тайтл и дескрипшн и т.п. (полный парсер будет ниже), то если использовать один Гугл Клауд аккаунт и 12 проектов на этом аккаунте (а я напоминаю, на одном аккаунте можно создавать до 12 проектов), то 1000 страниц я буду обрабатывать примерно неделю… А у меня этих страниц около 40к, вот и считайте, без смекалочки тут не обойтись.

Варианты парсинга (было два)

Сперва я пробовал парсить в лоб, то есть,я создал 3 аккаунта, на каждом по 12 проектов, итого 36 ключей АПИ, каждый из которых имеет уникальную квоту в 10000 запросов, один сервисный аккаунт (для запуска парсинга он нам потребуется) и включил парсинг на тест. Такой парсер прожил примерно 3 дня, пока Гугл не понял что к чему и не сложил 2+2. Он заблокировал мой сервисный аккаунт и вместе с ним обнулил квоты на каждом из 36 ключей до 1 запроса в сутки…

Пу пу пу…

Формулировка — вы пытаетесь обойти наши квоты путем разбиения на несколько проектов. Так нельзя — мы вас забанили.

Ок, следующая гипотеза — использование уникальных сервисных аккаунтов — один ключ АПИ на один аккаунт. Тут пришлось чуть повозиться, но я сделал 36 севрисных аккаунтов и 36 новых ключей АПИ. Чуть переписал скрипт, добавил ротацию прокси и запустил. Короче долго мусолить тему не буду — скрипт умер также через 3 дня.

Получается, если нет разницы, нужно просто увеличить количество ключей АПИ и реализовать задуманное быстрыми темпами — три дня до того, как сервисный аккаунт умрет, а вместе с ним и все АПИ.

В общем, я использовал 8 Гугло аккаунтов — 50 рублей на рынке акаунтов, подошли самые обычные недорогие аккаунты, главное, чтобы там можно было авторизироваться в Гугл Клауд консоль и создавать новые проекты.

Порядок действий следующий: Авторизируемся в Гугл Клауд Консоль — создаем проект — в поисковой строке вводим Youtube API — выбираем Youtube Data API v3 и нажимаем «enable» — для одного аккаунта проделываем тоже самое для Google Sheet API — нажимаем слева Api & Services и жмем в меню credentials — Нажимаем Create credentials — API key и создаем ключ АПИ — копируем его и сохраняем там, откуда можно будет его быстро достать (можно сразу сохранять ключи друг за другом через запятую — получится большая такая строка).

Теперь краткий гайд по тому, как создавать профили (вернее не совсем гайд, а рекомендация как делаю я). В целом все достаточно прозаично, связка простая:

1. Антидетект браузер — Gologin для генерации уникальных профилей.
2. Сервис резидентных прокси — Floppydata.com — тут прикупаю проксей, когда кончаются предоплаченные прокси от Gologin.
3. Терпение и немного времени для выполнения однотипных действий.

Конечно, можно было бы потратить пару часов на написание бота, на условном БАСе, чтобы он сам генерил АПИ ключи, в связке с антиком и прокси сервисом могло бы получится что то интересное (но так как для меня это локальная задача, делать я этого конечно же не стал, а просто потратил примерно час времени и все сделал ручками).

Что хочется отметить — каждый профиль в идеале должен выглядеть, как уникальная личность, чтобы у Гугла не возникло подозрений уже на подготовительном этапе. Конечно, можно забить, тем более если у вас не так много данных, которые надо спарсить, но если их больше чем 10к, я бы заморочился с профилями через антидетект браузер.

Для первого аккаунта (там где мы подключили Google Sheet API) создаем тем же способом сервисный акаунт — все те же шаги, но после нажатия на create credentials надо выбрать Service account. Называем аккаунт, даем ему доступ как владельцу и жмем создать. Снова переходим в созданный ключ и жмем Keys — там выбираем Add Key — Create new key — JSON — жмем создать и наш ключ скачивается на ваш компьютер. Запомните где он находится, его нужно будет поместить в папку с парсером (данные оттуда нам тоже понадобятся, в частности почта).

 

Все, теперь механическая работа — создаем новый проект, активируем Youtube Data API V3 и создаем АПИ ключ, сохраняя его к остальным ключам.

На каждом аккаунте 12 проектов, ну а дальше, насколько вас хватит, меня хватило примерно на 10 аккаунтов, и я закрыл свою задачу за 3 дня.

Как парсить Youtube легально… Ну почти…

Технически мы ничего не нарушаем. Вернее, каждый аккаунт ничего не нарушает, а владелец аккаунтов обходит квоту. Способ работоспособный, я за три дня вытащил около 300к видео (не скачал сами видео, а получил урл с информацией о каждом видео). Теперь, как это реализуется в скрипте.

Вот, собственно сам парсер (полный код есть на моем Github):

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
YouTube-парсер: Shorts + обычные ролики, без live/upcoming.
Считывает ключевые слова (col A) и исходные URL (col J) из Google Sheets и сохраняет результаты обратно в лист "Results".
Продолжает с места остановки через shelve.
При исчерпании квоты всех ключей останавливается и записывает уже собранные данные.
Сохраняет прогресс каждые BATCH_SIZE ключей.
"""
import os
import sys
import shelve
import time
import logging
import socket
from pathlib import Path
from datetime import datetime, timezone

import pandas as pd
import isodate
from langdetect import detect, LangDetectException
from dotenv import load_dotenv, dotenv_values
from googleapiclient.discovery import build
from googleapiclient.errors import HttpError
from google.oauth2 import service_account

# ─── Logging ─────────────────────────────────────────────────
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s %(levelname)s %(message)s',
    datefmt='%Y-%m-%d %H:%M:%S'
)
logger = logging.getLogger(__name__)

# ─── Конфигурация (из .env) ─────────────────────────────────────
# Явно указываем путь к .env рядом со скриптом и перезаписываем системные переменные
env_path = Path(__file__).parent / ".env"
if not env_path.exists():
    logger.error(f"❌ .env файл не найден по пути {env_path}")
    sys.exit(1)

load_dotenv(dotenv_path=env_path, override=True)
SHEET_ID = os.getenv("SHEET_ID")
logger.info(f"🚩 Loaded SHEET_ID from {env_path}: {SHEET_ID}")

KEYWORDS_SHEET       = os.getenv("KEYWORDS_SHEET", "Keywords")
RESULTS_SHEET        = os.getenv("RESULTS_SHEET", "Results")
SERVICE_ACCOUNT_JSON = os.getenv("SERVICE_ACCOUNT_JSON")
NUM_RESULTS          = int(os.getenv("NUM_RESULTS", 10))
REGION               = os.getenv("REGION", "US")
MAX_PAGES            = int(os.getenv("MAX_PAGES", 1))
OUTPUT_CSV           = os.getenv("OUTPUT_CSV", "yt_results.csv")
BATCH_SIZE           = int(os.getenv("BATCH_SIZE", 50))

if MAX_PAGES < 1:
    sys.exit("❌ MAX_PAGES должен быть ≥ 1")

# ─── Google Sheets API ─────────────────────────────────────────
SCOPES = ['https://www.googleapis.com/auth/spreadsheets']
creds_sheets = service_account.Credentials.from_service_account_file(
    SERVICE_ACCOUNT_JSON, scopes=SCOPES
)
sheets_service = build(
    'sheets', 'v4',
    credentials=creds_sheets,
    cache_discovery=False
)

# ─── Хранилище кеша ─────────────────────────────────────────────
CACHE = shelve.open("yt_cache.db")

# ─── Ошибка исчерпания квоты всех ключей ────────────────────────
class QuotaExceededAllKeys(Exception):
    """Все ключи YouTube API исчерпали квоту"""
    pass

# ─── Классы для управления API-ключами ────────────────────────────
class APIKey:
    def __init__(self, key: str):
        self.key = key
        self.service = build(
            "youtube", "v3",
            developerKey=key,
            cache_discovery=False
        )
        self.used_units = 0
        self.active = True

class KeyManager:
    def __init__(self, keys: list[str]):
        self.keys = [APIKey(k) for k in keys]
        self.index = 0

    def get_key(self) -> APIKey:
        n = len(self.keys)
        for _ in range(n):
            api = self.keys[self.index]
            self.index = (self.index + 1) % n
            if api.active:
                return api
        raise QuotaExceededAllKeys()

    def deactivate(self, api: APIKey):
        api.active = False
        logger.warning(f"Деактивирован ключ: {api.key}")

    def record(self, api: APIKey, units: int):
        api.used_units += units
        logger.info(f"Ключ {api.key}: расход {units} units (итого {api.used_units})")

    def execute(self, fn, units=0, backoff_max=3):
        attempt = 0
        while True:
            api = self.get_key()
            try:
                resp = fn(api.service).execute()
                used = units(resp) if callable(units) else units
                self.record(api, used)
                return resp
            except HttpError as e:
                err_str = str(e)
                if 'quotaExceeded' in err_str or 'dailyLimitExceeded' in err_str:
                    self.deactivate(api)
                    continue
                if 'rateLimitExceeded' in err_str:
                    if attempt < backoff_max:
                        delay = 2 ** attempt
                        logger.warning(
                            f"Rate limitExceeded на ключе {api.key}, жду {delay}s (попытка {attempt+1})"
                        )
                        time.sleep(delay)
                        attempt += 1
                        continue
                    else:
                        self.deactivate(api)
                        continue
                logger.error(f"Unexpected HttpError на ключе {api.key}: {e}")
                raise
            except (ConnectionResetError, socket.error) as e:
                if attempt < backoff_max:
                    delay = 2 ** attempt
                    logger.warning(
                        f"Connection error на ключе {api.key}: {e}, жду {delay}s (попытка {attempt+1})"
                    )
                    time.sleep(delay)
                    attempt += 1
                    continue
                else:
                    self.deactivate(api)
                    continue

# ─── Настройка ключей YouTube API ────────────────────────────────
raw_keys = os.getenv("YT_API_KEYS") or dotenv_values(env_path).get("YT_API_KEYS", "")
KEYS = [k.strip() for k in raw_keys.split(",") if k.strip()]
if not KEYS:
    sys.exit("❌ Нет API-ключей (YT_API_KEYS).")
logger.info(f"🔑 Найдено {len(KEYS)} ключей")
key_manager = KeyManager(KEYS)

VIDEO_PARTS = "snippet,contentDetails,status,player,statistics"

# ─── Утилиты ─────────────────────────────────────────────────────
def iso2hms(iso: str):
    if not iso:
        return "", 0
    td = isodate.parse_duration(iso)
    s = int(td.total_seconds())
    h, r = divmod(s, 3600)
    m, sec = divmod(r, 60)
    return f"{h:02d}:{m:02d}:{sec:02d}", s

def fmt_age(pub: str):
    if not pub:
        return ""
    dt = datetime.fromisoformat(pub.replace('Z', '+00:00'))
    d = (datetime.now(timezone.utc) - dt).days
    y, rem = divmod(d, 365)
    mo = rem // 30
    return f"{y} years {mo} months" if y else f"{mo} months"

def safe_detect(txt: str):
    try:
        return detect(txt) if txt.strip() else ""
    except LangDetectException:
        return ""

# ─── Функции для записи в Google Sheets и CSV ───────────────────
def append_to_sheets_with_retry(service, spreadsheet_id, sheet_range, values,
                                max_retries=5, base_delay=1):
    for attempt in range(1, max_retries+1):
        try:
            service.spreadsheets().values().append(
                spreadsheetId=spreadsheet_id,
                range=sheet_range,
                valueInputOption='RAW',
                body={'values': values}
            ).execute()
            logger.info("✅ Успешно записали результаты в Google Sheets")
            return True
        except (HttpError, ConnectionResetError, socket.error) as e:
            delay = base_delay * 2**(attempt-1)
            logger.warning(
                f"⚠️ Попытка {attempt}/{max_retries} записи в Sheets упала: {e}. Ждём {delay}s."
            )
            time.sleep(delay)
    logger.error("❌ Не удалось записать в Google Sheets после всех попыток")
    return False

def save_csv_with_retry(df: pd.DataFrame, path: str, max_retries=3, base_delay=1):
    for attempt in range(1, max_retries+1):
        try:
            df.to_csv(path, index=False)
            logger.info(f"✅ CSV сохранён: {path}")
            return True
        except Exception as e:
            delay = base_delay * 2**(attempt-1)
            logger.warning(
                f"⚠️ Попытка {attempt}/{max_retries} сохранения CSV упала: {e}. Ждём {delay}s."
            )
            time.sleep(delay)
    logger.error("❌ Не удалось сохранить CSV после всех попыток")
    return False

def append_csv_batch(df: pd.DataFrame, path: str, max_retries=3, base_delay=1) -> bool:
    """Дозаписывает DataFrame в CSV, создавая файл с заголовком, если не существует."""
    p = Path(path)
    header = not p.exists()
    for attempt in range(1, max_retries+1):
        try:
            df.to_csv(path, index=False, header=header, mode='a')
            logger.info(f"✅ CSV батч сохранён ({len(df)} строк): {path}")
            return True
        except Exception as e:
            delay = base_delay * 2**(attempt-1)
            logger.warning(f"⚠️ Попытка {attempt}/{max_retries} дозаписи CSV упала: {e}. Ждём {delay}s.")
            time.sleep(delay)
    logger.error("❌ Не удалось дозаписать CSV после всех попыток")
    return False

# ─── Поиск YouTube видео по ключу ────────────────────────────────
def search_once(keyword: str, input_url: str) -> list:
    rows, token, page = [], None, 0
    while len(rows) < NUM_RESULTS and page < MAX_PAGES:
        page += 1
        sk = f"S:{keyword}:{REGION}:{token or ''}"
        if sk in CACHE:
            sr = CACHE[sk]
        else:
            sr = key_manager.execute(
                lambda svc: svc.search().list(
                    q=keyword,
                    part="id",
                    type="video",
                    order="relevance",
                    regionCode=REGION,
                    maxResults=min(50, NUM_RESULTS - len(rows)),
                    pageToken=token or "",
                    safeSearch="none"
                ),
                units=100
            )
            CACHE[sk] = sr
        vids = [it.get('id', {}).get('videoId') for it in sr.get('items', []) if it.get('id', {}).get('videoId')]
        if not vids:
            break
        vk = f"V:{','.join(vids)}"
        if vk in CACHE:
            vr = CACHE[vk]
        else:
            vr = key_manager.execute(
                lambda svc: svc.videos().list(
                    id=",".join(vids),
                    part=VIDEO_PARTS
                ),
                units=lambda resp: len(resp.get('items', []))
            )
            CACHE[vk] = vr
        cids = list({it['snippet']['channelId'] for it in vr.get('items', [])})
        amap = {}
        if cids:
            ck = f"C:{','.join(cids)}"
            if ck in CACHE:
                ch = CACHE[ck]
            else:
                ch = key_manager.execute(
                    lambda svc: svc.channels().list(
                        part="snippet",
                        id=",".join(cids)
                    ),
                    units=1
                )
                CACHE[ck] = ch
            amap = {c['id']: c['snippet']['thumbnails']['default']['url'] for c in ch.get('items', [])}
        for it in vr.get('items', []):
            sn = it.get('snippet', {})
            if sn.get('liveBroadcastContent') in {'live', 'upcoming'}:
                continue
            det = it.get('contentDetails', {})
            dur_str, dur_s = iso2hms(det.get('duration', ''))
            st = it.get('status', {})
            stats = it.get('statistics', {})
            author = sn.get('channelTitle', '')
            avatar = amap.get(sn.get('channelId', ''), '')
            if sn.get('defaultAudioLanguage'):
                ls, lang = 'audio', sn['defaultAudioLanguage']
            elif sn.get('defaultLanguage'):
                ls, lang = 'default', sn['defaultLanguage']
            else:
                ls, lang = 'detect', safe_detect(sn.get('title', '') + ' ' + sn.get('description', '')) or 'unknown'
            lic = st.get('license', '')
            emb = st.get('embeddable', False)
            allowed = emb and lic == 'creativeCommon'
            rows.append([
                keyword, input_url, it['id'], sn.get('title', ''), sn.get('description', ''),
                lang, ls, 'short' if dur_s < 60 else 'video',
                dur_str, dur_s, fmt_age(sn.get('publishedAt', '')), author, avatar,
                stats.get('viewCount', ''), stats.get('likeCount', ''), stats.get('dislikeCount', ''),
                emb, lic, allowed,
                it.get('player', {}).get('embedHtml', f"<iframe src=\"https://www.youtube.com/embed/{it['id']}\" allowfullscreen></iframe>")
            ])
        token = sr.get('nextPageToken') or None
    return rows

# ─── Главная функция ───────────────────────────────────────────
r1 = sheets_service.spreadsheets().values().get(
    spreadsheetId=SHEET_ID,
    range=f"{KEYWORDS_SHEET}!A2:A"
).execute()
r2 = sheets_service.spreadsheets().values().get(
    spreadsheetId=SHEET_ID,
    range=f"{KEYWORDS_SHEET}!J2:J"
).execute()
keywords = [r[0] for r in r1.get('values', [])]
input_urls = [r[0] for r in r2.get('values', [])]

prog = shelve.open('progress.db')
start = prog.get('last_index', 0)
all_res = []
batch_res = []

try:
    for idx in range(start, len(keywords)):
        kw = keywords[idx]
        input_url = input_urls[idx] if idx < len(input_urls) else ''
        logger.info(f"🔍 [{idx}] {kw}")
        res = search_once(kw, input_url)

        all_res.extend(res)
        batch_res.extend(res)
        prog['last_index'] = idx + 1

        # Сохранение батча каждые BATCH_SIZE ключей
        if (idx + 1 - start) % BATCH_SIZE == 0:
            df_batch = pd.DataFrame(batch_res, columns=[
                'keyword','input_url','videoId','title','description','language','language_source',
                'video_type','duration','duration_seconds','age','author','author_avatar',
                'view_count','like_count','dislike_count','embeddable','license',
                'allowed_on_third_party','iframe'
            ])
            append_csv_batch(df_batch, OUTPUT_CSV)
            append_to_sheets_with_retry(
                sheets_service,
                SHEET_ID,
                f"{RESULTS_SHEET}!A2",
                batch_res
            )
            logger.info(f"🔄 Батч из {len(batch_res)} результатов сохранён после {(idx+1-start)} ключей")
            batch_res = []

except QuotaExceededAllKeys:
    logger.warning("⚠️ Все ключи исчерпали квоту, завершаем работу")
finally:
    prog.close()
    CACHE.close()

# Сохраняем остаток батча после завершения
if batch_res:
    df_batch = pd.DataFrame(batch_res, columns=[
        'keyword','input_url','videoId','title','description','language','language_source',
        'video_type','duration','duration_seconds','age','author','author_avatar',
        'view_count','like_count','dislike_count','embeddable','license',
        'allowed_on_third_party','iframe'
    ])
    append_csv_batch(df_batch, OUTPUT_CSV)
    append_to_sheets_with_retry(
        sheets_service,
        SHEET_ID,
        f"{RESULTS_SHEET}!A2",
        batch_res
    )
    logger.info(f"🔄 Финальный батч из {len(batch_res)} строк сохранён")

# Итоговый расход квоты
total_units = sum(api.used_units for api in key_manager.keys)
logger.info(f"ℹ️ Всего расход: {total_units} units")

Скрипт берёт ключевые слова и URL-источники из таблицы в Google Sheets, по каждому ключу делает поиск через YouTube API (короткие видео и обычные ролики, без live/upcoming), ротирует несколько API-ключей до исчерпания квоты, кэширует все ответы, каждые BATCH_SIZE запросов пишет результаты обратно в лист «Results» и дополнительно в CSV, запоминает последний обработанный индекс в shelve, чтобы при рестарте продолжить с того же места.

1. Импорты и базовые зависимости

• shelve, os, sys, time, logging, socket, Path, datetime — стандартные модули для файлов, времени, логирования и сети.
• pandas, isodate, langdetect, python-dotenv, google-api-python-client, google-oauth2 — сторонние библиотеки, обеспечивают парсинг ISO-длительностей, определение языка, чтение .env и работу с Google/YouTube API.

2. Логирование

logging.basicConfig(...) выводит однострочное сообщение уровня INFO с таймстампом UTC. Это позволяет сразу видеть ход выполнения и ошибки без сторонних инструментов.

3. Загрузка конфигурации из .env

pythonCopyEditenv_path = Path(__file__).parent / ".env"
load_dotenv(dotenv_path=env_path, override=True)

Скрипт принудительно ищет файл .env рядом с .py, обрывается, если не находит.
Переменные:
SHEET_ID, KEYWORDS_SHEET, RESULTS_SHEET, SERVICE_ACCOUNT_JSON, NUM_RESULTS, REGION, MAX_PAGES, OUTPUT_CSV, BATCH_SIZE, YT_API_KEYS.
Все числовые — сразу переводятся в int, что исключает «строка вместо числа».

Таким образом вам потребуется дополнительный файл, называющийся — .env. В нем такая вот структура

YT_API_KEYS=AIzaSyBluR2XJc9yt9ylHBMcSNUKQpZuV7hGHyY,AIzaSyCWMhhJbGi90bL9Hhq8H3DLfsy3YyP07ig
SHEET_ID=строка из урла Гугл таблиц начинается после /spreadsheets/d/ и заканчивается /edit? (все что между этими параметрами втавляете сюда)
KEYWORDS_SHEET=Название листа, где исходные данные
RESULTS_SHEET=Results #лист куда будет выводится результат
SERVICE_ACCOUNT_JSON=credentials.json #название файла, где у вас данные сервисного аккаунт, можете его апереиминовать или оставить как он скачался, но  тогда тут замените название

4. Авторизация Google Sheets API

service_account.Credentials.from_service_account_file() читает JSON-ключ сервисного аккаунта. Уровень доступа ограничен только spreadsheets, поэтому ключ безопасен для read/write без доступа к Google Drive.

5. Кэш на основе shelve

CACHE = shelve.open("yt_cache.db") — файловый key-value-store в формате Berkeley DB. Хранит:

• поисковую выдачу по ключу (S:),
• детальный ответ по списку видео-ID (V:),
• детальный ответ по списку каналов (C:).
  Позволяет экономить квоту: запрос повторяется — сначала смотрим в кэше, сетевой вызов не нужен.

6. Исключение QuotaExceededAllKeys

Специальный класс-сигнал. Как только все API-ключи «выгорают», исключение перехватывается в try/except главного цикла, чтобы корректно закрыть файлы и сохранить прогресс.

7. Ротация YouTube API-ключей (APIKey, KeyManager)

• APIKey — обёртка над строкой ключа + счётчик израсходованных units.
• KeyManager
  • get_key() выдаёт следующий активный ключ по кругу (round-robin).
  • deactivate() помечает ключ как неактивный.
  • record() накапливает потраченные units.
  • execute() — универсальный вызов-обёртка: принимает лямбду-вызов googleapiclient, считает units, обрабатывает HttpError (quotaExceeded, dailyLimitExceeded, rateLimitExceeded) и сетевые ошибки socket.error/ConnectionResetError с экспоненциальным back-off.
    Таким образом, скрипт устойчив к перерывам сети и умеет «дожидаться» временного rate-limit, не теряя данные.

8. Утилиты

• iso2hms() — перевод ISO 8601 PT#H#M#S → HH:MM:SS и длительность в секундах.
• fmt_age() — считает «сколько лет | месяцев прошло со дня публикации».
• safe_detect() — обёртка вокруг langdetect.detect, гасит LangDetectException и возвращает пустую строку при ошибке.
• Запись в Sheets / CSV
  • append_to_sheets_with_retry() и save_csv_with_retry() — одинаковая логика retry (5 или 3 попытки, delay = 2^n секунд).
  • append_csv_batch() — дозаписывает батчи, создаёт файл, если не существует, и пишет без заголовка на последующих итерациях.

9. Функция search_once(keyword, input_url)

Пошагово:

1. Пагинация: до MAX_PAGES страниц, пока не наберём NUM_RESULTS.
2. Кэш ключевого запроса: ключ S:{keyword}:{REGION}:{token}.
3. YouTube Search API (search().list) возвращает до 50 videoId. Стоимость 100 units.
4. Кэш списка видео: ключ V:videoId1,videoId2,….
5. YouTube Videos API (videos().list) тянет snippet,contentDetails,status,player,statistics. Стоимость = 1 unit × кол-во роликов.
6. Сбор каналов: уникальные channelId → отдельный запрос channels().list (1 unit). Даёт аватар канала.
7. Фильтры
   • liveBroadcastContent ∉ {live, upcoming} — убираем стримы и премьеры.
   • duration_seconds < 60 → video_type = 'short'.
8. Обогащение записи: язык (по трём источникам), лицензия, embeddable, возможен ли показ на стороннем сайте (allowed_on_third_party).
9. Формирование строки (rows.append([...])) из 21 столбца, включая полностью готовый <iframe>.

10. Главный цикл

pythonCopyEditkeywords = sheets.values().get(... "A2:A").execute()
input_urls = sheets.values().get(... "J2:J").execute()

• Идём по индексам start … len(keywords); start достаётся из progress.db, поэтому перезапуск не повторяет пройденное.
• Каждые BATCH_SIZE (по умолчанию 50) ключей:
  1. Делаем DataFrame df_batch = pd.DataFrame(batch_res, columns=[…]).
  2. append_csv_batch(df_batch, OUTPUT_CSV) — дозаписываем в CSV.
  3. append_to_sheets_with_retry(..., f"{RESULTS_SHEET}!A2", batch_res) — пушим в Sheets.
  4. Очищаем batch_res = [].
• Если вылетает QuotaExceededAllKeys, предупреждаем, закрываем файлы и выходим — часть данных уже сохранена.

11. Завершение

• После цикла проверяем «хвост» batch_res — пишем в CSV и Sheets точно так же, чтобы ничего не потерять.
• Закрываем все shelve-базы (progress.db, yt_cache.db).
• Итоговый лог: ℹ️ Всего расход: {total_units} units — сумма api.used_units по каждому ключу.

12. Что получаем на выходе

Файл/лист	Содержимое	Обновляется
Results (Google Sheets)	Полная таблица 21 столбец: ключ, videoId, язык, тип, длительность, <iframe> и т. д.	Каждые BATCH_SIZE ключей
yt_results.csv	Тот же набор данных локально	Каждые BATCH_SIZE ключей
progress.db	Последний обработанный индекс	После каждого ключа
yt_cache.db	Кэш API-ответов	При первой выдаче / запросе

Таким образом, один запуск скрипта устойчив к обрывам, rate-limit, падениям сети и перезапускам, а данные всегда частично сохранены — минимум за последний завершённый батч.

Что требуется чтобы парсер Youtube заработал

Помимо самого скрипта (его можно сохранить в файл main.py) потребуется — файл .env (выше его раскрыл), файл сервисного аккаунта, в нем только данные которые скачались у вас с Google Cloude Console, файл Google Sheets — там обязательно столбец с названием Keywords (колонка А: ключи, колонка J: URL-источник). У меня в столбце J собраны напротив каждого ключа исходный УРЛ, чтобы в дальнейшем не путаться и не сопостовлять формулами к какой же странице принадлежит спаршеный УРЛ.

Все, из Гугл таблицы переносим необходимые названия в .env и переходим к предпоследнему и последнему шагам.

Ставим требуемые зависимости:

pip install google-api-python-client google-auth google-auth-httplib2 \
            google-auth-oauthlib pandas isodate langdetect python-dotenv

Запускаем скрипт:

Python main.py

Естественно — мы получаем кучу мусора, а вот как мы будем чистить это все — это тема отдельной статьи, который уже конечно же реализован, иначе зачем бы я вам тут все это расписывал, правда?