做新闻聚合和媒体监测，验证码几乎每一步都会挡在前面：通讯社常挂 reCAPTCHA v2，主流新闻站爱用 Cloudflare Turnstile，付费媒体靠 reCAPTCHA v3 打分。人工点验证码撑不起每天几十上百个信源的抓取节奏，CaptchaAI 用同一套 API 把它们接进流水线，识别到验证码就自动提交、拿 token、继续跑。
海外通讯社走 reCAPTCHA v2，国内新闻站更常见 Cloudflare 检测或 GeeTest（支持到 v3）；reCAPTCHA 靠 Google 托管脚本，国内网络访问海外信源常见加载慢，超时判断要留够余量。
新闻站点验证码速览
- 主流媒体：Cloudflare Turnstile，机器人检测。
- 通讯社（美联社、路透社）：reCAPTCHA v2，批量访问。
- 付费媒体：reCAPTCHA v3，频率评分。
- 地方新闻站：reCAPTCHA v2，限流触发。
- 新闻聚合站：Cloudflare 验证流程，抓取检测。
- 通稿网站：图片验证码，下载页面。
抓取实战：预算规划与聚合器代码
先定抓取预算
按信源更新频率和验证码触发强度分别定预算，不要所有域名共用一套节奏；触发频繁的域名先降速，同时记录识别成败数据，反过来调整预算。
新闻聚合器代码
import requests
import time
import re
from bs4 import BeautifulSoup
from datetime import datetime
import json
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("Timeout")
class NewsAggregator:
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 (Windows NT 10.0; Win64; x64) "
"AppleWebKit/537.36 Chrome/126.0.0.0 Safari/537.36",
"Accept-Language": "en-US,en;q=0.9",
})
def collect_headlines(self, source_url, section=None):
"""Collect headlines from a news source."""
url = f"{source_url}/{section}" if section else source_url
resp = self.session.get(url, timeout=30)
if self._has_captcha(resp.text):
resp = self._solve_and_retry(resp.text, url)
soup = BeautifulSoup(resp.text, "html.parser")
articles = []
for item in soup.select("article, .story, .headline-item, h2 a, h3 a"):
link = item if item.name == "a" else item.select_one("a")
if link:
articles.append({
"title": link.get_text(strip=True),
"url": self._abs_url(source_url, link.get("href", "")),
"source": source_url,
"collected_at": datetime.now().isoformat(),
})
return articles
def get_article(self, article_url):
"""Fetch full article content."""
resp = self.session.get(article_url, timeout=30)
if self._has_captcha(resp.text):
resp = self._solve_and_retry(resp.text, article_url)
soup = BeautifulSoup(resp.text, "html.parser")
# Remove unwanted elements
for tag in soup.select("script, style, nav, footer, .ad, .sidebar"):
tag.decompose()
content_el = soup.select_one(
"article, .article-body, .story-body, .entry-content"
)
return {
"url": article_url,
"title": self._text(soup, "h1, .article-title"),
"author": self._text(soup, ".author, .byline, [rel='author']"),
"date": self._text(soup, "time, .publish-date, .article-date"),
"content": content_el.get_text(separator="\n", strip=True) if content_el else "",
"word_count": len(content_el.get_text().split()) if content_el else 0,
}
def aggregate_sources(self, sources, max_articles_per=20):
"""Aggregate headlines across multiple sources."""
all_articles = []
for source in sources:
try:
articles = self.collect_headlines(source["url"], source.get("section"))
all_articles.extend(articles[:max_articles_per])
print(f"{source['name']}: {len(articles)} headlines")
except Exception as e:
print(f"{source['name']}: Error - {e}")
time.sleep(3)
return all_articles
def _has_captcha(self, html):
return any(tag in html.lower() for tag in [
'data-sitekey', 'g-recaptcha', 'cf-turnstile',
])
def _solve_and_retry(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})
token = solve_captcha("userrecaptcha", sitekey, url)
return self.session.post(url, data={"g-recaptcha-response": token})
def _text(self, soup, selector):
el = soup.select_one(selector)
return el.get_text(strip=True) if el else ""
def _abs_url(self, base, href):
if href.startswith("http"):
return href
return base.rstrip("/") + "/" + href.lstrip("/")
# Usage
aggregator = NewsAggregator(
proxy="http://user:[email protected]:5000"
)
sources = [
{"name": "Tech News A", "url": "https://technews-a.example.com", "section": "latest"},
{"name": "Business B", "url": "https://business-b.example.com", "section": "tech"},
{"name": "Industry C", "url": "https://industry-c.example.com"},
]
headlines = aggregator.aggregate_sources(sources)
print(f"Total: {len(headlines)} headlines collected")
关键词监控
在聚合器基础上加一层关键词匹配，命中就报警，适合品牌监测、竞品追踪。
class NewsMonitor:
def __init__(self, keywords, sources, proxy=None):
self.keywords = [kw.lower() for kw in keywords]
self.aggregator = NewsAggregator(proxy=proxy)
self.sources = sources
self.seen_urls = set()
def scan(self):
"""Scan for articles matching keywords."""
headlines = self.aggregator.aggregate_sources(self.sources)
matches = []
for article in headlines:
if article["url"] in self.seen_urls:
continue
title_lower = article["title"].lower()
matched_kws = [kw for kw in self.keywords if kw in title_lower]
if matched_kws:
article["matched_keywords"] = matched_kws
matches.append(article)
self.seen_urls.add(article["url"])
return matches
def continuous_monitor(self, interval_min=30):
"""Run continuous monitoring with alerts."""
while True:
matches = self.scan()
if matches:
print(f"\n=== {len(matches)} new matches found ===")
for m in matches:
print(f" [{', '.join(m['matched_keywords'])}] {m['title']}")
print(f" {m['url']}")
else:
print(f"No new matches at {datetime.now().strftime('%H:%M')}")
time.sleep(interval_min * 60)
# Monitor for specific topics
monitor = NewsMonitor(
keywords=["captcha", "bot detection", "web scraping", "automation"],
sources=sources,
proxy="http://user:[email protected]:5000",
)
matches = monitor.scan()
命中关键词的文章会按 URL 去重，避免同一条新闻反复报警。
常见故障排查
| 问题 | 原因 | 处理方式 |
|---|---|---|
| Cloudflare 拦截所有请求 | 检测策略过严 | 自有服务器基础设施 + 真实 UA |
| 打开付费墙而非文章 | 内容在订阅后 | 检测付费墙特征，跳过或按权限处理 |
| 每个页面都触发验证码 | IP 被标记 | 切换 IP，间隔加到 5 秒以上 |
| 文章正文抓空 | JS 渲染内容 | SPA 页面改用 Selenium/Puppeteer |
| 同一条新闻反复出现 | 多信源报道同一事件 | 按标题相似度去重 |
常见问题
高频问题：
新闻聚合、媒体监测这类抓取合法吗？
抓取标题、摘要研究是常见做法，整篇复制版权正文则不行——保留摘要片段并链接回原文更稳妥，境内数据处理留意《网络安全法》《数据安全法》。
CaptchaAI 能处理新闻网站上的哪些验证码？
覆盖 reCAPTCHA v2/v3、Cloudflare Turnstile 和验证流程，国内常见的 GeeTest 支持到 v3；hCaptcha、FunCaptcha 暂不支持。
识别一次验证码要等多久，会拖慢抓取吗？
Turnstile 通常 10 秒内，reCAPTCHA v3 约 4 秒，reCAPTCHA v2 上限约 60 秒，并发抓取不会拖慢节奏。
遇到付费墙内容该怎么处理？
识别付费墙特征（CSS 类名或偏短正文），命中打标记，只处理有权限内容，付费正文不抓取完整版。
相关阅读
自有服务器基础设施看轮换指南，社交媒体信源看验证码处理。
不管新闻来源是通讯社还是聚合站，拿到你的 CaptchaAI API Key，接入你的采集流程。
