Twitter（推特）爬虫，​实时采集 X 的喜欢、转帖和查看次数

import html
import json
import os
import re
import time
from datetime import datetime, timezone

import requests

username = "NASA"


def scrape(url):
    for attempt in range(3):
        address = url if attempt == 0 else f"{url}?fresh={int(time.time())}"
        response = requests.get(
            "https://parsing.webunlocker.gologin.com/v1/scrape",
            params={"url": address, "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 'role="group"' in page:
            return page
        time.sleep(3)
    raise RuntimeError("The page came back without data, try again later")


def meta(page, name):
    found = re.search(rf'<meta (?:property|name)="{name}" content="([^"]*)"', page)
    return html.unescape(found.group(1)) if found else None


def count(page, word):
    found = re.search(rf">([\d.,]+[KMB]?)</span>\s*<span[^>]*>{word}<", page)
    return found.group(1) if found else None


def text_of(chunk):
    start = chunk.find('aria-label="More"')
    if start < 0:
        return ""
    start = chunk.find("</button>", start) + len("</button>")
    ends = [chunk.find(mark, start) for mark in ('inert=""', "data-press-control", "<video", 'role="group"', "<article")]
    end = min([chunk.rfind("<", 0, at) for at in ends if at > 0] or [len(chunk)])
    words = re.sub(r"<script.*?</script>|<[^>]+>", "", chunk[start:end], flags=re.S)
    return html.unescape(words).strip()


page = scrape(f"https://x.com/{username}")

posts = []
start = 0
for match in re.finditer(r'aria-label="([^"]*)"[^>]*role="group"', page):
    chunk = page[start : match.start()]
    start = match.end()
    link = re.search(r'href="/(\w+)/status/(\d+)', chunk)
    stamp = re.search(r'"timestamp":(\d+)', chunk)
    counts = {word.rstrip("s").replace("replie", "reply"): int(value) for value, word in re.findall(r"(\d+) (\w+)", match.group(1))}
    posts.append(
        {
            "url": f"https://x.com/{link.group(1)}/status/{link.group(2)}",
            "time": datetime.fromtimestamp(int(stamp.group(1)) / 1000, timezone.utc).isoformat() if stamp else None,
            "pinned": ">Pinned</span>" in chunk[: chunk.find("data-post-avatar")],
            "text": text_of(chunk),
            "replies": counts.get("reply", 0),
            "reposts": counts.get("repost", 0),
            "likes": counts.get("like", 0),
            "bookmarks": counts.get("bookmark", 0),
            "views": counts.get("view"),
        }
    )

profile = {
    "username": meta(page, "profile:username"),
    "name": meta(page, "author"),
    "bio": meta(page, "description"),
    "followers": count(page, "Followers"),
    "following": count(page, "Following"),
    "posts": meta(page, "twitter:data1"),
    "joined": meta(page, "twitter:data2"),
    "latest_posts": posts,
}

print(json.dumps(profile, indent=2, ensure_ascii=False))
print(f"@{username}: {profile['followers']} followers, {len(posts)} latest posts with exact counts from one page")

import { writeFileSync } from 'node:fs';

const links = [
  'https://x.com/NASA/status/2120305504295510080',
  'https://x.com/SpaceX/status/2120808224324510133',
  'https://x.com/Spotify/status/2120607242374504556',
];

async function scrape(url) {
  for (let attempt = 1; attempt <= 3; attempt++) {
    const address = attempt === 1 ? url : `${url}?fresh=${Date.now()}`;
    const params = new URLSearchParams({ url: address, 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('role="group"')) return page;
    await new Promise((resolve) => setTimeout(resolve, 3000));
  }
  throw new Error(`${url} came back without data, try again later`);
}

const meta = (page, name) => page.match(new RegExp(`<meta (?:property|name)="${name}" content="([^"]*)"`))?.[1] ?? '';
const count = (label, word) => Number(label.match(new RegExp(`(\\d+) ${word}`))?.[1] ?? 0);

const posts = [];
for (const url of links) {
  const page = await scrape(url);
  const label = page.match(/aria-label="([^"]*)"[^>]*role="group"/)[1];
  posts.push({
    url,
    author: meta(page, 'og:title').replace(/ on X$/, ''),
    date: meta(page, 'article:published_time').slice(0, 10),
    replies: count(label, 'repl'),
    reposts: count(label, 'repost'),
    likes: count(label, 'like'),
    bookmarks: count(label, 'bookmark'),
    views: count(label, 'view'),
  });
}

writeFileSync('posts.csv', ['url,author,date,replies,reposts,likes,bookmarks,views', ...posts.map((post) => Object.values(post).map((value) => JSON.stringify(value)).join(','))].join('\n'));
console.table(posts);
console.log(`${posts.length} posts in ${posts.length} pages: $${(posts.length * 0.00098).toFixed(4)} on the Growth plan`);

import { writeFileSync } from 'node:fs';

const accounts = ['NASA', 'NatGeo', 'SpaceX', 'Spotify', 'YouTube'];

async function scrape(url) {
  for (let attempt = 1; attempt <= 3; attempt++) {
    const address = attempt === 1 ? url : `${url}?fresh=${Date.now()}`;
    const params = new URLSearchParams({ url: address, 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('role="group"')) return page;
    await new Promise((resolve) => setTimeout(resolve, 3000));
  }
  return '';
}

const count = (label, word) => Number(label.match(new RegExp(`(\\d+) ${word}`))?.[1] ?? 0);

const rows = [];
for (const account of accounts) {
  const page = await scrape(`https://x.com/${account}`);
  const followers = page.match(/>([\d.,]+[KMB]?)<\/span>\s*<span[^>]*>Followers</)?.[1] ?? '';
  let start = 0;
  for (const group of page.matchAll(/aria-label="([^"]*)"[^>]*role="group"/g)) {
    const chunk = page.slice(start, group.index);
    start = group.index + group[0].length;
    const [, author, id] = chunk.match(/href="\/(\w+)\/status\/(\d+)/) ?? [];
    const time = chunk.match(/"timestamp":(\d+)/)?.[1];
    rows.push({
      account,
      followers,
      post: `https://x.com/${author}/status/${id}`,
      date: time ? new Date(Number(time)).toISOString().slice(0, 10) : '',
      replies: count(group[1], 'repl'),
      reposts: count(group[1], 'repost'),
      likes: count(group[1], 'like'),
      views: count(group[1], 'view'),
    });
  }
}

writeFileSync('latest-posts.csv', ['account,followers,post,date,replies,reposts,likes,views', ...rows.map((row) => Object.values(row).map((value) => JSON.stringify(value)).join(','))].join('\n'));
console.table(rows);
console.log(`${rows.length} posts from ${accounts.length} accounts in ${accounts.length} pages: $${(accounts.length * 0.00098).toFixed(4)} on the Growth plan`);