Парсим YouTube на Python по-взрослому: скрипт без падений и с автоматической ротацией ключей

# ID вашей Google-таблицы SHEET_ID="..." # Лист с ключевиками KEYWORDS_SHEET="Keywords" # Лист для результатов RESULTS_SHEET="Results" # JSON сервисного аккаунта SERVICE_ACCOUNT_JSON="credentials.json" # YouTube API-ключи через запятую YT_API_KEYS="KEY1,KEY2,KEY3" # Параметры парсинга NUM_RESULTS=10 # сколько видео на запрос REGION="US" # регион поиска MAX_PAGES=1 # глубина пагинации BATCH_SIZE=50 # размер батча для записи OUTPUT_CSV="yt_results.csv"

class APIKey: """Один API-ключ плюс его состояние""" 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: for _ in range(len(self.keys)): api = self.keys[self.index] self.index = (self.index + 1) % len(self.keys) if api.active: return api raise QuotaExceededAllKeys() def deactivate(self, api: APIKey): api.active = False logger.warning(f"Ключ выведен из оборота: {api.key}") def execute(self, fn, units=0, backoff_max=3): attempt = 0 while True: api = self.get_key() try: resp = fn(api.service).execute() # учёт расхода квоты return resp except HttpError as e: err = str(e) if 'quotaExceeded' in err: self.deactivate(api) continue if 'rateLimitExceeded' in err and attempt < backoff_max: delay = 2 ** attempt logger.warning(f"Rate limit, жду {delay}s…") time.sleep(delay) attempt += 1 continue raise

key_manager.execute(lambda svc: svc.search().list(...))

#!/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")