Как парсить комментарии YouTube для аналитики

import csv
import json
import queue
import threading
import time
from dataclasses import dataclass, field
from typing import List, Optional

import requests


# ============================================================
# CONFIG
# ============================================================

CONFIG = {
    # Количество параллельных потоков
    "threads": 2,

    # Максимальное число запросов на один API-ключ
    # Лимит YouTube Data API v3 обычно 10 000 units/day
    "quota_limit": 9990,

    # Сколько комментариев получать:
    # None = до конца
    # любое число = лимит комментариев
    "max_comments_per_video": None,

    # Таймаут запросов
    "request_timeout": 30,

    # Пауза между запросами
    "sleep_between_requests": 0.1,

    # Файл со списком видео
    "videos_file": "videos.txt",

    # CSV результат
    "output_csv": "youtube_comments.csv",
}


# ============================================================
# API / PROXY CONFIG
# ============================================================
# Один прокси = один API ключ
#
# proxy:
#   None -> без прокси
#   http://user:pass@host:port
#
# ============================================================

API_ACCOUNTS = [
    {
        "api_key": "YOUTUBE_API_KEY_1",
        "proxy": "http://login:password@127.0.0.1:8080",
    },
    {
        "api_key": "YOUTUBE_API_KEY_2",
        "proxy": "http://login:password@127.0.0.1:8081",
    },
    {
        "api_key": "YOUTUBE_API_KEY_3",
        "proxy": None,
    },
]


# ============================================================
# DATA CLASSES
# ============================================================

@dataclass
class ApiAccount:
    api_key: str
    proxy: Optional[str]
    requests_used: int = 0
    lock: threading.Lock = field(default_factory=threading.Lock)

    def is_available(self, quota_limit: int) -> bool:
        return self.requests_used < quota_limit

    def add_request(self):
        with self.lock:
            self.requests_used += 1


# ============================================================
# ACCOUNT MANAGER
# ============================================================

class AccountManager:
    def __init__(self, accounts_data, quota_limit):
        self.accounts: List[ApiAccount] = [
            ApiAccount(
                api_key=a["api_key"],
                proxy=a["proxy"]
            )
            for a in accounts_data
        ]

        self.quota_limit = quota_limit
        self.index = 0
        self.lock = threading.Lock()

    def get_account(self) -> ApiAccount:
        with self.lock:
            while self.index < len(self.accounts):
                account = self.accounts[self.index]

                if account.is_available(self.quota_limit):
                    return account

                print(
                    f"[INFO] API key exhausted: "
                    f"{account.api_key[:10]}..."
                )

                self.index += 1

            raise RuntimeError("No API accounts left")


# ============================================================
# CSV WRITER
# ============================================================

class CsvWriter:
    def __init__(self, filename):
        self.filename = filename
        self.lock = threading.Lock()

        with open(self.filename, "w", newline="", encoding="utf-8-sig") as f:
            writer = csv.writer(f)
            writer.writerow([
                "user",
                "comment",
                "published_at"
            ])

    def write_comment(self, user, comment, published_at):
        with self.lock:
            with open(
                self.filename,
                "a",
                newline="",
                encoding="utf-8-sig"
            ) as f:
                writer = csv.writer(f)

                writer.writerow([
                    user,
                    comment,
                    published_at
                ])


# ============================================================
# YOUTUBE PARSER
# ============================================================

class YouTubeCommentsParser:
    API_URL = (
        "https://www.googleapis.com/youtube/v3/commentThreads"
    )

    def __init__(self, account_manager, csv_writer):
        self.account_manager = account_manager
        self.csv_writer = csv_writer

    def fetch_comments(self, video_id):
        total_comments = 0
        next_page_token = None

        while True:
            account = self.account_manager.get_account()

            params = {
                "part": "snippet",
                "videoId": video_id,
                "maxResults": 100,
                "textFormat": "plainText",
                "key": account.api_key,
            }

            if next_page_token:
                params["pageToken"] = next_page_token

            proxies = None

            if account.proxy:
                proxies = {
                    "http": account.proxy,
                    "https": account.proxy,
                }

            try:
                response = requests.get(
                    self.API_URL,
                    params=params,
                    proxies=proxies,
                    timeout=CONFIG["request_timeout"]
                )

                account.add_request()

                if response.status_code != 200:
                    print(
                        f"[ERROR] Video={video_id} "
                        f"HTTP={response.status_code} "
                        f"Response={response.text}"
                    )
                    break

                data = response.json()

                items = data.get("items", [])

                if not items:
                    break

                for item in items:
                    snippet = (
                        item["snippet"]
                        ["topLevelComment"]
                        ["snippet"]
                    )

                    author = snippet.get(
                        "authorDisplayName",
                        ""
                    )

                    text = snippet.get(
                        "textDisplay",
                        ""
                    )

                    published_at = snippet.get(
                        "publishedAt",
                        ""
                    )

                    self.csv_writer.write_comment(
                        author,
                        text,
                        published_at
                    )

                    total_comments += 1

                    limit = CONFIG["max_comments_per_video"]

                    if limit is not None:
                        if total_comments >= limit:
                            print(
                                f"[INFO] Limit reached "
                                f"for video {video_id}"
                            )
                            return

                next_page_token = data.get("nextPageToken")

                print(
                    f"[INFO] Video={video_id} "
                    f"Collected={total_comments} "
                    f"Requests={account.requests_used}"
                )

                if not next_page_token:
                    break

                time.sleep(CONFIG["sleep_between_requests"])

            except Exception as e:
                print(
                    f"[ERROR] Video={video_id} "
                    f"Exception={e}"
                )
                break

        print(
            f"[DONE] Video={video_id} "
            f"Total comments={total_comments}"
        )


# ============================================================
# THREAD WORKER
# ============================================================

def worker(video_queue, parser):
    while True:
        try:
            video_id = video_queue.get_nowait()
        except queue.Empty:
            return

        try:
            parser.fetch_comments(video_id)
        finally:
            video_queue.task_done()


# ============================================================
# LOAD VIDEOS
# ============================================================

def load_videos(filename):
    videos = []

    with open(filename, "r", encoding="utf-8") as f:
        for line in f:
            line = line.strip()

            if not line:
                continue

            videos.append(line)

    return videos


# ============================================================
# MAIN
# ============================================================

def main():
    videos = load_videos(CONFIG["videos_file"])

    print(f"[INFO] Loaded videos: {len(videos)}")

    video_queue = queue.Queue()

    for video_id in videos:
        video_queue.put(video_id)

    account_manager = AccountManager(
        API_ACCOUNTS,
        CONFIG["quota_limit"]
    )

    csv_writer = CsvWriter(
        CONFIG["output_csv"]
    )

    parser = YouTubeCommentsParser(
        account_manager,
        csv_writer
    )

    threads = []

    for i in range(CONFIG["threads"]):
        t = threading.Thread(
            target=worker,
            args=(video_queue, parser),
            daemon=True
        )

        t.start()
        threads.append(t)

    video_queue.join()

    for t in threads:
        t.join(timeout=1)

    print("\n[FINISHED]")
    print("API usage statistics:")

    for idx, account in enumerate(
        account_manager.accounts,
        start=1
    ):
        print(
            f"{idx}. "
            f"Requests={account.requests_used} "
            f"Proxy={account.proxy}"
        )


if __name__ == "__main__":
    main()

dQw4w9WgXcQ
aqz-KE-bpKQ
M7lc1UVf-VE