使用验证码处理进行社交媒体研究数据收集

适用范围： 只采公开可见数据；示例中的登录流程仅用于你自有或已授权的账号。

结论先行：社媒采集里的验证码不是异常，而是一条常驻分支。每拿到一个响应先判断有没有 CAPTCHA，命中就走“取 sitekey → 提交 CaptchaAI → 轮询 token → 回填重放”四步。示例用 Python + requests。

各平台分别用哪种验证码

识别方法按类型选；节奏一列是起点值，按触发率再调：

国内站点不一样：微博、小红书更常见 GeeTest（极验）、网易易盾、腾讯防水墙。CaptchaAI 覆盖 GeeTest v3，v4 还不支持，示例集中在 reCAPTCHA 与 Turnstile。

采集器骨架：检测、识别、回填

solve_captcha() 封装两个接口：in.php 提交拿任务 ID，res.php 每 5 秒轮询，60 轮后判超时。_has_captcha() 只做关键字判断，每个响应都能跑。

import requests
import time
import re

CAPTCHAAI_KEY = "YOUR_API_KEY"
CAPTCHAAI_URL = "https://ocr.captchaai.com"


def solve_captcha(method, sitekey, pageurl, **kwargs):
    data = {
        "key": CAPTCHAAI_KEY,
        "method": method,
        "googlekey": sitekey,
        "pageurl": pageurl,
        "json": 1,
    }
    data.update(kwargs)
    resp = requests.post(f"{CAPTCHAAI_URL}/in.php", data=data)
    task_id = resp.json()["request"]

    for _ in range(60):
        time.sleep(5)
        result = requests.get(f"{CAPTCHAAI_URL}/res.php", params={
            "key": CAPTCHAAI_KEY, "action": "get",
            "id": task_id, "json": 1,
        })
        r = result.json()
        if r["request"] != "CAPCHA_NOT_READY":
            return r["request"]
    raise TimeoutError("Solve timeout")


class SocialMediaResearcher:
    def __init__(self, proxy=None):
        self.session = requests.Session()
        if proxy:
            self.session.proxies = {"http": proxy, "https": proxy}
        self.session.headers.update({
            "User-Agent": "Mozilla/5.0 (iPhone; CPU iPhone OS 17_5 like Mac OS X) "
            "AppleWebKit/605.1.15 (KHTML, like Gecko) Version/17.5 "
            "Mobile/15E148 Safari/604.1",
            "Accept-Language": "en-US,en;q=0.9",
        })

    def authenticate(self, login_url, credentials, sitekey):
        """Login with CAPTCHA handling."""
        # Load login page
        self.session.get(login_url)

        # Solve CAPTCHA
        token = solve_captcha("userrecaptcha", sitekey, login_url)

        # Submit login
        resp = self.session.post(login_url, data={
            **credentials,
            "g-recaptcha-response": token,
        })
        return resp.status_code == 200

    def collect_profiles(self, profile_urls):
        """Collect public profile data with CAPTCHA handling."""
        profiles = []

        for url in profile_urls:
            try:
                resp = self.session.get(url, timeout=30)

                # Handle CAPTCHA if triggered
                if self._has_captcha(resp.text):
                    resp = self._handle_captcha(resp.text, url)

                profiles.append({
                    "url": url,
                    "data": self._parse_profile(resp.text),
                    "status": "success",
                })
                time.sleep(5)  # Slow down between profiles

            except Exception as e:
                profiles.append({
                    "url": url,
                    "error": str(e),
                    "status": "failed",
                })

        return profiles

    def _has_captcha(self, html):
        return any(tag in html.lower() for tag in [
            'data-sitekey', 'g-recaptcha', 'cf-turnstile',
            'challenge-platform', 'captcha',
        ])

    def _handle_captcha(self, html, url):
        match = re.search(r'data-sitekey="([^"]+)"', html)
        if not match:
            return self.session.get(url)

        sitekey = match.group(1)

        if 'cf-turnstile' in html:
            token = solve_captcha("turnstile", sitekey, url)
            return self.session.post(url, data={"cf-turnstile-response": token})
        else:
            token = solve_captcha("userrecaptcha", sitekey, url)
            return self.session.post(url, data={"g-recaptcha-response": token})

    def _parse_profile(self, html):
        from bs4 import BeautifulSoup
        soup = BeautifulSoup(html, "html.parser")
        return {
            "name": self._safe_text(soup, "h1, .profile-name"),
            "bio": self._safe_text(soup, ".bio, .profile-bio"),
            "followers": self._safe_text(soup, "[data-followers], .followers"),
            "posts": self._safe_text(soup, "[data-posts], .posts-count"),
        }

    def _safe_text(self, soup, selector):
        el = soup.select_one(selector)
        return el.get_text(strip=True) if el else ""

token 字段跟着类型走：reCAPTCHA 回填 g-recaptcha-response，Turnstile 回填 cf-turnstile-response，写错只会被静默拒绝。失败条目写进结果列表而不抛异常，单页失败不影响整轮。

话题标签与趋势采样

标签页是舆情分析性价比最高的入口。翻页函数复用采集器实例，验证码分支一致。

def research_hashtag(hashtag, platform_url, pages=5):
    """Collect posts for a specific hashtag."""
    researcher = SocialMediaResearcher(
        proxy="http://user:[email protected]:5000"
    )

    all_posts = []
    for page in range(pages):
        url = f"{platform_url}/explore/tags/{hashtag}?page={page}"
        resp = researcher.session.get(url, timeout=30)

        if researcher._has_captcha(resp.text):
            resp = researcher._handle_captcha(resp.text, url)

        from bs4 import BeautifulSoup
        soup = BeautifulSoup(resp.text, "html.parser")
        posts = soup.select(".post-item, article")
        for post in posts:
            all_posts.append({
                "text": post.get_text(strip=True)[:500],
                "hashtag": hashtag,
                "page": page,
            })

        time.sleep(5)

    return all_posts

品牌提及的每日巡检

关键词和平台列表配置化后，就是每天定时跑的巡检任务，输出可入库的 JSON。

import json
from datetime import datetime


class BrandMonitor:
    def __init__(self, brand_name, keywords, proxy=None):
        self.brand = brand_name
        self.keywords = keywords
        self.researcher = SocialMediaResearcher(proxy=proxy)

    def daily_scan(self, platform_urls):
        """Run daily brand mention scan across platforms."""
        report = {
            "brand": self.brand,
            "date": datetime.now().isoformat(),
            "platforms": {},
        }

        for name, url in platform_urls.items():
            mentions = []
            for keyword in self.keywords:
                search_url = f"{url}/search?q={keyword}"
                try:
                    resp = self.researcher.session.get(search_url, timeout=30)

                    if self.researcher._has_captcha(resp.text):
                        resp = self.researcher._handle_captcha(
                            resp.text, search_url,
                        )

                    from bs4 import BeautifulSoup
                    soup = BeautifulSoup(resp.text, "html.parser")
                    results = soup.select(".search-result, .post")
                    mentions.append({
                        "keyword": keyword,
                        "count": len(results),
                    })
                    time.sleep(5)
                except Exception as e:
                    mentions.append({
                        "keyword": keyword,
                        "error": str(e),
                    })

            report["platforms"][name] = mentions

        return report


# Usage
monitor = BrandMonitor(
    brand_name="CaptchaAI",
    keywords=["captchaai", "captcha ai", "captcha solver"],
    proxy="http://user:[email protected]:5000",
)
report = monitor.daily_scan({
    "twitter": "https://twitter-alternative.example.com",
    "reddit": "https://www.reddit.com",
})
print(json.dumps(report, indent=2))

出口 IP 与会话一致性

• 用自有或正规采购的服务器基础设施做出口，别把流量全压在一个 IP 上。
• 一个会话内 IP、User-Agent、Accept-Language 不要中途跳变，特征忽变比慢更容易触发验证码。

排错对照表

成本按线程算，不按次数算

CaptchaAI 按并发线程计费，套餐内识别次数不限。研究型采集低并发、长周期：单机跑一两个平台，BASIC（$15/月，5 线程）够用；多平台每日巡检看 STANDARD（$30/月，15 线程）；几十个关键词同时扫再上 ADVANCE（$90/月，50 线程）。

常见问题

CaptchaAI 能识别国内站点的极验滑块吗？

GeeTest v3 可以，用 geetest 方法提交，需要页面上的 gt 和 challenge。v4 还不支持，易盾和防水墙也不在范围内。

任务一直返回 CAPCHA_NOT_READY 怎么办？

这是排队状态，继续轮询即可，间隔保持 5 秒。超时后重新提交，并把 sitekey 记进日志方便复查。

采集速度上不去，该加线程还是降频率？

先看日志里等 token 的时间占比。占比低说明卡在请求间隔，加线程没用；只有识别任务互相排队才需要更多线程。

采到的个人信息该怎么处理？

只保留研究口径需要的字段，其余入库前脱敏。国内团队按《网络安全法》《数据安全法》和 PIPL 走，开工前读一遍目标站点的 robots 协议。

相关指南

• 出口 IP 与识别成功率
• 浏览器会话的保持与复用
• 浏览器配置隔离与 CaptchaAI 集成

把这条分支交给服务端 ——领取 CaptchaAI API Key，脚本自己跑完。