YouTube 爬虫，搜索不设上限

import json
import os
import time
from urllib.parse import quote_plus

import requests

search = "home espresso"


def scrape(url):
    for attempt in range(3):
        response = requests.get(
            "https://parsing.webunlocker.gologin.com/v1/scrape",
            params={"url": url, "geo": "us", "format": "json"},
            headers={"apikey": os.environ["GOLOGIN_SCRAPING_API_KEY"]},
            timeout=90,
        )
        page = response.json().get("data", {}).get("html", "") if response.ok else ""
        if '"videoRenderer"' in page:
            return page
        time.sleep(5)
    raise RuntimeError("No videos came back, try again later")


def find(node, key):
    if isinstance(node, dict):
        for name, value in node.items():
            if name == key:
                yield value
            yield from find(value, key)
    elif isinstance(node, list):
        for item in node:
            yield from find(item, key)


def text(field):
    if not field:
        return None
    return field.get("simpleText") or "".join(run["text"] for run in field.get("runs", []))


page = scrape(f"https://www.youtube.com/results?search_query={quote_plus(search)}")
data = json.loads(page.split("var ytInitialData = ")[1].split(";</script>")[0])

videos = {}
for video in find(data, "videoRenderer"):
    videos[video["videoId"]] = {
        "title": text(video.get("title")),
        "url": f"https://www.youtube.com/watch?v={video['videoId']}",
        "channel": text(video.get("ownerText")),
        "views": text(video.get("viewCountText")),
        "published": text(video.get("publishedTimeText")),
        "length": text(video.get("lengthText")),
    }

print(json.dumps(list(videos.values()), indent=2, ensure_ascii=False))
print(f"{len(videos)} videos from one search page")

const channelId = 'UCBJycsmduvYEL83R_U4JriQ';
const wanted = 800;
const uploads = `UU${channelId.slice(2)}`;

async function scrape(url) {
  for (let attempt = 1; attempt <= 3; attempt++) {
    const params = new URLSearchParams({ url, geo: 'us', format: 'json' });
    const response = await fetch(`https://parsing.webunlocker.gologin.com/v1/scrape?${params}`, {
      headers: { apikey: process.env.GOLOGIN_SCRAPING_API_KEY },
    });
    const page = response.ok ? (await response.json()).data?.html : '';
    if (page?.includes('var ytInitialData = ')) return JSON.parse(page.split('var ytInitialData = ')[1].split(';</script>')[0]);
    await new Promise((resolve) => setTimeout(resolve, 5000));
  }
  throw new Error('The page came back without videos, try again later');
}

function* find(node, key) {
  if (Array.isArray(node)) for (const item of node) yield* find(item, key);
  else if (node && typeof node === 'object') {
    for (const [name, value] of Object.entries(node)) {
      if (name === key) yield value;
      yield* find(value, key);
    }
  }
}

const videos = new Map();
for (const item of find(await scrape(`https://www.youtube.com/playlist?list=${uploads}`), 'lockupViewModel')) {
  if (item.contentType === 'LOCKUP_CONTENT_TYPE_VIDEO') videos.set(item.contentId, item.metadata.lockupMetadataViewModel.title.content);
}
let pages = 1;

while (videos.size < wanted) {
  const last = [...videos.keys()].at(-1);
  const before = videos.size;
  for (const item of find(await scrape(`https://www.youtube.com/watch?v=${last}&list=${uploads}`), 'playlistPanelVideoRenderer')) {
    videos.set(item.videoId, item.title.simpleText);
  }
  pages++;
  if (videos.size === before) break;
}

console.table([...videos].slice(0, 5).map(([id, title]) => ({ title, url: `https://www.youtube.com/watch?v=${id}` })));
console.log(`${videos.size} videos in ${pages} pages: $${(pages * 0.00098).toFixed(4)} on the Growth plan`);

const video = 'https://www.youtube.com/watch?v=jNQXAC9IVRw';

async function scrape(url) {
  for (let attempt = 1; attempt <= 3; attempt++) {
    const params = new URLSearchParams({ url, geo: 'us', format: 'json' });
    const response = await fetch(`https://parsing.webunlocker.gologin.com/v1/scrape?${params}`, {
      headers: { apikey: process.env.GOLOGIN_SCRAPING_API_KEY },
    });
    const page = response.ok ? (await response.json()).data?.html : '';
    if (page?.includes('var ytInitialPlayerResponse = ')) return page;
    await new Promise((resolve) => setTimeout(resolve, 5000));
  }
  throw new Error('The video page came back empty, try again later');
}

const page = await scrape(video);
const player = JSON.parse(page.split('var ytInitialPlayerResponse = ')[1].split(';var meta')[0]);
const { videoDetails: details, microformat, captions } = player;

console.log({
  title: details.title,
  channel: details.author,
  views: Number(details.viewCount),
  likes: Number(page.match(/along with ([\d,]+) other/)?.[1].replaceAll(',', '')),
  comments: page.match(/"contextualInfo":\{"runs":\[\{"text":"([^"]+)"/)?.[1],
  published: microformat.playerMicroformatRenderer.publishDate,
  seconds: Number(details.lengthSeconds),
  tags: details.keywords ?? [],
  captions: captions?.playerCaptionsTracklistRenderer.captionTracks.map((track) => track.languageCode) ?? [],
});